Skip to content

tux.database.controllers.base

Shared SQLModel-powered base controller for database operations.

Classes:

Name Description
BaseController

A thin asynchronous repository / DAO layer.

Classes

BaseController(model: type[ModelT])

A thin asynchronous repository / DAO layer.

The goal is not to hide SQLAlchemy entirely but to provide a small, opinionated convenience wrapper for common database operations.

The public API exposes familiar methods such as find_many or create for clean and consistent database access patterns.

Methods:

Name Description
find_one

Return the first row that matches where or None.
find_many

Return a list of rows matching where (or all rows).
execute_transaction

Execute callback inside a database session / transaction block.
safe_get_attr

Return getattr(obj, attr, default) - keeps old helper available.
connect_or_create_relation

Return a dict with a single key that can be merged into data dicts.
Source code in tux/database/controllers/base.py
Python
def __init__(self, model: type[ModelT]):
    self.model = model
    self.model_name = model.__name__.lower()
    # Legacy support - many controllers used `self.table` to access
    # the underlying ORM model (mainly for `.upsert`).
    self.table = model

Functions

_execute_query(op: Callable[[AsyncSession], Any], span_desc: str) -> Any async

Run op inside a managed session & sentry span (if enabled).

Source code in tux/database/controllers/base.py
Python
async def _execute_query(self, op: Callable[[AsyncSession], Any], span_desc: str) -> Any:
    """Run *op* inside a managed session & sentry span (if enabled)."""

    if sentry_sdk.is_initialized():
        with sentry_sdk.start_span(op="db.query", description=span_desc) as span:
            span.set_tag("db.table", self.model_name)
            try:
                async with db.session() as _session:
                    result = await op(_session)
                span.set_status("ok")
                return result  # noqa: TRY300 - maintain behaviour
            except Exception as exc:
                span.set_status("internal_error")
                span.set_data("error", str(exc))
                logger.error(f"{span_desc}: {exc}")
                raise
    else:
        async with db.session() as _session:
            return await op(_session)
find_one(*, where: dict[str, Any], include: dict[str, bool] | None = None, **__: Any) -> ModelT | None async

Return the first row that matches where or None.

Source code in tux/database/controllers/base.py
Python
async def find_one(
    self,
    *,
    where: dict[str, Any],
    include: dict[str, bool] | None = None,  # ignored but kept for API consistency
    **__: Any,
) -> ModelT | None:
    """Return the first row that matches *where* or *None*."""

    async def _op(session: AsyncSession):
        stmt = select(self.model).filter_by(**where).limit(1)
        result = await session.execute(stmt)  # type: ignore[attr-defined]
        return result.scalar_one_or_none()  # type: ignore[attr-defined]

    return await self._execute_query(_op, f"find_one {self.model_name}")
find_many(*, where: dict[str, Any] | None = None, include: dict[str, bool] | None = None, order: dict[str, str] | None = None, take: int | None = None, skip: int | None = None) -> list[ModelT] async

Return a list of rows matching where (or all rows).

Source code in tux/database/controllers/base.py
Python
async def find_many(
    self,
    *,
    where: dict[str, Any] | None = None,
    include: dict[str, bool] | None = None,  # ignored
    order: dict[str, str] | None = None,
    take: int | None = None,
    skip: int | None = None,
) -> list[ModelT]:
    """Return a list of rows matching *where* (or all rows)."""

    async def _op(session: AsyncSession):
        stmt = select(self.model)
        if where:
            stmt = stmt.filter_by(**where)
        if order:
            # Expecting {"col": "asc"|"desc"}
            for col, direction in order.items():
                column_attr = getattr(self.model, col)
                stmt = stmt.order_by(column_attr.desc() if direction.lower() == "desc" else column_attr.asc())
        if take is not None:
            stmt = stmt.limit(take)
        if skip is not None:
            stmt = stmt.offset(skip)
        res = await session.execute(stmt)  # type: ignore[attr-defined]
        return res.scalars().all()  # type: ignore[attr-defined]

    return await self._execute_query(_op, f"find_many {self.model_name}")
execute_transaction(callback: Callable[[], Any]) -> Any async

Execute callback inside a database session / transaction block.

Source code in tux/database/controllers/base.py
Python
async def execute_transaction(self, callback: Callable[[], Any]) -> Any:
    """Execute *callback* inside a database session / transaction block."""

    try:
        async with db.transaction() as _session:  # db.transaction yields AsyncSession
            # Provide session to callback by setting attribute maybe? The old
            # code expected none, so we just call callback; operations inside
            # will create their own sessions where needed.  This still gives
            # us atomicity because we're running in a SAVEPOINT transaction.
            return await callback()
    except Exception as exc:
        logger.error(f"Transaction failed in {self.model_name}: {exc}")
        raise
safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any staticmethod

Return getattr(obj, attr, default) - keeps old helper available.

Source code in tux/database/controllers/base.py
Python
@staticmethod
def safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any:
    """Return ``getattr(obj, attr, default)`` - keeps old helper available."""
    return getattr(obj, attr, default)
connect_or_create_relation(id_field: str, model_id: Any, *_: Any, **__: Any) -> dict[str, Any] staticmethod

Return a dict with a single key that can be merged into data dicts.

The calling code does something like::

Text Only
data = {"guild": connect_or_create_relation("guild_id", guild_id)}

We map that pattern to a very small helper that collapses to {"guild_id": guild_id}.

Source code in tux/database/controllers/base.py
Python
@staticmethod
def connect_or_create_relation(id_field: str, model_id: Any, *_: Any, **__: Any) -> dict[str, Any]:
    """Return a dict with a single key that can be merged into *data* dicts.

    The calling code does something like::

        data = {"guild": connect_or_create_relation("guild_id", guild_id)}

    We map that pattern to a very small helper that collapses to `{"guild_id": guild_id}`.
    """
    return {id_field: model_id}