Database

This is the layer every Tempest service uses to talk to PostgreSQL (production) or SQLite (development/tests) over SQLAlchemy 2.0 async. It exists so you never rewrite the same engine, the same per-request session, the same CRUD and the same pagination in every project.

Installation

The database core ships with tempest-fastapi-sdk. The async drivers come via extras — uv add "tempest-fastapi-sdk[postgres]" (PostgreSQL, pulls in asyncpg) or [sqlite] (SQLite in dev, pulls in aiosqlite).

There are four pieces, and you'll meet them one at a time:

Piece	Symbol	What for
Base model	BaseModel	The four canonical columns (id / is_active / created_at / updated_at) + serialization helpers.
Connection	AsyncDatabaseManager	Engine, pool, per-request session, health_check.
Repository	BaseRepository[Model]	Async CRUD, convention-based filters, bulk ops, pagination.
Migrations	AlembicHelper	Alembic bootstrap, autogenerate, CI drift gate.

Plus three opt-ins that show up when the domain asks for them: the mixins (SoftDeleteMixin, AuditMixin, MFAMixin), cursor pagination, and the SlowQueryLogger.

How to read this page

It's progressive. Start with the model, connect the database, stand up a repository, learn the filters, then pagination, migrations and observability. Every code block is a complete file — copy, paste, run. If you only want the API reference, jump to Reference ».

---

1. The base model

Every model in your service inherits from BaseModel. You get four columns without writing any:

# src/db/models/user.py
from sqlalchemy.orm import Mapped, mapped_column

from tempest_fastapi_sdk import BaseModel


class UserModel(BaseModel):
    """Users table."""

    name: Mapped[str] = mapped_column()
    email: Mapped[str] = mapped_column(unique=True)
    password_hash: Mapped[str] = mapped_column()

That already creates the user table with seven columns: your three (name, email, password_hash) plus the four inherited ones:

Column	Type	Default	Role
id	UUID (v4)	uuid4()	Primary key, portable across Postgres/SQLite/MySQL/MSSQL.
is_active	bool	True	Fast soft-delete flag.
created_at	datetime (tz-aware)	utcnow() on flush	Creation timestamp.
updated_at	datetime (tz-aware)	utcnow() on onupdate	Last-write timestamp.

Why is the table named user and not UserModel?

BaseModel derives __tablename__ from the class automatically: it strips the Model suffix and converts to snake_case. UserModel → user, OrderItemModel → order_item. You can always pin __tablename__ = "users" explicitly — the explicit declaration wins over the automatic one.

Pinning the name is not just taste

USER is a reserved word in standard SQL. SQLAlchemy always quotes the identifier, so your application works — but a SELECT * FROM user typed by hand in psql returns the database user, not your table, and raises nothing. The plural (users) sidesteps that, and it is the convention the SDK itself assumes for the token tables (user_tokens, user_refresh_tokens).

Centralizing table names

A table name almost never appears just once. It is in __tablename__ and it comes back as a string in every ForeignKey pointing at it:

# src/db/models/user_token.py
from uuid import UUID

from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column

from tempest_fastapi_sdk import BaseModel


class UserTokenModel(BaseModel):
    __tablename__ = "user_tokens"

    user_id: Mapped[UUID] = mapped_column(
        ForeignKey("users.id", ondelete="CASCADE"),
    )

Renaming users now depends on you remembering every place the string shows up. And missing a FK does not blow up right away: SQLAlchemy only resolves the target when it configures the mappers, so the error lands at application startup — or, worse, in a migration pointing at a table that no longer exists.

The fix is a module that holds nothing but names:

# src/db/configs/names.py
"""Table names for this project. Single source of truth."""

USER_TABLE_NAME = "users"
USER_TOKEN_TABLE_NAME = "user_tokens"
USER_REFRESH_TOKEN_TABLE_NAME = "user_refresh_tokens"
ORDER_TABLE_NAME = "orders"
ORDER_ITEM_TABLE_NAME = "order_items"

The _TABLE_NAME suffix keeps the constant self-explanatory at the call site, far from this file: ForeignKey(f"{USER_TABLE_NAME}.id") says what that string is on its own. The prefix follows the model, in the singular (UserTokenModel → USER_TOKEN_TABLE_NAME), even when the value is plural.

Every model imports from there, on both sides of the relationship:

# src/db/models/user.py
from sqlalchemy.orm import Mapped, mapped_column

from tempest_fastapi_sdk import BaseModel

from src.db.configs.names import USER_TABLE_NAME


class UserModel(BaseModel):
    __tablename__ = USER_TABLE_NAME

    email: Mapped[str] = mapped_column(unique=True)

# src/db/models/user_token.py
from uuid import UUID

from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column

from tempest_fastapi_sdk import BaseModel

from src.db.configs.names import USER_TABLE_NAME, USER_TOKEN_TABLE_NAME


class UserTokenModel(BaseModel):
    __tablename__ = USER_TOKEN_TABLE_NAME

    user_id: Mapped[UUID] = mapped_column(
        ForeignKey(f"{USER_TABLE_NAME}.id", ondelete="CASCADE"),
        index=True,
    )

The payoff shows up when you change something. Renaming a table becomes one line in names.py. Your editor's "find usages" locates everyone who depends on it, because it is now a symbol instead of a loose string. And no ForeignKey can point at a table that no longer exists without the import breaking first.

Why db/configs/ and not core/constants.py?

A table name is a database detail, and db/models/ is what consumes it. Keeping it under db/ keeps the dependency inside its own layer, and the module ends up importing nothing from the project — it is only strings. That is what guarantees it never joins an import cycle: models imports configs, and configs imports nobody.

It applies to the SDK's tables too

The abstract models the SDK ships (BaseUserModel, BaseUserTokenModel, BaseUserRefreshTokenModel, BaseWebPushSubscriptionModel, BaseOutboxModel) deliberately leave __tablename__ and the FK for the concrete project to declare — precisely so both can come out of your names.py, under your naming convention.

Explicit __tablename__ with Pyright

BaseModel declares __tablename__ as a @declared_attr.directive, SQLAlchemy 2.0's mechanism for deriving the name from the class. mypy understands a subclass overriding that with a string and does not complain — it is the checker tempest type runs, so the default gate stays clean.

Pyright is stricter: it reads the inherited attribute as a mutable variable with an invariant type and flags the assignment.

reportIncompatibleVariableOverride in Pyright/Pylance

Type "Literal['users']" is not assignable to declared type
"_declared_directive[str]" (reportAssignmentType)

This is not a defect in your model: the assignment works at runtime, and it is the form used throughout these docs. It is Pyright being stricter than mypy about overriding a descriptor.

If your editor runs Pyright and you want the file clean, declare the name through the same mechanism the base class uses:

from sqlalchemy.orm import declared_attr
from tempest_fastapi_sdk import BaseModel

from src.db.configs.names import USER_TABLE_NAME


class UserModel(BaseModel):
    @declared_attr.directive
    def __tablename__(cls) -> str:  # noqa: N805
        """Pin the table name."""
        return USER_TABLE_NAME

More verbose, and equivalent at runtime. Pick by the project's checker: with mypy (or no static checking in the editor), prefer __tablename__ = USER_TABLE_NAME, which reads more directly.

Constraint naming convention

BaseModel.metadata ships configured with NAMING_CONVENTION. That makes every PK/FK/index/unique/check get a deterministic name — ix_user_email, uq_user_email, fk_order_user_id_user — identical on every machine and every engine.

The real win is in the migrations

Without deterministic names, alembic revision --autogenerate invents random identifiers and each developer generates a different diff for the same schema. With the convention, autogenerate only emits real schema diffs — no name churn.

Composite unique — the name carries every column

A uniqueness rule spanning more than one column ("the same title cannot repeat in the same year") is a UniqueConstraint with several columns in __table_args__. You do not need to name it — the convention does:

# src/db/models/book.py
from sqlalchemy import Index, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column

from tempest_fastapi_sdk import BaseModel


class BookModel(BaseModel):
    """Catalog books."""

    __tablename__ = "books"
    __table_args__ = (
        UniqueConstraint("isbn"),
        UniqueConstraint("title", "release_year"),
        UniqueConstraint("title", "author"),
        Index(None, "author", "title"),
    )

    isbn: Mapped[str] = mapped_column()
    title: Mapped[str] = mapped_column()
    release_year: Mapped[int] = mapped_column()
    author: Mapped[str] = mapped_column()


names: list[str] = sorted(
    str(item.name)
    for item in [*BookModel.__table__.constraints, *BookModel.__table__.indexes]
)
print(names)

Output:

['ix_books_author_books_title', 'pk_books', 'uq_books_isbn', 'uq_books_title_author', 'uq_books_title_release_year']

Every composite constraint gets all its columns in the name, joined with _: uq_books_title_release_year and uq_books_title_author are two different rules with two different names. The templates use SQLAlchemy's column_0_N_* tokens, which for a single column render exactly what the column_0_* tokens rendered — so uq_books_isbn, ix_user_email and fk_order_user_id_user stay the same.

A distinct name is what makes the constraint Postgres returns through parse_integrity_error useful: a name → error code dict now points at exactly one rule.

from typing import Any

from sqlalchemy.exc import IntegrityError

from tempest_fastapi_sdk import parse_integrity_error

CONFLICT_CODES: dict[str, str] = {
    "uq_books_isbn": "ISBN_TAKEN",
    "uq_books_title_release_year": "TITLE_TAKEN_THIS_YEAR",
    "uq_books_title_author": "AUTHOR_ALREADY_HAS_TITLE",
}


def to_conflict(error: IntegrityError) -> dict[str, Any]:
    """Translate the violation into the code the client branches on."""
    failure = parse_integrity_error(error)
    return {
        "code": CONFLICT_CODES.get(failure.constraint or "", "CONFLICT"),
        "fields": list(failure.columns),
    }

Before this fix, the two uniques shared one name

Up to the previous version, the uq, ix and fk templates read only the first column. UniqueConstraint("title") and UniqueConstraint("title", "release_year") both became uq_books_title. PostgreSQL refuses the CREATE TABLE with relation "uq_books_title" already exists; SQLite accepts both, which is why the defect slipped past the test database. A database created under the old convention still holds the old names for its composite constraints — the step-by-step rename is in Migrate composite constraints from the old convention.

Names past 63 characters

PostgreSQL accepts identifiers of up to 63 characters. When a convention name is longer, SQLAlchemy shortens it while emitting the DDL: the first 55 characters, _, and the last 4 hex digits of the full name's MD5 — uq_subscription_billing_events_customer_identifier_bill_f06e for a 90-character unique. It is the same on every run, and autogenerate reports no difference afterwards (measured against PostgreSQL 16). SQLite has no limit and keeps the full name.

The one ambiguity left

Joining with _ leaves one case the convention cannot tell apart: columns ("a_b", "c") and ("a", "b_c") on the same table render the same name. If your schema has such a pair, give one of them an explicit name= — an explicit name wins over the convention.

Helpers you get for free

Every BaseModel instance gets:

# Serialize to a dict (handy in logs/tests)

from typing import Any

from src.db.models import UserModel
from src.schemas import UserUpdateSchema

