API reference

Generated from the source. See Usage for a narrative introduction.

Repository

Bases: Generic[Model]

Repository over a model, bound to a database or cluster.

The database resolves to a :meth:using override if given, otherwise to the process-wide default built from DATABASE_* settings -- so in the common single-database case no binding is needed at all. Repositories accessed through a unit of work are bound to the unit's database.

__init__ deliberately takes no arguments so subclasses work directly as FastAPI dependencies (Depends(UserRepository)) without leaking routing knobs as query parameters.

Source code in sqlargon/repository.py

class SQLAlchemyRepository(Generic[Model]):
    """Repository over a model, bound to a database or cluster.

    The database resolves to a :meth:`using` override if given, otherwise to
    the process-wide default built from ``DATABASE_*`` settings -- so in the
    common single-database case no binding is needed at all. Repositories
    accessed through a unit of work are bound to the unit's database.

    ``__init__`` deliberately takes no arguments so subclasses work directly
    as FastAPI dependencies (``Depends(UserRepository)``) without leaking
    routing knobs as query parameters.
    """

    __slots__ = ("_query", "routing")

    database: ClassVar[Database | None] = None
    model: type[Model]
    default_order_by: ClassVar[str | SQLColumnExpression[Any] | None] = None

    def __init__(self) -> None:
        self._query: Any = None
        self.routing = RoutingOptions()

    def __init_subclass__(
        cls,
        *,
        abstract: bool = False,
        model: type[Model] | None = None,
        **kwargs: Any,
    ) -> None:
        if model is not None:
            cls.model = model
        elif not abstract:
            if not hasattr(cls, "model"):
                cls.model = cls.__orig_bases__[0].__args__[0]  # type: ignore[attr-defined]
            if not issubclass(cls.model, Base):
                msg = f"Could not resolve model for {cls.__name__}"
                raise TypeError(msg)
        super().__init_subclass__(**kwargs)

    @property
    def db(self) -> AnyDatabase:
        if self.routing.db is not None:
            return self.routing.db
        if self.database is not None:
            return self.database
        return get_default_database()

    @property
    def qb(self) -> QueryBuilder:
        return self.db.query_builder

    @property
    def query(self) -> Any:
        if self._query is None:
            query = self.qb.select(self.model)
            # read off the class: a bare mapped column is a descriptor, and
            # resolving it against a repository instance would fall through
            # to __getattr__
            order_by = type(self).default_order_by
            if order_by is not None:
                query = query.order_by(order_by)
            self._query = query
        return self._query

    def __getattr__(self, attribute: str) -> Self:
        self._query = getattr(self.query, attribute)
        return self

    def __call__(self, *args: Any, **kwargs: Any) -> Self:
        self._query = self.query(*args, **kwargs)
        return self

    def __await__(self) -> Generator[Any, None, Result]:
        return self.execute().__await__()

    @classmethod
    def _get_default_index_elements(cls) -> set[str]:
        return {
            c.name
            for c in cls.model.__table__.primary_key.columns  # type: ignore[attr-defined]
        }

    @classmethod
    def _get_default_set(cls) -> set[str]:
        return {
            c.name
            for c in cls.model.__table__.columns  # type: ignore[attr-defined]
            if c.name not in cls._get_default_index_elements()
        }

    @property
    def on_conflict(self) -> OnConflictOptions:
        return {
            "index_elements": self._get_default_index_elements(),
            "set_": self._get_default_set(),
        }

    def use_query(self, query: Any) -> Self:
        self._query = query
        return self

    def copy(self, query: Any) -> Self:
        clone = self.__class__().use_query(query)
        clone.routing = self.routing
        return clone

    def using(
        self,
        hint: str | None = None,
        *,
        db: AnyDatabase | None = None,
        read_only: bool | None = None,
        shard_key: Any | None = None,
    ) -> Self:
        """Return a copy of this repository with a routing preference baked in.

        The returned repository routes its statements to the given database,
        replica or shard, unless a transaction already pins another one::

            await repo.using("replica_0").all()
            await repo.using(read_only=True).count()
            await repo.using(shard_key=tenant_id).create(**values)
            await repo.using(db=other_database).all()

        ``read_only=False`` explicitly forces the primary even when the
        repository inherited ``read_only=True``; ``None`` keeps it.
        """
        clone = self.copy(self._query)
        clone.routing = self.routing.merge(
            hint, db=db, read_only=read_only, shard_key=shard_key
        )
        return clone

    def insert(
        self,
        values: Values,
        *,
        return_results: bool = False,
        ignore_conflicts: bool = False,
        **options: Unpack[OnConflictOptions],
    ) -> Self:
        on_conflict = None
        if ignore_conflicts:
            on_conflict = OnConflict(do="ignore", options=self.on_conflict | options)
        query = self.qb.insert(
            self.model, values, return_results=return_results, on_conflict=on_conflict
        )
        return self.copy(query)

    def upsert(
        self,
        values: Values,
        *,
        return_results: bool = False,
        **options: Unpack[OnConflictOptions],
    ) -> Self:
        query = self.qb.insert(
            self.model,
            values,
            return_results=return_results,
            on_conflict=OnConflict(do="update", options=self.on_conflict | options),
        )
        return self.copy(query)

    def select(
        self,
        *args: Any,
        with_for_update: bool | WithForUpdate | None = None,
        options: tuple[Any, ...] | None = None,
    ) -> Self:
        args = args or (self.model,)
        query = self.qb.select(*args, with_for_update=with_for_update, options=options)
        return self.copy(query)

    def update(self, values: Values, *, return_results: bool = False) -> Self:
        query = self.qb.update(self.model, values, return_results=return_results)
        return self.copy(query)

    def delete(self, *, return_results: bool = False) -> Self:
        query = self.qb.delete(self.model, return_results=return_results)
        return self.copy(query)

    def where(self, *args: _ColumnExpressionArgument[bool], **kwargs: Any) -> Self:
        query = self.qb.filter(self.query, *args, **kwargs)
        return self.copy(query)

    def filter(self, *args: _ColumnExpressionArgument[bool], **kwargs: Any) -> Self:
        return self.where(*args, **kwargs)

    def join(
        self,
        target: _JoinTargetArgument,
        onclause: _OnClauseArgument | None = None,
        *,
        isouter: bool = False,
        full: bool = False,
    ) -> Self:
        query = self.query.join(target, onclause, isouter=isouter, full=full)  # type: ignore[attr-defined]
        return self.copy(query)

    def load(self, *keys: QueryableAttribute) -> Self:
        query = self.query.options(selectinload(*keys))
        return self.copy(query)

    def routing_context(
        self, statement: Any = None, *, read_only: bool | None = None
    ) -> RoutingContext:
        return self.routing.context(
            statement, getattr(type(self), "model", None), read_only=read_only
        )

    @asynccontextmanager
    async def session(
        self, statement: Any = None, *, read_only: bool | None = None
    ) -> AsyncGenerator[AsyncSession]:
        context = self.routing_context(statement, read_only=read_only)
        async with self.db.session_context(context) as session:
            yield session

    async def execute_query(
        self,
        query: Executable | TypedReturnsRows,
        params: Params | None = None,
        *,
        read_only: bool | None = None,
        **kwargs: Any,
    ) -> Result:
        async with self.session(query, read_only=read_only) as session:
            return await session.execute(query, params, **kwargs)

    async def execute(
        self,
        params: Params | None = None,
        *,
        read_only: bool | None = None,
        **kwargs: Any,
    ) -> Result:
        return await self.execute_query(
            self.query, params, read_only=read_only, **kwargs
        )

    async def execute_many(
        self, *queries: Executable | TypedReturnsRows, **kwargs: Any
    ) -> None:
        async with self.session() as session:
            for query in queries:
                await session.execute(query, **kwargs)

    async def stream_query(
        self,
        query: Executable | TypedReturnsRows,
        params: Params | None = None,
        *,
        read_only: bool | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[Row[Any]]:
        async with self.session(query, read_only=read_only) as session:
            rows = await session.stream(query, params, **kwargs)
            async for row in rows:
                yield row

    def stream(
        self,
        params: Params | None = None,
        *,
        read_only: bool | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[Row[Any]]:
        return self.stream_query(self.query, params, read_only=read_only, **kwargs)

    async def mappings(self) -> MappingResult:
        return (await self.execute()).mappings()

    async def scalar(self) -> Any:
        return (await self.execute()).scalar()

    async def scalars(self) -> ScalarResult[Model]:
        return (await self.execute()).scalars()

    async def unique(
        self, strategy: Callable[[Any], Any] | None = None
    ) -> ScalarResult[Model]:
        return (await self.scalars()).unique(strategy)

    async def all(
        self, *, unique: bool | Callable[[Any], Any] = False
    ) -> Sequence[Model]:
        if not unique:
            return (await self.scalars()).all()
        strategy = unique if callable(unique) else None
        return (await self.unique(strategy)).all()

    async def one(self) -> Model:
        return (await self.scalars()).one()

    async def one_or_none(self) -> Model | None:
        return (await self.scalars()).one_or_none()

    async def first(self) -> Model | None:
        return (await self.scalars()).first()

    async def get(
        self, *args: _ColumnExpressionArgument[bool], **kwargs: Any
    ) -> Model | None:
        return await self.filter(*args, **kwargs).one_or_none()

    @classmethod
    def _column(cls, name: str) -> Any:
        return cls.model.__table__.columns[name]  # type: ignore[attr-defined]

    @classmethod
    def _identity_elements(
        cls, options: OnConflictOptions | None = None
    ) -> tuple[str, ...]:
        elements = (options or {}).get("index_elements")
        return tuple(elements or cls._get_default_index_elements())

    @staticmethod
    def _identity(values: SingleValue, elements: tuple[str, ...]) -> tuple[Any, ...]:
        return tuple(values[name] for name in elements)

    def _identified(
        self, values: SingleValue, elements: tuple[str, ...]
    ) -> dict[str, Any]:
        row = dict(values)
        for name in elements:
            if row.get(name) is not None:
                continue
            default = self._column(name).default
            if default is None or not (default.is_scalar or default.is_callable):
                msg = (
                    f"{self.model.__name__}.{name} is filled by the server, so the "
                    f"{self.db.dialect} dialect cannot return the written rows "
                    "without a RETURNING clause"
                )
                raise UnsupportedOption(msg)
            row[name] = default.arg(None) if default.is_callable else default.arg
        return row

    def _identity_filter(
        self, rows: Sequence[SingleValue], elements: tuple[str, ...]
    ) -> Any:
        if not rows:
            return false()
        if len(elements) == 1:
            name = elements[0]
            return self._column(name).in_([row[name] for row in rows])
        return or_(
            *(
                and_(*(self._column(name) == row[name] for name in elements))
                for row in rows
            )
        )

    async def _identities(
        self, elements: tuple[str, ...], where: Any = None
    ) -> list[tuple[Any, ...]]:
        query = self.qb.select(*(self._column(name) for name in elements))
        if where is not None:
            query = query.where(where)
        return [tuple(row) for row in (await self.execute_query(query)).all()]

    async def _fetch(self, where: Any = None) -> ScalarResult[Model]:
        query = self.qb.select(self.model)
        if where is not None:
            query = query.where(where)
        return (await self.execute_query(query)).scalars()

    def _insert_statement(
        self,
        values: Values,
        do: Literal["ignore", "update"] | None,
        options: OnConflictOptions,
        *,
        return_results: bool,
    ) -> Self:
        if do == "update":
            return self.upsert(values, return_results=return_results, **options)
        return self.insert(
            values,
            return_results=return_results,
            ignore_conflicts=do == "ignore",
            **options,
        )

    async def _insert_returning(
        self,
        values: MultipleValues,
        *,
        do: Literal["ignore", "update"] | None = None,
        **options: Unpack[OnConflictOptions],
    ) -> ScalarResult[Model]:
        if self.qb.supports(Option.RETURNING):
            statement = self._insert_statement(values, do, options, return_results=True)
            return await statement.scalars()

        elements = self._identity_elements(self.on_conflict | options)
        rows = [self._identified(row, elements) for row in values]
        statement = self._insert_statement(rows, do, options, return_results=False)
        async with self.session(statement.query):
            written = rows
            if do == "ignore":
                identities = await self._identities(
                    elements, self._identity_filter(rows, elements)
                )
                stored = set(identities)
                written = [
                    row for row in rows if self._identity(row, elements) not in stored
                ]
            await statement.execute()
            return await self._fetch(self._identity_filter(written, elements))

    async def _write_returning(self, statement: Self) -> ScalarResult[Model]:
        query = statement.query
        async with self.session(query):
            if isinstance(query, Delete):
                rows = await self._fetch(query.whereclause)
                await statement.execute()
                return rows
            elements = self._identity_elements()
            identities = await self._identities(elements, query.whereclause)
            await statement.execute()
            written = [
                dict(zip(elements, identity, strict=True)) for identity in identities
            ]
            return await self._fetch(self._identity_filter(written, elements))

    async def _update_returning(
        self, values: Values, *args: _ColumnExpressionArgument[bool], **kwargs: Any
    ) -> ScalarResult[Model]:
        if self.qb.supports(Option.RETURNING):
            statement = self.update(values, return_results=True)
            return await statement.filter(*args, **kwargs).scalars()
        return await self._write_returning(self.update(values).filter(*args, **kwargs))

    async def _delete_returning(
        self, *args: _ColumnExpressionArgument[bool], **kwargs: Any
    ) -> ScalarResult[Model]:
        if self.qb.supports(Option.RETURNING):
            statement = self.delete(return_results=True)
            return await statement.filter(*args, **kwargs).scalars()
        return await self._write_returning(self.delete().filter(*args, **kwargs))

    async def create_or_update(self, **kwargs: Any) -> Model:
        result = await self._insert_returning([kwargs], do="update")
        return result.one()

    async def get_or_create(
        self, defaults: SingleValue | None = None, **kwargs: Any
    ) -> Model:
        async with self.session():
            values = {**(defaults or {}), **kwargs}
            result = await self._insert_returning([values], do="ignore")
            created = result.one_or_none()
            if created is not None:
                return created
            return await self.select().filter(**kwargs).one()

    async def create(self, **kwargs: Any) -> Model | None:
        result = await self._insert_returning([kwargs], do="ignore")
        return result.one_or_none()

    async def remove(
        self, *args: _ColumnExpressionArgument[bool], **kwargs: Any
    ) -> None:
        await self.delete().filter(*args, **kwargs).execute()

    async def delete_one(
        self, *args: _ColumnExpressionArgument[bool], **kwargs: Any
    ) -> Model | None:
        result = await self._delete_returning(*args, **kwargs)
        return result.one_or_none()

    async def update_one(
        self, values: SingleValue, *args: _ColumnExpressionArgument[bool], **kwargs: Any
    ) -> Model | None:
        result = await self._update_returning(values, *args, **kwargs)
        return result.one_or_none()

    async def update_many(
        self, values: SingleValue, *args: _ColumnExpressionArgument[bool], **kwargs: Any
    ) -> Sequence[Model]:
        result = await self._update_returning(values, *args, **kwargs)
        return result.all()

    async def delete_many(
        self, *args: _ColumnExpressionArgument[bool], **kwargs: Any
    ) -> None:
        await self.delete(return_results=False).filter(*args, **kwargs).execute()

    async def list(
        self, *args: _ColumnExpressionArgument[bool], **kwargs: Any
    ) -> Sequence[Model]:
        return await self.select().filter(*args, **kwargs).all()

    @overload
    async def bulk_create_or_update(
        self,
        values: MultipleValues,
        *,
        return_results: Literal[False] = ...,
        **options: Unpack[OnConflictOptions],
    ) -> Result: ...

    @overload
    async def bulk_create_or_update(
        self,
        values: MultipleValues,
        *,
        return_results: Literal[True],
        **options: Unpack[OnConflictOptions],
    ) -> Sequence[Model]: ...

    async def bulk_create_or_update(
        self,
        values: MultipleValues,
        *,
        return_results: bool = False,
        **options: Unpack[OnConflictOptions],
    ) -> Sequence[Model] | Result:
        if return_results:
            result = await self._insert_returning(values, do="update", **options)
            return result.all()
        return await self.upsert(values, **options).execute()

    @overload
    async def bulk_create(
        self,
        values: MultipleValues,
        *,
        ignore_conflicts: bool = ...,
        return_results: Literal[False] = ...,
    ) -> None: ...

    @overload
    async def bulk_create(
        self,
        values: MultipleValues,
        *,
        ignore_conflicts: bool = ...,
        return_results: Literal[True],
    ) -> Sequence[Model]: ...

    async def bulk_create(
        self,
        values: MultipleValues,
        *,
        ignore_conflicts: bool = True,
        return_results: bool = False,
        **options: Unpack[OnConflictOptions],
    ) -> Sequence[Model] | None:
        if return_results:
            result = await self._insert_returning(
                values, do="ignore" if ignore_conflicts else None, **options
            )
            return result.all()
        await self.insert(
            values, ignore_conflicts=ignore_conflicts, **options
        ).execute()
        return None

    async def create_many(
        self,
        items: MultipleValues,
        *,
        ignore_conflicts: bool = False,
        **kwargs: Any,
    ) -> Sequence[Model]:
        return await self.bulk_create(
            items, return_results=True, ignore_conflicts=ignore_conflicts, **kwargs
        )

    async def bulk_update(
        self,
        values: MultipleValues,
        *args: Any,
        on_: set[str] | None = None,
        **kwargs: Any,
    ) -> None:
        """Update many rows in a single executemany statement.

        An executemany cannot return rows; use :meth:`update_many` when the
        updated models are needed.
        """
        on_ = on_ or self._get_default_index_elements()
        where = [getattr(self.model, field) == bindparam(f"u_{field}") for field in on_]
        values = [
            {key if key not in on_ else f"u_{key}": value for key, value in row.items()}
            for row in values
        ]
        query = self.qb.update(self.model).where(*args, *where).filter_by(**kwargs)
        async with self.session() as session:
            connection = await session.connection()
            await connection.execute(query, values)

    async def count(self, *args: _ColumnExpressionArgument[bool], **kwargs: Any) -> int:
        query = self.qb.count(self.model, *args, **kwargs)
        return (await self.execute_query(query)).scalar()  # type: ignore[return-value]

    @asynccontextmanager
    async def get_chunk_for_update(
        self,
        values: SingleValue | None,
        limit: int = 100,
        on_: str = "id",
        order_by: Any = None,
        *args: Any,
        **kwargs: Any,
    ) -> AsyncGenerator[Sequence[Model]]:
        pk = getattr(self.model, on_)
        order_by = order_by or pk
        async with self.session():
            results = (
                await self.select(with_for_update={"skip_locked": True})
                .filter(*args, **kwargs)
                .order_by(order_by)
                .limit(limit)
                .all()
            )

            yield results
            if values:
                await self.update_many(
                    values, pk.in_([getattr(r, on_) for r in results])
                )
            else:
                await self.remove(pk.in_([getattr(r, on_) for r in results]))

bulk_update(values: MultipleValues, *args: Any, on_: set[str] | None = None, **kwargs: Any) -> None async

Update many rows in a single executemany statement.

An executemany cannot return rows; use :meth:update_many when the updated models are needed.

Source code in sqlargon/repository.py

async def bulk_update(
    self,
    values: MultipleValues,
    *args: Any,
    on_: set[str] | None = None,
    **kwargs: Any,
) -> None:
    """Update many rows in a single executemany statement.

    An executemany cannot return rows; use :meth:`update_many` when the
    updated models are needed.
    """
    on_ = on_ or self._get_default_index_elements()
    where = [getattr(self.model, field) == bindparam(f"u_{field}") for field in on_]
    values = [
        {key if key not in on_ else f"u_{key}": value for key, value in row.items()}
        for row in values
    ]
    query = self.qb.update(self.model).where(*args, *where).filter_by(**kwargs)
    async with self.session() as session:
        connection = await session.connection()
        await connection.execute(query, values)

using(hint: str | None = None, *, db: AnyDatabase | None = None, read_only: bool | None = None, shard_key: Any | None = None) -> Self

Return a copy of this repository with a routing preference baked in.

The returned repository routes its statements to the given database, replica or shard, unless a transaction already pins another one::

await repo.using("replica_0").all()
await repo.using(read_only=True).count()
await repo.using(shard_key=tenant_id).create(**values)
await repo.using(db=other_database).all()

read_only=False explicitly forces the primary even when the repository inherited read_only=True; None keeps it.

Source code in sqlargon/repository.py

def using(
    self,
    hint: str | None = None,
    *,
    db: AnyDatabase | None = None,
    read_only: bool | None = None,
    shard_key: Any | None = None,
) -> Self:
    """Return a copy of this repository with a routing preference baked in.

    The returned repository routes its statements to the given database,
    replica or shard, unless a transaction already pins another one::

        await repo.using("replica_0").all()
        await repo.using(read_only=True).count()
        await repo.using(shard_key=tenant_id).create(**values)
        await repo.using(db=other_database).all()

    ``read_only=False`` explicitly forces the primary even when the
    repository inherited ``read_only=True``; ``None`` keeps it.
    """
    clone = self.copy(self._query)
    clone.routing = self.routing.merge(
        hint, db=db, read_only=read_only, shard_key=shard_key
    )
    return clone

Bases: SQLAlchemyRepository[SoftDeleteModel]

Repository that tombstones rows instead of deleting them.

Every statement is scoped to live rows: selects and :meth:count skip tombstoned rows, :meth:update refuses to touch them, and :meth:delete is rewritten into an update raising the flag -- so remove, delete_one and delete_many all soft delete::

class User(UUIDModelMixin, SoftDeleteBase): ...


class UserRepository(SoftDeleteRepository[User]): ...


users = UserRepository()

await users.remove(User.id == user_id)  # UPDATE ... SET tombstone = true
await users.list()  # the row is gone from reads
await users.restore(User.id == user_id)  # and back again

The flag belongs to :meth:delete and :meth:restore alone: it is left out of the default ON CONFLICT DO UPDATE set, so an upsert cannot silently resurrect a deleted row. Reach past the scope with :meth:with_deleted, :meth:only_deleted and :meth:hard_delete.

The model type variable is bound to :class:~sqlargon.mixins.SoftDeleteBase, so a type checker rejects a model that cannot be soft deleted. At runtime the looser :class:~sqlargon.mixins.SoftDeleteMixin is enough; anything else raises TypeError on subclassing.

Source code in sqlargon/repository.py

class SoftDeleteRepository(SQLAlchemyRepository[SoftDeleteModel], abstract=True):
    """Repository that tombstones rows instead of deleting them.

    Every statement is scoped to live rows: selects and :meth:`count` skip
    tombstoned rows, :meth:`update` refuses to touch them, and :meth:`delete`
    is rewritten into an update raising the flag -- so ``remove``,
    ``delete_one`` and ``delete_many`` all soft delete::

        class User(UUIDModelMixin, SoftDeleteBase): ...


        class UserRepository(SoftDeleteRepository[User]): ...


        users = UserRepository()

        await users.remove(User.id == user_id)  # UPDATE ... SET tombstone = true
        await users.list()  # the row is gone from reads
        await users.restore(User.id == user_id)  # and back again

    The flag belongs to :meth:`delete` and :meth:`restore` alone: it is left
    out of the default ``ON CONFLICT DO UPDATE`` set, so an upsert cannot
    silently resurrect a deleted row. Reach past the scope with
    :meth:`with_deleted`, :meth:`only_deleted` and :meth:`hard_delete`.

    The model type variable is bound to
    :class:`~sqlargon.mixins.SoftDeleteBase`, so a type checker rejects a
    model that cannot be soft deleted. At runtime the looser
    :class:`~sqlargon.mixins.SoftDeleteMixin` is enough; anything else raises
    ``TypeError`` on subclassing.
    """

    __slots__ = ("deleted_only", "include_deleted")

    def __init__(self) -> None:
        super().__init__()
        self.include_deleted = False
        self.deleted_only = False

    def __init_subclass__(cls, *, abstract: bool = False, **kwargs: Any) -> None:
        super().__init_subclass__(abstract=abstract, **kwargs)
        if not abstract and not issubclass(cls.model, SoftDeleteMixin):
            msg = (
                f"{cls.model.__name__} must inherit from SoftDeleteMixin "
                f"to be used with {cls.__name__}"
            )
            raise TypeError(msg)

    @classmethod
    def _get_default_set(cls) -> set[str]:
        return super()._get_default_set() - {"tombstone"}

    @property
    def _not_deleted(self) -> _ColumnExpressionArgument[bool]:
        return self.model.not_deleted

    @property
    def _is_deleted(self) -> _ColumnExpressionArgument[bool]:
        return self.model.is_deleted

    @property
    def _scope(self) -> _ColumnExpressionArgument[bool] | None:
        """The predicate every statement is narrowed to, if any."""
        if self.deleted_only:
            return self._is_deleted
        if self.include_deleted:
            return None
        return self._not_deleted

    def _scoped(self, query: Any) -> Any:
        """Narrow ``query`` to the rows this repository is scoped to."""
        scope = self._scope
        if scope is None:
            return query
        return query.where(scope)

    def copy(self, query: Any) -> Self:
        clone = super().copy(query)
        clone.include_deleted = self.include_deleted
        clone.deleted_only = self.deleted_only
        return clone

    @property
    def query(self) -> Any:
        if self._query is None:
            self.use_query(self._scoped(super().query))
        return self._query

    def select(
        self,
        *args: Any,
        with_for_update: bool | WithForUpdate | None = None,
        options: tuple[Any, ...] | None = None,
    ) -> Self:
        clone = super().select(*args, with_for_update=with_for_update, options=options)
        return clone.use_query(self._scoped(clone.query))

    def update(self, values: Values, *, return_results: bool = False) -> Self:
        clone = super().update(values, return_results=return_results)
        return clone.use_query(self._scoped(clone.query))

    def delete(self, *, return_results: bool = False) -> Self:
        """Raise the tombstone on the matched rows instead of removing them."""
        return self.update({"tombstone": True}, return_results=return_results)

    async def count(self, *args: _ColumnExpressionArgument[bool], **kwargs: Any) -> int:
        scope = self._scope
        if scope is not None:
            args = (*args, scope)
        return await super().count(*args, **kwargs)

    async def bulk_update(
        self,
        values: MultipleValues,
        *args: Any,
        on_: set[str] | None = None,
        **kwargs: Any,
    ) -> None:
        scope = self._scope
        if scope is not None:
            args = (*args, scope)
        await super().bulk_update(values, *args, on_=on_, **kwargs)

    async def get_or_create(
        self, defaults: SingleValue | None = None, **kwargs: Any
    ) -> SoftDeleteModel:
        """Get the live row matching ``kwargs`` or create it.

        Raises :class:`DeletedRowExistsError` when the row exists but is
        tombstoned: creating it would violate the unique key, and returning it
        would resurrect a deleted row behind the caller's back.
        """
        async with self.session():
            values = {**(defaults or {}), **kwargs}
            result = await self._insert_returning([values], do="ignore")
            created = result.one_or_none()
            if created is not None:
                return created
            obj = await self.select().filter(**kwargs).one_or_none()
            if obj is None:
                msg = (
                    f"A deleted {self.model.__name__} row already matches "
                    f"{kwargs}; restore or hard delete it first"
                )
                raise DeletedRowExistsError(msg)
            return obj

    def with_deleted(self) -> Self:
        """Return a copy whose statements cover tombstoned rows as well."""
        clone = self.copy(self._query)
        clone.include_deleted = True
        clone.deleted_only = False
        return clone

    def only_deleted(self) -> Self:
        """Return a copy scoped to tombstoned rows.

        The scope holds for every statement the copy builds, so
        ``only_deleted().count()`` counts the trash and
        ``only_deleted().hard_delete()`` empties it.
        """
        clone = self.copy(self._query)
        clone.include_deleted = True
        clone.deleted_only = True
        return clone

    async def restore(
        self, *args: _ColumnExpressionArgument[bool], **kwargs: Any
    ) -> Sequence[SoftDeleteModel]:
        """Clear the tombstone on the matched rows and return them."""
        return await self.only_deleted().update_many(
            {"tombstone": False}, *args, **kwargs
        )

    async def hard_delete(
        self, *args: _ColumnExpressionArgument[bool], **kwargs: Any
    ) -> None:
        """Physically delete the matched rows, bypassing the tombstone."""
        if self.deleted_only:
            args = (*args, self._is_deleted)
        await super().delete().filter(*args, **kwargs).execute()

delete(*, return_results: bool = False) -> Self

Raise the tombstone on the matched rows instead of removing them.

Source code in sqlargon/repository.py

def delete(self, *, return_results: bool = False) -> Self:
    """Raise the tombstone on the matched rows instead of removing them."""
    return self.update({"tombstone": True}, return_results=return_results)

get_or_create(defaults: SingleValue | None = None, **kwargs: Any) -> SoftDeleteModel async

Get the live row matching kwargs or create it.

Raises :class:DeletedRowExistsError when the row exists but is tombstoned: creating it would violate the unique key, and returning it would resurrect a deleted row behind the caller's back.

Source code in sqlargon/repository.py

async def get_or_create(
    self, defaults: SingleValue | None = None, **kwargs: Any
) -> SoftDeleteModel:
    """Get the live row matching ``kwargs`` or create it.

    Raises :class:`DeletedRowExistsError` when the row exists but is
    tombstoned: creating it would violate the unique key, and returning it
    would resurrect a deleted row behind the caller's back.
    """
    async with self.session():
        values = {**(defaults or {}), **kwargs}
        result = await self._insert_returning([values], do="ignore")
        created = result.one_or_none()
        if created is not None:
            return created
        obj = await self.select().filter(**kwargs).one_or_none()
        if obj is None:
            msg = (
                f"A deleted {self.model.__name__} row already matches "
                f"{kwargs}; restore or hard delete it first"
            )
            raise DeletedRowExistsError(msg)
        return obj

hard_delete(*args: _ColumnExpressionArgument[bool], **kwargs: Any) -> None async

Physically delete the matched rows, bypassing the tombstone.

Source code in sqlargon/repository.py

async def hard_delete(
    self, *args: _ColumnExpressionArgument[bool], **kwargs: Any
) -> None:
    """Physically delete the matched rows, bypassing the tombstone."""
    if self.deleted_only:
        args = (*args, self._is_deleted)
    await super().delete().filter(*args, **kwargs).execute()

only_deleted() -> Self

Return a copy scoped to tombstoned rows.

The scope holds for every statement the copy builds, so only_deleted().count() counts the trash and only_deleted().hard_delete() empties it.

Source code in sqlargon/repository.py

def only_deleted(self) -> Self:
    """Return a copy scoped to tombstoned rows.

    The scope holds for every statement the copy builds, so
    ``only_deleted().count()`` counts the trash and
    ``only_deleted().hard_delete()`` empties it.
    """
    clone = self.copy(self._query)
    clone.include_deleted = True
    clone.deleted_only = True
    return clone

restore(*args: _ColumnExpressionArgument[bool], **kwargs: Any) -> Sequence[SoftDeleteModel] async

Clear the tombstone on the matched rows and return them.

Source code in sqlargon/repository.py

async def restore(
    self, *args: _ColumnExpressionArgument[bool], **kwargs: Any
) -> Sequence[SoftDeleteModel]:
    """Clear the tombstone on the matched rows and return them."""
    return await self.only_deleted().update_many(
        {"tombstone": False}, *args, **kwargs
    )

with_deleted() -> Self

Return a copy whose statements cover tombstoned rows as well.

Source code in sqlargon/repository.py

def with_deleted(self) -> Self:
    """Return a copy whose statements cover tombstoned rows as well."""
    clone = self.copy(self._query)
    clone.include_deleted = True
    clone.deleted_only = False
    return clone

Bases: RuntimeError

A tombstoned row holds the unique key a new row was to be created with.

Source code in sqlargon/repository.py

class DeletedRowExistsError(RuntimeError):
    """A tombstoned row holds the unique key a new row was to be created with."""

Run a repository method within a single transaction.

All queries issued by the decorated method share one session_context on the repository's bound database, committed on success and rolled back on error. Objects without a routing_context method (only the db property) are still accepted; their statements route with ambient :func:using markers alone.

Source code in sqlargon/functools.py

def atomic(
    fn: Callable[Concatenate[S, P], Awaitable[R]],
) -> Callable[Concatenate[S, P], Awaitable[R]]:
    """Run a repository method within a single transaction.

    All queries issued by the decorated method share one ``session_context``
    on the repository's bound database, committed on success and rolled back
    on error. Objects without a ``routing_context`` method (only the ``db``
    property) are still accepted; their statements route with ambient
    :func:`using` markers alone.
    """

    @wraps(fn)
    async def wrapper(self: S, *args: P.args, **kwargs: P.kwargs) -> R:
        routing_context = getattr(self, "routing_context", RoutingContext.create)
        async with self.db.session_context(routing_context()):
            return await fn(self, *args, **kwargs)

    return wrapper

Unit of work

Bases: AbstractUnitOfWork

Unit of work over a single database or cluster.

The database is a :meth:using override if given, otherwise the process-wide default built from DATABASE_* settings. Declared repositories are bound to that database, so all work inside the unit shares one transaction -- a unit of work never spans databases (there is no two-phase commit).

__init__ deliberately takes no arguments so subclasses work directly as FastAPI dependencies (Depends(OrdersUow)) without leaking routing knobs as query parameters. Per-use routing (e.g. a shard) is chosen with :meth:using; on a cluster the member database is resolved once on __aenter__ and pinned for the whole transaction::

async with OrdersUow().using(shard_key=tenant_id) as uow:
    await uow.orders.create(**values)

Source code in sqlargon/uow.py

class SQLAlchemyUnitOfWork(AbstractUnitOfWork):
    """Unit of work over a single database or cluster.

    The database is a :meth:`using` override if given, otherwise the
    process-wide default built from ``DATABASE_*`` settings. Declared
    repositories are bound to that database, so all work inside the unit
    shares one transaction -- a unit of work never spans databases (there is
    no two-phase commit).

    ``__init__`` deliberately takes no arguments so subclasses work directly
    as FastAPI dependencies (``Depends(OrdersUow)``) without leaking routing
    knobs as query parameters. Per-use routing (e.g. a shard) is chosen with
    :meth:`using`; on a cluster the member database is resolved once on
    ``__aenter__`` and pinned for the whole transaction::

        async with OrdersUow().using(shard_key=tenant_id) as uow:
            await uow.orders.create(**values)
    """

    database: ClassVar[AnyDatabase | None] = None
    _repository_models: ClassVar[list[type[Base]]] = []

    def __init_subclass__(cls, **kwargs: Any) -> None:
        super().__init_subclass__(**kwargs)
        models: list[type[Base]] = []
        for name, hint in get_type_hints(cls).items():
            if not (isinstance(hint, type) and issubclass(hint, SQLAlchemyRepository)):
                continue
            if not isinstance(cls.__dict__.get(name), _RepositoryDescriptor):
                setattr(cls, name, _RepositoryDescriptor(hint))
            model = getattr(hint, "model", None)
            if model is not None and model not in models:
                models.append(model)
        cls._repository_models = models

    def __init__(self) -> None:
        self._stack = AsyncExitStack()
        self._session: AsyncSession | None = None
        self.routing = RoutingOptions()

    def using(
        self,
        hint: str | None = None,
        *,
        db: AnyDatabase | None = None,
        read_only: bool | None = None,
        shard_key: Any | None = None,
    ) -> Self:
        """Return a fresh unit of work with a routing preference baked in.

        The returned unit is not yet entered, so this is safe to call on a
        dependency-injected instance::

            async with uow.using(shard_key=tenant_id):
                ...
        """
        clone = self.__class__()
        clone.routing = self.routing.merge(
            hint, db=db, read_only=read_only, shard_key=shard_key
        )
        return clone

    @property
    def db(self) -> AnyDatabase:
        if self.routing.db is not None:
            return self.routing.db
        if self.database is not None:
            return self.database
        return get_default_database()

    @property
    def session(self) -> AsyncSession:
        if self._session is None:
            msg = "Session not initialized"
            raise ValueError(msg)
        return self._session

    def _routing_context(self, model: type[Base] | None = None) -> RoutingContext:
        return self.routing.context(model=model)

    def _model_home_probe(self, model: type[Base]) -> RoutingContext:
        # read-intent with no statement and no read_only flag: model and
        # shard routers reveal the model's home database, while replica
        # routers answer with the primary -- deterministically and without
        # picking replicas or setting their sticky-read flag
        return replace(self._routing_context(model), operation="read", read_only=False)

    async def __aenter__(self) -> Self:
        if self._session is not None:
            msg = "Cannot open the same unit of work more than once"
            raise ValueError(msg)
        models = self._repository_models
        if len(models) > 1:
            routed = {self.db.route(self._model_home_probe(model)) for model in models}
            if len(routed) > 1:
                msg = (
                    "Repositories of this unit of work route to different "
                    "databases; a unit of work cannot span databases"
                )
                raise RoutingError(msg)
        await self._stack.__aenter__()
        self._session = await self._stack.enter_async_context(
            self.db.session_context(
                self._routing_context(models[0] if models else None)
            )
        )
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        self._session = None
        await self._stack.__aexit__(exc_type, exc_val, exc_tb)

    async def commit(self) -> None:
        await self.session.commit()

    async def rollback(self) -> None:
        await self.session.rollback()

using(hint: str | None = None, *, db: AnyDatabase | None = None, read_only: bool | None = None, shard_key: Any | None = None) -> Self

Return a fresh unit of work with a routing preference baked in.

The returned unit is not yet entered, so this is safe to call on a dependency-injected instance::

async with uow.using(shard_key=tenant_id):
    ...

Source code in sqlargon/uow.py

def using(
    self,
    hint: str | None = None,
    *,
    db: AnyDatabase | None = None,
    read_only: bool | None = None,
    shard_key: Any | None = None,
) -> Self:
    """Return a fresh unit of work with a routing preference baked in.

    The returned unit is not yet entered, so this is safe to call on a
    dependency-injected instance::

        async with uow.using(shard_key=tenant_id):
            ...
    """
    clone = self.__class__()
    clone.routing = self.routing.merge(
        hint, db=db, read_only=read_only, shard_key=shard_key
    )
    return clone

Bases: AbstractAsyncContextManager

Source code in sqlargon/uow.py

class AbstractUnitOfWork(AbstractAsyncContextManager):
    @abstractmethod
    async def commit(self) -> None:
        raise NotImplementedError

    @abstractmethod
    async def rollback(self) -> None:
        raise NotImplementedError

Databases

Bases: ABC

Shared interface of a single :class:Database and a :class:~sqlargon.DatabaseCluster.

Everything built purely on sessions -- statement execution, named locks and the atomic/with_lock/inject_session decorators -- lives here, so code can accept either implementation interchangeably.

Source code in sqlargon/database.py

class BaseDatabase(ABC):
    """Shared interface of a single :class:`Database` and a
    :class:`~sqlargon.DatabaseCluster`.

    Everything built purely on sessions -- statement execution, named locks
    and the ``atomic``/``with_lock``/``inject_session`` decorators -- lives
    here, so code can accept either implementation interchangeably.
    """

    Model = Base

    dialect: str
    query_builder: QueryBuilder

    def __init__(self) -> None:
        self._locks: dict[str, _NamedLock] = {}

    @abstractmethod
    def route(self, context: RoutingContext | None = None) -> Database:
        """Return the concrete database a statement should run against."""

    @abstractmethod
    def session(
        self, context: RoutingContext | None = None
    ) -> AbstractAsyncContextManager[AsyncSession]:
        """Open a fresh session, committed on success and rolled back on error."""

    @abstractmethod
    def session_context(
        self, context: RoutingContext | None = None
    ) -> AbstractAsyncContextManager[AsyncSession]:
        """Yield the context-local session, opening one if none is active."""

    @abstractmethod
    async def verify_connection(self) -> None:
        """Check every underlying engine can execute a trivial statement."""

    @abstractmethod
    async def dispose(self) -> None:
        """Dispose every underlying engine and its connection pool."""

    @abstractmethod
    async def create_all(self) -> None:
        """Create all tables of :attr:`Model` metadata on writable databases."""

    @abstractmethod
    async def drop_all(self) -> None:
        """Drop all tables of :attr:`Model` metadata on writable databases."""

    async def execute(
        self,
        query: Executable | TypedReturnsRows,
        params: Params | None = None,
        **kwargs: Any,
    ) -> Result[Any]:
        """Execute a statement on the routed context-local session."""
        context = RoutingContext.create(statement=query)
        async with self.session_context(context) as session:
            return await session.execute(query, params, **kwargs)

    @asynccontextmanager
    async def lock(self, name: str) -> AsyncGenerator[None]:
        """Hold a named lock: a native (advisory) lock when the dialect
        supports one, a process-local asyncio lock otherwise."""
        if self.query_builder.supports(Option.LOCKS):
            async with self.native_lock(name):
                yield
        else:
            # the entry is reference counted: removing it while another task
            # still waits would let a third task create a fresh lock under
            # the same name and enter concurrently
            entry = self._locks.setdefault(name, _NamedLock())
            entry.refs += 1
            try:
                async with entry.lock:
                    yield
            finally:
                entry.refs -= 1
                if entry.refs == 0:
                    self._locks.pop(name, None)

    @asynccontextmanager
    async def native_lock(self, name: str) -> AsyncGenerator[None]:
        """Hold a database-native advisory lock for the duration of the block."""
        _lock, _release = self.query_builder.get_lock_pair(name)
        async with self.session_context() as session:
            try:
                await session.execute(_lock)
                yield
            finally:
                await session.execute(_release)

    def inject_session(
        self, func: Callable[P, Awaitable[R]]
    ) -> Callable[P, Awaitable[R]]:
        """Provide the context-local session as the ``session`` keyword
        argument unless the caller already passed one."""

        @wraps(func)
        async def wrapped(*args: P.args, **kwargs: P.kwargs) -> R:
            if kwargs.get("session") is None:
                async with self.session_context() as session:
                    kwargs["session"] = session
                    return await func(*args, **kwargs)
            else:
                return await func(*args, **kwargs)

        return wrapped

    def atomic(self, fn: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
        """Run the decorated coroutine within a single transaction."""

        @wraps(fn)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            async with self.session_context():
                return await fn(*args, **kwargs)

        return wrapper

    @overload
    def with_lock(
        self, fn: Callable[P, Awaitable[R]], *, key: str
    ) -> Callable[P, Awaitable[R]]: ...

    @overload
    def with_lock(
        self, fn: None = None, *, key: str
    ) -> Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[R]]]: ...

    def with_lock(
        self,
        fn: Callable[P, Awaitable[R]] | None = None,
        *,
        key: str,
    ) -> (
        Callable[P, Awaitable[R]]
        | Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[R]]]
    ):
        """Run the decorated coroutine while holding the named :meth:`lock`."""

        def wrapper(fn: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
            @wraps(fn)
            async def wrapped(*args: P.args, **kwargs: P.kwargs) -> R:
                async with self.lock(key):
                    return await fn(*args, **kwargs)

            return wrapped

        if fn is not None:
            return wrapper(fn)
        return wrapper

atomic(fn: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]

Run the decorated coroutine within a single transaction.

Source code in sqlargon/database.py

def atomic(self, fn: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
    """Run the decorated coroutine within a single transaction."""

    @wraps(fn)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        async with self.session_context():
            return await fn(*args, **kwargs)

    return wrapper

create_all() -> None abstractmethod async

Create all tables of :attr:Model metadata on writable databases.

Source code in sqlargon/database.py

@abstractmethod
async def create_all(self) -> None:
    """Create all tables of :attr:`Model` metadata on writable databases."""

dispose() -> None abstractmethod async

Dispose every underlying engine and its connection pool.

Source code in sqlargon/database.py

@abstractmethod
async def dispose(self) -> None:
    """Dispose every underlying engine and its connection pool."""

drop_all() -> None abstractmethod async

Drop all tables of :attr:Model metadata on writable databases.

Source code in sqlargon/database.py

@abstractmethod
async def drop_all(self) -> None:
    """Drop all tables of :attr:`Model` metadata on writable databases."""

execute(query: Executable | TypedReturnsRows, params: Params | None = None, **kwargs: Any) -> Result[Any] async

Execute a statement on the routed context-local session.

Source code in sqlargon/database.py

async def execute(
    self,
    query: Executable | TypedReturnsRows,
    params: Params | None = None,
    **kwargs: Any,
) -> Result[Any]:
    """Execute a statement on the routed context-local session."""
    context = RoutingContext.create(statement=query)
    async with self.session_context(context) as session:
        return await session.execute(query, params, **kwargs)

inject_session(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]

Provide the context-local session as the session keyword argument unless the caller already passed one.

Source code in sqlargon/database.py

def inject_session(
    self, func: Callable[P, Awaitable[R]]
) -> Callable[P, Awaitable[R]]:
    """Provide the context-local session as the ``session`` keyword
    argument unless the caller already passed one."""

    @wraps(func)
    async def wrapped(*args: P.args, **kwargs: P.kwargs) -> R:
        if kwargs.get("session") is None:
            async with self.session_context() as session:
                kwargs["session"] = session
                return await func(*args, **kwargs)
        else:
            return await func(*args, **kwargs)

    return wrapped

lock(name: str) -> AsyncGenerator[None] async

Hold a named lock: a native (advisory) lock when the dialect supports one, a process-local asyncio lock otherwise.

Source code in sqlargon/database.py

@asynccontextmanager
async def lock(self, name: str) -> AsyncGenerator[None]:
    """Hold a named lock: a native (advisory) lock when the dialect
    supports one, a process-local asyncio lock otherwise."""
    if self.query_builder.supports(Option.LOCKS):
        async with self.native_lock(name):
            yield
    else:
        # the entry is reference counted: removing it while another task
        # still waits would let a third task create a fresh lock under
        # the same name and enter concurrently
        entry = self._locks.setdefault(name, _NamedLock())
        entry.refs += 1
        try:
            async with entry.lock:
                yield
        finally:
            entry.refs -= 1
            if entry.refs == 0:
                self._locks.pop(name, None)

native_lock(name: str) -> AsyncGenerator[None] async

Hold a database-native advisory lock for the duration of the block.

Source code in sqlargon/database.py

@asynccontextmanager
async def native_lock(self, name: str) -> AsyncGenerator[None]:
    """Hold a database-native advisory lock for the duration of the block."""
    _lock, _release = self.query_builder.get_lock_pair(name)
    async with self.session_context() as session:
        try:
            await session.execute(_lock)
            yield
        finally:
            await session.execute(_release)

route(context: RoutingContext | None = None) -> Database abstractmethod

Return the concrete database a statement should run against.

Source code in sqlargon/database.py

@abstractmethod
def route(self, context: RoutingContext | None = None) -> Database:
    """Return the concrete database a statement should run against."""

session(context: RoutingContext | None = None) -> AbstractAsyncContextManager[AsyncSession] abstractmethod

Open a fresh session, committed on success and rolled back on error.

Source code in sqlargon/database.py

@abstractmethod
def session(
    self, context: RoutingContext | None = None
) -> AbstractAsyncContextManager[AsyncSession]:
    """Open a fresh session, committed on success and rolled back on error."""

session_context(context: RoutingContext | None = None) -> AbstractAsyncContextManager[AsyncSession] abstractmethod

Yield the context-local session, opening one if none is active.

Source code in sqlargon/database.py

@abstractmethod
def session_context(
    self, context: RoutingContext | None = None
) -> AbstractAsyncContextManager[AsyncSession]:
    """Yield the context-local session, opening one if none is active."""

verify_connection() -> None abstractmethod async

Check every underlying engine can execute a trivial statement.

Source code in sqlargon/database.py

@abstractmethod
async def verify_connection(self) -> None:
    """Check every underlying engine can execute a trivial statement."""

with_lock(fn: Callable[P, Awaitable[R]] | None = None, *, key: str) -> Callable[P, Awaitable[R]] | Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[R]]]

