fastapi-toolsets/docs/module/db.md

# DB

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

!!! info
    This module has been coded and tested to be compatible with PostgreSQL only.

## Overview

The `db` module is built around one object, [`Database`](../reference/db.md#fastapi_toolsets.db.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`.

```python
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`](https://docs.pydantic.dev/latest/api/networks/#pydantic.networks.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`):

```python
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)`](../reference/db.md#fastapi_toolsets.db.Database) 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.

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

!!! warning "Streaming / SSE endpoints"
    For a `StreamingResponse` / `EventSourceResponse`, the commit fires at the **start** of the stream. A stream that **writes** must open a short-lived session per write with [`db.session()`](#session-context-manager); the start-time commit will not flush writes made later during the stream.

## Lifespan

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

```python
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`](../reference/db.md#fastapi_toolsets.db.Database) 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()`](../reference/db.md#fastapi_toolsets.db.Database) for sessions outside request handlers (e.g. background tasks, CLI commands). It commits on clean exit and rolls back on exception:

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

## Transactions

[`transaction`](../reference/db.md#fastapi_toolsets.db.transaction) opens a transaction on a session, using a savepoint when one is already open so it nests safely:

```python
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()`](../reference/db.md#fastapi_toolsets.db.Database) opens a session already inside a transaction:

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

## Table locking

[`db.lock_tables`](../reference/db.md#fastapi_toolsets.db.Database) acquires PostgreSQL table-level locks for a critical section. It opens a dedicated session internally and releases the lock when the context exits:

```python
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`](../reference/db.md#fastapi_toolsets.db.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`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.LockTimeoutError) is raised instead of a raw database error:

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

## Advisory locking

[`advisory_lock`](../reference/db.md#fastapi_toolsets.db.advisory_lock) acquires a PostgreSQL session-level advisory lock on a session you provide. The lock is released when the context exits:

```python
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)):
    ...
```

!!! note
    Advisory locks use PostgreSQL session-level functions (`pg_advisory_lock` / `pg_advisory_unlock`). The lock is tied to the database connection, not the SQLAlchemy transaction, so it is released when the context exits even if the surrounding transaction is still open.

## Row-change polling

[`wait_for_row_change`](../reference/db.md#fastapi_toolsets.db.wait_for_row_change) polls a row until a specific column changes value:

```python
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`](../reference/db.md#fastapi_toolsets.db.testing.create_database) (in `fastapi_toolsets.db.testing`) connects to *server_url* and issues a `CREATE DATABASE` statement:

```python
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`](../reference/pytest.md#fastapi_toolsets.pytest.utils.create_worker_database) from the `pytest` module, which handles drop-before, create, and drop-after.

## Cleaning up tables

[`cleanup_tables`](../reference/db.md#fastapi_toolsets.db.testing.cleanup_tables) (in `fastapi_toolsets.db.testing`) truncates all tables:

```python
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`](../reference/db.md#fastapi_toolsets.db.m2m_add) inserts one or more rows into a secondary table:

```python
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:

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

### `m2m_remove`: delete associations

[`m2m_remove`](../reference/db.md#fastapi_toolsets.db.m2m_remove) deletes specific association rows. Removing a non-existent association is a no-op:

```python
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`](../reference/db.md#fastapi_toolsets.db.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:

```python
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).

---

[:material-api: API Reference](../reference/db.md)