payload = UserUpdateSchema(name="Ana Paula")
user = UserModel(name="Ana", email="ana@example.com")


data: dict[str, Any] = user.to_dict(exclude=["password_hash"])

# Assign many fields at once, with a whitelist against mass-assignment
user.update_from_dict(
    payload.model_dump(exclude_unset=True),
    allowed_fields={"name", "email"},   # id/role never get written
)

__eq__ and __hash__ compare by (type, id), so the same row loaded across different sessions compares equal — handy in tests and sets. Rows not yet persisted (id is None) fall back to Python identity.

Always use allowed_fields on external payloads

update_from_dict without allowed_fields accepts any mapped column. For PATCH bodies coming from the client, pass the whitelist — it's the defense against mass-assignment on sensitive columns (id, role, is_active).

The table name — get_table_name()

A dict key, a panel lookup, a test assertion: when you need the table name, ask the model, not the dunder:

from tempest_fastapi_sdk import AdminModel, AdminSite, BaseModel


class OrderItemModel(BaseModel):
    """An order line."""


print(OrderItemModel.get_table_name())  # order_item

site = AdminSite(title="Panel")
site.register(AdminModel(model=OrderItemModel))
admin = site.get(OrderItemModel)

get_table_name() reads the table the mapper built, not __tablename__: it answers the name derived from the class, an explicit __tablename__, and also a model declared with __table__ = Table(...), which has no __tablename__ at all. Under single-table inheritance a subclass answers its parent's table, where its rows live. The return type is str for the type checker, and AdminSite.get / require / unregister take the class itself in place of the string.

Recap: inherit BaseModel, declare only your domain columns, and the SDK delivers id/timestamps/soft-delete, deterministic constraint names (a composite unique carries every column in its name) and serialization helpers.

---

2. Connecting to the database

AsyncDatabaseManager is instantiated once per application and owns the engine, the pool and the session factory. Put it in the infrastructure dependencies, not inside app.py:

# src/api/dependencies/resources.py
from tempest_fastapi_sdk import AsyncDatabaseManager

from src.core.settings import settings

db = AsyncDatabaseManager(
    settings.DATABASE_URL,
    echo=settings.DEBUG,        # echo SQL to stdout in dev
    pool_size=10,               # ignored for SQLite
    max_overflow=20,
    pool_recycle=3600,
)

It detects the backend from the URL (make_url), so SQLite gets check_same_thread=False automatically and the pool parameters are ignored — no substring tricks.

One session per request

Use session_dependency as the FastAPI dependency. It hands out one session per request and does not commit on success — committing is the repository/service layer's responsibility:

# src/api/dependencies/resources.py (continued)

from typing import Annotated

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from src.api.dependencies.resources import db


SessionDep = Annotated[AsyncSession, Depends(db.session_dependency)]

# src/api/routers/user.py
from uuid import UUID

from fastapi import APIRouter

from src.api.dependencies.resources import SessionDep
from src.db.repositories import UserRepository
from src.schemas import UserResponse

router = APIRouter(prefix="/api/users", tags=["users"])


@router.get("/{user_id}", response_model=UserResponse)
async def get_user(user_id: UUID, session: SessionDep) -> UserResponse:
    """Fetch a single user by id."""
    repository = UserRepository(session)
    return repository.map_to_response(await repository.get_by_id(user_id))

A manager built on demand — session_dependency_for

Depends(db.session_dependency) needs db to exist when the module is imported. Some services prefer to build the manager on first use: to read settings after the boot guard, and so tests can swap DATABASE_URL before any engine exists. In that case, hand the factory to session_dependency_for:

# src/api/dependencies/resources.py
import os
from functools import lru_cache
from typing import Annotated

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from tempest_fastapi_sdk import AsyncDatabaseManager, session_dependency_for


@lru_cache
def get_db() -> AsyncDatabaseManager:
    """Build the manager on first use, never at import."""
    return AsyncDatabaseManager(os.environ["DATABASE_URL"])


get_session = session_dependency_for(get_db)
SessionDep = Annotated[AsyncSession, Depends(get_session)]

The returned dependency has the same semantics as session_dependency: one session per request, no commit on success, and the session closed when the request ends — also when the endpoint raises. get_db is called on each request, never at import; with @lru_cache, every request gets the same manager.

Build the dependency once, at module level

FastAPI caches a dependency per request by the callable's identity. Two calls to session_dependency_for(get_db) return two different functions — and, used in the same request, two sessions. Keep the result under a name (get_session) and reuse it.

Lifecycle in the lifespan

Open and close the engine alongside the application:

# src/api/app.py
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from src.api.dependencies.resources import db


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    """Open the database on startup, dispose it on shutdown."""
    await db.connect()
    yield
    await db.disconnect()

Health check

health_check() runs a SELECT 1 and swallows any exception, returning only True/False — perfect for /health:

from fastapi import APIRouter

from src.api.dependencies.resources import db

router = APIRouter()


@router.get("/health")
async def health() -> dict[str, object]:
    """Liveness + database probe."""
    return {
        "status": "ok",
        "database": await db.health_check(),
        "url": db.db_url_safe,   # credentials masked
    }

Other ways to get a session

• db.get_session_context() / db.transaction() — the same context manager under two names: it commits on exit and rolls back on error. Use it in scripts and background tasks.
• db.get_session() — a raw session; you close it.
• db.create_tables() / db.drop_tables() — tests and local dev only; in production the schema is Alembic's. Without an argument, only the models of the SDK's BaseModel — see Your own DeclarativeBase.

get_session_context() commits — never in a request dependency

The code after a dependency's yield runs after the response went out. Measured on FastAPI 0.141.1, with an ASGI middleware recording the messages: http.response.start and http.response.body are sent before the COMMIT. A dependency wrapping get_session_context() answers 204 to a DELETE whose commit has not happened yet — the client's next GET may still see the row — and, if that commit fails, the status the client already received can no longer change. For requests use session_dependency (or session_dependency_for) and commit in the service/repository layer.

create_tables() is a silent no-op on an existing table

create_all is CREATE TABLE IF NOT EXISTS: against a table that already exists it adds no column at all, does not fail and does not warn. Using it at boot and stamping head afterwards leaves an old schema with Alembic declaring itself up to date — the failure surfaces weeks later, far from the cause. The correct bootstrap, for all three states a database can be in, is in Migrations ».

Never log the raw URL, always db_url_safe

The raw URL carries user and password. db_url_safe replaces only the password: postgresql+asyncpg://user:pass@host:5432/db comes out as postgresql+asyncpg://user:***@host:5432/db — user, host, port, database and query string stay visible, so treat the result as internal diagnostics if the database user is itself sensitive. The raw URL lives on a private attribute precisely so it doesn't leak through repr() or accidental logging.

Your own DeclarativeBase

create_tables() without an argument creates only what inherits the SDK's BaseModel. A service with its own DeclarativeBase — because it addresses rows by a natural key, say, and BaseModel imposes a UUID and timestamps — gets no table at all: the first query fails with no such table (SQLite). Pass your base's metadata:

import asyncio

from sqlalchemy import Integer, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from tempest_fastapi_sdk import AsyncDatabaseManager


class Base(DeclarativeBase):
    """A base of your own: natural key, no BaseModel UUID or timestamps."""


class ObjectModel(Base):
    __tablename__ = "objects"

    bucket: Mapped[str] = mapped_column(String(63), primary_key=True)
    key: Mapped[str] = mapped_column(String(1024), primary_key=True)
    size: Mapped[int] = mapped_column(Integer, nullable=False)


async def main() -> None:
    db = AsyncDatabaseManager("sqlite+aiosqlite:///:memory:")
    await db.create_tables(Base.metadata)
    async with db.transaction() as session:
        session.add(ObjectModel(bucket="photos", key="cat.jpg", size=2048))
    async with db.transaction() as session:
        rows = (await session.scalars(select(ObjectModel))).all()
        print([(row.bucket, row.key, row.size) for row in rows])
    await db.disconnect()


asyncio.run(main())

Output: [('photos', 'cat.jpg', 2048)]. drop_tables(Base.metadata) goes the other way. Both remain tools for tests and local dev.

Outside a request

Not every consumer has a request to hang Depends off. An agent tool, a TaskIQ task, a FastStream consumer and a maintenance script all run outside the HTTP cycle — and they all use get_session_context(), which opens the session, commits on exit and rolls back on error:

# src/tasks/cleanup.py
from src.api.dependencies.resources import db
from src.db.repositories import UserRepository


async def count_inactive_users() -> int:
    """Count the users that were deactivated."""
    async with db.get_session_context() as session:
        repository = UserRepository(session)
        return len(await repository.list(filters={"is_active": False}))

The rule is to open as late as possible and close as early as possible: a process holding the session while it waits on something else — a model generating tokens, an external API replying — occupies a pool connection without using it.

An agent tool is the trickiest case

An agent run spans several steps and can take minutes. AI agents (database) » shows why the session is opened inside each tool, what the automatic commit means for a tool that writes, and why two AsyncDatabaseManager instances in one process are two pools.

SQLite with a worker: WAL and the busy timeout

The day the application grows a worker, the development SQLite has two processes writing one file. In the default rollback journal (delete) a reader and a writer exclude each other, so the second one dies:

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) database is locked
[SQL: INSERT INTO budget_drafts (...) VALUES (?, ?, ...)]

Measured across two processes — one holding a read transaction open while the other inserts:

journal_mode	What happens to the writer
delete	waits out the whole busy_timeout, then database is locked
wal	commits immediately

So AsyncDatabaseManager opens every SQLite file in WAL, with a 30-second busy_timeout. You don't have to ask:

from tempest_fastapi_sdk import AsyncDatabaseManager

db = AsyncDatabaseManager("sqlite+aiosqlite:///./app.db")
# journal_mode = wal, busy_timeout = 30000 (ms)

Both are tunable, and ignored on every other backend:

from tempest_fastapi_sdk import AsyncDatabaseManager

db = AsyncDatabaseManager(
    "sqlite+aiosqlite:///./app.db",
    sqlite_wal=False,           # filesystems without working shared memory
    sqlite_busy_timeout=5.0,    # seconds
)

From the environment, through DatabaseSettings: DATABASE_SQLITE_WAL and DATABASE_SQLITE_BUSY_TIMEOUT.

WAL is a property of the file

Turning it on once is enough: the mode outlives the process and every later connection, from any process, opens the file already in WAL. On a :memory: database the pragma is inert — SQLite answers memory and carries on.

:memory: gets a real connection per session (v0.252.0)

sqlite+aiosqlite:///:memory: makes SQLAlchemy pick StaticPool: one DBAPI connection shared by every session. Together with the explicit BEGIN the manager has emitted since v0.200.0 — needed so RELEASE SAVEPOINT stops committing on SQLite — that broke any pair of overlapping sessions with cannot start a transaction within a transaction. It hits the test pattern this SDK recommends, and it hits an endpoint that answers and finishes its work in a BackgroundTasks.

The manager now rewrites the URL to a shared-cache in-memory database (file:<name>?mode=memory&cache=shared&uri=true), with a normal pool, and holds one connection open for as long as the manager lives — a shared-cache in-memory database is destroyed when the last connection closes. Each manager gets its own name, so two managers stay isolated.

