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"}},
)

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. With the middleware installed, the dependency does not commit again.

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.

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. It opens a dedicated session internally and releases the lock when the context exits:

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:
    ...

Advisory locking¶

advisory_lock acquires a PostgreSQL session-level advisory lock on a session you provide. The lock is 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)):
    ...

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