with_lock(
    fn: Callable[P, Awaitable[R]], *, key: str
) -> Callable[P, Awaitable[R]]

with_lock(
    fn: None = None, *, key: str
) -> Callable[
    [Callable[P, Awaitable[R]]], Callable[P, Awaitable[R]]
]

Run the decorated coroutine while holding the named :meth:lock.

Source code in sqlargon/database.py

def with_lock(
    self,
    fn: Callable[P, Awaitable[R]] | None = None,
    *,
    key: str,
) -> (
    Callable[P, Awaitable[R]]
    | Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[R]]]
):
    """Run the decorated coroutine while holding the named :meth:`lock`."""

    def wrapper(fn: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
        @wraps(fn)
        async def wrapped(*args: P.args, **kwargs: P.kwargs) -> R:
            async with self.lock(key):
                return await fn(*args, **kwargs)

        return wrapped

    if fn is not None:
        return wrapper(fn)
    return wrapper

Bases: BaseDatabase

A single (writable) database: one engine, sessions and query builder.

Routing policy lives outside this class -- to front replicas or shards, compose Database objects into a :class:~sqlargon.DatabaseCluster.

Source code in sqlargon/database.py

class Database(BaseDatabase):
    """A single (writable) database: one engine, sessions and query builder.

    Routing policy lives outside this class -- to front replicas or shards,
    compose ``Database`` objects into a :class:`~sqlargon.DatabaseCluster`.
    """

    def __init__(
        self,
        url: str,
        *,
        enable_tracker: bool = False,
        **kwargs: Any,
    ) -> None:
        super().__init__()
        self.engine = create_async_engine(url, **kwargs)
        self.session_maker = async_sessionmaker(
            bind=self.engine, expire_on_commit=False
        )
        self._current_session: ContextVar[AsyncSession | None] = ContextVar(
            "_current_session", default=None
        )
        self.dialect = self.engine.url.get_dialect().name
        self.query_builder = get_query_builder(self.dialect)

        if enable_tracker:
            from .tracker import TRACKER

            TRACKER.track_pool(self.engine.sync_engine)

        if SQLAlchemyInstrumentor is not None:
            SQLAlchemyInstrumentor().instrument(engine=self.engine.sync_engine)

    def route(self, context: RoutingContext | None = None) -> Database:  # noqa: ARG002
        """Return the database a statement should run against (always ``self``).

        Exists so a single database and a cluster expose the same interface.
        """
        return self

    @asynccontextmanager
    async def session(
        self,
        context: RoutingContext | None = None,  # noqa: ARG002
    ) -> AsyncGenerator[AsyncSession]:
        """Open a fresh session, committed on success and rolled back on error.

        ``context`` is accepted for interface parity with
        :class:`~sqlargon.DatabaseCluster` and ignored.
        """
        session = self.session_maker()

        try:
            yield session
            await session.commit()
        except:
            await session.rollback()
            raise
        finally:
            await session.close()

    @asynccontextmanager
    async def session_context(
        self,
        context: RoutingContext | None = None,  # noqa: ARG002
    ) -> AsyncGenerator[AsyncSession]:
        """Yield the context-local session, opening one if none is active.

        ``context`` is accepted for interface parity with
        :class:`~sqlargon.DatabaseCluster` and ignored -- there is only one
        database to route to.

        The current session lives in a :class:`~contextvars.ContextVar`, so
        it is task-local by construction; no lock is needed around the
        get-or-open below.
        """
        current_session = self._current_session.get()
        if current_session is not None:
            yield current_session
            return

        async with self.session() as session:
            token = self._current_session.set(session)
            try:
                yield session
            finally:
                self._current_session.reset(token)

    async def verify_connection(self) -> None:
        async with self.session() as session:
            await session.execute(text("SELECT 1;"))

    async def dispose(self) -> None:
        await self.engine.dispose()

    async def create_all(self) -> None:
        async with self.engine.begin() as conn:
            await conn.run_sync(self.Model.metadata.create_all)

    async def drop_all(self) -> None:
        async with self.engine.begin() as conn:
            await conn.run_sync(self.Model.metadata.drop_all)

    @classmethod
    def from_env(cls, **kwargs: Any) -> Self:
        """Build a single database from ``DATABASE_*`` settings.

        To route across replicas, build a :class:`~sqlargon.DatabaseCluster`
        with :meth:`~sqlargon.DatabaseCluster.from_env` instead.
        """
        from .settings import DatabaseSettings

        settings = DatabaseSettings(**kwargs)
        return cls(**settings.to_kwargs() | kwargs)

from_env(**kwargs: Any) -> Self classmethod

Build a single database from DATABASE_* settings.

To route across replicas, build a :class:~sqlargon.DatabaseCluster with :meth:~sqlargon.DatabaseCluster.from_env instead.

Source code in sqlargon/database.py

@classmethod
def from_env(cls, **kwargs: Any) -> Self:
    """Build a single database from ``DATABASE_*`` settings.

    To route across replicas, build a :class:`~sqlargon.DatabaseCluster`
    with :meth:`~sqlargon.DatabaseCluster.from_env` instead.
    """
    from .settings import DatabaseSettings

    settings = DatabaseSettings(**kwargs)
    return cls(**settings.to_kwargs() | kwargs)

route(context: RoutingContext | None = None) -> Database

Return the database a statement should run against (always self).

Exists so a single database and a cluster expose the same interface.

Source code in sqlargon/database.py

def route(self, context: RoutingContext | None = None) -> Database:  # noqa: ARG002
    """Return the database a statement should run against (always ``self``).

    Exists so a single database and a cluster expose the same interface.
    """
    return self

session(context: RoutingContext | None = None) -> AsyncGenerator[AsyncSession] async

Open a fresh session, committed on success and rolled back on error.

context is accepted for interface parity with :class:~sqlargon.DatabaseCluster and ignored.

Source code in sqlargon/database.py

@asynccontextmanager
async def session(
    self,
    context: RoutingContext | None = None,  # noqa: ARG002
) -> AsyncGenerator[AsyncSession]:
    """Open a fresh session, committed on success and rolled back on error.

    ``context`` is accepted for interface parity with
    :class:`~sqlargon.DatabaseCluster` and ignored.
    """
    session = self.session_maker()

    try:
        yield session
        await session.commit()
    except:
        await session.rollback()
        raise
    finally:
        await session.close()

session_context(context: RoutingContext | None = None) -> AsyncGenerator[AsyncSession] async

Yield the context-local session, opening one if none is active.

context is accepted for interface parity with :class:~sqlargon.DatabaseCluster and ignored -- there is only one database to route to.

The current session lives in a :class:~contextvars.ContextVar, so it is task-local by construction; no lock is needed around the get-or-open below.

Source code in sqlargon/database.py

@asynccontextmanager
async def session_context(
    self,
    context: RoutingContext | None = None,  # noqa: ARG002
) -> AsyncGenerator[AsyncSession]:
    """Yield the context-local session, opening one if none is active.

    ``context`` is accepted for interface parity with
    :class:`~sqlargon.DatabaseCluster` and ignored -- there is only one
    database to route to.

    The current session lives in a :class:`~contextvars.ContextVar`, so
    it is task-local by construction; no lock is needed around the
    get-or-open below.
    """
    current_session = self._current_session.get()
    if current_session is not None:
        yield current_session
        return

    async with self.session() as session:
        token = self._current_session.set(session)
        try:
            yield session
        finally:
            self._current_session.reset(token)

Bases: Database

A read replica that rejects any write statement at the engine level.

Compiled DML and textual write statements are rejected client side (a best-effort early error); the server-side guard is authoritative: on PostgreSQL the connection is put into read-only transaction mode, on SQLite and MySQL/MariaDB every new connection is switched to a read-only session mode.

Source code in sqlargon/database.py

class ReadOnlyDatabase(Database):
    """A read replica that rejects any write statement at the engine level.

    Compiled DML and textual write statements are rejected client side (a
    best-effort early error); the server-side guard is authoritative: on
    PostgreSQL the connection is put into read-only transaction mode, on
    SQLite and MySQL/MariaDB every new connection is switched to a
    read-only session mode.
    """

    def __init__(self, url: str, **kwargs: Any) -> None:
        if make_url(url).get_dialect().name == "postgresql":
            execution_options = dict(kwargs.get("execution_options") or {})
            execution_options.setdefault("postgresql_readonly", True)
            kwargs["execution_options"] = execution_options
        super().__init__(url, **kwargs)
        event.listen(self.engine.sync_engine, "before_execute", self._reject_writes)
        statement = _READ_ONLY_SESSION_STATEMENTS.get(self.dialect)
        if statement is not None:

            def _make_connection_read_only(dbapi_connection: Any, _: Any) -> None:
                cursor = dbapi_connection.cursor()
                try:
                    cursor.execute(statement)
                finally:
                    cursor.close()

            event.listen(self.engine.sync_engine, "connect", _make_connection_read_only)

    @staticmethod
    def _reject_writes(
        _conn: Any,
        clauseelement: Any,
        _multiparams: Any,
        _params: Any,
        _execution_options: Any,
    ) -> None:
        is_write = isinstance(clauseelement, UpdateBase) or (
            isinstance(clauseelement, TextClause) and _is_text_write(clauseelement.text)
        )
        if is_write:
            msg = f"Cannot execute {type(clauseelement).__name__} on a read replica"
            raise ReadOnlyError(msg)

Bases: RuntimeError

Raised when a write statement is executed against a read replica.

Source code in sqlargon/database.py

class ReadOnlyError(RuntimeError):
    """Raised when a write statement is executed against a read replica."""

Bases: BaseDatabase

Named databases behind a routing policy, usable wherever a :class:~sqlargon.Database is.

Repositories and units of work bind to the cluster; each statement is routed to a member database by, in order of precedence:

1. the database pinned by an open transaction in the current context,
2. an explicit hint (using(...), repository binding, unit of work),
3. the :class:~sqlargon.routing.Router,

A hint conflicting with the pinned database raises :class:~sqlargon.RoutingError instead of silently splitting the transaction. All members must share one SQL dialect, since queries are built before they are routed.

Source code in sqlargon/cluster.py

class DatabaseCluster(BaseDatabase):
    """Named databases behind a routing policy, usable wherever a
    :class:`~sqlargon.Database` is.

    Repositories and units of work bind to the cluster; each statement is
    routed to a member database by, in order of precedence:

    1. the database pinned by an open transaction in the current context,
    2. an explicit hint (``using(...)``, repository binding, unit of work),
    3. the :class:`~sqlargon.routing.Router`,

    A hint conflicting with the pinned database raises
    :class:`~sqlargon.RoutingError` instead of silently splitting the
    transaction. All members must share one SQL dialect, since queries are
    built before they are routed.
    """

    def __init__(
        self,
        databases: Mapping[str, Database],
        *,
        router: Router | None = None,
        default: str = "primary",
    ) -> None:
        super().__init__()
        if not databases:
            msg = "A database cluster requires at least one database"
            raise ValueError(msg)
        if default not in databases:
            msg = f"Default database {default!r} not in {sorted(databases)}"
            raise ValueError(msg)
        dialects = {database.dialect for database in databases.values()}
        if len(dialects) > 1:
            msg = (
                "All databases in a cluster must share one dialect "
                f"(queries are built before routing); got {sorted(dialects)}"
            )
            raise ValueError(msg)
        self.databases = dict(databases)
        self.default = default
        self.dialect = next(iter(dialects))
        self.query_builder = self.databases[default].query_builder
        self.router: Router = router if router is not None else DefaultRouter(default)
        self._pinned: ContextVar[Database | None] = ContextVar("_pinned", default=None)

    @classmethod
    def with_replicas(
        cls,
        url: str,
        *,
        read_replicas: Sequence[ReplicaConfig] | None = None,
        auto_route: bool = False,
        replica_strategy: Literal["random", "round_robin"] = "random",
        **kwargs: Any,
    ) -> DatabaseCluster:
        """Build a primary/replica cluster from connection URLs.

        Each replica is either a URL (inheriting the primary's engine
        options) or a mapping with a ``url`` key plus per-replica engine
        option overrides.
        """
        databases: dict[str, Database] = {"primary": Database(url, **kwargs)}
        for index, spec in enumerate(read_replicas or []):
            if isinstance(spec, str):
                replica_url, overrides = spec, {}
            else:
                overrides = dict(spec)
                replica_url = overrides.pop("url")
            databases[f"replica_{index}"] = ReadOnlyDatabase(
                replica_url, **{**kwargs, **overrides}
            )
        router = PrimaryReplicaRouter(
            replicas=[name for name in databases if name != "primary"],
            auto_route=auto_route,
            strategy=replica_strategy,
        )
        return cls(databases, router=router, default="primary")

    @classmethod
    def from_env(
        cls, *, router: Router | None = None, **kwargs: Any
    ) -> DatabaseCluster:
        """Build a cluster from ``DATABASE_*`` settings.

        Always returns a cluster: primary plus any ``DATABASE_READ_REPLICAS``
        (a single-primary cluster when none are configured, ready to grow).
        Pass ``router`` to replace the primary/replica policy, e.g. with a
        :class:`~sqlargon.ShardRouter` over the configured databases.
        """
        from .settings import DatabaseClusterSettings

        settings = DatabaseClusterSettings(**kwargs)
        cluster = cls.with_replicas(**settings.to_kwargs() | kwargs)
        if router is not None:
            cluster.router = router
        return cluster

    @property
    def default_database(self) -> Database:
        return self.databases[self.default]

    @property
    def replicas(self) -> list[Database]:
        return [
            database
            for database in self.databases.values()
            if isinstance(database, ReadOnlyDatabase)
        ]

    def _database_for(self, name: str) -> Database:
        try:
            return self.databases[name]
        except KeyError:
            msg = f"Unknown database {name!r}; available: {sorted(self.databases)}"
            raise RoutingError(msg) from None

    def route(self, context: RoutingContext | None = None) -> Database:
        """Return the member database the given statement should run against."""
        if context is None:
            context = RoutingContext.create()
        pinned = self._pinned.get()
        if pinned is not None:
            if (
                context.hint is not None
                and self._database_for(context.hint) is not pinned
            ):
                msg = (
                    f"Hint {context.hint!r} conflicts with the database pinned by "
                    "the open transaction; a transaction cannot span databases"
                )
                raise RoutingError(msg)
            return pinned
        if context.hint is not None:
            return self._database_for(context.hint)
        return self.router.route(self.databases, context)

    @asynccontextmanager
    async def session_context(
        self, context: RoutingContext | None = None
    ) -> AsyncGenerator[AsyncSession]:
        """Yield a session on the routed database, pinning it for the context.

        While the session is open, every statement in the same async context
        routes to the same database, so a unit of work never straddles two
        databases.
        """
        database = self.route(context)
        token = None
        if self._pinned.get() is None:
            token = self._pinned.set(database)
        try:
            async with database.session_context() as session:
                yield session
        finally:
            if token is not None:
                self._pinned.reset(token)

    @asynccontextmanager
    async def session(
        self, context: RoutingContext | None = None
    ) -> AsyncGenerator[AsyncSession]:
        async with self.route(context).session() as session:
            yield session

    def using(
        self,
        hint: str | None = None,
        *,
        read_only: bool | None = None,
        shard_key: Any | None = None,
    ) -> UsingContext:
        """Return a :func:`~sqlargon.using` marker validated against this cluster."""
        if hint is not None and hint not in self.databases:
            msg = f"Unknown database {hint!r}; available: {sorted(self.databases)}"
            raise RoutingError(msg)
        return _using(hint, read_only=read_only, shard_key=shard_key)

    async def verify_connection(self) -> None:
        for database in self.databases.values():
            await database.verify_connection()

    async def dispose(self) -> None:
        for database in self.databases.values():
            await database.dispose()

    async def create_all(self) -> None:
        for database in self._writable_databases():
            await database.create_all()

    async def drop_all(self) -> None:
        for database in self._writable_databases():
            await database.drop_all()

    def _writable_databases(self) -> list[Database]:
        seen: list[Database] = []
        for database in self.databases.values():
            if isinstance(database, ReadOnlyDatabase):
                continue
            if database not in seen:
                seen.append(database)
        return seen

from_env(*, router: Router | None = None, **kwargs: Any) -> DatabaseCluster classmethod

Build a cluster from DATABASE_* settings.

Always returns a cluster: primary plus any DATABASE_READ_REPLICAS (a single-primary cluster when none are configured, ready to grow). Pass router to replace the primary/replica policy, e.g. with a :class:~sqlargon.ShardRouter over the configured databases.

Source code in sqlargon/cluster.py

@classmethod
def from_env(
    cls, *, router: Router | None = None, **kwargs: Any
) -> DatabaseCluster:
    """Build a cluster from ``DATABASE_*`` settings.

    Always returns a cluster: primary plus any ``DATABASE_READ_REPLICAS``
    (a single-primary cluster when none are configured, ready to grow).
    Pass ``router`` to replace the primary/replica policy, e.g. with a
    :class:`~sqlargon.ShardRouter` over the configured databases.
    """
    from .settings import DatabaseClusterSettings

    settings = DatabaseClusterSettings(**kwargs)
    cluster = cls.with_replicas(**settings.to_kwargs() | kwargs)
    if router is not None:
        cluster.router = router
    return cluster

route(context: RoutingContext | None = None) -> Database

Return the member database the given statement should run against.

Source code in sqlargon/cluster.py

def route(self, context: RoutingContext | None = None) -> Database:
    """Return the member database the given statement should run against."""
    if context is None:
        context = RoutingContext.create()
    pinned = self._pinned.get()
    if pinned is not None:
        if (
            context.hint is not None
            and self._database_for(context.hint) is not pinned
        ):
            msg = (
                f"Hint {context.hint!r} conflicts with the database pinned by "
                "the open transaction; a transaction cannot span databases"
            )
            raise RoutingError(msg)
        return pinned
    if context.hint is not None:
        return self._database_for(context.hint)
    return self.router.route(self.databases, context)

session_context(context: RoutingContext | None = None) -> AsyncGenerator[AsyncSession] async

Yield a session on the routed database, pinning it for the context.

While the session is open, every statement in the same async context routes to the same database, so a unit of work never straddles two databases.

Source code in sqlargon/cluster.py

@asynccontextmanager
async def session_context(
    self, context: RoutingContext | None = None
) -> AsyncGenerator[AsyncSession]:
    """Yield a session on the routed database, pinning it for the context.

    While the session is open, every statement in the same async context
    routes to the same database, so a unit of work never straddles two
    databases.
    """
    database = self.route(context)
    token = None
    if self._pinned.get() is None:
        token = self._pinned.set(database)
    try:
        async with database.session_context() as session:
            yield session
    finally:
        if token is not None:
            self._pinned.reset(token)

using(hint: str | None = None, *, read_only: bool | None = None, shard_key: Any | None = None) -> UsingContext

Return a :func:~sqlargon.using marker validated against this cluster.

Source code in sqlargon/cluster.py

def using(
    self,
    hint: str | None = None,
    *,
    read_only: bool | None = None,
    shard_key: Any | None = None,
) -> UsingContext:
    """Return a :func:`~sqlargon.using` marker validated against this cluster."""
    if hint is not None and hint not in self.databases:
        msg = f"Unknown database {hint!r}; available: {sorted(self.databases)}"
        raise RoutingError(msg)
    return _using(hint, read_only=read_only, shard_key=shard_key)

with_replicas(url: str, *, read_replicas: Sequence[ReplicaConfig] | None = None, auto_route: bool = False, replica_strategy: Literal['random', 'round_robin'] = 'random', **kwargs: Any) -> DatabaseCluster classmethod

Build a primary/replica cluster from connection URLs.

Each replica is either a URL (inheriting the primary's engine options) or a mapping with a url key plus per-replica engine option overrides.

Source code in sqlargon/cluster.py

@classmethod
def with_replicas(
    cls,
    url: str,
    *,
    read_replicas: Sequence[ReplicaConfig] | None = None,
    auto_route: bool = False,
    replica_strategy: Literal["random", "round_robin"] = "random",
    **kwargs: Any,
) -> DatabaseCluster:
    """Build a primary/replica cluster from connection URLs.

    Each replica is either a URL (inheriting the primary's engine
    options) or a mapping with a ``url`` key plus per-replica engine
    option overrides.
    """
    databases: dict[str, Database] = {"primary": Database(url, **kwargs)}
    for index, spec in enumerate(read_replicas or []):
        if isinstance(spec, str):
            replica_url, overrides = spec, {}
        else:
            overrides = dict(spec)
            replica_url = overrides.pop("url")
        databases[f"replica_{index}"] = ReadOnlyDatabase(
            replica_url, **{**kwargs, **overrides}
        )
    router = PrimaryReplicaRouter(
        replicas=[name for name in databases if name != "primary"],
        auto_route=auto_route,
        strategy=replica_strategy,
    )
    return cls(databases, router=router, default="primary")

Return the default database, building it from DATABASE_* settings on first use.

The built default is a plain :class:~sqlargon.Database, or a primary/replica :class:~sqlargon.DatabaseCluster when DATABASE_READ_REPLICAS is configured. Building is guarded by a lock, so concurrent first calls share one instance instead of leaking engines.

Source code in sqlargon/registry.py

def get_default_database() -> AnyDatabase:
    """Return the default database, building it from ``DATABASE_*`` settings
    on first use.

    The built default is a plain :class:`~sqlargon.Database`, or a
    primary/replica :class:`~sqlargon.DatabaseCluster` when
    ``DATABASE_READ_REPLICAS`` is configured. Building is guarded by a lock,
    so concurrent first calls share one instance instead of leaking engines.
    """
    global _default_database  # noqa: PLW0603
    if _default_database is not None:
        return _default_database
    with _default_database_lock:
        if _default_database is None:
            _default_database = _build_default_database()
        return _default_database

Set the process-wide default database (or clear it with None).

Repositories and units of work that are not explicitly bound to a database fall back to this one.

Returns the previously set database (if any) so the caller can dispose() it once it is no longer used -- replacing the default does not close the old engine's connection pool.

Source code in sqlargon/registry.py

def set_default_database(database: AnyDatabase | None) -> AnyDatabase | None:
    """Set the process-wide default database (or clear it with ``None``).

    Repositories and units of work that are not explicitly bound to a
    database fall back to this one.

    Returns the previously set database (if any) so the caller can
    ``dispose()`` it once it is no longer used -- replacing the default does
    not close the old engine's connection pool.
    """
    global _default_database  # noqa: PLW0603
    with _default_database_lock:
        previous = _default_database
        _default_database = database
    return previous

Routing

Bases: Protocol

Routing policy: pick a database for a statement.

Routers are pure policy objects -- they never own engines and are only consulted when no transaction is pinned and no explicit hint is set.

Source code in sqlargon/routing.py

class Router(Protocol):
    """Routing policy: pick a database for a statement.

    Routers are pure policy objects -- they never own engines and are only
    consulted when no transaction is pinned and no explicit hint is set.
    """

    def route(
        self, databases: Mapping[str, Database], context: RoutingContext
    ) -> Database: ...

Everything a :class:Router may consult to pick a database.

operation is derived from the statement type (writes are any DML), read_only records an explicit user request for a read replica, and hint/shard_key carry explicit user choices made via :func:using, repository binding or unit-of-work arguments.

Source code in sqlargon/routing.py

@dataclass(frozen=True, slots=True)
class RoutingContext:
    """Everything a :class:`Router` may consult to pick a database.

    ``operation`` is derived from the statement type (writes are any DML),
    ``read_only`` records an explicit user request for a read replica, and
    ``hint``/``shard_key`` carry explicit user choices made via
    :func:`using`, repository binding or unit-of-work arguments.
    """

    statement: Any | None = None
    model: type[Base] | None = None
    operation: Operation = "write"
    read_only: bool = False
    hint: str | None = None
    shard_key: Any | None = None

    @classmethod
    def create(
        cls,
        statement: Any | None = None,
        model: type[Base] | None = None,
        *,
        read_only: bool | None = None,
        hint: str | None = None,
        shard_key: Any | None = None,
    ) -> RoutingContext:
        """Build a context, merging explicit arguments with :func:`using` markers.

        Explicit arguments win over ambient context variables: ``read_only``
        is tri-state, so an explicit ``False`` overrides an ambient
        ``using(read_only=True)`` scope while ``None`` inherits it. The
        operation defaults to ``write`` for statements that cannot be proven
        read-only, so unknown statements never leak onto a replica.
        """
        options = _routing_options.get()
        forced = bool(options.read_only if read_only is None else read_only)
        if forced or isinstance(statement, SelectBase):
            operation: Operation = "read"
        elif isinstance(statement, UpdateBase):
            operation = "write"
        else:
            operation = "write"
        return cls(
            statement=statement,
            model=model,
            operation=operation,
            read_only=forced,
            hint=hint if hint is not None else options.hint,
            shard_key=shard_key if shard_key is not None else options.shard_key,
        )

create(statement: Any | None = None, model: type[Base] | None = None, *, read_only: bool | None = None, hint: str | None = None, shard_key: Any | None = None) -> RoutingContext classmethod

Build a context, merging explicit arguments with :func:using markers.

Explicit arguments win over ambient context variables: read_only is tri-state, so an explicit False overrides an ambient using(read_only=True) scope while None inherits it. The operation defaults to write for statements that cannot be proven read-only, so unknown statements never leak onto a replica.

Source code in sqlargon/routing.py

@classmethod
def create(
    cls,
    statement: Any | None = None,
    model: type[Base] | None = None,
    *,
    read_only: bool | None = None,
    hint: str | None = None,
    shard_key: Any | None = None,
) -> RoutingContext:
    """Build a context, merging explicit arguments with :func:`using` markers.

    Explicit arguments win over ambient context variables: ``read_only``
    is tri-state, so an explicit ``False`` overrides an ambient
    ``using(read_only=True)`` scope while ``None`` inherits it. The
    operation defaults to ``write`` for statements that cannot be proven
    read-only, so unknown statements never leak onto a replica.
    """
    options = _routing_options.get()
    forced = bool(options.read_only if read_only is None else read_only)
    if forced or isinstance(statement, SelectBase):
        operation: Operation = "read"
    elif isinstance(statement, UpdateBase):
        operation = "write"
    else:
        operation = "write"
    return cls(
        statement=statement,
        model=model,
        operation=operation,
        read_only=forced,
        hint=hint if hint is not None else options.hint,
        shard_key=shard_key if shard_key is not None else options.shard_key,
    )

Explicit, ambient routing preferences (as opposed to the per-statement :class:RoutingContext, which is derived fresh for every execution).

Carried by repository instances and by the context-local :func:using marker. Immutable so copies can share it safely; derive a changed variant with :meth:merge.

Source code in sqlargon/routing.py

@dataclass(frozen=True, slots=True)
class RoutingOptions:
    """Explicit, ambient routing preferences (as opposed to the per-statement
    :class:`RoutingContext`, which is derived fresh for every execution).

    Carried by repository instances and by the context-local :func:`using`
    marker. Immutable so copies can share it safely; derive a changed variant
    with :meth:`merge`.
    """

    db: Database | DatabaseCluster | None = None
    hint: str | None = None
    read_only: bool | None = None
    shard_key: Any | None = None

    def merge(
        self,
        hint: str | None = None,
        *,
        db: Database | DatabaseCluster | None = None,
        read_only: bool | None = None,
        shard_key: Any | None = None,
    ) -> RoutingOptions:
        """Return a copy with the given preferences overriding unset ones.

        ``read_only`` is tri-state: ``None`` keeps the inherited value, while
        an explicit ``False`` overrides it, so a primary read can be forced
        from within a ``read_only=True`` scope.
        """
        return RoutingOptions(
            db=db if db is not None else self.db,
            hint=hint if hint is not None else self.hint,
            read_only=self.read_only if read_only is None else read_only,
            shard_key=shard_key if shard_key is not None else self.shard_key,
        )

    def context(
        self,
        statement: Any | None = None,
        model: type[Base] | None = None,
        *,
        read_only: bool | None = None,
    ) -> RoutingContext:
        """Build the per-statement :class:`RoutingContext` for these options."""
        return RoutingContext.create(
            statement=statement,
            model=model,
            read_only=self.read_only if read_only is None else read_only,
            hint=self.hint,
            shard_key=self.shard_key,
        )

context(statement: Any | None = None, model: type[Base] | None = None, *, read_only: bool | None = None) -> RoutingContext

Build the per-statement :class:RoutingContext for these options.

Source code in sqlargon/routing.py

def context(
    self,
    statement: Any | None = None,
    model: type[Base] | None = None,
    *,
    read_only: bool | None = None,
) -> RoutingContext:
    """Build the per-statement :class:`RoutingContext` for these options."""
    return RoutingContext.create(
        statement=statement,
        model=model,
        read_only=self.read_only if read_only is None else read_only,
        hint=self.hint,
        shard_key=self.shard_key,
    )

merge(hint: str | None = None, *, db: Database | DatabaseCluster | None = None, read_only: bool | None = None, shard_key: Any | None = None) -> RoutingOptions

Return a copy with the given preferences overriding unset ones.

read_only is tri-state: None keeps the inherited value, while an explicit False overrides it, so a primary read can be forced from within a read_only=True scope.

Source code in sqlargon/routing.py

def merge(
    self,
    hint: str | None = None,
    *,
    db: Database | DatabaseCluster | None = None,
    read_only: bool | None = None,
    shard_key: Any | None = None,
) -> RoutingOptions:
    """Return a copy with the given preferences overriding unset ones.

    ``read_only`` is tri-state: ``None`` keeps the inherited value, while
    an explicit ``False`` overrides it, so a primary read can be forced
    from within a ``read_only=True`` scope.
    """
    return RoutingOptions(
        db=db if db is not None else self.db,
        hint=hint if hint is not None else self.hint,
        read_only=self.read_only if read_only is None else read_only,
        shard_key=shard_key if shard_key is not None else self.shard_key,
    )

Bases: RuntimeError

Raised when a statement cannot be routed to a database.

Source code in sqlargon/routing.py

class RoutingError(RuntimeError):
    """Raised when a statement cannot be routed to a database."""

Mark statements in the wrapped scope with a routing preference.

Use it as a context manager (plain with -- the marker is a context variable, so it composes with async code) or as a decorator for async functions::

with using("replica_eu"):
    users = await repo.all()


@using(read_only=True)
async def report() -> None: ...

Source code in sqlargon/routing.py

def using(
    hint: str | None = None,
    *,
    read_only: bool | None = None,
    shard_key: Any | None = None,
) -> UsingContext:
    """Mark statements in the wrapped scope with a routing preference.

    Use it as a context manager (plain ``with`` -- the marker is a context
    variable, so it composes with async code) or as a decorator for async
    functions::

        with using("replica_eu"):
            users = await repo.all()


        @using(read_only=True)
        async def report() -> None: ...
    """
    return UsingContext(hint, read_only=read_only, shard_key=shard_key)

Routing marker returned by :func:using.

A synchronous context manager (safe inside async code -- the marker is a context variable) and a decorator for async functions. Not reentrant and not safe to share between concurrently running tasks; create one per scope (the decorator form does this automatically on every call).

Source code in sqlargon/routing.py

class UsingContext:
    """Routing marker returned by :func:`using`.

    A synchronous context manager (safe inside async code -- the marker is a
    context variable) and a decorator for async functions. Not reentrant and
    not safe to share between concurrently running tasks; create one per
    scope (the decorator form does this automatically on every call).
    """

    __slots__ = ("_token", "hint", "read_only", "shard_key")

    def __init__(
        self,
        hint: str | None = None,
        *,
        read_only: bool | None = None,
        shard_key: Any | None = None,
    ) -> None:
        self.hint = hint
        self.read_only = read_only
        self.shard_key = shard_key
        self._token: Token[RoutingOptions] | None = None

    def __enter__(self) -> None:
        if self._token is not None:
            msg = "using() marker is not reentrant; create a new one per scope"
            raise RuntimeError(msg)
        merged = _routing_options.get().merge(
            self.hint, read_only=self.read_only, shard_key=self.shard_key
        )
        self._token = _routing_options.set(merged)

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        if self._token is not None:
            _routing_options.reset(self._token)
            self._token = None

    def __call__(self, fn: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
        @wraps(fn)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            with using(self.hint, read_only=self.read_only, shard_key=self.shard_key):
                return await fn(*args, **kwargs)

        return wrapper

Build a dependency applying :func:using for the span of a request.

Designed for DI frameworks with generator dependencies (e.g. FastAPI): the returned zero-argument async generator sets the routing marker before the endpoint runs and resets it afterwards, so every repository or unit of work resolved in the same request routes accordingly::

@app.get("/users", dependencies=[Depends(use_context("replica_0"))])
async def list_users(repo: UserRepository = Depends()) -> list[UserOut]:
    return await repo.all()

Source code in sqlargon/routing.py

def use_context(
    hint: str | None = None,
    *,
    read_only: bool | None = None,
    shard_key: Any | None = None,
) -> Callable[[], AsyncIterator[None]]:
    """Build a dependency applying :func:`using` for the span of a request.

    Designed for DI frameworks with generator dependencies (e.g. FastAPI):
    the returned zero-argument async generator sets the routing marker before
    the endpoint runs and resets it afterwards, so every repository or unit
    of work resolved in the same request routes accordingly::

        @app.get("/users", dependencies=[Depends(use_context("replica_0"))])
        async def list_users(repo: UserRepository = Depends()) -> list[UserOut]:
            return await repo.all()
    """

    async def dependency() -> AsyncIterator[None]:
        with using(hint, read_only=read_only, shard_key=shard_key):
            yield

    return dependency

Route reads issued within fn to a read replica when one is available.

Use it to opt a repository method into replica routing without enabling auto_route globally::

class UserRepository(SQLAlchemyRepository[User]):
    @read_only
    async def active(self) -> Sequence[User]:
        return await self.filter(is_active=True).all()

Source code in sqlargon/routing.py

def read_only(fn: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
    """Route reads issued within ``fn`` to a read replica when one is available.

    Use it to opt a repository method into replica routing without enabling
    ``auto_route`` globally::

        class UserRepository(SQLAlchemyRepository[User]):
            @read_only
            async def active(self) -> Sequence[User]:
                return await self.filter(is_active=True).all()
    """
    return using(read_only=True)(fn)

Route every statement to a single named database.

Source code in sqlargon/routing.py

class DefaultRouter:
    """Route every statement to a single named database."""

    def __init__(self, name: str) -> None:
        self.name = name

    def route(
        self,
        databases: Mapping[str, Database],
        context: RoutingContext,  # noqa: ARG002
    ) -> Database:
        return _pick(databases, self.name)

Writes go to the primary, reads may go to a replica.

auto_route diverts SELECT statements to replicas; an explicit read_only request always forces a replica. With sticky_reads (the default) reads issued after a write in the same async context stay on the primary so the caller can read its own writes; disable it for long-lived worker tasks where the flag would never reset.

Source code in sqlargon/routing.py

class PrimaryReplicaRouter:
    """Writes go to the primary, reads may go to a replica.

    ``auto_route`` diverts ``SELECT`` statements to replicas; an explicit
    ``read_only`` request always forces a replica. With ``sticky_reads``
    (the default) reads issued after a write in the same async context stay
    on the primary so the caller can read its own writes; disable it for
    long-lived worker tasks where the flag would never reset.
    """

    def __init__(
        self,
        *,
        primary: str = "primary",
        replicas: Sequence[str],
        strategy: Literal["random", "round_robin"] = "random",
        auto_route: bool = True,
        sticky_reads: bool = True,
    ) -> None:
        self.primary = primary
        self.replicas = list(replicas)
        self.auto_route = auto_route
        self.sticky_reads = sticky_reads
        self.strategy = strategy
        self._cycle = itertools.cycle(self.replicas)
        self._wrote: ContextVar[bool] = ContextVar("_wrote", default=False)

    def _pick_replica(self) -> str:
        if self.strategy == "round_robin":
            return next(self._cycle)
        return random.choice(self.replicas)  # noqa: S311  # nosec B311

    def route(
        self, databases: Mapping[str, Database], context: RoutingContext
    ) -> Database:
        if not self.replicas or context.operation == "write":
            if context.operation == "write" and self.sticky_reads:
                self._wrote.set(True)
            return _pick(databases, self.primary)
        if context.read_only:
            return _pick(databases, self._pick_replica())
        if self.sticky_reads and self._wrote.get():
            return _pick(databases, self.primary)
        if self.auto_route and isinstance(context.statement, SelectBase):
            return _pick(databases, self._pick_replica())
        return _pick(databases, self.primary)

Route by model: vertical partitioning of tables across databases.

Resolution order: the explicit mapping (keyed by model class or table name), then the model's __database__ attribute, then default.

Source code in sqlargon/routing.py

class ModelRouter:
    """Route by model: vertical partitioning of tables across databases.

    Resolution order: the explicit ``mapping`` (keyed by model class or table
    name), then the model's ``__database__`` attribute, then ``default``.
    """

    def __init__(
        self,
        mapping: Mapping[type[Any] | str, str] | None = None,
        *,
        default: str,
        attribute: str = "__database__",
    ) -> None:
        self.mapping = dict(mapping or {})
        self.default = default
        self.attribute = attribute

    def _name_for(self, model: type[Any] | None) -> str:
        if model is None:
            return self.default
        if model in self.mapping:
            return self.mapping[model]
        tablename = getattr(model, "__tablename__", None)
        if tablename is not None and tablename in self.mapping:
            return self.mapping[tablename]
        return getattr(model, self.attribute, None) or self.default

    def route(
        self, databases: Mapping[str, Database], context: RoutingContext
    ) -> Database:
        return _pick(databases, self._name_for(context.model))

Route by shard key: horizontal partitioning across databases.

The shard key comes from the routing context (set explicitly or via using(shard_key=...)); shards maps it to a database name, either as a mapping or as a callable (e.g. hash-mod). A missing key falls back to default when given, otherwise raises :class:RoutingError -- sharded writes without a key are a caller bug, not a soft default.

Source code in sqlargon/routing.py

class ShardRouter:
    """Route by shard key: horizontal partitioning across databases.

    The shard key comes from the routing context (set explicitly or via
    ``using(shard_key=...)``); ``shards`` maps it to a database name, either
    as a mapping or as a callable (e.g. hash-mod). A missing key falls back
    to ``default`` when given, otherwise raises :class:`RoutingError` --
    sharded writes without a key are a caller bug, not a soft default.
    """

    def __init__(
        self,
        shards: Mapping[Any, str] | Callable[[Any], str],
        *,
        default: str | None = None,
    ) -> None:
        self.shards = shards
        self.default = default

    def _name_for(self, shard_key: Any) -> str:
        if callable(self.shards):
            return self.shards(shard_key)
        try:
            return self.shards[shard_key]
        except KeyError:
            msg = f"No shard configured for key {shard_key!r}"
            raise RoutingError(msg) from None

    def route(
        self, databases: Mapping[str, Database], context: RoutingContext
    ) -> Database:
        if context.shard_key is None:
            if self.default is not None:
                return _pick(databases, self.default)
            msg = (
                "No shard key in routing context; set one with "
                "using(shard_key=...) or configure a default shard"
            )
            raise RoutingError(msg)
        return _pick(databases, self._name_for(context.shard_key))

Query builder

Source code in sqlargon/query_builder.py

class QueryBuilder:
    supported_options: Option = Option.NONE

    @classmethod
    def supports(cls, option: Option) -> bool:
        return option in cls.supported_options

    def select(
        self,
        *args: Any,
        with_for_update: bool | WithForUpdate | None = None,
        options: tuple[Any, ...] | None = None,
    ) -> sa.Select:
        query = sa.select(*args)
        if with_for_update is True:
            query = query.with_for_update()
        elif with_for_update:
            query = query.with_for_update(**with_for_update)
        if options:
            query = query.options(*options)
        return query

    def excluded(self, table: _DMLTableArgument) -> Any:
        msg = f"Cannot reference the excluded row of {table}"
        raise UnsupportedOption(msg)

    def conflict_set(self, options: OnConflictOptions, excluded: Any) -> dict[str, Any]:
        set_ = options.get("set_") or ()
        exclude = options.get("exclude_set") or ()
        overrides = set_ if isinstance(set_, Mapping) else {}
        return {
            key: overrides[key] if key in overrides else getattr(excluded, key)
            for key in set_
            if key not in exclude
        }

    def _insert(
        self,
        table: _DMLTableArgument,
        values: Values | None = None,
        on_conflict: OnConflict | None = None,
    ) -> sa.Insert:
        if on_conflict and not self.supports(Option.CONFLICTS):
            raise UnsupportedOption
        query = sa.insert(table)
        if values is not None:
            query = query.values(values)
        return query

    @overload
    def insert(
        self,
        table: _DMLTableArgument,
        values: Values | None = ...,
        *,
        return_results: Literal[False] = ...,
        on_conflict: OnConflict | None = ...,
    ) -> sa.Insert: ...

    @overload
    def insert(
        self,
        table: _DMLTableArgument,
        values: Values | None = ...,
        *,
        return_results: Literal[True],
        on_conflict: OnConflict | None = ...,
    ) -> ReturningInsert: ...

    @overload
    def insert(
        self,
        table: _DMLTableArgument,
        values: Values | None = ...,
        *,
        return_results: bool = ...,
        on_conflict: OnConflict | None = ...,
    ) -> sa.Insert | ReturningInsert: ...

    def insert(
        self,
        table: _DMLTableArgument,
        values: Values | None = None,
        *,
        return_results: bool = False,
        on_conflict: OnConflict | None = None,
    ) -> sa.Insert | ReturningInsert:
        if return_results and not self.supports(Option.RETURNING):
            raise UnsupportedOption
        query = self._insert(table, values, on_conflict)
        if return_results:
            return query.returning(table)
        return query

    @overload
    def update(
        self,
        table: _DMLTableArgument,
        values: Values | None = ...,
        *,
        return_results: Literal[False] = ...,
    ) -> sa.Update: ...

    @overload
    def update(
        self,
        table: _DMLTableArgument,
        values: Values | None = ...,
        *,
        return_results: Literal[True],
    ) -> ReturningUpdate: ...

    @overload
    def update(
        self,
        table: _DMLTableArgument,
        values: Values | None = ...,
        *,
        return_results: bool,
    ) -> sa.Update | ReturningUpdate: ...

    def update(
        self,
        table: _DMLTableArgument,
        values: Values | None = None,
        *,
        return_results: bool = False,
    ) -> sa.Update | ReturningUpdate:

        if return_results and not self.supports(Option.RETURNING):
            raise UnsupportedOption
        query = sa.update(table)
        if values is not None:
            query = query.values(values)
        if return_results:
            return query.returning(table)
        return query

    @overload
    def delete(
        self, table: _DMLTableArgument, *, return_results: Literal[False] = ...
    ) -> sa.Delete: ...

    @overload
    def delete(
        self, table: _DMLTableArgument, *, return_results: Literal[True]
    ) -> ReturningDelete: ...

    @overload
    def delete(
        self, table: _DMLTableArgument, *, return_results: bool = ...
    ) -> sa.Delete | ReturningDelete: ...

    def delete(
        self, table: _DMLTableArgument, *, return_results: bool = False
    ) -> sa.Delete | ReturningDelete:
        if return_results and not self.supports(Option.RETURNING):
            raise UnsupportedOption
        query = sa.delete(table)
        if return_results:
            return query.returning(table)
        return query

    def filter(
        self, query: Any, *args: _ColumnExpressionArgument[bool], **kwargs: Any
    ) -> Any:
        if args:
            query = query.filter(*args)
        if kwargs:
            query = query.filter_by(**kwargs)
        return query

    def count(
        self,
        from_: _FromClauseArgument,
        *args: _ColumnExpressionArgument[bool],
        **kwargs: Any,
    ) -> sa.Select[tuple[int, ...]]:
        query = self.select(sa.func.count()).select_from(from_)
        return self.filter(query, *args, **kwargs)

    @overload
    def page(
        self,
        query: sa.Select,
        *,
        offset: int = ...,
        limit: int = ...,
        include_total: Literal[True],
    ) -> tuple[sa.Select, sa.Select[tuple[int, ...]]]: ...

    @overload
    def page(
        self,
        query: sa.Select,
        *,
        offset: int = ...,
        limit: int = ...,
        include_total: Literal[False] = ...,
    ) -> tuple[sa.Select, None]: ...

    @overload
    def page(
        self,
        query: sa.Select,
        *,
        offset: int = ...,
        limit: int = ...,
        include_total: bool = ...,
    ) -> tuple[sa.Select, None] | tuple[sa.Select, sa.Select[tuple[int, ...]]]: ...

    def page(
        self,
        query: sa.Select,
        *,
        offset: int = 1,
        limit: int = 100,
        include_total: bool = False,
    ) -> tuple[sa.Select, sa.Select[tuple[int, ...]] | None]:
        page_query = query.offset(offset).limit(limit)
        total_query = self.count(query.subquery()) if include_total else None
        return page_query, total_query

    def lock(self, key: str) -> sa.TextClause:
        msg = f"Cannot obtain lock for key {key}"
        raise NotImplementedError(msg)

    def unlock(self, key: str) -> sa.TextClause:
        msg = f"Cannot release lock for key {key}"
        raise NotImplementedError(msg)

    def get_lock_pair(self, key: str) -> tuple[sa.TextClause, sa.TextClause]:
        if not self.supports(Option.LOCKS):
            raise UnsupportedOption
        return self.lock(key), self.unlock(key)

Bases: Flag

Source code in sqlargon/query_builder.py

class Option(Flag):
    NONE = 0
    RETURNING = auto()
    CONFLICTS = auto()
    LOCKS = auto()

Bases: Exception

Source code in sqlargon/query_builder.py

class QueryBuilderError(Exception):
    pass

Bases: QueryBuilderError

Source code in sqlargon/query_builder.py

class UnsupportedOption(QueryBuilderError):
    pass

Source code in sqlargon/query_builder.py

@functools.cache
def get_query_builder(dialect: str) -> QueryBuilder:
    if dialect == "postgresql":
        from .dialects.postgres import PostgresqlQueryBuilder

        return PostgresqlQueryBuilder()
    if dialect == "sqlite":
        from .dialects.sqlite import SQLiteQueryBuilder

        return SQLiteQueryBuilder()
    if dialect == "mysql":
        from .dialects.mysql import MysqlQueryBuilder

        return MysqlQueryBuilder()
    return QueryBuilder()

Pagination

Bases: ABC

Immutable pagination configuration, attached to a repository class.

Concrete strategies act as descriptors: accessed on a repository instance they return a :class:Paginator bound to it, typed with the repository's model, so a repository declares its pagination once::

class UserRepository(SQLAlchemyRepository[User]):
    paginate = PageNumberPagination(default_page_size=25)


page = await UserRepository().paginate(page=2)  # NumberedPage[User]

Paging inputs are not validated; constrain them at the API boundary, e.g. with pydantic-typed FastAPI parameters.

Source code in sqlargon/pagination/abc.py

@dataclass(frozen=True, kw_only=True, slots=True)
class PaginationStrategy(ABC):
    """Immutable pagination configuration, attached to a repository class.

    Concrete strategies act as descriptors: accessed on a repository instance
    they return a :class:`Paginator` bound to it, typed with the repository's
    model, so a repository declares its pagination once::

        class UserRepository(SQLAlchemyRepository[User]):
            paginate = PageNumberPagination(default_page_size=25)


        page = await UserRepository().paginate(page=2)  # NumberedPage[User]

    Paging inputs are not validated; constrain them at the API boundary,
    e.g. with pydantic-typed FastAPI parameters.
    """

    default_page_size: int = 100
    unique: bool = False

    @abstractmethod
    def bind(self, source: SupportsPagination[Any]) -> Paginator[Any]:
        """Bind this strategy to a repository, returning a callable paginator."""

    def resolve_page_size(self, page_size: int | None) -> int:
        """Return the requested page size, or the configured default."""
        return self.default_page_size if page_size is None else page_size

bind(source: SupportsPagination[Any]) -> Paginator[Any] abstractmethod

Bind this strategy to a repository, returning a callable paginator.

Source code in sqlargon/pagination/abc.py

@abstractmethod
def bind(self, source: SupportsPagination[Any]) -> Paginator[Any]:
    """Bind this strategy to a repository, returning a callable paginator."""

resolve_page_size(page_size: int | None) -> int

Return the requested page size, or the configured default.

Source code in sqlargon/pagination/abc.py

def resolve_page_size(self, page_size: int | None) -> int:
    """Return the requested page size, or the configured default."""
    return self.default_page_size if page_size is None else page_size

Bases: Generic[ModelT]

A pagination strategy bound to a repository instance.

Concrete paginators are awaitable callables returning a page and expose a pages() async iterator walking consecutive pages until exhaustion.

Source code in sqlargon/pagination/abc.py

class Paginator(Generic[ModelT]):
    """A pagination strategy bound to a repository instance.

    Concrete paginators are awaitable callables returning a page and expose
    a ``pages()`` async iterator walking consecutive pages until exhaustion.
    """

    __slots__ = ("_config", "_source")

    def __init__(
        self, source: SupportsPagination[ModelT], config: PaginationStrategy
    ) -> None:
        self._source = source
        self._config = config

    def _items(self, result: Result[Any], *, as_model: bool) -> Sequence[Any]:
        results: ScalarResult[Any] | MappingResult = (
            result.scalars() if as_model else result.mappings()
        )
        if self._config.unique:
            results = results.unique()
        return results.all()

Bases: Protocol[ModelT]

Structural view of a repository, as required by pagination strategies.

Source code in sqlargon/pagination/abc.py

class SupportsPagination(Protocol[ModelT]):
    """Structural view of a repository, as required by pagination strategies."""

    model: type[ModelT]

    @property
    def query(self) -> Select[tuple[ModelT]]: ...

    @property
    def qb(self) -> QueryBuilder: ...

    def session(
        self, statement: Any = None, *, read_only: bool | None = None
    ) -> AbstractAsyncContextManager[AsyncSession]: ...

Bases: PaginationStrategy

Page-number pagination: ?page=3&page_size=25.

Page numbers are 1-based. The returned :class:NumberedPage carries has_more detected by over-fetching a single row, so no COUNT query is issued.

Source code in sqlargon/pagination/page.py

@dataclass(frozen=True, kw_only=True, slots=True)
class PageNumberPagination(PaginationStrategy):
    """Page-number pagination: ``?page=3&page_size=25``.

    Page numbers are 1-based. The returned :class:`NumberedPage` carries
    ``has_more`` detected by over-fetching a single row, so no COUNT query
    is issued.
    """

    @overload
    def __get__(self, obj: None, objtype: type | None = None, /) -> Self: ...

    @overload
    def __get__(
        self, obj: SupportsPagination[ModelT], objtype: type | None = None, /
    ) -> PageNumberPaginator[ModelT]: ...

    def __get__(
        self,
        obj: SupportsPagination[ModelT] | None,
        _objtype: type | None = None,
        /,
    ) -> Self | PageNumberPaginator[ModelT]:
        return self if obj is None else self.bind(obj)

    def bind(self, source: SupportsPagination[ModelT]) -> PageNumberPaginator[ModelT]:
        """Bind this strategy to a repository, returning a callable paginator."""
        return PageNumberPaginator(source, self)

bind(source: SupportsPagination[ModelT]) -> PageNumberPaginator[ModelT]

Bind this strategy to a repository, returning a callable paginator.

Source code in sqlargon/pagination/page.py

def bind(self, source: SupportsPagination[ModelT]) -> PageNumberPaginator[ModelT]:
    """Bind this strategy to a repository, returning a callable paginator."""
    return PageNumberPaginator(source, self)

Bases: PaginationStrategy

Page-number pagination that also reports total rows and pages.

Runs an additional COUNT query in the same session as the page query and returns a :class:TotalNumberedPage.

Source code in sqlargon/pagination/page.py

@dataclass(frozen=True, kw_only=True, slots=True)
class TotalPageNumberPagination(PaginationStrategy):
    """Page-number pagination that also reports total rows and pages.

    Runs an additional COUNT query in the same session as the page query and
    returns a :class:`TotalNumberedPage`.
    """

    @overload
    def __get__(self, obj: None, objtype: type | None = None, /) -> Self: ...

    @overload
    def __get__(
        self, obj: SupportsPagination[ModelT], objtype: type | None = None, /
    ) -> TotalPageNumberPaginator[ModelT]: ...

    def __get__(
        self,
        obj: SupportsPagination[ModelT] | None,
        _objtype: type | None = None,
        /,
    ) -> Self | TotalPageNumberPaginator[ModelT]:
        return self if obj is None else self.bind(obj)

    def bind(
        self, source: SupportsPagination[ModelT]
    ) -> TotalPageNumberPaginator[ModelT]:
        """Bind this strategy to a repository, returning a callable paginator."""
        return TotalPageNumberPaginator(source, self)

bind(source: SupportsPagination[ModelT]) -> TotalPageNumberPaginator[ModelT]

Bind this strategy to a repository, returning a callable paginator.

Source code in sqlargon/pagination/page.py

def bind(
    self, source: SupportsPagination[ModelT]
) -> TotalPageNumberPaginator[ModelT]:
    """Bind this strategy to a repository, returning a callable paginator."""
    return TotalPageNumberPaginator(source, self)

Bases: PaginationStrategy

Offset/limit pagination: ?offset=200&limit=100.

Suited for background processing and internal APIs. The returned :class:OffsetPage carries has_more detected by over-fetching a single row, so no COUNT query is issued.

Source code in sqlargon/pagination/offset_limit.py

@dataclass(frozen=True, kw_only=True, slots=True)
class LimitOffsetPagination(PaginationStrategy):
    """Offset/limit pagination: ``?offset=200&limit=100``.

    Suited for background processing and internal APIs. The returned
    :class:`OffsetPage` carries ``has_more`` detected by over-fetching a
    single row, so no COUNT query is issued.
    """

    @overload
    def __get__(self, obj: None, objtype: type | None = None, /) -> Self: ...

    @overload
    def __get__(
        self, obj: SupportsPagination[ModelT], objtype: type | None = None, /
    ) -> LimitOffsetPaginator[ModelT]: ...

    def __get__(
        self,
        obj: SupportsPagination[ModelT] | None,
        _objtype: type | None = None,
        /,
    ) -> Self | LimitOffsetPaginator[ModelT]:
        return self if obj is None else self.bind(obj)

    def bind(self, source: SupportsPagination[ModelT]) -> LimitOffsetPaginator[ModelT]:
        """Bind this strategy to a repository, returning a callable paginator."""
        return LimitOffsetPaginator(source, self)

bind(source: SupportsPagination[ModelT]) -> LimitOffsetPaginator[ModelT]

Bind this strategy to a repository, returning a callable paginator.

Source code in sqlargon/pagination/offset_limit.py

def bind(self, source: SupportsPagination[ModelT]) -> LimitOffsetPaginator[ModelT]:
    """Bind this strategy to a repository, returning a callable paginator."""
    return LimitOffsetPaginator(source, self)

Bases: PaginationStrategy

Offset/limit pagination that also reports the total number of rows.

Runs an additional COUNT query in the same session as the page query and returns a :class:TotalOffsetPage.

Source code in sqlargon/pagination/offset_limit.py

@dataclass(frozen=True, kw_only=True, slots=True)
class TotalLimitOffsetPagination(PaginationStrategy):
    """Offset/limit pagination that also reports the total number of rows.

    Runs an additional COUNT query in the same session as the page query and
    returns a :class:`TotalOffsetPage`.
    """

    @overload
    def __get__(self, obj: None, objtype: type | None = None, /) -> Self: ...

    @overload
    def __get__(
        self, obj: SupportsPagination[ModelT], objtype: type | None = None, /
    ) -> TotalLimitOffsetPaginator[ModelT]: ...

    def __get__(
        self,
        obj: SupportsPagination[ModelT] | None,
        _objtype: type | None = None,
        /,
    ) -> Self | TotalLimitOffsetPaginator[ModelT]:
        return self if obj is None else self.bind(obj)

    def bind(
        self, source: SupportsPagination[ModelT]
    ) -> TotalLimitOffsetPaginator[ModelT]:
        """Bind this strategy to a repository, returning a callable paginator."""
        return TotalLimitOffsetPaginator(source, self)

bind(source: SupportsPagination[ModelT]) -> TotalLimitOffsetPaginator[ModelT]

Bind this strategy to a repository, returning a callable paginator.

Source code in sqlargon/pagination/offset_limit.py

def bind(
    self, source: SupportsPagination[ModelT]
) -> TotalLimitOffsetPaginator[ModelT]:
    """Bind this strategy to a repository, returning a callable paginator."""
    return TotalLimitOffsetPaginator(source, self)

Bases: PaginationStrategy

Keyset (cursor) pagination for large datasets: ?cursor=<token>.

Cursors are opaque bookmarks produced by sqlakeyset; paging is stable under concurrent inserts and cheap on large offsets. The unique option does not apply to this strategy.

Source code in sqlargon/pagination/cursor.py

@dataclass(frozen=True, kw_only=True, slots=True)
class CursorPagination(PaginationStrategy):
    """Keyset (cursor) pagination for large datasets: ``?cursor=<token>``.

    Cursors are opaque bookmarks produced by ``sqlakeyset``; paging is stable
    under concurrent inserts and cheap on large offsets. The ``unique``
    option does not apply to this strategy.
    """

    @overload
    def __get__(self, obj: None, objtype: type | None = None, /) -> Self: ...

    @overload
    def __get__(
        self, obj: SupportsPagination[ModelT], objtype: type | None = None, /
    ) -> CursorPaginator[ModelT]: ...

    def __get__(
        self,
        obj: SupportsPagination[ModelT] | None,
        _objtype: type | None = None,
        /,
    ) -> Self | CursorPaginator[ModelT]:
        return self if obj is None else self.bind(obj)

    def bind(self, source: SupportsPagination[ModelT]) -> CursorPaginator[ModelT]:
        """Bind this strategy to a repository, returning a callable paginator."""
        return CursorPaginator(source, self)

bind(source: SupportsPagination[ModelT]) -> CursorPaginator[ModelT]

Bind this strategy to a repository, returning a callable paginator.

Source code in sqlargon/pagination/cursor.py

def bind(self, source: SupportsPagination[ModelT]) -> CursorPaginator[ModelT]:
    """Bind this strategy to a repository, returning a callable paginator."""
    return CursorPaginator(source, self)

CursorPage dataclass

Bases: Page[T]

Page of results addressed by opaque keyset cursors.

Source code in sqlargon/pagination/models.py

@dataclass(frozen=True, slots=True)
class CursorPage(Page[T]):
    """Page of results addressed by opaque keyset cursors."""

    cursor: str | None
    next_page: str | None
    previous_page: str | None

    @property
    def has_next(self) -> bool:
        return self.next_page is not None

    @property
    def has_previous(self) -> bool:
        return self.previous_page is not None

NumberedPage dataclass

Bases: Page[T]

Page of results addressed by a 1-based page number.

Source code in sqlargon/pagination/models.py

@dataclass(frozen=True, slots=True)
class NumberedPage(Page[T]):
    """Page of results addressed by a 1-based page number."""

    current_page: int
    page_size: int
    has_more: bool

OffsetPage dataclass

Bases: Page[T]

Page of results addressed by offset/limit.

Source code in sqlargon/pagination/models.py

@dataclass(frozen=True, slots=True)
class OffsetPage(Page[T]):
    """Page of results addressed by ``offset``/``limit``."""

    offset: int
    limit: int
    has_more: bool

Page dataclass

Bases: Generic[T]

Base page shape shared by all pagination strategies.

Source code in sqlargon/pagination/models.py

@dataclass(frozen=True, slots=True)
class Page(Generic[T]):
    """Base page shape shared by all pagination strategies."""

    items: Sequence[T]

    @property
    def items_on_page(self) -> int:
        return len(self.items)

TotalNumberedPage dataclass

Bases: NumberedPage[T]

Numbered page including the total number of matching rows and pages.

Source code in sqlargon/pagination/models.py

@dataclass(frozen=True, slots=True)
class TotalNumberedPage(NumberedPage[T]):
    """Numbered page including the total number of matching rows and pages."""

    total_pages: int
    total_items: int

TotalOffsetPage dataclass

Bases: OffsetPage[T]

Offset page including the total number of matching rows.

Source code in sqlargon/pagination/models.py

@dataclass(frozen=True, slots=True)
class TotalOffsetPage(OffsetPage[T]):
    """Offset page including the total number of matching rows."""

    total_items: int

Cron

Database-backed cron scheduler for a single namespace.

Declarative mode: decorate functions with :meth:task and call :meth:run (or :meth:sync); declared tasks are created or updated and tasks no longer declared are deleted. Imperative mode: manage schedules at runtime with :meth:schedule and :meth:unschedule.

Multiple instances may run the same namespace concurrently; due tasks are claimed with FOR UPDATE SKIP LOCKED so each run executes once.

Source code in sqlargon/cron/manager.py

class Cron:
    """Database-backed cron scheduler for a single namespace.

    Declarative mode: decorate functions with :meth:`task` and call
    :meth:`run` (or :meth:`sync`); declared tasks are created or updated and
    tasks no longer declared are deleted. Imperative mode: manage schedules
    at runtime with :meth:`schedule` and :meth:`unschedule`.

    Multiple instances may run the same namespace concurrently; due tasks
    are claimed with ``FOR UPDATE SKIP LOCKED`` so each run executes once.
    """

    def __init__(
        self,
        namespace: str = "default",
        *,
        repository: CronTaskRepository | None = None,
        poll_interval: float = 1.0,
        max_concurrency: int = 32,
        batch_size: int = 100,
    ) -> None:
        self.namespace = namespace
        self.poll_interval = poll_interval
        self.batch_size = batch_size
        # a repository carries the statement it is building, so each scheduler
        # needs its own; pass one bound elsewhere with
        # ``CronTaskRepository().using(db=...)`` to use another database
        self.repository = repository if repository is not None else CronTaskRepository()
        self.max_concurrency = max_concurrency
        self._registry: dict[str, _Registration] = {}
        # the reverse index of the registry, keyed by the function as given
        self._names: dict[Any, str] = {}
        self._limiter: CapacityLimiter | None = None

    @property
    def limiter(self) -> CapacityLimiter:
        """The bound on concurrent executions.

        Built on first execution rather than in ``__init__``, so a scheduler
        can be constructed at import time, before an event loop exists.
        """
        if self._limiter is None:
            self._limiter = CapacityLimiter(self.max_concurrency)
        return self._limiter

    def task(
        self, schedule: str | None = None, *, name: str | None = None
    ) -> Callable[[F], F]:
        """Register the decorated function; with ``schedule`` it becomes a
        declarative task reconciled by :meth:`sync`."""

        def decorator(func: F) -> F:
            task_name = self.register(func, name=name)
            if schedule is not None:
                self._registry[task_name] = replace(
                    self._registry[task_name], schedule=validate_schedule(schedule)
                )
            return func

        return decorator

    def register(self, func: TaskFunc, *, name: str | None = None) -> str:
        """Make ``func`` executable by this scheduler and return its name.

        Sync functions are wrapped once here to run in a worker thread. A name
        registered again keeps the schedule it was declared with.
        """
        name = name or _default_name(func)
        declared = self._registry.get(name)
        self._registry[name] = _Registration(
            func if _is_async(func) else _in_thread(func),
            declared.schedule if declared else None,
        )
        self._names[func] = name
        return name

    async def sync(self) -> None:
        """Reconcile declarative tasks with the database: create missing
        ones, update changed schedules and delete undeclared ones.

        A declared name already held by an imperative row takes that row over
        rather than leaving the declaration unapplied.
        """
        declared = {
            name: registration.schedule
            for name, registration in self._registry.items()
            if registration.schedule is not None
        }
        await self.repository.reconcile(self.namespace, declared, utc_now())

    async def schedule(
        self,
        func: TaskFunc | str,
        schedule: str,
        *args: Any,
        name: str | None = None,
        **kwargs: Any,
    ) -> CronTask:
        """Create or update a task at runtime.

        ``func`` may be a callable (registered automatically) or the name of
        a function registered on another instance of this namespace. Extra
        ``args`` and ``kwargs`` must be JSON-serializable; they are stored
        with the task and passed to the function on execution.
        """
        name = self.register(func, name=name) if callable(func) else name or func
        values = {
            "namespace": self.namespace,
            "name": name,
            "schedule": validate_schedule(schedule),
            "next_run_at": next_run_time(schedule, utc_now()),
            "args": list(args) or None,
            "kwargs": kwargs or None,
        }
        return await self.repository.upsert(
            values,
            return_results=True,
            index_elements={"namespace", "name"},
            set_={"schedule", "next_run_at", "args", "kwargs"},
        ).one()

    async def unschedule(self, func: TaskFunc | str) -> None:
        """Delete the task from the database."""
        await self.repository.remove(
            namespace=self.namespace, name=self._task_name(func)
        )

    async def pause(self, func: TaskFunc | str) -> None:
        """Stop scheduling the task without deleting it."""
        await self._set_enabled(self._task_name(func), enabled=False)

    async def resume(self, func: TaskFunc | str) -> None:
        """Re-enable a paused task."""
        await self._set_enabled(self._task_name(func), enabled=True)

    async def tasks(self) -> Sequence[CronTask]:
        """Return all tasks stored in this namespace."""
        return await self.repository.list(namespace=self.namespace)

    async def run(self, *, task_status: TaskStatus[None] = TASK_STATUS_IGNORED) -> None:
        """Sync declarative tasks, then poll and execute due ones until
        cancelled.

        Polling errors are logged and retried on the next tick, so a database
        blip never takes the scheduler down.
        """
        await self.sync()
        async with anyio.create_task_group() as tg:
            task_status.started()
            while True:
                await self._poll(tg)
                await anyio.sleep(self.poll_interval)

    @asynccontextmanager
    async def running(self) -> AsyncGenerator[Cron]:
        """Run the scheduler in the background, e.g. in an ASGI lifespan."""
        async with anyio.create_task_group() as tg:
            await tg.start(self.run)
            try:
                yield self
            finally:
                tg.cancel_scope.cancel()

    def _task_name(self, func: TaskFunc | str) -> str:
        """Resolve ``func`` to the name it was registered under."""
        if isinstance(func, str):
            return func
        return self._names.get(func) or _default_name(func)

    async def _poll(self, tg: TaskGroup) -> None:
        try:
            tasks = await self.repository.claim_due(
                self.namespace,
                utc_now(),
                names=set(self._registry),
                limit=self.batch_size,
            )
        except Exception:
            logger.exception("Claiming due cron tasks failed")
            return
        for task in tasks:
            tg.start_soon(self._execute, task)

    async def _set_enabled(self, name: str, *, enabled: bool) -> None:
        await self.repository.update_one(
            {"enabled": enabled}, namespace=self.namespace, name=name
        )

    async def _execute(self, task: CronTask) -> None:
        registration = self._registry.get(task.name)
        if registration is None:
            logger.warning("No function registered for cron task %r", task.name)
            return
        async with self.limiter:
            try:
                await registration.func(*(task.args or ()), **(task.kwargs or {}))
            except Exception:
                logger.exception("Cron task %r failed", task.name)

limiter: CapacityLimiter property

The bound on concurrent executions.

Built on first execution rather than in __init__, so a scheduler can be constructed at import time, before an event loop exists.

pause(func: TaskFunc | str) -> None async

Stop scheduling the task without deleting it.

Source code in sqlargon/cron/manager.py

async def pause(self, func: TaskFunc | str) -> None:
    """Stop scheduling the task without deleting it."""
    await self._set_enabled(self._task_name(func), enabled=False)

register(func: TaskFunc, *, name: str | None = None) -> str

Make func executable by this scheduler and return its name.

Sync functions are wrapped once here to run in a worker thread. A name registered again keeps the schedule it was declared with.

Source code in sqlargon/cron/manager.py

def register(self, func: TaskFunc, *, name: str | None = None) -> str:
    """Make ``func`` executable by this scheduler and return its name.

    Sync functions are wrapped once here to run in a worker thread. A name
    registered again keeps the schedule it was declared with.
    """
    name = name or _default_name(func)
    declared = self._registry.get(name)
    self._registry[name] = _Registration(
        func if _is_async(func) else _in_thread(func),
        declared.schedule if declared else None,
    )
    self._names[func] = name
    return name

resume(func: TaskFunc | str) -> None async

Re-enable a paused task.

Source code in sqlargon/cron/manager.py

async def resume(self, func: TaskFunc | str) -> None:
    """Re-enable a paused task."""
    await self._set_enabled(self._task_name(func), enabled=True)

run(*, task_status: TaskStatus[None] = TASK_STATUS_IGNORED) -> None async

Sync declarative tasks, then poll and execute due ones until cancelled.

Polling errors are logged and retried on the next tick, so a database blip never takes the scheduler down.

Source code in sqlargon/cron/manager.py

async def run(self, *, task_status: TaskStatus[None] = TASK_STATUS_IGNORED) -> None:
    """Sync declarative tasks, then poll and execute due ones until
    cancelled.

    Polling errors are logged and retried on the next tick, so a database
    blip never takes the scheduler down.
    """
    await self.sync()
    async with anyio.create_task_group() as tg:
        task_status.started()
        while True:
            await self._poll(tg)
            await anyio.sleep(self.poll_interval)

running() -> AsyncGenerator[Cron] async

Run the scheduler in the background, e.g. in an ASGI lifespan.

Source code in sqlargon/cron/manager.py

@asynccontextmanager
async def running(self) -> AsyncGenerator[Cron]:
    """Run the scheduler in the background, e.g. in an ASGI lifespan."""
    async with anyio.create_task_group() as tg:
        await tg.start(self.run)
        try:
            yield self
        finally:
            tg.cancel_scope.cancel()

schedule(func: TaskFunc | str, schedule: str, *args: Any, name: str | None = None, **kwargs: Any) -> CronTask async

Create or update a task at runtime.

func may be a callable (registered automatically) or the name of a function registered on another instance of this namespace. Extra args and kwargs must be JSON-serializable; they are stored with the task and passed to the function on execution.

Source code in sqlargon/cron/manager.py

async def schedule(
    self,
    func: TaskFunc | str,
    schedule: str,
    *args: Any,
    name: str | None = None,
    **kwargs: Any,
) -> CronTask:
    """Create or update a task at runtime.

    ``func`` may be a callable (registered automatically) or the name of
    a function registered on another instance of this namespace. Extra
    ``args`` and ``kwargs`` must be JSON-serializable; they are stored
    with the task and passed to the function on execution.
    """
    name = self.register(func, name=name) if callable(func) else name or func
    values = {
        "namespace": self.namespace,
        "name": name,
        "schedule": validate_schedule(schedule),
        "next_run_at": next_run_time(schedule, utc_now()),
        "args": list(args) or None,
        "kwargs": kwargs or None,
    }
    return await self.repository.upsert(
        values,
        return_results=True,
        index_elements={"namespace", "name"},
        set_={"schedule", "next_run_at", "args", "kwargs"},
    ).one()

sync() -> None async

Reconcile declarative tasks with the database: create missing ones, update changed schedules and delete undeclared ones.

A declared name already held by an imperative row takes that row over rather than leaving the declaration unapplied.

Source code in sqlargon/cron/manager.py

async def sync(self) -> None:
    """Reconcile declarative tasks with the database: create missing
    ones, update changed schedules and delete undeclared ones.

    A declared name already held by an imperative row takes that row over
    rather than leaving the declaration unapplied.
    """
    declared = {
        name: registration.schedule
        for name, registration in self._registry.items()
        if registration.schedule is not None
    }
    await self.repository.reconcile(self.namespace, declared, utc_now())

task(schedule: str | None = None, *, name: str | None = None) -> Callable[[F], F]

Register the decorated function; with schedule it becomes a declarative task reconciled by :meth:sync.

Source code in sqlargon/cron/manager.py

def task(
    self, schedule: str | None = None, *, name: str | None = None
) -> Callable[[F], F]:
    """Register the decorated function; with ``schedule`` it becomes a
    declarative task reconciled by :meth:`sync`."""

    def decorator(func: F) -> F:
        task_name = self.register(func, name=name)
        if schedule is not None:
            self._registry[task_name] = replace(
                self._registry[task_name], schedule=validate_schedule(schedule)
            )
        return func

    return decorator

tasks() -> Sequence[CronTask] async

Return all tasks stored in this namespace.

Source code in sqlargon/cron/manager.py

async def tasks(self) -> Sequence[CronTask]:
    """Return all tasks stored in this namespace."""
    return await self.repository.list(namespace=self.namespace)

unschedule(func: TaskFunc | str) -> None async

Delete the task from the database.

Source code in sqlargon/cron/manager.py

async def unschedule(self, func: TaskFunc | str) -> None:
    """Delete the task from the database."""
    await self.repository.remove(
        namespace=self.namespace, name=self._task_name(func)
    )

Bases: UUIDModelMixin, CreatedUpdatedMixin, Base

A scheduled task persisted per namespace.

Source code in sqlargon/cron/models.py

class CronTask(UUIDModelMixin, CreatedUpdatedMixin, Base):
    """A scheduled task persisted per namespace."""

    __tablename__ = "cron_tasks"
    __table_args__ = (
        sa.UniqueConstraint("namespace", "name"),
        sa.Index("idx_cron_tasks_namespace_next_run_at", "namespace", "next_run_at"),
    )

    namespace: Mapped[str] = mapped_column(sa.String(255), nullable=False)
    name: Mapped[str] = mapped_column(sa.String(255), nullable=False)
    schedule: Mapped[str] = mapped_column(sa.String(255), nullable=False)
    declarative: Mapped[bool] = mapped_column(
        sa.Boolean(), nullable=False, default=False, server_default=sa.sql.false()
    )
    enabled: Mapped[bool] = mapped_column(
        sa.Boolean(), nullable=False, default=True, server_default=sa.sql.true()
    )
    next_run_at: Mapped[datetime] = mapped_column(Timestamp(), nullable=False)
    last_run_at: Mapped[datetime | None] = mapped_column(Timestamp(), nullable=True)
    args: Mapped[list[Any] | None] = mapped_column(JSON(), nullable=True)
    kwargs: Mapped[dict[str, Any] | None] = mapped_column(JSON(), nullable=True)

Bases: SQLAlchemyRepository[CronTask]

Repository for :class:CronTask rows.

Source code in sqlargon/cron/repository.py

class CronTaskRepository(SQLAlchemyRepository[CronTask]):
    """Repository for :class:`CronTask` rows."""

    async def reconcile(
        self, namespace: str, schedules: Mapping[str, str], now: datetime
    ) -> None:
        """Make the declarative tasks of ``namespace`` exactly ``schedules``.

        The declared tasks are upserted -- creating the missing ones and taking
        over any row already holding their name -- and one delete removes the
        declarative rows no longer declared. Rows scheduled imperatively are
        left alone.

        ``next_run_at`` is only recomputed for a task whose schedule changed,
        so a run that came due while the scheduler was down still fires.
        """
        async with self.session():
            if schedules:
                await self._upsert_declared(namespace, schedules, now)
            await self.remove(
                CronTask.namespace == namespace,
                CronTask.declarative,
                CronTask.name.not_in(list(schedules)),
            )

    async def _upsert_declared(
        self, namespace: str, schedules: Mapping[str, str], now: datetime
    ) -> None:
        """Create or update the declared tasks of ``namespace``.

        The tasks whose schedule changed are upserted apart from the rest, so
        that only they have their ``next_run_at`` overwritten. Which group a
        task belongs to is decided here, from the schedules read back first,
        rather than by a conditional assignment in the upsert itself: MySQL
        applies the assignments of an ``ON DUPLICATE KEY UPDATE`` in order, so
        one reading ``schedule`` would already see the value the same statement
        had just written to it.
        """
        stored = await self._stored_schedules(namespace, schedules)
        unchanged: list[dict[str, Any]] = []
        rescheduled: list[dict[str, Any]] = []
        for name, schedule in schedules.items():
            group = unchanged if stored.get(name) == schedule else rescheduled
            group.append(
                {
                    "namespace": namespace,
                    "name": name,
                    "schedule": schedule,
                    "declarative": True,
                    "next_run_at": next_run_time(schedule, now),
                }
            )
        if unchanged:
            await self.bulk_create_or_update(
                unchanged,
                index_elements=CONFLICT_KEY,
                set_={"schedule", "declarative"},
            )
        if rescheduled:
            await self.bulk_create_or_update(
                rescheduled,
                index_elements=CONFLICT_KEY,
                set_={"schedule", "declarative", "next_run_at"},
            )

    async def _stored_schedules(
        self, namespace: str, names: Collection[str]
    ) -> dict[str, str]:
        """The schedule each of ``names`` currently holds in ``namespace``.

        Only the two columns are read, so the rows the upsert is about to
        overwrite are not left stale in the session identity map.
        """
        query = self.select(CronTask.name, CronTask.schedule).filter(
            CronTask.namespace == namespace, CronTask.name.in_(list(names))
        )
        return {row["name"]: row["schedule"] for row in await query.mappings()}

    async def claim_due(
        self,
        namespace: str,
        now: datetime,
        *,
        names: Collection[str] | None = None,
        limit: int = 100,
    ) -> Sequence[CronTask]:
        """Claim tasks due at ``now`` and return them.

        Due rows are locked with ``FOR UPDATE SKIP LOCKED`` and their
        ``next_run_at`` advanced within the same transaction, so concurrent
        instances polling the same namespace never claim the same run.

        With ``names``, only those tasks are claimed -- a caller that cannot
        execute a task leaves its run to an instance that can, instead of
        consuming the slot and dropping it.
        """
        filters = [
            CronTask.namespace == namespace,
            CronTask.enabled,
            CronTask.next_run_at <= now,
        ]
        if names is not None:
            filters.append(CronTask.name.in_(names))
        async with self.session():
            tasks = await (
                self.select(with_for_update={"skip_locked": True})
                .filter(*filters)
                .order_by(CronTask.next_run_at)
                .limit(limit)
                .all()
            )
            for task in tasks:
                task.last_run_at = now
                task.next_run_at = next_run_time(task.schedule, now)
            return tasks

claim_due(namespace: str, now: datetime, *, names: Collection[str] | None = None, limit: int = 100) -> Sequence[CronTask] async

Claim tasks due at now and return them.

Due rows are locked with FOR UPDATE SKIP LOCKED and their next_run_at advanced within the same transaction, so concurrent instances polling the same namespace never claim the same run.

With names, only those tasks are claimed -- a caller that cannot execute a task leaves its run to an instance that can, instead of consuming the slot and dropping it.

Source code in sqlargon/cron/repository.py

async def claim_due(
    self,
    namespace: str,
    now: datetime,
    *,
    names: Collection[str] | None = None,
    limit: int = 100,
) -> Sequence[CronTask]:
    """Claim tasks due at ``now`` and return them.

    Due rows are locked with ``FOR UPDATE SKIP LOCKED`` and their
    ``next_run_at`` advanced within the same transaction, so concurrent
    instances polling the same namespace never claim the same run.

    With ``names``, only those tasks are claimed -- a caller that cannot
    execute a task leaves its run to an instance that can, instead of
    consuming the slot and dropping it.
    """
    filters = [
        CronTask.namespace == namespace,
        CronTask.enabled,
        CronTask.next_run_at <= now,
    ]
    if names is not None:
        filters.append(CronTask.name.in_(names))
    async with self.session():
        tasks = await (
            self.select(with_for_update={"skip_locked": True})
            .filter(*filters)
            .order_by(CronTask.next_run_at)
            .limit(limit)
            .all()
        )
        for task in tasks:
            task.last_run_at = now
            task.next_run_at = next_run_time(task.schedule, now)
        return tasks

reconcile(namespace: str, schedules: Mapping[str, str], now: datetime) -> None async

Make the declarative tasks of namespace exactly schedules.

The declared tasks are upserted -- creating the missing ones and taking over any row already holding their name -- and one delete removes the declarative rows no longer declared. Rows scheduled imperatively are left alone.

next_run_at is only recomputed for a task whose schedule changed, so a run that came due while the scheduler was down still fires.

Source code in sqlargon/cron/repository.py

async def reconcile(
    self, namespace: str, schedules: Mapping[str, str], now: datetime
) -> None:
    """Make the declarative tasks of ``namespace`` exactly ``schedules``.

    The declared tasks are upserted -- creating the missing ones and taking
    over any row already holding their name -- and one delete removes the
    declarative rows no longer declared. Rows scheduled imperatively are
    left alone.

    ``next_run_at`` is only recomputed for a task whose schedule changed,
    so a run that came due while the scheduler was down still fires.
    """
    async with self.session():
        if schedules:
            await self._upsert_declared(namespace, schedules, now)
        await self.remove(
            CronTask.namespace == namespace,
            CronTask.declarative,
            CronTask.name.not_in(list(schedules)),
        )

Return schedule unchanged, raising ValueError if it is not a valid cron expression.

Source code in sqlargon/cron/utils.py

def validate_schedule(schedule: str) -> str:
    """Return ``schedule`` unchanged, raising ``ValueError`` if it is not a
    valid cron expression."""
    if not croniter.is_valid(schedule):
        msg = f"Invalid cron expression: {schedule!r}"
        raise ValueError(msg)
    return schedule

Return the first run time of schedule strictly after after.

Source code in sqlargon/cron/utils.py

def next_run_time(schedule: str, after: datetime) -> datetime:
    """Return the first run time of ``schedule`` strictly after ``after``."""
    return croniter(schedule, after).get_next(datetime)

ORM and types

Bases: DeclarativeBase

Source code in sqlargon/orm.py

class Base(DeclarativeBase):
    # required in order to access columns with server defaults
    # or SQL expression defaults, after a flush, without
    # triggering an expired load
    #
    # this allows us to load attributes with a server default after
    # an INSERT, for example
    #
    # https://docs.sqlalchemy.org/en/14/orm/extensions/asyncio.html#preventing-implicit-io-when-using-asyncsession
    metadata = MetaData(naming_convention=naming_convention)

    __mapper_args__: Any = {"eager_defaults": True}  # noqa: RUF012

    if TYPE_CHECKING:
        __tablename__: str
    else:

        @declared_attr.directive
        def __tablename__(cls) -> str:
            """
            By default, turn the model's camel-case class name
            into a snake-case table name. Override by providing
            an explicit `__tablename__` class property.
            """
            return camel_to_snake.sub("_", cls.__name__).lower()

__tablename__() -> str

By default, turn the model's camel-case class name into a snake-case table name. Override by providing an explicit __tablename__ class property.

Source code in sqlargon/orm.py

@declared_attr.directive
def __tablename__(cls) -> str:
    """
    By default, turn the model's camel-case class name
    into a snake-case table name. Override by providing
    an explicit `__tablename__` class property.
    """
    return camel_to_snake.sub("_", cls.__name__).lower()

GUID

Bases: TypeDecorator

Platform-independent UUID type.

Uses PostgreSQL's UUID type, otherwise uses CHAR(36), storing as stringified hex values with hyphens.

Source code in sqlargon/types/uuid.py

class GUID(TypeDecorator):
    """
    Platform-independent UUID type.

    Uses PostgreSQL's UUID type, otherwise uses
    CHAR(36), storing as stringified hex values with
    hyphens.
    """

    impl = UUID
    cache_ok = True

    def load_dialect_impl(self, dialect: Dialect) -> TypeEngine[Any]:
        if dialect.name == "postgresql":
            return dialect.type_descriptor(postgresql.UUID(as_uuid=True))
        return dialect.type_descriptor(CHAR(36))

    def process_bind_param(self, value: Any, dialect: Dialect) -> Any:
        if value is None:
            return None
        if dialect.name == "postgresql":
            if isinstance(value, str):
                value = uuid.UUID(value)
            return value
        if isinstance(value, uuid.UUID):
            return str(value)
        return str(uuid.UUID(value))

    def process_result_value(self, value: Any, dialect: Dialect) -> uuid.UUID | None:
        if value is None:
            return value
        if not isinstance(value, uuid.UUID):
            value = uuid.UUID(value)
        return value

Timestamp

Bases: TypeDecorator

TypeDecorator that ensures that timestamps have a timezone.

For SQLite, all timestamps are converted to UTC (since they are stored as naive timestamps without timezones) and recovered as UTC.

Source code in sqlargon/types/datetime.py

class Timestamp(TypeDecorator):
    """TypeDecorator that ensures that timestamps have a timezone.

    For SQLite, all timestamps are converted to UTC (since they are stored
    as naive timestamps without timezones) and recovered as UTC.
    """

    impl = sa.TIMESTAMP(timezone=True)
    cache_ok = True

    def load_dialect_impl(self, dialect: Dialect) -> TypeEngine[datetime]:
        if dialect.name == "postgresql":
            return dialect.type_descriptor(postgresql.TIMESTAMP(timezone=True))
        if dialect.name == "sqlite":
            return dialect.type_descriptor(sqlite.DATETIME())
        if dialect.name == "mysql":
            return dialect.type_descriptor(mysql.DATETIME(fsp=6))
        return dialect.type_descriptor(sa.TIMESTAMP(timezone=True))

    def process_bind_param(self, value: Any, dialect: Dialect) -> Any:
        if value is None:
            return None
        if value.tzinfo is None:
            msg = "Timestamps must have a timezone."
            raise ValueError(msg)
        if dialect.name in ("sqlite", "mysql"):
            return value.astimezone(timezone.utc)
        return value

    def process_result_value(self, value: Any, dialect: Dialect) -> Any:
        # retrieve timestamps in their native timezone (or UTC)
        if value is not None:
            return value.replace(tzinfo=timezone.utc)
        return None

now

Bases: FunctionElement

Platform-independent "now" generator.

Source code in sqlargon/types/datetime.py

class now(FunctionElement):
    """
    Platform-independent "now" generator.
    """

    type = Timestamp()
    name = "now"
    # see https://docs.sqlalchemy.org/en/14/core/compiler.html#enabling-caching-support-for-custom-constructs
    inherit_cache = True

Bases: TypeDecorator

JSON type that returns SQLAlchemy's dialect-specific JSON types, where possible. Uses generic JSON otherwise.

The "base" type is postgresql.JSONB to expose useful methods prior to SQL compilation

Source code in sqlargon/types/json.py

class JSON(TypeDecorator):
    """
    JSON type that returns SQLAlchemy's dialect-specific JSON types, where
    possible. Uses generic JSON otherwise.

    The "base" type is postgresql.JSONB to expose useful methods prior
    to SQL compilation
    """

    impl = postgresql.JSONB
    cache_ok = True

    def load_dialect_impl(self, dialect: Dialect) -> TypeEngine[Any]:
        if dialect.name == "postgresql":
            return dialect.type_descriptor(postgresql.JSONB(none_as_null=True))
        if dialect.name == "sqlite":
            return dialect.type_descriptor(sqlite.JSON(none_as_null=True))
        return dialect.type_descriptor(sa.JSON(none_as_null=True))

    class ComparatorFactory(sa.JSON.Comparator):
        def contains(self, other: Any, **_kw: Any) -> json_contains:
            return json_contains(self, other)

        def has_any_key(self, other: Any) -> json_has_any_key:
            return json_has_any_key(self, other)

        def has_all_keys(self, other: Any) -> json_has_all_keys:
            return json_has_all_keys(self, other)

        def json_value(self, other: Any) -> json_value:
            return json_value(self, other)

    comparator_factory = ComparatorFactory

Settings

Bases: BaseSettings

Source code in sqlargon/settings.py

class DatabaseSettings(BaseSettings):
    url: str = "postgresql+asyncpg://localhost:5432"
    echo: bool = False
    isolation_level: str | None = None
    json_serializer: ImportedType[Callable[[Any], str]] | None = json_dumps
    json_deserializer: ImportedType[Callable[[str], Any]] | None = json_loads
    connect_args: dict[str, Any] | None = None
    enable_tracker: bool = False
    poolclass: ImportedType[type[Pool]] | None = None
    pool_size: int | None = None
    max_overflow: int | None = None
    echo_pool: bool | None = None
    pool_recycle: int | None = None
    pool_pre_ping: bool | None = None
    pool_timeout: int | None = None
    pool_use_lifo: bool | None = None

    model_config = SettingsConfigDict(
        extra="allow",
        arbitrary_types_allowed=True,
        validate_default=True,
        env_prefix="DATABASE_",
    )

    def to_kwargs(self) -> dict[str, Any]:
        """Keyword arguments for the matching constructor."""
        return self.model_dump(exclude_none=True)

to_kwargs() -> dict[str, Any]

Keyword arguments for the matching constructor.

Source code in sqlargon/settings.py

def to_kwargs(self) -> dict[str, Any]:
    """Keyword arguments for the matching constructor."""
    return self.model_dump(exclude_none=True)

Bases: DatabaseSettings

Source code in sqlargon/settings.py

class DatabaseClusterSettings(DatabaseSettings):
    read_replicas: list[str] | None = None
    auto_route: bool = True
    replica_strategy: Literal["random", "round_robin"] = "random"

Typing helpers