Measured on both properties: an overlapping session works and a nested block that exits cleanly is still not durable after an outer rollback. Dropping the BEGIN — the obvious way out — buys the first and loses the second.

Need the old topology? Pass poolclass=StaticPool explicitly: a pool the caller names is never overridden.

What waiting does not fix

WAL admits one writer at a time; the others wait out the busy_timeout. What no timeout fixes is a transaction that reads first and writes later: promoting the lock fails at once if another connection wrote in between, and busy_timeout does not apply because there is nothing to wait for. For long work: claim the row, do the work with no session open, and only then persist.

A refused COMMIT no longer leaves the pooled connection dirty

Without WAL, a reader open on another connection makes COMMIT fail with database is locked. The transaction stays open on the driver, and SQLAlchemy does not close it: the connection went back to the pool inside its BEGIN, and its next use failed with cannot start a transaction within a transaction. enable_sqlite_savepoints — which the manager and create_test_engine apply — rolls back on check-in whenever the driver is still in a transaction. The refused COMMIT still raises OperationalError; what changes is that the next connection comes out clean, and the refused write does not become durable. An engine you build by hand must call enable_sqlite_savepoints to get the same reset.

SQLite with foreign keys: PRAGMA foreign_keys

SQLite parses the REFERENCES clause but only checks the foreign key on a connection that ran PRAGMA foreign_keys=ON — off is its default, and the pragma lives on the connection, not on the file. Left off, three things pass on SQLite that fail (or do something else) on PostgreSQL:

Operation	FK off	FK on
insert a child pointing at a missing parent	accepted	IntegrityError
delete the parent with ON DELETE CASCADE	child left orphaned	child deleted
add_all([child, parent]) with no relationship() ordering the flush	accepted	IntegrityError

Measured on SQLAlchemy 2.0.52 and 2.1.3, aiosqlite 0.22.1, SQLite 3.47.1. So AsyncDatabaseManager turns the pragma on for every SQLite engine — file and :memory: alike —, the same way it does WAL:

import asyncio

from sqlalchemy import ForeignKey, func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from tempest_fastapi_sdk import AsyncDatabaseManager


class Base(DeclarativeBase):
    """Declarative base for this example only."""


class Org(Base):
    """Organization: the parent row."""

    __tablename__ = "orgs"

    id: Mapped[int] = mapped_column(primary_key=True)


class Member(Base):
    """Member: deleted along with its organization."""

    __tablename__ = "members"

    id: Mapped[int] = mapped_column(primary_key=True)
    org_id: Mapped[int] = mapped_column(ForeignKey("orgs.id", ondelete="CASCADE"))


async def main() -> None:
    """Show the orphan refused and the cascade applied."""
    db: AsyncDatabaseManager = AsyncDatabaseManager("sqlite+aiosqlite:///:memory:")
    await db.create_tables(Base.metadata)

    async with await db.get_session() as session:
        session.add(Member(id=1, org_id=999))
        try:
            await session.commit()
        except IntegrityError as exc:
            await session.rollback()
            print("orphan:", exc.orig)

        session.add(Org(id=1))
        await session.flush()
        session.add(Member(id=2, org_id=1))
        await session.commit()

        org: Org | None = await session.get(Org, 1)
        await session.delete(org)
        await session.commit()
        left: int = (
            await session.execute(select(func.count()).select_from(Member))
        ).scalar_one()
        print("members after deleting the organization:", left)

    await db.disconnect()


asyncio.run(main())

orphan: FOREIGN KEY constraint failed
members after deleting the organization: 0

Out-of-order INSERT

With no relationship() between the two models, SQLAlchemy's unit of work does not know the member depends on the organization and sends the INSERTs in add_all order. Declare the relationship(), or flush() after the parent, as the example does.

To turn it off — a legacy database with orphan rows you cannot clean up yet:

from tempest_fastapi_sdk import AsyncDatabaseManager

db: AsyncDatabaseManager = AsyncDatabaseManager(
    "sqlite+aiosqlite:///./legacy.db",
    sqlite_foreign_keys=False,
)

From the environment, through DatabaseSettings: DATABASE_SQLITE_FOREIGN_KEYS=false. In tests, create_test_engine(foreign_keys=False) — see Testing.

Why the pragma runs on connect

SQLite ignores PRAGMA foreign_keys inside a transaction, without an error. Since the manager emits an explicit BEGIN (that is what stops RELEASE SAVEPOINT from committing), a pragma issued through the session would never take effect: measured, it keeps answering 0. The SDK registers a listener on the engine's connect event that runs the pragma through the driver cursor before any transaction. An engine you build by hand gets the same behaviour from enable_sqlite_foreign_keys(engine).

Migrations do not run with FK on

Alembic's batch mode rebuilds the table on SQLite, and with FK on the DROP TABLE of the old copy fires ON DELETE CASCADE: measured, the children go from 3 rows to 0, with no error. The SDK's migration engines keep FK off, and the generated env.py refuses a handed-over connection that has it on — see Migrations.

Recap: one AsyncDatabaseManager per app, in resources.py; session_dependency injects the per-request session (session_dependency_for when the manager is built on demand), while get_session_context()/ transaction(), which commit, stay outside requests; create_tables(metadata) accepts your own DeclarativeBase; connect/disconnect in the lifespan; health_check + db_url_safe on /health; on SQLite, WAL and the busy timeout ship on so web and worker can share the file, and foreign keys are checked the way PostgreSQL checks them.

---

3. The repository

BaseRepository[Model] is the heart of the layer. It encapsulates async CRUD, filters, bulk ops and pagination. There are two ways to use it.

Direct mode — plain CRUD

When you have no custom query, instantiate directly:

import asyncio
from uuid import UUID

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from tempest_fastapi_sdk import BaseRepository

from src.db.models import UserModel

# In a service the session comes from `db.get_session_context()`; here, SQLite.
session = AsyncSession(create_async_engine("sqlite+aiosqlite:///:memory:"))

user_id = UUID("2b1d0c2e-7f3a-4c56-9d18-2f9a4c5b6d70")


repository = BaseRepository(session, model=UserModel)


async def main() -> None:
    """Run this example."""
    user = await repository.get_by_id(user_id)


asyncio.run(main())

Subclass mode — when you have your own queries

Subclass it to add domain queries and the mappers that translate ORM ↔ DTO. There are three: map_to_model (default self.model(**data)), map_to_response and map_to_schema — the last two raise NotImplementedError until you override them, so implement the ones your code calls (the example implements map_to_response and map_to_model). The constructor is the contract — you forward model to super().__init__, there are no magic class attributes:

# src/db/repositories/user.py
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from tempest_fastapi_sdk import BaseRepository

from src.db.models import UserModel
from src.schemas import UserResponse


class UserRepository(BaseRepository[UserModel]):
    """Data access for the user domain."""

    def __init__(self, session: AsyncSession) -> None:
        """Bind the repository to a session and the user model.

        Args:
            session (AsyncSession): The async database session.
        """
        super().__init__(
            session,
            model=UserModel,
            not_found_message="User not found",
            create_conflict_message="Email already registered",
        )

    def map_to_response(self, instance: UserModel) -> UserResponse:
        """Map an ORM row to its API response schema.

        Args:
            instance (UserModel): The persisted user row.

        Returns:
            UserResponse: The serializable response DTO.
        """
        return UserResponse.model_validate(instance)

    def map_to_model(self, data: dict[str, Any]) -> UserModel:
        """Build an ORM instance from a plain payload.

        Args:
            data (dict[str, Any]): Column-value pairs.

        Returns:
            UserModel: The unpersisted instance.
        """
        return UserModel(**data)

Per-repository error messages

The kwargs not_found_message, create_conflict_message, update_conflict_message, bulk_create_conflict_message and bulk_update_conflict_message customize the exception text. Without them, the SDK generates messages from Model.__name__ — the whole class name, suffix included: "UserModel not found", "Conflict creating UserModel".

Per-repository exception classes

Every *_message has a matching *_exception. A message alone gives the client nothing to branch on: the default ConflictException answers code = "CONFLICT", so a duplicate coin pack name is indistinguishable from any other 409 — and error_responses() cannot document it. Passing a domain subclass (one that declares its own code in the class body) makes the 409 identifiable without the repository knowing anything about the domain:

from tempest_fastapi_sdk import BaseRepository, ConflictException

from src.core.exceptions import CoinPackNotFoundException
from src.db.models import CoinPackModel


class CoinPackAlreadyExistsException(ConflictException):
    """Raised when a coin pack name is already taken."""

    code: str = "COIN_PACK_ALREADY_EXISTS"


class CoinPackRepository(BaseRepository[CoinPackModel]):
    """Data access for coin packs."""

    def __init__(self, session: AsyncSession) -> None:
        """Initialize the repository.

        Args:
            session (AsyncSession): The async database session.
        """
        super().__init__(
            session,
            model=CoinPackModel,
            not_found_exception=CoinPackNotFoundException,
            create_conflict_exception=CoinPackAlreadyExistsException,
        )

Resolution runs most-specific-first — create_conflict_exception if given, else conflict_exception, else ConflictException — so a single kwarg (conflict_exception=) covers every write, or each write can carry its own:

Kwarg	Covers
create_conflict_exception	add, save_with_outbox, add_audited
update_conflict_exception	update, update_audited
bulk_create_conflict_exception	add_all, bulk_create_values, bulk_upsert
bulk_update_conflict_exception	update_many, bulk_update, update_returning
conflict_exception	fallback for all four

The class is instantiated as cls(message=...), the same contract not_found_exception already has, so it must accept a message keyword. Declaring code in the class body and taking message optionally satisfies both. Every kwarg is optional: omit them and the behavior is exactly what it was (the generic ConflictException). Available from 0.169.0.

Which constraint refused — parse_integrity_error

Swapping ConflictException for a class of your own answers there was a conflict. Answering on which field means reading what the server said — and what it says has the driver's shape, not your application's.

from typing import Any

from sqlalchemy.exc import IntegrityError

from tempest_fastapi_sdk import parse_integrity_error


def to_response(error: IntegrityError) -> dict[str, Any]:
    """Translate an integrity error into the envelope clients parse."""
    failure = parse_integrity_error(error)
    return {
        "code": failure.kind.value,
        "field": failure.column,
        "constraint": failure.constraint,
    }

IntegrityViolation classifies into unique, foreign_key, not_null, check and unknown. IntegrityFailure carries what the server reported: constraint, table, columns (in the order listed) and column, the shortcut for the single-column case.

An empty field means 'this server did not say'

Never 'there is none'. The two dialects speak differently, and the differences are not cosmetic:

	Postgres	SQLite
unique	names constraint, table and columns	names table.column; no constraint
not null	names column and table	names table.column
check	names the constraint and the table	returns the expression when the constraint is unnamed
foreign key	names constraint, table and column	says only FOREIGN KEY constraint failed

A measured absence worth repeating: a SQLite foreign key carries nothing beyond the kind.

On Postgres, the driver's fields come before the text

