DB¶

SQLAlchemy async session management with transactions, table locking, advisory locking, and row-change polling.

Overview¶

The db module is built around one object, Database, which owns the engine and sessionmaker and exposes the FastAPI dependency, a commit-before-response middleware, session/transaction context managers, and table locking. Free helpers cover savepoint-aware transactions, advisory locks, many-to-many association tables, and row-change polling.

Setup¶

Create one Database for your app. Provide a URL (the facade builds and disposes the engine) or pass an existing engine= you own (e.g. for Alembic or event.listen). The session factory is built internally with expire_on_commit=False.

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

from fastapi_toolsets.db import Database

db = Database("postgresql+asyncpg://postgres:postgres@localhost/app")

app = FastAPI()
db.install(app)  # commit middleware + engine disposal on shutdown


@app.get("/users")
async def list_users(session: AsyncSession = Depends(db)): ...

The Database instance is the dependency: use it directly as Depends(db). The whole request runs as a single transaction (CRUD writes use savepoints under it).

The URL may be a plain string or a Pydantic PostgresDsn. In URL mode you can tune the engine: pass connect_args for DBAPI-level options and any other keyword for create_async_engine (e.g. pool_size, echo, pool_pre_ping):

from pydantic_settings import BaseSettings
from pydantic import PostgresDsn


class Settings(BaseSettings):
    database_url: PostgresDsn


settings = Settings()

db = Database(
    settings.database_url,
    pool_size=20,
    pool_pre_ping=True,
    connect_args={"server_settings": {"application_name": "myapp"}},
)

Session class¶

session_class selects the class the internal factory builds sessions from. It defaults to plain AsyncSession.

To use the lifecycle events of the models module, pass EventSession here. It is the only place to wire them in v5: Database owns the session factory, so a factory you build yourself is no longer the one serving requests.

from fastapi_toolsets.models import EventSession

db = Database("postgresql+asyncpg://...", session_class=EventSession)

Session execution options¶

session_execution_options applies execution options to every statement the sessions run, including the flush and eager loads. It works with url and with engine, so several Database objects can share one engine with different options, for example one schema per tenant:

engine = create_async_engine("postgresql+asyncpg://...")

tenant_a = Database(
    engine=engine,
    session_execution_options={"schema_translate_map": {None: "tenant_a"}},
)

Options passed as extra keywords in URL mode (Database(url, execution_options=...)) still configure the engine, not the sessions.

Committing before the response¶

db.install(app) adds a middleware that commits the request's session when the response starts, after the endpoint returns and before the body is sent. The dependency commits only if the middleware did not: when a function-scoped dependency unwinds before the response, or when the response never passes through the middleware. Either way the request is committed exactly once.

The request is committed as a single transaction:

• Read-after-write: a follow-up request sees the write.
• Atomicity: multi-write endpoints roll back as a unit on failure.
• Errors roll back: on a raised exception the session rolls back and nothing is committed.

If the commit itself fails, for example on a unique constraint that was never flushed in the endpoint, the middleware answers with the app's exception handler for that error, such as the constraint violation handler of init_exceptions_handlers. Without a handler for that error class, the request ends in a 500.

Without install, the session commits in the dependency teardown, which runs after the response has been sent.

Lifespan¶

db.install(app) disposes the engine on shutdown, composing around your own lifespan:

from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app):
    await warm_cache()  # your startup
    yield
    await flush_metrics()  # your shutdown


app = FastAPI(lifespan=lifespan)
db.install(app)  # your shutdown runs first, then the engine is disposed

If you have no lifespan of your own, db.lifespan works standalone as FastAPI(lifespan=db.lifespan). Engine disposal is idempotent and is a no-op when you passed your own engine=.

Session context manager¶

Use db.session() for sessions outside request handlers (e.g. background tasks, CLI commands). It commits on clean exit and rolls back on exception:

async def seed():
    async with db.session() as session:
        ...

Transactions¶

transaction opens a transaction on a session, using a savepoint when one is already open so it nests safely:

from fastapi_toolsets.db import transaction


async def create_user_with_role(session):
    async with transaction(session):
        ...
        async with transaction(session):  # uses a savepoint
            ...

When you have a Database, db.begin() opens a session already inside a transaction:

async with db.begin() as session:
    session.add(User(name="ada"))  # commits on exit, rolls back on exception

Table locking¶

db.lock_tables acquires PostgreSQL table-level locks for a critical section. By default it opens a dedicated session internally and releases the lock when the context exits (or locks on a session you pass):

from fastapi_toolsets.db import LockMode

async with db.lock_tables([User], mode=LockMode.EXCLUSIVE) as session:
    # No other transaction can modify User until this block exits
    ...

Available lock modes are defined in LockMode: ACCESS_SHARE, ROW_SHARE, ROW_EXCLUSIVE, SHARE_UPDATE_EXCLUSIVE, SHARE, SHARE_ROW_EXCLUSIVE, EXCLUSIVE, ACCESS_EXCLUSIVE.

Pass timeout to limit how long the lock waits. On timeout, a LockTimeoutError is raised instead of a raw database error:

async with db.lock_tables([Order], timeout="2s") as session:
    ...

Locking on the request's own session¶

The dedicated session above is a second connection, on top of the one the request already holds through Depends(db), and it is a different backend from the request's. Two consequences:

• The request needs two simultaneous connections. Size the pool for concurrent locking requests, not concurrent users. Behind a transaction-mode pooler (PgBouncer, a CNPG Pooler) where default_pool_size is below peak concurrency, this self-deadlocks: every server connection is pinned by a request transaction waiting for a second one. SQLAlchemy's pool_timeout never fires, because the wait happens inside the pooler.
• The block must not touch the request session under a conflicting mode (SHARE, SHARE_ROW_EXCLUSIVE, EXCLUSIVE, ACCESS_EXCLUSIVE). timeout applies only to the dedicated session, so the request session waits with no bound at all, and PostgreSQL sees no deadlock, because the lock holder is merely idle in transaction.

Pass session= to avoid both: the lock is taken on the transaction the request already owns, on one connection, and the block is free to use that same session.

@app.post("/orders")
async def create_order(session=Depends(db)):
    async with db.lock_tables([Order], session=session, mode=LockMode.EXCLUSIVE):
        # Same session, same connection: writing through it cannot self-block
        return await OrderCrud.create(session, data)

Whichever path you take, connect_args={"server_settings": {"statement_timeout": "30s", "idle_in_transaction_session_timeout": "60s"}} turns a freeze into an error.

Advisory locking¶

advisory_lock acquires a PostgreSQL advisory lock on a session you provide. By default it is session-level and released when the context exits:

from fastapi_toolsets.db import advisory_lock

# Blocking exclusive lock: waits until the lock is free
async with advisory_lock(session=session, key=42):
    ...

# Non-blocking: yields False immediately if already held
async with advisory_lock(session=session, key=42, nowait=True) as acquired:
    if not acquired:
        raise HTTPException(409, "Resource is locked")

# Blocking with a timeout: raises LockTimeoutError if not acquired in time
async with advisory_lock(session=session, key=42, timeout="5s"):
    ...

# Shared lock: multiple readers allowed simultaneously, blocks exclusive writers
async with advisory_lock(session=session, key=42, shared=True):
    ...

# Two-integer key for namespacing (e.g. lock_type + resource_id)
async with advisory_lock(session=session, key=(1, user_id)):
    ...

# Transaction-level: released by the commit, not at block exit
async with advisory_lock(session=session, key=(1, user_id), xact=True):
    ...

Choosing between session-level and transaction-level¶

xact=True switches to pg_advisory_xact_lock, which PostgreSQL releases at commit or rollback. Reach for it whenever the protected read has to see the previous holder's write. A check-then-insert is the usual case:

@app.post("/claims")
async def claim(session=Depends(db)):
    async with advisory_lock(session=session, key=(team_id, question_id), xact=True):
        if await ClaimCrud.first(session, [Claim.question_id == question_id]):
            raise ConflictError(detail="Already claimed")
        return await ClaimCrud.create(session, data)

A session-level lock is wrong here. db.install commits the request session at http.response.start, which is after the handler returns and therefore after the async with block released the lock. Two concurrent requests both read "not present" and both insert. The lock is held over exactly the wrong interval.

Keep the session-level default when the critical section is self-contained, for example one worker at a time over work that is not read back through the same transaction.

Row-change polling¶

wait_for_row_change polls a row until a specific column changes value:

from fastapi_toolsets.db import wait_for_row_change

# Wait up to 30s for order.status to change
await wait_for_row_change(
    session=session,
    model=Order,
    pk_value=order_id,
    columns=["status"],
    interval=1.0,
    timeout=30.0,
)

Creating a database¶

create_database (in fastapi_toolsets.db.testing) connects to server_url and issues a CREATE DATABASE statement:

from fastapi_toolsets.db.testing import create_database

SERVER_URL = "postgresql+asyncpg://postgres:postgres@localhost/postgres"

await create_database(db_name="myapp_test", server_url=SERVER_URL)

For test isolation with automatic cleanup, use create_worker_database from the pytest module, which handles drop-before, create, and drop-after.

Cleaning up tables¶

cleanup_tables (in fastapi_toolsets.db.testing) truncates all tables:

from fastapi_toolsets.db.testing import cleanup_tables


@pytest.fixture(autouse=True)
async def clean(db_session):
    yield
    await cleanup_tables(session=db_session, base=Base)

Many-to-Many helpers¶

The three m2m_* helpers modify a many-to-many association table with direct SQL, without loading the ORM collection.

m2m_add: insert associations¶

m2m_add inserts one or more rows into a secondary table:

from fastapi_toolsets.db import m2m_add

async with db.lock_tables([Tag]) as session:
    tag = await TagCrud.create(session, TagCreate(name="python"))
    await m2m_add(session, post, Post.tags, tag)

Pass ignore_conflicts=True to skip associations that already exist:

await m2m_add(session, post, Post.tags, tag, ignore_conflicts=True)

m2m_remove: delete associations¶

m2m_remove deletes specific association rows. Removing a non-existent association is a no-op:

from fastapi_toolsets.db import m2m_remove, transaction

async with transaction(session):
    await m2m_remove(session, post, Post.tags, tag1, tag2)

m2m_set: replace the full set¶

m2m_set replaces all associations: it deletes every existing row for the owner instance then inserts the new set. Passing no related instances clears the association:

from fastapi_toolsets.db import m2m_set, transaction

# Replace all tags
async with transaction(session):
    await m2m_set(session, post, Post.tags, tag_a, tag_b)

# Clear all tags
async with transaction(session):
    await m2m_set(session, post, Post.tags)

All three helpers raise TypeError if the relationship attribute is not a Many-to-Many (i.e. has no secondary table).

API Reference