Skip to content

PyDrocsid.database

db_context async 

db_context() -> AsyncIterator[None]

Async context manager for database sessions.

Source code in PyDrocsid/database/__init__.py 
12
13
14
15
16
17
18
19
20
21
@asynccontextmanager
async def db_context() -> AsyncIterator[None]:
    """Async context manager for database sessions."""

    db.create_session()
    try:
        yield
    finally:
        await db.commit()
        await db.close()

db_wrapper 

db_wrapper(f: Callable[P, Awaitable[T]]) -> Callable[P, Awaitable[T]]

Decorator which wraps an async function in a database context.

Source code in PyDrocsid/database/__init__.py 
24
25
26
27
28
29
30
31
32
def db_wrapper(f: Callable[P, Awaitable[T]]) -> Callable[P, Awaitable[T]]:
    """Decorator which wraps an async function in a database context."""

    @wraps(f)
    async def inner(*args: P.args, **kwargs: P.kwargs) -> T:
        async with db_context():
            return await f(*args, **kwargs)

    return inner

delete 

delete(table: Any) -> Delete

Shortcut for :meth:sqlalchemy.sql.expression.delete

Source code in PyDrocsid/database/database.py 
79
80
81
82
def delete(table: Any) -> Delete:
    """Shortcut for :meth:`sqlalchemy.sql.expression.delete`"""

    return sa_delete(table)

exists 

exists(statement: Executable, *entities: Column[Any], **kwargs: Any) -> Exists

Shortcut for :meth:sqlalchemy.future.select

Source code in PyDrocsid/database/database.py 
73
74
75
76
def exists(statement: Executable, *entities: Column[Any], **kwargs: Any) -> Exists:
    """Shortcut for :meth:`sqlalchemy.future.select`"""

    return sa_exists(statement, *entities, **kwargs)

filter_by 

filter_by(cls: Any, *args: Column[Any], **kwargs: Any) -> Select

Shortcut for :meth:sqlalchemy.future.Select.filter_by

Source code in PyDrocsid/database/database.py 
67
68
69
70
def filter_by(cls: Any, *args: Column[Any], **kwargs: Any) -> Select:
    """Shortcut for :meth:`sqlalchemy.future.Select.filter_by`"""

    return select(cls, *args).filter_by(**kwargs)

select 

select(entity: Any, *args: Column[Any]) -> Select

Shortcut for :meth:sqlalchemy.future.select

Source code in PyDrocsid/database/database.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def select(entity: Any, *args: Column[Any]) -> Select:
    """Shortcut for :meth:`sqlalchemy.future.select`"""

    if not args:
        return sa_select(entity)

    options = []
    for arg in args:
        if isinstance(arg, (tuple, list)):
            head, *tail = arg
            opt = selectinload(head)
            for x in tail:
                opt = opt.selectinload(x)
            options.append(opt)
        else:
            options.append(selectinload(arg))

    return sa_select(entity).options(*options)