asyncpg exposes the server's diagnostics as attributes of the exception — sqlstate, detail, constraint_name, table_name, column_name — and parse_integrity_error reads those first, falling back to the sentence only for what they leave empty. The text is not a stable carrier. Measured against Postgres 16 with asyncpg 0.31.0:

SQLAlchemy	str(error.orig)	where the driver exception is
2.0.52 (the SDK's floor)	first sentence plus the DETAIL: line	error.orig.__cause__
2.1.1	first sentence only	error.orig.orig (and __cause__)

Reading only the text, a unique under 2.1.1 came back with columns=(). The fields are read by attribute name: asyncpg is not imported, and an installation without [postgres] does not need it. The table of a unique comes from table_name; a bare message, without the driver exception behind it, still leaves it None, because the sentence does not say it and splitting the constraint name on a convention would be a guess. The sqlstate (23505, 23503, 23502, 23514) also classifies a violation whose sentence the server translated: with lc_messages=de_DE.utf8, on both SQLAlchemy versions, a unique came back unique with constraint and table, and with columns=(), because the DETAIL is translated too.

Reads error.orig, not str(error)

str() on a SQLAlchemy IntegrityError appends [SQL: <statement>], and a statement contains user data. A value carrying the text UNIQUE constraint failed: x.y would be parsed as a column name. The function reads the DBAPI exception, whose message is the server's sentence alone.

Never raises

An unknown message comes back as IntegrityViolation.UNKNOWN with the text in failure.message. A parser for error prose that raises turns a handled 409 into an unhandled 500, which is worse than the generic conflict it was added to improve on.

The CRUD you get

Recall the project's collection convention: single-record lookups raise 404; collection lookups return [].

import asyncio
from uuid import UUID, uuid4

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from tempest_fastapi_sdk import BaseRepository

from src.db.models import UserModel

# In a service the session comes from `db.get_session_context()`; here, SQLite.
session = AsyncSession(create_async_engine("sqlite+aiosqlite:///:memory:"))

id1, id2, id3 = uuid4(), uuid4(), uuid4()
repository = BaseRepository(session, model=UserModel)
user_id = UUID("2b1d0c2e-7f3a-4c56-9d18-2f9a4c5b6d70")
user_or_id = user_id


async def main() -> None:
    """Run this example."""
    # Read — single record (404 when missing)
    user = await repository.get_by_id(user_id)
    user = await repository.get({"email": "a@b.com"})

    # Read — may not exist (None, no 404)
    user = await repository.get_or_none({"email": "a@b.com"})
    first = await repository.first({"is_active": True})

    # Read — collection (always [], never 404)
    users = await repository.list({"is_active": True})

    # Existence / count
    exists = await repository.exists({"email": "a@b.com"})
    total = await repository.count({"is_active": True})

    # "Is this value already used by ANOTHER row?" — uniqueness check on update
    taken = await repository.exists_excluding({"email": "a@b.com"}, exclude_id=user.id)

    # id-or-instance → instance (no scattered if isinstance in services)
    user = await repository.resolve(user_or_id)

    # Write
    created = await repository.add(
        UserModel(name="Ana", email="ana@x.com", password_hash="...")
    )
    updated = await repository.update(user)  # commits mutations on an attached instance

    # Removal
    await repository.delete(user_id)  # hard delete (404 if missing)
    await repository.delete_many({"is_active": False})  # returns count
    await repository.delete_batch([id1, id2, id3])  # by PK, returns count

    # Soft-delete via the is_active flag (no SoftDeleteMixin needed)
    await repository.soft_delete(user_id)  # is_active = False
    await repository.restore(user_id)  # is_active = True


asyncio.run(main())

update expects an attached instance

The typical flow is: get_by_id → mutate with update_from_dict → repository.update(instance). Don't build a detached model and pass it to update — it persists mutations on something already loaded in the session.

resolve and exists_excluding — two helpers you'll reach for constantly

resolve(id_or_instance) settles the old dilemma: your method takes UUID | UserModel and you don't want to write if isinstance(x, UUID): ... else: ... in every service. resolve does it for you — pass a UUID and it fetches (404 if missing); pass an instance and it returns the same one. One line:

user_model = await self.repository.resolve(user)  # user is UUID OR UserModel

exists_excluding(filters, exclude_id=...) answers "is this email/phone/username already someone else's?" — exactly what you need when updating a unique field. Plain exists would say True even for the row itself; exists_excluding ignores the id you pass:

if await self.repository.exists_excluding(
    {"phone": new_phone}, exclude_id=user.id
):
    raise UserWithPhoneExistsException(phone=new_phone)

Pass exclude_id=None on create (when there's no row to exclude yet) — then it behaves just like exists.

Recap: instantiate directly for plain CRUD, subclass for queries + mappers. 404 only on single lookups; collections return []. soft_delete flips the is_active flag; SoftDeleteMixin (section 6) adds a deleted_at timestamp when you need temporal auditing.

Eager-loading relationships with with_

Touching a relationship (user.orders) after the async session closed raises the dreaded MissingGreenlet — SQLAlchemy would attempt a lazy query in a context that can no longer await I/O. The fix is to load the relationship up front, in the same query. Every read method (get, get_or_none, get_by_id, first, list) accepts with_=:

import asyncio
from uuid import UUID

from tempest_fastapi_sdk import BaseRepository

from db_setup import db
from src.db.models import UserModel


async def main() -> None:
    """Run this example."""
    user_id = UUID("2b1d0c2e-7f3a-4c56-9d18-2f9a4c5b6d70")
    async with db.get_session_context() as session:
        repository = BaseRepository(session, model=UserModel)
        # Load the user and its orders in a single round trip
        user = await repository.get_by_id(user_id, with_=["orders"])
        for order in user.orders:      # no lazy load, no MissingGreenlet
            print(order.total)

        # Several relationships + nested (dotted)
        user = await repository.get_by_id(
            user_id,
            with_=["profile", "orders.items"],   # orders → and each order's items
        )

        # Works on collections too
        users = await repository.list({"is_active": True}, with_=["orders"])


asyncio.run(main())

Each path uses selectinload: N related rows cost one extra query per level (a SELECT ... IN (...)), not N — no JOIN row multiplication, and it works for both collections and scalars.

A wrong name fails loudly

A with_ segment that is not a relationship on the model reached at that hop raises ValueError immediately — not a silent runtime error. with_=["orders.ghost"] → ValueError: Order has no relationship 'ghost'.

Lifecycle signals

When you want to react to a write — bust a cache, enqueue an event, sync a search index, fire a domain event — without scattering callbacks across every service, register a signal. The repository emits four moments on the unit-of-work path:

from tempest_fastapi_sdk import RepositorySignal, on_signal
from tempest_fastapi_sdk.cache import AsyncRedisManager
from tempest_fastapi_sdk.db import connect, disconnect

from src.core.settings import settings
from src.db.models import UserModel
from src.services.search import SearchIndex

cache = AsyncRedisManager(settings.REDIS_URL, decode_responses=True)
search_index = SearchIndex()


# Decorator form
@on_signal(UserModel, RepositorySignal.POST_SAVE)
async def index_user(user: UserModel) -> None:
    """Reindex the user in search after the row commits."""
    await search_index.upsert(user.id, user.name)


# Imperative form (same thing)
async def bust_cache(user: UserModel) -> None:
    """Drop the user's cache entry once the row has committed."""
    await cache.client.delete(f"user:{user.id}")

connect(UserModel, RepositorySignal.POST_SAVE, bust_cache)
disconnect(UserModel, RepositorySignal.POST_SAVE, bust_cache)  # remove

search_index is illustrative; cache is not

search_index.upsert(...) is a placeholder from your project (a search client) — it is not part of the SDK. Swap in your domain's real object.

cache is a real AsyncRedisManager, which is why the call goes through cache.client: the manager owns the lifecycle, and the Redis commands live on the client. Building the handler before the lifespan runs, use cache.client_proxy — cache.client raises RuntimeError until connect() has run.

The four moments:

Signal	Fires when	Typical use
PRE_SAVE	before the INSERT/UPDATE commits	cross-cutting validation; raising here vetoes the write (rollback + re-raise)
POST_SAVE	after commit + refresh	reindex, cache-bust, domain event
PRE_DELETE	before a single-row delete	clean up external dependencies
POST_DELETE	after the delete commits	notify that the row is gone

Handlers may be sync or async — an awaitable return value is awaited. Registering on a base model applies to its subclasses (resolved through the instance's MRO).

Signals cover the unit-of-work path only

add / add_all / update / update_many / soft_delete / restore / delete fire signals. The set-based bulk methods (bulk_update, bulk_create_values, bulk_upsert, delete_many, delete_batch) issue a single SQL statement and bypass signals by design — they never materialize the affected rows. soft_delete/restore fire PRE_SAVE/POST_SAVE (they are an UPDATE), not the delete signals.

Test isolation

The registry is process-global. In tests, call clear_signals() (from tempest_fastapi_sdk.db.signals) in a fixture teardown so one test's handler never leaks into the next.

F and Q expressions

For Django refugees: F references a column inside the query and Q composes conditions with OR/NOT. Both plug straight into the repository.

F — atomic in-database update. Decrementing stock with read-modify-write races: two requests read 10, both write 9. F("stock") - 1 computes in the database, in one statement — no lost update:

import asyncio
from uuid import UUID

from tempest_fastapi_sdk import BaseRepository

from db_setup import db
from src.db.models import UserModel

from tempest_fastapi_sdk import F


async def main() -> None:
    """Run this example."""
    product_id = UUID("6f1c3d84-2a55-4d0b-9d7e-0c1a2b3c4d5e")
    pid = product_id
    async with db.get_session_context() as session:
        repository = BaseRepository(session, model=UserModel)
        # stock = stock - 1, in the database
        await repository.bulk_update({"id": product_id}, {"stock": F("stock") - 1})

        # arithmetic from either side and between columns
        await repository.bulk_update({"id": pid}, {"stock": 100 - F("stock")})
        await repository.bulk_update({"id": pid}, {"total": F("price") * F("qty")})


asyncio.run(main())

update_returning — the decision and the new value in one statement. bulk_update returns only the count. When the write is conditional and you need its outcome — debit only if the balance covers it, with the new balance in hand —, update_returning runs the same UPDATE ... WHERE with RETURNING. No returned row means the condition did not match; there is no read first, so there is no window for another request:

import asyncio
from uuid import UUID

from tempest_fastapi_sdk import BaseRepository, F

from db_setup import db
from src.db.models import UserModel


async def main() -> None:
    """Run this example."""
    user_id = UUID("6f1c3d84-2a55-4d0b-9d7e-0c1a2b3c4d5e")
    async with db.get_session_context() as session:
        repository = BaseRepository(session, model=UserModel)
        rows = await repository.update_returning(
            {"id": user_id, "wallet_cents__gte": 500},
            {"wallet_cents": F("wallet_cents") - 500},
            returning=("wallet_cents",),
        )
        if rows:
            print("new balance:", rows[0]["wallet_cents"])
        else:
            print("refused: balance below 500")


asyncio.run(main())

where= takes a Q or a ready SQLAlchemy clause, for what the dict cannot express — the wallet discounts the held balance with a correlated subquery there. RETURNING needs PostgreSQL, or SQLite 3.35+.

Q — the OR / NOT the filter dict can't express. The dict ANDs everything; Q combines with & / | / ~ and enters via where=:

import asyncio

from tempest_fastapi_sdk import BaseRepository

from db_setup import db
from src.db.models import UserModel

from tempest_fastapi_sdk import Q


async def main() -> None:
    """Run this example."""
    async with db.get_session_context() as session:
        repository = BaseRepository(session, model=UserModel)
        # status open OR pending
        open_ = await repository.list(where=Q(status="open") | Q(status="pending"))

        # active and NOT guest
        active = await repository.list(where=Q(is_active=True) & ~Q(role="guest"))

        # combine with the dict (AND): stock >= 5 AND (open OR closed)
        rows = await repository.list(
            {"stock__gte": 5}, where=Q(status="open") | Q(status="closed")
        )


asyncio.run(main())

~Q(...) does not return the NULL row

The negation becomes role != 'guest' in SQL, and NULL != 'guest' is NULL, not true: the row whose role is null is left out along with the guest one. With three users (guest, admin and role=None), ~Q(role="guest") returns only admin. If the null row belongs in the result, ask for it explicitly: ~Q(role="guest") | Q(role=None) returns admin and the null one (None on a bare column is IS NULL). role__ne="guest" has the same semantics as ~Q here — it also drops the null.

Q uses the same conventions as the filter dict (name ILIKE, field__gte, iterable → IN, …), so Q(priority__gte=5, name="ana") is the AND of those conditions. where= works on list / first / get / get_or_none / count / exists / paginate / delete_many.

Available field__op suffix operators (in Q and the dict):

Suffix	SQL	Example
gt gte lt lte ne	comparison	Q(priority__gte=5)
in notin not_in	IN / NOT IN (value is any non-string iterable: list/set/tuple/generator, or a scalar, wrapped in a one-item list; not_in aliases notin)	Q(status__in={"open", "paid"})
between	col BETWEEN lo AND hi (value is an ordered pair (lo, hi) as a list/tuple)	Q(price__between=(10, 20))
iexact	case-insensitive equality (lower(col) == lower(v))	Q(email__iexact="Ana@X.com")
like ilike	raw LIKE / ILIKE with the caller's own %/_ wildcards, not escaped	Q(sku__ilike="ab_-%")
isnull	IS NULL (True) / IS NOT NULL (False)	Q(closed_at__isnull=True)
contains icontains	ILIKE %v% (value escaped)	Q(name__contains="ana")
startswith endswith	ILIKE v% / %v (value escaped)	Q(sku__startswith="SKU-")

like case-sensitivity is backend-defined

ilike is always case-insensitive. Plain like follows the database's LIKE semantics: SQLite folds ASCII case, PostgreSQL does not. For portable case handling, reach for ilike or iexact.

Raw SQLAlchemy is still there

F/Q are typed sugar over expressions SQLAlchemy already has. Need something they don't cover? Use select(...) directly — the repository doesn't get in the way.

---

4. Convention-based filters

Every method that takes filters: dict[str, Any] goes through the same engine. A None value means IS NULL, as in any ORM: {"deleted_at": None} matches the rows that were never deleted. The conventions:

Key / value	Generated SQL	Example
name (str)	case-insensitive ILIKE %value%	{"name": "ana"}
bool	col.is_(value)	{"is_active": True}
non-string iterable (list/set/tuple/frozenset/range/generator/dict view)	col.in_(values) — the iterable is materialized once, so passing a set needs no manual conversion to a list	{"id": {id1, id2}}
date	func.date(col) == value (whole day)	{"created_at": today}
start_in / end_in (date)	range on date/created_at	{"start_in": d1, "end_in": d2}
<col>__<op>	comparison gt/gte/lt/lte/ne	{"updated_at__gt": mark}
any other column	col == value	{"email": "a@b.com"}

import asyncio
from datetime import datetime, timedelta, timezone
from uuid import uuid4

from tempest_fastapi_sdk import BaseRepository

from db_setup import db
from src.db.models import UserModel


async def main() -> None:
    """Run this example."""
    end = datetime(2026, 1, 31, tzinfo=timezone.utc)
    selected_ids = [uuid4(), uuid4()]
    start = datetime(2026, 1, 1, tzinfo=timezone.utc)
    watermark = datetime.now(timezone.utc) - timedelta(hours=1)
    async with db.get_session_context() as session:
        repository = BaseRepository(session, model=UserModel)
        # "active rows updated after the watermark" — timestamp precision
        changed = await repository.list({
            "is_active": True,
            "updated_at__gt": watermark,
        })

        # "created between two dates" — whole day
        report = await repository.list({"start_in": start, "end_in": end})

        # text search + membership in a set
        hits = await repository.list({"name": "silva", "id": selected_ids})


asyncio.run(main())

start_in/end_in vs __gt/__lt

start_in/end_in match by whole day (func.date) against the model's date column (or created_at if absent). The __op suffixes are timestamp-precise — that's what delta-sync queries use. Choose by precision.

Filters come from a schema, not loose strings

In practice you don't assemble this dict by hand. BasePaginationFilterSchema (and its subclasses) expose .get_conditions(), which returns the dict already stripped of None. The router receives the filter via Depends().

An unknown key is refused

A key that is not a column of the model ({"usr_id": 1}, a typo) or a suffix that is not an operator ({"user_id__bogus": 1}) raises UnknownFilterKeyException before any statement runs — in every method that takes filters (list, count, exists, first, get, get_or_none, paginate, cursor_paginate, bulk_update, update_returning, delete_many), in Q, and in the layers that only pass it along (BaseService, BaseController, TenantScopedRepository):

import asyncio

from sqlalchemy import String
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import Mapped, mapped_column
from tempest_fastapi_sdk import BaseModel, BaseRepository, UnknownFilterKeyException


class TicketModel(BaseModel):
    __tablename__ = "tickets_unknown_key_demo"

    user_id: Mapped[int] = mapped_column()
    status: Mapped[str] = mapped_column(String(16))


async def main() -> None:
    """Run this example."""
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(TicketModel.metadata.create_all)
    async with AsyncSession(engine) as session:
        repo: BaseRepository[TicketModel] = BaseRepository(session, model=TicketModel)
        for user_id in (1, 2, 3):
            await repo.add(TicketModel(user_id=user_id, status="open"))
        try:
            await repo.delete_many({"usr_id": 1})
        except UnknownFilterKeyException as exc:
            print(exc.status_code, exc.code, exc.details)
        try:
            await repo.list({"user_id__bogus": 1})
        except UnknownFilterKeyException as exc:
            print(exc.details)
        print(await repo.count())
    await engine.dispose()


asyncio.run(main())

Output:

422 UNKNOWN_FILTER_KEY {'filter': 'usr_id'}
{'filter': 'user_id__bogus'}
3

All three rows are still there: delete_many deleted nothing.

• details carries only the rejected key, never the model's column list — when the dict comes from a query string, that list is a map of what to probe.
• A Python attribute with no SQL reading (a method, a @property, the metadata itself) counts as unknown. A column, a column_property and the class-level expression of a hybrid_property all resolve.
• start_in / end_in remain repository sugar, not columns.
• In the admin, list_filter, search_fields and every Lens' filters are checked when the AdminModel is built: a typo becomes a ValueError at boot, naming the option, instead of a 422 on the first click.

Behavior change (#465)

Up to 0.308.0 an unknown key was silently ignored: with only {"usr_id": 1} in the dict, list/count returned every row and bulk_update/delete_many changed the whole table. Anyone relying on that — for instance a filter schema with a non-column field (search, include_archived) forwarded whole through get_conditions() — now gets a 422. Take the field out of the dict before forwarding (conditions = f.get_conditions() then search = conditions.pop("search", None)) and handle it in the service.

A value that does not fit the operator is refused

A value with the wrong shape for its operator — a between that is not the (lo, hi) pair — raises InvalidFilterValueException before any statement runs. It was the rest of the #465 defect: the condition vanished from the WHERE and, on bulk_update, delete_many and update_returning, the write hit exactly the rows the filter did not name.

in is not on that list: a bare scalar is wrapped in a one-item list ({"status__in": "open"} → IN ('open')), as it always was, because the condition never vanished from the WHERE.

import asyncio

from sqlalchemy import String
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import Mapped, mapped_column
from tempest_fastapi_sdk import (
    BaseModel,
    BaseRepository,
    InvalidFilterValueException,
)


class TicketModel(BaseModel):
    __tablename__ = "tickets_bad_value_demo"

    status: Mapped[str] = mapped_column(String(16))


async def main() -> None:
    """Run this example."""
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(TicketModel.metadata.create_all)
    async with AsyncSession(engine) as session:
        repo: BaseRepository[TicketModel] = BaseRepository(
            session, model=TicketModel
        )
        for status in ("open", "paid"):
            await repo.add(TicketModel(status=status))
        try:
            await repo.delete_many({"status__between": ["open"]})
        except InvalidFilterValueException as exc:
            print(exc.status_code, exc.code, exc.details)
        print(await repo.count())
    await engine.dispose()


asyncio.run(main())

Output:

422 INVALID_FILTER_VALUE {'filter': 'status__between', 'value': "['open']"}
2

details["value"] echoes a repr of the value, truncated at 64 characters: a mistyped filter usually carries a large value (a payload, a list of ids), and the error is no place to repeat all of it.

Behavior change (#484)

Up to 0.312.0 a malformed between was ignored — the condition disappeared from the WHERE. Anyone sending a wrong pair and relying on that silence now gets a 422; fix the shape of the (lo, hi) pair. A scalar in in is still wrapped in a one-item list, as it always was.

Every paginated listing inherits the operators

Because get_conditions() only strips the pagination keys (page, page_size, order_by, ascending) and forwards everything else to the same engine, any BasePaginationFilterSchema subclass gets the operators for free: just declare a field named <column>__<op>. No extra inheritance, no mixin — the field name is the operator.

from tempest_fastapi_sdk import BasePaginationFilterSchema
from pydantic import Field


class ProductFilter(BasePaginationFilterSchema):
    """Product listing filter — each field becomes one condition."""

    name: str | None = Field(default=None)                 # ILIKE %name%
    category_id__in: set[int] | None = Field(default=None)  # IN (a set!)
    price__between: tuple[float, float] | None = Field(default=None)  # BETWEEN
    sku__ilike: str | None = Field(default=None)            # raw ILIKE
    created_at__gte: str | None = Field(default=None)       # >=

import asyncio

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from tempest_fastapi_sdk import BaseRepository

from src.db.models import UserModel
from src.schemas import ProductFilterSchema

# In a service the session comes from `db.get_session_context()`; here, SQLite.
session = AsyncSession(create_async_engine("sqlite+aiosqlite:///:memory:"))

f = ProductFilterSchema(name="silva", page=1, page_size=20)
repo = BaseRepository(session, model=UserModel)


async def main() -> None:
    """Run this example."""
    # In the service/repo the whole schema becomes filters + pagination:
    data = await repo.paginate(
        filters=f.get_conditions(),  # name/category_id__in/price__between/…
        **f.get_pagination_conditions(),  # page/page_size/order_by/ascending
    )


asyncio.run(main())

The frontend calls ?category_id__in=1&category_id__in=2&price__between=10&price__between=20 and FastAPI builds the schema via Depends(). The schema's get_conditions() drops the fields that were not filled before the dict is built (to_dict(exclude_none=True)), so the client sends only the fields it wants and a None from the schema never reaches the repository as IS NULL.

None changed meaning in v0.292.0

It used to drop the key — the filter vanished and the query returned everything. That is the most expensive failure shape available: silent, and in the direction of more rows. Measured in a consumer, {"left_at": None} (the obvious spelling of "still a member") also matched people who had left, and the WebSocket fan-out built from it kept delivering the group's messages to them.

If you built the dict by hand with optional keys and relied on the drop, leave the key out instead — or use get_conditions(), which already does. Where None has no reading at all (__gt, __between, __in), the condition is still dropped, now with a DroppedFilterWarning naming the key.

Recap: one dict, predictable conventions, None is IS NULL. Strings on name become ILIKE searches; __op suffixes give precise comparisons. Every paginated listing inherits these operators just by declaring the field.

---

The wire name of the page size

By default the envelope publishes page_size, both in the response and as the query parameter. For a new service that costs nothing — but a service that already publishes {"total", "items", "page", "size", "pages"} cannot rename it without breaking every paginated endpoint at once, and an app already in a store does not update in lockstep with its backend.

As of v0.261.0 there is a ready-made variant:

from tempest_fastapi_sdk import (
    CompactPaginationFilterSchema,
    CompactPaginationSchema,
)

# GET /users?size=50
filters = CompactPaginationFilterSchema.model_validate({"size": 50})
filters.page_size   # 50 — the Python name does not move

page = CompactPaginationSchema[int](
    items=[1], total=1, page=1, page_size=20, pages=1
)
page.model_dump(by_alias=True)["size"]   # 20

FastAPI serialises a response_model with by_alias=True, so declaring CompactPaginationSchema[Item] as the response_model is all it takes.

The repository does not change

BaseRepository.paginate keeps returning {"items", "total", "page", "page_size", "pages"} whatever the envelope publishes. The rename lives in the schema and nowhere else — there is nothing to rename in the data layer, and moving a service from one envelope to the other touches one line.

If your wire name is neither page_size nor size, both base classes turn populate_by_name on, so you can override just the field:

from pydantic import Field

from tempest_fastapi_sdk import BasePaginationSchema


class MyPage(BasePaginationSchema[int]):
    """An envelope of your own, renaming only the page size."""

    page_size: int = Field(
        validation_alias="perPage",
        serialization_alias="perPage",
    )

The wire name is written twice, validation_alias to read and serialization_alias to write. Field(alias=...) alone looks like it works and mypy accepts it, but pyright and basedpyright flag it in the consumer — the SDK does not use that form anywhere.

5. Bulk operations

For volume, row-by-row ORM is expensive. The repository offers two families: those that keep the unit-of-work (instances refreshed back) and those that bypass it (a single statement, no refresh).

import asyncio

from tempest_fastapi_sdk import BaseRepository

from db_setup import db
from src.db.models import UserModel


async def main() -> None:
    """Run this example."""
    m1, m2, m3 = (UserModel(name=n, email=f"{n}@x.com") for n in "abc")
    async with db.get_session_context() as session:
        repository = BaseRepository(session, model=UserModel)
        # Keeps the UoW — attached, refreshed instances
        created = await repository.add_all([m1, m2, m3])      # several INSERTs, 1 tx
        u1, u2 = created[0], created[1]
        updated = await repository.update_many([u1, u2])      # several UPDATEs, 1 tx

        # Bypasses the UoW — one statement, scales better (>= 50 rows)
        n = await repository.bulk_create_values([
            {"name": "A", "email": "a@x.com", "password_hash": "..."},
            {"name": "B", "email": "b@x.com", "password_hash": "..."},
        ])  # INSERT ... VALUES (...), (...) — returns row count

        n = await repository.bulk_update(
            filters={"is_active": False},
            values={"is_active": True},
        )  # UPDATE ... WHERE — returns affected row count

        n = await repository.bulk_upsert(
            rows=[{"sku": "ABC", "price": 10}, {"sku": "DEF", "price": 20}],
            conflict_columns=["sku"],          # requires a UNIQUE index
            update_columns=["price"],          # None = update everything but PK + conflict
        )  # INSERT ... ON CONFLICT DO UPDATE — Postgres and SQLite


asyncio.run(main())

bulk_update refuses an empty filter

Passing filters={} raises ValueError — it's the guard against an accidental table-wide UPDATE. To genuinely update every row, pass an explicit always-true condition.

bulk_* does not refresh the session

bulk_create_values, bulk_update and bulk_upsert emit a raw statement and do not refresh or attach instances to the session. Use them when you don't need the ORM objects back. If you need the instances, use add_all / update_many.

bulk_upsert is dialect-specific

Postgres and SQLite have native upsert. Other dialects raise NotImplementedError — fall back to a SELECT FOR UPDATE + UPDATE loop.

Recap: add_all/update_many when you want the instances back; bulk_* when you want throughput. An empty filter on bulk_update is a deliberate error.

---

6. Soft-delete and auditing (mixins)

The mixins are opt-in: you mix them alongside BaseModel only when the domain asks. SoftDeleteMixin adds deleted_at (+ mark_deleted() / mark_restored() / is_deleted). AuditMixin adds created_by / updated_by (+ stamp_created_by / stamp_updated_by).

# src/db/models/user.py
from sqlalchemy.orm import Mapped, mapped_column

from tempest_fastapi_sdk import AuditMixin, BaseModel, SoftDeleteMixin


class UserModel(BaseModel, SoftDeleteMixin, AuditMixin):
    """Users — soft-deletable and audited."""

    name: Mapped[str] = mapped_column()
    email: Mapped[str] = mapped_column(unique=True)
    password_hash: Mapped[str] = mapped_column()

Filtering is the caller's responsibility — the mixin does not install a global filter. Hide soft-deleted rows by passing deleted_at=None, or filter in the subclass. Stamping audit columns belongs in the service, where the current user is in scope:

# src/services/user.py
from uuid import UUID

from tempest_fastapi_sdk import BaseService

from src.db.repositories import UserRepository
from src.schemas import UserResponse, UserUpdateSchema


class UserService(BaseService[UserRepository, UserResponse]):
    """Business logic for the user domain."""

    async def list_alive(self) -> list[UserResponse]:
        """Return only rows where ``deleted_at IS NULL``.

        Returns:
            list[UserResponse]: The alive users.
        """
        instances = await self.repository.list(filters={"deleted_at": None})
        return [self.repository.map_to_response(i) for i in instances]

    async def update(
        self,
        user_id: UUID,
        data: UserUpdateSchema,
        *,
        actor_id: UUID,
    ) -> UserResponse:
        """Apply a partial update and stamp ``updated_by`` with the actor.

        Args:
            user_id (UUID): Primary key of the row to update.
            data (UserUpdateSchema): The partial payload.
            actor_id (UUID): The acting user, written to ``updated_by``.

        Returns:
            UserResponse: The updated user.
        """
        instance = await self.repository.get_by_id(user_id)
        instance.update_from_dict(data.model_dump(exclude_unset=True))
        instance.stamp_updated_by(actor_id)
        updated = await self.repository.update(instance)
        return self.repository.map_to_response(updated)

Two delete stamps, different purposes

Use repository.soft_delete(id) (the is_active flag) when the boolean is enough. Use the SoftDeleteMixin helpers (mark_deleted → deleted_at) when you need to know when the delete happened — auditing, retention policies.

MFA is another opt-in mixin

MFAMixin adds totp_secret / totp_enabled_at / totp_last_step to the user model when the project turns on the bundled MFA flow. Details in MFA (TOTP / 2FA) ».

Locale — the user's preferred language

LocaleColumnMixin adds a locale column (BCP-47, e.g. "pt-BR", "en-US", nullable) so a model can carry the language its notifications and localized text should render in — without every project re-declaring the same column. Mix it in like any other mixin:

# src/db/models/user.py
from sqlalchemy.orm import Mapped, mapped_column

from tempest_fastapi_sdk import BaseModel, LocaleColumnMixin


class UserModel(BaseModel, LocaleColumnMixin):
    """Users — carry a notification locale."""

    name: Mapped[str] = mapped_column()
    email: Mapped[str] = mapped_column(unique=True)

To write the value, use the Locale enum (a curated list of BCP-47 tags) instead of typing the string by hand — each member is the tag itself, so it compares and stores as that tag:

from tempest_fastapi_sdk import Locale

from src.db.models import UserModel

user = UserModel(name="Ana", email="ana@example.com")


user.locale = Locale.PT_BR          # stores "pt-BR"
user.locale = "en-US"               # a raw string works too
assert Locale.PT_BR == "pt-BR"      # a member is a str

A NULL locale means "no preference": resolve it to your app's default when rendering (typically via MessageCatalog), not as an error. This is exactly the pair the Web Push recipe » uses to localize each notification's title/body by the recipient's locale.

Locale is curated, not exhaustive

The enum covers the most widely used locales (pt/en/es/fr/de/… plus regional variants). Need a tag outside the list? The column is a str, so store the raw string and propose the new member upstream once it becomes common.

Recap: mixins enter only when the domain needs them; soft-delete filtering is yours (deleted_at IS NULL via a raw query); the audit stamp lives in the service; the user's locale comes from LocaleColumnMixin + the Locale enum.

---

7. Pagination

The SDK paginates two ways, both built into the repository. You almost never write the pagination query by hand.

Offset — when the client wants "page 3 of 12"

# src/db/repositories/user.py — convenience method

from typing import Any

from tempest_fastapi_sdk import BasePaginationSchema, BaseRepository

from src.db.models import UserModel
from src.schemas import UserResponse


UserPage = BasePaginationSchema[UserResponse]


class UserRepository(BaseRepository[UserModel]):
    # ... __init__ + mappers ...

    async def list_page(
        self,
        *,
        filters: dict[str, Any] | None = None,
        page: int = 1,
        page_size: int = 20,
    ) -> UserPage:
        """Return one offset-paginated page of users.

        Args:
            filters (dict[str, Any] | None): Filter conditions.
            page (int): 1-indexed page number.
            page_size (int): Items per page.

        Returns:
            UserPage: Items + total + page metadata.
        """
        result = await self.paginate(
            filters=filters,
            page=page,
            page_size=page_size,
        )
        return UserPage(
            items=[self.map_to_response(i) for i in result["items"]],
            total=result["total"],
            page=result["page"],
            page_size=result["page_size"],
            pages=result["pages"],
        )

BaseRepository.paginate returns a dict with items / total / page / page_size / pages. The total is computed from the same filtered query, so custom joins still report a correct total. When order_by is None, it orders by created_at desc.

order_by is validated against the model's columns

It arrives straight from a query parameter (BasePaginationFilterSchema declares a str), so it is untrusted input. paginate and cursor_paginate resolve the name through the mapper and raise OrderByNotAllowedException (422, a ValidationException subclass) when it is not a mapped column — including an attribute that exists on the class but is not a column, like metadata. The refusal never lists the model's columns: since 0.298.0 details["allowed"] only appears when you declared orderable_columns, and then it holds only that set. See the next section.

Forward the schema without unpacking it by hand

The get_conditions() / get_pagination_conditions() pair covers both sides of the filter: the former returns only the domain filters, the latter only the pagination keys (page, page_size, order_by, ascending). The service forwards the filter straight through, no **f — which would leak domain filters (is_active, etc.) as kwargs the repository does not accept:

data = await repo.paginate(
    filters=f.get_conditions(),
    **f.get_pagination_conditions(),
)

CursorPaginationFilterSchema exposes the same pair (with cursor / limit instead of page / page_size).

Which columns sort, and how many rows fit in a page

Checking that order_by is a column is not enough on a public listing. Sorting is an oracle: ?order_by=wallet ranks everyone by balance, and a caller who controls their own row (writes their own email, their own tax id) learns the neighbouring row's value by binary search over where theirs lands. And with no ceiling on page_size, one request reads the whole table.

Declare both on the listing's filter:

from typing import Annotated

from fastapi import FastAPI, Query
from tempest_fastapi_sdk import BasePaginationFilterSchema, register_exception_handlers


class ProducerFilterSchema(BasePaginationFilterSchema):
    """Filter of the public producer listing."""

    orderable_columns = frozenset({"created_at", "name"})
    max_page_size = 50


app = FastAPI()
register_exception_handlers(app)


@app.get("/producers")
def list_producers(
    filters: Annotated[ProducerFilterSchema, Query()],
) -> dict[str, int | str | None]:
    """Echo the validated pagination keys."""
    return {"page_size": filters.page_size, "order_by": filters.order_by}

Piece by piece:

• orderable_columns is a ClassVar[frozenset[str] | None]. With the set declared, any other value becomes OrderByNotAllowedException during schema validation, before it reaches the repository:

  {
    "detail": "Cannot order by this field",
    "code": "ORDER_BY_NOT_ALLOWED",
    "details": {"order_by": "hashed_password", "allowed": ["created_at", "name"]},
    "field": "order_by"
  }
  

  details["allowed"] is only the declared set, sorted. The default is None, which accepts any value and leaves the check to the repository — the previous behavior. An empty order_by (?order_by=) means absent and becomes None, never a 422.

• max_page_size is the inclusive ceiling on page_size. The base default is DEFAULT_MAX_PAGE_SIZE (100); declaring it on a subclass rewrites the field's le=, so the refusal is pydantic's own less_than_equal and OpenAPI publishes maximum: 50 on the parameter — with Depends() and with Annotated[..., Query()]. None removes the ceiling.

The ceiling of 100 is new in 0.298.0

Up to 0.297.x page_size only had ge=1. A client asking for page_size=500 now gets a 422 — declare max_page_size with the value the listing really needs (or None) instead of redeclaring the field.

The second line is the repository, for callers that reach it without the schema (a script, the panel, another route). The same names exist on BaseRepository, as a class attribute or a constructor argument (the constructor wins):

import asyncio
from typing import ClassVar

from sqlalchemy import Integer, String
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import Mapped, mapped_column
from tempest_fastapi_sdk import (
    BaseModel,
    BaseRepository,
    OrderByNotAllowedException,
    PageSizeTooLargeException,
)


class ProducerModel(BaseModel):
    __tablename__ = "producers_order_demo"

    name: Mapped[str] = mapped_column(String(80))
    wallet: Mapped[int] = mapped_column(Integer, default=0)


class ProducerRepository(BaseRepository[ProducerModel]):
    orderable_columns: ClassVar[frozenset[str] | None] = frozenset(
        {"created_at", "name"},
    )
    max_page_size: ClassVar[int | None] = 50

    def __init__(self, session: AsyncSession) -> None:
        super().__init__(session, model=ProducerModel)


async def main() -> None:
    """Run this example."""
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(ProducerModel.metadata.create_all)
    async with AsyncSession(engine) as session:
        repo = ProducerRepository(session)
        await repo.paginate(order_by="name", page_size=50)
        try:
            await repo.paginate(order_by="wallet")
        except OrderByNotAllowedException as exc:
            print(exc.details)
        try:
            await repo.paginate(page_size=51)
        except PageSizeTooLargeException as exc:
            print(exc.code, exc.details)
    await engine.dispose()


asyncio.run(main())

Output:

{'order_by': 'wallet', 'allowed': ['created_at', 'name']}
PAGE_SIZE_TOO_LARGE {'page_size': 51, 'max_page_size': 50}

• Without orderable_columns on the repository, any mapped column still sorts — but refusing a name that is not a column answers details with just {"order_by": ...}, no allowed key.
• A declared name that is not a model column raises ValueError when the repository is built: a typo in the declaration would otherwise refuse, on every request, exactly the column you meant to allow.
• The repository's max_page_size defaults to None (no ceiling), because internal callers such as the /admin panel export page by the thousand. Once declared, it applies to paginate, cursor_paginate (over limit) and GeoRepositoryMixin.paginate_nearby.

The repository's set governs every path

orderable_columns on the repository covers paginate, cursor_paginate and changes_since — include the watermark column (updated_at) if you use sync. And the /admin panel sorts through the same repository: if it uses this class via repository_class, sorting by a column outside the set is refused there too.

CursorPaginationFilterSchema has the same orderable_columns, and max_limit (default DEFAULT_MAX_CURSOR_LIMIT, 500 — the le=500 the field always had) in the role of max_page_size.

Recap: declare orderable_columns and max_page_size on the filter of every public listing; the 422 lists only what you declared. On the repository, the same names are the second line.

Cursor — when the table is large

Cursor pagination scales better than offset on large tables (no COUNT(*), stable under concurrent inserts) at the cost of losing random access. It's already built in as cursor_paginate — it orders by (order_by, id) and encodes the opaque cursor automatically:

# src/db/repositories/user.py

from typing import Any

from tempest_fastapi_sdk import BaseRepository, CursorPaginationSchema

from src.db.models import UserModel
from src.schemas import UserResponse


UserCursorPage = CursorPaginationSchema[UserResponse]


class UserRepository(BaseRepository[UserModel]):
    # ... __init__ + mappers ...

    async def cursor_page(
        self,
        *,
        cursor: str | None = None,
        limit: int = 20,
        ascending: bool = False,
        filters: dict[str, Any] | None = None,
    ) -> UserCursorPage:
        """Return one cursor-paginated page of users.

        Args:
            cursor (str | None): Opaque cursor from the previous page.
            limit (int): Max items in the page.
            ascending (bool): Sort direction.
            filters (dict[str, Any] | None): Filter conditions.

        Returns:
            UserCursorPage: Items + next_cursor + has_more.
        """
        result = await self.cursor_paginate(
            filters=filters,
            cursor=cursor,
            limit=limit,
            order_by="created_at",
            ascending=ascending,
        )
        return UserCursorPage(
            items=[self.map_to_response(i) for i in result["items"]],
            next_cursor=result["next_cursor"],
            has_more=result["has_more"],
            limit=result["limit"],
        )

Router, with the filter coming from a schema via Depends():

# src/api/routers/user.py
from fastapi import APIRouter, Depends

from tempest_fastapi_sdk import CursorPaginationFilterSchema

from src.api.dependencies.resources import SessionDep
from src.db.repositories import UserCursorPage, UserRepository

router = APIRouter(prefix="/api/users", tags=["users"])


class UserCursorFilter(CursorPaginationFilterSchema):
    """Cursor filter for the user listing."""

    name: str | None = None   # ILIKE %value% by the repository convention


@router.get("/", response_model=UserCursorPage)
async def list_users(
    session: SessionDep,
    f: UserCursorFilter = Depends(),
) -> UserCursorPage:
    """List users, cursor-paginated."""
    repository = UserRepository(session)
    return await repository.cursor_page(
        cursor=f.cursor,
        limit=f.limit,
        ascending=f.ascending,
        filters=f.get_conditions(),
    )

The cursor is opaque

next_cursor is url-safe base64 JSON. The client never inspects it; it echoes the value back verbatim until next_cursor becomes null. Under the hood, cursor_paginate uses encode_cursor/decode_cursor and a (order_by, id) tuple comparison that's stable on Postgres.

The cursor value comes back in the column's type

The cursor is JSON, so a datetime leaves page 1 as an ISO string. Before building the WHERE, cursor_paginate rehydrates the value into the column's Python type (datetime, date, time, UUID, Decimal, enum). Without that, PostgreSQL refused the whole comparison on the second page:

operator does not exist: timestamp with time zone < character varying

And SQLite did not complain — it compared the two strings and returned a wrong page: measured, a 7-row walk in pages of 3 repeated r02 and skipped r05. A text column is left untouched, and so is a TypeDecorator that declares no python_type.

For offline-first sync there's a third mode

changes_since + SyncPaginationSchema do delta pagination (rows changed since a watermark). See Offline sync ».

Recap: paginate (offset) for page navigation; cursor_paginate for feeds/large tables. Both ready — you only map the result to the response schema.

---

8. Alembic migrations

AlembicHelper wraps Alembic with a curated config (UTC timezone, date-prefixed files, target_metadata already wired, batch mode). Full flow: bootstrap → revision → apply → CI gate.

Bootstrap, once per project

# scripts/alembic_init.py
from tempest_fastapi_sdk import AlembicHelper

from src.core.settings import settings

helper = AlembicHelper(config_path="alembic.ini", db_url=settings.DATABASE_URL)
helper.init(
    directory="alembic",
    metadata_module="src.db.models",   # exposes BaseModel
    metadata_attr="BaseModel",
    db_url=settings.DATABASE_URL,
)

uv run python scripts/alembic_init.py

Creates:

alembic.ini                 # SDK-curated config (UTC, date prefix, post-write hooks, prepend_sys_path)
alembic/
├── env.py                  # run_alembic_env(target_metadata): hooks, compare_type, batch
├── script.py.mako
└── versions/

Generate revisions

# scripts/make_migration.py
import sys

from tempest_fastapi_sdk import AlembicHelper

from src.core.settings import settings

helper = AlembicHelper("alembic.ini", db_url=settings.DATABASE_URL)
helper.revision(message=sys.argv[1], autogenerate=True)

uv run python scripts/make_migration.py "add users table"

The file lands in alembic/versions/2026_05_16_1432-ae12cd34_add_users_table.py — the date prefix orders files chronologically and makes merge conflicts obvious.

Migrations come out lint-clean

The alembic.ini that init() writes includes [post_write_hooks] that runs ruff format and then ruff check --fix on every revision. Without it, Alembic's files fail tempest lint (W291 on the empty Revises:, E501 on long sa.Column(...) lines) — which is why the formatter goes first: it is the one that wraps the long line and strips the trailing space, so running the linter first would report errors the next hook is about to fix. The hooks use your project's ruff config. Requires ruff on PATH — already a dev dependency in every tempest new scaffold.

Apply on startup

# src/api/app.py — inside the lifespan

from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from tempest_fastapi_sdk import AlembicHelper

from src.api.dependencies.resources import db
from src.core.settings import settings


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    """Run pending migrations, then serve."""
    helper = AlembicHelper("alembic.ini", db_url=settings.DATABASE_URL)
    await helper.upgrade_async()
    await db.connect()
    yield
    await db.disconnect()

upgrade_async() runs the migration in a worker thread: the SDK's env.py calls asyncio.run, which cannot nest inside the lifespan's loop, so a plain helper.upgrade() there raises a RuntimeError naming upgrade_async. Every command that runs env.py has the same _async twin — see Migrations ».

Destructive migrations: use safe_upgrade

helper.pending_destructive_ops() lists pending column/table DROPs; helper.safe_upgrade() raises DestructiveMigrationError instead of silently dropping data. The full deploy guide (migration + graceful shutdown) is in Safe deploys ».

CI gate — the schema must match the models

# scripts/check_migrations.py
import sys

from tempest_fastapi_sdk import AlembicHelper

from src.core.settings import settings

helper = AlembicHelper("alembic.ini", db_url=settings.DATABASE_URL)
if not helper.check():
    print("Schema drift detected — run make_migration.py and commit.")
    sys.exit(1)
print("Schema is in sync.")

# .github/workflows/ci.yml
- name: Check migrations are in sync
  run: uv run python scripts/check_migrations.py

check() no longer reports phantom enum drift on SQLite (v0.313.0, #483)

On SQLite an enum column is a VARCHAR plus a CHECK (ck_<table>_<enum>). Alembic's CHECK comparator read the constraint from the database but dropped it from the metadata side — they are _type_bound, attached to the column type — so check() proposed a phantom remove_constraint right after a clean upgrade head, and the drift gate stayed off. run_alembic_env passes an include_object that ignores that CHECK, keyed by (table, name) so two tables with a same-named CHECK cannot hide each other's drift. Real drift is still reported — a new column, and also a changed enum member: sync_enum_types appends a replace_enum, and ReplaceEnumOp gained to_diff_tuple, without which check() raised NotImplementedError while building the diff instead of reporting it.

Base columns always first

The SDK's env.py installs the reorder_base_columns_first hook, so every generated migration lists id / is_active / created_at / updated_at ahead of your columns — consistent diffs across people.

A new NOT NULL column no longer explodes (v0.67.0)

Adding a NOT NULL column to a table that already has rows blows up on PostgreSQL with NotNullViolationError: column "x" contains null values — because a Python-side default= only fires on ORM inserts, never as DDL. The SDK now installs a second hook, backfill_non_nullable_defaults: every added column that is nullable=False, has no server_default, but does declare a scalar Python default gets a server_default derived from that default — so the generated migration backfills existing rows in the same statement.

# In the model — just the Python default:
is_professional: Mapped[bool] = mapped_column(default=False)

# The generated migration now reads (note server_default):
op.add_column(
    "users",
    sa.Column(
        "is_professional",
        sa.Boolean(),
        nullable=False,
        server_default=sa.text("false"),
    ),
)

Covers bool / int / float / str / Enum (uses .value). It does not act when the default is a callable (uuid4, func.now()) or absent — those need a hand-written data migration, since the SDK cannot infer a safe backfill value.

Already have an old env.py? Update the import + wiring to compose both hooks:

# alembic/env.py
from tempest_fastapi_sdk.db.alembic_hooks import (
    backfill_non_nullable_defaults,
    compose_hooks,
    reorder_base_columns_first,
)

_process_revision_directives = compose_hooks(
    reorder_base_columns_first,
    backfill_non_nullable_defaults,
)

# ...and pass it to context.configure(process_revision_directives=...)

For a migration that already exploded, add server_default=sa.text("...") by hand to the op.add_column (or backfill + alter_column to drop the default afterwards).

Migrate composite constraints from the old convention

If your database was created before the composite naming fix, every unique, index and foreign key with more than one column is still named after its first column only. Single-column names did not change. After upgrading the SDK the drift gate reports a difference — and what autogenerate proposes is not what you want to apply.

On a database with UniqueConstraint("title", "release_year"), Index(None, "author", "title") and a composite FK, alembic revision --autogenerate renders (measured against PostgreSQL 16 and Alembic 1.19.1):

from alembic import op


def upgrade() -> None:
    """What autogenerate renders on the legacy database."""
    with op.batch_alter_table("books", schema=None) as batch_op:
        batch_op.drop_index(batch_op.f("ix_books_author"))
        batch_op.drop_constraint(batch_op.f("uq_books_title"), type_="unique")
        batch_op.create_index(
            batch_op.f("ix_books_author_books_title"),
            ["author", "title"],
            unique=False,
        )
        batch_op.create_unique_constraint(
            batch_op.f("uq_books_title_release_year"),
            ["title", "release_year"],
        )

Do not apply that drop + create on PostgreSQL

Two measured problems, and a cost:

• It breaks when an FK depends on the unique. With a composite FK pointing at the unique it drops, PostgreSQL refuses: cannot drop constraint uq_authors_tenant_id on table authors because other objects depend on it.
• It does not see the foreign keys. A composite FK whose name changed does not show up in the diff; it keeps its old name in the database, and that is the name parse_integrity_error returns.
• It rebuilds the index. Creating the unique again builds its index by reading the whole table. RENAME only changes the name — and on PostgreSQL renaming the constraint also renames the index behind it (measured in pg_indexes).

legacy_constraint_renames compares the metadata against the old convention and lists every composite constraint whose name changed — foreign keys included — with the RENAME ready. Run it once, with every model imported:

# scripts/constraint_renames.py
from sqlalchemy.dialects import postgresql

import src.db.models
from tempest_fastapi_sdk import BaseModel, legacy_constraint_renames

dialect = postgresql.dialect()
renames = legacy_constraint_renames(BaseModel.metadata)

print("def upgrade() -> None:")
for rename in renames:
    print(f"    op.execute({rename.statement(dialect)!r})")

print("def downgrade() -> None:")
for rename in renames:
    print(f"    op.execute({rename.inverse().statement(dialect)!r})")

Create an empty revision (helper.revision(message="rename composite constraints", autogenerate=False)) and paste the output into it:

"""rename composite constraints to the per-column convention."""

from alembic import op

revision: str = "b7e1c2d3a4f5"
down_revision: str | None = "ae12cd34"


def upgrade() -> None:
    """Rename composite constraints to the per-column convention."""
    op.execute("ALTER INDEX ix_books_author RENAME TO ix_books_author_books_title")
    op.execute(
        "ALTER TABLE books RENAME CONSTRAINT uq_books_title "
        "TO uq_books_title_release_year"
    )


def downgrade() -> None:
    """Go back to the old convention's names."""
    op.execute("ALTER INDEX ix_books_author_books_title RENAME TO ix_books_author")
    op.execute(
        "ALTER TABLE books RENAME CONSTRAINT uq_books_title_release_year "
        "TO uq_books_title"
    )

After the upgrade, autogenerate sees no difference at all, and the single-column constraints were left untouched (measured: compare_metadata returns [] on the renamed database, in tests/db/test_naming_live.py).

Why paste, instead of calling the function inside the migration

A migration has to do tomorrow what it does today. Called inside upgrade(), the function would compute the renames from the models on the day the migration runs — a composite constraint added later would leak into an old file. Earlier revisions need nothing: autogenerate wrote the name into each of them with op.f(...).

Details worth knowing

• The SDK has one too. make_user_oauth_account_model declares UniqueConstraint("provider", "subject") and UniqueConstraint("user_id", "provider"): uq_<table>_provider becomes uq_<table>_provider_subject, and uq_<table>_user_id becomes uq_<table>_user_id_provider. With the model imported, the function lists them next to yours.
• Two entries with the same old_name are the old collision: PostgreSQL refused the second CREATE, so the database holds only one of them. Keep the entry whose columns are those of the constraint that exists.
• SQLite cannot rename a constraint — the name lives in the CREATE TABLE text —, and statement() raises ValueError outside PostgreSQL. On a development SQLite, apply the batch migration autogenerate renders (it rebuilds the table and works, measured) or recreate the database.
• Want to postpone? Give the constraint the name it already has in the database: UniqueConstraint("title", "release_year", name="uq_books_title"). An explicit name wins over the convention, and the function then skips it.

Recap: init once, revision --autogenerate per change, upgrade on startup, check in CI, safe_upgrade to protect data. A database created before the composite naming fix: legacy_constraint_renames writes the RENAMEs, and you paste them into an empty revision.

---

9. Detecting slow queries

SlowQueryLogger registers an engine listener and emits a log line for every statement above a threshold. Attach it once at boot:

# src/api/app.py — after db.connect()

from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from tempest_fastapi_sdk.db import SlowQueryLogger

from src.api.dependencies.resources import db


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    """Connect, instrument slow queries, then serve."""
    await db.connect()
    slow = SlowQueryLogger(db.engine, threshold_ms=200.0)
    slow.attach()
    yield
    await db.disconnect()

Parameter	Default	What for
threshold_ms	500.0	Statements at or above this duration are logged.
level	logging.WARNING	Level of the slow-query lines.
log_parameters	False	Includes bind params in the line. Dev only — they may carry PII.
explain	False	Runs EXPLAIN and appends the plan. Costs a round-trip per slow query.

log_parameters=True in development only

Bind parameters may contain secrets and PII. Keep it False in production — the default is already safe.

Recap: SlowQueryLogger(db.engine, threshold_ms=...).attach() in the lifespan turns slow queries into actionable log lines, with optional EXPLAIN to investigate plans.

---

Recap

• BaseModel brings id, is_active, created_at and updated_at; you declare only your domain columns, and the Alembic hook keeps that order in generated migrations.
• A composite constraint is named after every column (uq_books_title_release_year); a database created before that fix migrates with the RENAMEs legacy_constraint_renames writes.
• One AsyncDatabaseManager per application, in resources.py — not one per request.
• BaseRepository works instantiated for plain CRUD and subclassed once real queries appear; a filter is a dict with predictable conventions, and None on a bare column is IS NULL.
• Bulk work comes in two families: the one that hands the instances back (add_all, update_many) and the one that does not, but is a single trip to the database.
• A mixin joins when the domain asks for it: soft-delete and auditing cost a column, and the filter is yours — SoftDeleteMixin installs no global filter, so hiding a deleted row means passing {"deleted_at": None}.
• Pagination has two shapes with different purposes: paginate to walk pages, cursor_paginate for a list that grows while the user reads it.
• Migrations are init once, revision --autogenerate per change, upgrade on deploy — and SlowQueryLogger on the engine shows the slow query with EXPLAIN before a user complains.

Next steps

This page covered the core. The advanced database features have dedicated recipes:

• Multi-tenant » — TenantScopedRepository for per-tenant isolation.
• Audit trail » — BaseAuditLogModel, add_audited / update_audited / delete_audited (who changed what, in the same tx).
• Transactional outbox » — BaseOutboxModel + OutboxRelay, save_with_outbox to publish events atomically with the write.
• Offline sync » — changes_since + delta pagination for offline-first clients.
• Safe deploys » — destructive migrations + graceful shutdown.
• Testing » — in-memory SQLite, fixtures, create_tables.