`db`¶

Here's the reference for the Database facade, the transaction helper, locking functions, many-to-many helpers, and row-watching utilities.

You can import them directly from fastapi_toolsets.db:

from fastapi_toolsets.db import (
    Database,
    LockMode,
    advisory_lock,
    lock_tables,
    m2m_add,
    m2m_remove,
    m2m_set,
    transaction,
    wait_for_row_change,
)

`fastapi_toolsets.db.Database` ¶

One object that owns the engine, sessions, dependency, and middleware.

Provide exactly one of url (the facade builds and disposes the engine) or engine (an engine you own, e.g. for Alembic or event.listen, left untouched).

Parameters:

Name	Type	Description	Default
`url`	`str \| PostgresDsn \| None`	Database connection URL. Accepts a plain string or a Pydantic :class:`~pydantic.PostgresDsn`.	`None`
`engine`	`AsyncEngine \| None`	An existing :class:`AsyncEngine` to reuse instead of url.	`None`
`session_class`	`type[AsyncSession]`	Session class for the sessionmaker (e.g. `EventSession`).	`AsyncSession`
`expire_on_commit`	`bool`	Expire attributes after commit. Defaults to `False`.	`False`
`autoflush`	`bool`	Autoflush the session before queries. Defaults to `True`.	`True`
`connect_args`	`dict[str, Any] \| None`	DBAPI-level connection arguments forwarded to :func:`create_async_engine` (URL mode only).	`None`
`**engine_options`	`Any`	Extra keyword arguments forwarded to :func:`create_async_engine` (URL mode only).	`{}`

Raises:

Type	Description
`TypeError`	If neither or both of url and engine are given, or if connect_args/engine_options are passed together with engine.

Example

from fastapi import Depends, FastAPI
from fastapi_toolsets.db import Database

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

app = FastAPI()
db.install(app)

@app.get("/users/{user_id}")
async def get_user(user_id: int, session=Depends(db)):
    return await UserCrud.get(session, [User.id == user_id])

`call(request)` `async` ¶

FastAPI dependency: yield a session and commit once at the right time.

Parameters:

Name	Type	Description	Default
`request`	`Request`	The incoming request (injected by FastAPI).	required

Yields:

Type	Description
`AsyncGenerator[AsyncSession, None]`	An AsyncSession for the duration of the request.

Example

@app.get("/users/{user_id}")
async def get_user(user_id: int, session=Depends(db)):
    return await UserCrud.get(session, [User.id == user_id])

`begin()` `async` ¶

Open a session already inside a transaction (sugar for the common case).

Equivalent to session() + :func:transaction. Commits on clean exit, rolls back on exception.

Yields:

Type	Description
`AsyncGenerator[AsyncSession, None]`	An AsyncSession open within a transaction.

Example

async with db.begin() as session:
    session.add(User(name="ada"))

`install(app)` ¶

Wire the commit middleware and engine disposal onto app.

Parameters:

Name	Type	Description	Default
`app`	`Any`	The FastAPI/Starlette application to wire.	required

Example

@asynccontextmanager
async def lifespan(app):
    ...  # your startup
    yield
    ...  # your shutdown

app = FastAPI(lifespan=lifespan)
db.install(app)

`lifespan(app)` `async` ¶

Dispose the engine on shutdown; use as FastAPI(lifespan=db.lifespan).

Parameters:

Name	Type	Description	Default
`app`	`Any`	The ASGI application (unused; required by the lifespan protocol).	required

Yields:

Type	Description
`AsyncGenerator[None, None]`	Control to the application for its lifetime.

Example

app = FastAPI(lifespan=db.lifespan)

`lock_tables(tables, *, mode=LockMode.SHARE_UPDATE_EXCLUSIVE, timeout='5s')` ¶

Lock PostgreSQL tables for the duration of a dedicated transaction.

Opens its own session from the facade's sessionmaker, changes are committed when the context exits.

Parameters:

Name	Type	Description	Default
`tables`	`list[type[DeclarativeBase]]`	List of SQLAlchemy model classes to lock.	required
`mode`	`LockMode`	Lock mode (default: `SHARE UPDATE EXCLUSIVE`).	`SHARE_UPDATE_EXCLUSIVE`
`timeout`	`str`	Lock timeout (default: `"5s"`).	`'5s'`

Yields:

Type	Description
`AbstractAsyncContextManager[AsyncSession]`	The dedicated session, open within the locked transaction.

Raises:

Type	Description
`LockTimeoutError`	If the lock cannot be acquired within timeout.
`PoolExhaustedError`	If the connection pool is exhausted.

Example

async with db.lock_tables([User, Account]) as session:
    user = await UserCrud.get(session, [User.id == 1])
    user.balance += 100

`session()` `async` ¶

Open a session outside request handlers (background tasks, CLI, tests).

Commits on clean exit, rolls back on exception.

Yields:

Type	Description
`AsyncGenerator[AsyncSession, None]`	An AsyncSession ready for database operations.

Example

async with db.session() as session:
    user = await UserCrud.get(session, [User.id == 1])

`fastapi_toolsets.db.transaction(session)` `async` ¶

Run a block inside a savepoint-aware transaction.

If session is already in a transaction, a nested transaction (savepoint) is opened so the block can roll back independently. Otherwise a top-level transaction is started. Commits on clean exit, rolls back on exception.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSession`	AsyncSession instance.	required

Yields:

Type	Description
`AsyncGenerator[AsyncSession, None]`	The session within the transaction context.

Example

from fastapi_toolsets.db import transaction

async with transaction(session):
    session.add(model)

`fastapi_toolsets.db.LockMode` ¶

Bases: str, Enum

PostgreSQL table lock modes.

See: https://www.postgresql.org/docs/current/explicit-locking.html

`fastapi_toolsets.db.lock_tables(session_maker, tables, *, mode=LockMode.SHARE_UPDATE_EXCLUSIVE, timeout='5s')` ¶

Lock PostgreSQL tables for the duration of a transaction.

Prefer the method on a :class:Database instance; use this directly only when you manage your own session factory.

Parameters:

Name	Type	Description	Default
`session_maker`	`async_sessionmaker[_SessionT]`	Async session factory used to create the dedicated session.	required
`tables`	`list[type[DeclarativeBase]]`	List of SQLAlchemy model classes to lock.	required
`mode`	`LockMode`	Lock mode (default: SHARE UPDATE EXCLUSIVE).	`SHARE_UPDATE_EXCLUSIVE`
`timeout`	`str`	Lock timeout (default: "5s").	`'5s'`

Yields:

Type	Description
`AbstractAsyncContextManager[_SessionT]`	The dedicated session, open within the locked transaction.

Raises:

Type	Description
`LockTimeoutError`	If the lock cannot be acquired within timeout.
`PoolExhaustedError`	If the connection pool is exhausted.

Example

from fastapi_toolsets.db import lock_tables

async with lock_tables(session_maker, [User, Account]) as session:
    user = await UserCrud.get(session, [User.id == 1])
    user.balance += 100

`fastapi_toolsets.db.advisory_lock(session, key, *, shared=False, nowait=False, timeout=None)` `async` ¶

Acquire a PostgreSQL session-level advisory lock.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSession`	AsyncSession instance.	required
`key`	`int \| tuple[int, int]`	Lock key, either a single `int` (bigint) or a `(int, int)` pair for namespacing.	required
`shared`	`bool`	Acquire a shared lock (multiple holders allowed). Default is exclusive.	`False`
`nowait`	`bool`	Return `False` immediately if the lock is unavailable instead of waiting.	`False`
`timeout`	`str \| None`	Maximum wait time (e.g. `"5s"`, `"500ms"`). Raises `DBAPIError` if exceeded. Ignored when nowait is `True`.	`None`

Yields:

Type	Description
`AsyncGenerator[bool, None]`	`True` if the lock was acquired, `False` if nowait is `True` and the lock
`AsyncGenerator[bool, None]`	is already held.

Raises:

Type	Description
`LockTimeoutError`	If timeout is set and the lock cannot be acquired in time.

Example

from fastapi_toolsets.db import advisory_lock

async with advisory_lock(session, 42):
    ...

async with advisory_lock(session, 42, nowait=True) as acquired:
    if not acquired:
        raise HTTPException(409, "Resource is locked")

async with advisory_lock(session, 42, timeout="5s"):
    ...

async with advisory_lock(session, (1, user_id), shared=True):
    ...

`fastapi_toolsets.db.m2m_add(session, instance, rel_attr, *related, ignore_conflicts=False)` `async` ¶

Insert rows into a Many-to-Many association table without loading the ORM collection.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSession`	DB async session.	required
`instance`	`DeclarativeBase`	The "owner" side model instance (e.g. the `A` in `A.b_list`).	required
`rel_attr`	`QueryableAttribute`	The M2M relationship attribute on the model class (e.g. `A.b_list`).	required
`*related`	`DeclarativeBase`	One or more related instances to associate with `instance`.	`()`
`ignore_conflicts`	`bool`	When `True`, silently skip rows that already exist in the association table (`ON CONFLICT DO NOTHING`).	`False`

Raises:

Type	Description
`TypeError`	If `rel_attr` is not a Many-to-Many relationship.

Example

from fastapi_toolsets.db import m2m_add, transaction

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

`fastapi_toolsets.db.m2m_remove(session, instance, rel_attr, *related)` `async` ¶

Remove rows from a Many-to-Many association table without loading the ORM collection.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSession`	DB async session.	required
`instance`	`DeclarativeBase`	The "owner" side model instance (e.g. the `A` in `A.b_list`).	required
`rel_attr`	`QueryableAttribute`	The M2M relationship attribute on the model class (e.g. `A.b_list`).	required
`*related`	`DeclarativeBase`	One or more related instances to disassociate from `instance`.	`()`

Raises:

Type	Description
`TypeError`	If `rel_attr` is not a Many-to-Many relationship.

Example

from fastapi_toolsets.db import m2m_remove, transaction

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

`fastapi_toolsets.db.m2m_set(session, instance, rel_attr, *related)` `async` ¶

Replace the entire Many-to-Many association set atomically.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSession`	DB async session.	required
`instance`	`DeclarativeBase`	The "owner" side model instance (e.g. the `A` in `A.b_list`).	required
`rel_attr`	`QueryableAttribute`	The M2M relationship attribute on the model class (e.g. `A.b_list`).	required
`*related`	`DeclarativeBase`	The new complete set of related instances.	`()`

Raises:

Type	Description
`TypeError`	If `rel_attr` is not a Many-to-Many relationship.

Example

from fastapi_toolsets.db import m2m_set, transaction

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

`fastapi_toolsets.db.wait_for_row_change(session, model, pk_value, *, columns=None, interval=0.5, timeout=None)` `async` ¶

Poll a database row until a change is detected.

Queries the row every interval seconds and returns the model instance once a change is detected in any column (or only the specified columns).

Parameters:

Name	Type	Description	Default
`session`	`AsyncSession`	AsyncSession instance.	required
`model`	`type[_M]`	SQLAlchemy model class.	required
`pk_value`	`Any`	Primary key value of the row to watch.	required
`columns`	`list[str] \| None`	Optional list of column names to watch. If None, all columns are watched.	`None`
`interval`	`float`	Polling interval in seconds (default: 0.5).	`0.5`
`timeout`	`float \| None`	Maximum time to wait in seconds. None means wait forever.	`None`

Returns:

Type	Description
`_M`	The refreshed model instance with updated values.

Raises:

Type	Description
`NotFoundError`	If the row does not exist or is deleted during polling.
`TimeoutError`	If timeout expires before a change is detected.

Example

from fastapi_toolsets.db import wait_for_row_change

# Wait for any column to change
updated = await wait_for_row_change(session, User, user_id)

# Watch specific columns with a timeout
updated = await wait_for_row_change(
    session, User, user_id,
    columns=["status", "email"],
    interval=1.0,
    timeout=30.0,
)

Admin and test helpers live in fastapi_toolsets.db.testing:

from fastapi_toolsets.db.testing import cleanup_tables, create_database

`fastapi_toolsets.db.testing.create_database(db_name, *, server_url)` `async` ¶

Create a database.

Connects to server_url using AUTOCOMMIT isolation and issues a CREATE DATABASE statement for db_name.

Parameters:

Name	Type	Description	Default
`db_name`	`str`	Name of the database to create.	required
`server_url`	`str`	URL used for server-level DDL (must point to an existing database on the same server).	required

Example

from fastapi_toolsets.db.testing import create_database

SERVER_URL = "postgresql+asyncpg://postgres:postgres@localhost/postgres"
await create_database("myapp_test", server_url=SERVER_URL)

`fastapi_toolsets.db.testing.cleanup_tables(session, base)` `async` ¶

Truncate all tables for fast between-test cleanup.

Executes a single TRUNCATE … RESTART IDENTITY CASCADE statement across every table in base's metadata.

This is a no-op when the metadata contains no tables.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSession`	An active async database session.	required
`base`	`type[DeclarativeBase]`	SQLAlchemy DeclarativeBase class containing model metadata.	required

Example

@pytest.fixture
async def db_session(worker_db_url):
    async with create_db_session(worker_db_url, Base) as session:
        yield session
        await cleanup_tables(session, Base)

db¶

fastapi_toolsets.db.Database ¶

__call__(request) async ¶

begin() async ¶

install(app) ¶

lifespan(app) async ¶

lock_tables(tables, *, mode=LockMode.SHARE_UPDATE_EXCLUSIVE, timeout='5s') ¶

session() async ¶

fastapi_toolsets.db.transaction(session) async ¶

fastapi_toolsets.db.LockMode ¶

fastapi_toolsets.db.lock_tables(session_maker, tables, *, mode=LockMode.SHARE_UPDATE_EXCLUSIVE, timeout='5s') ¶

fastapi_toolsets.db.advisory_lock(session, key, *, shared=False, nowait=False, timeout=None) async ¶

fastapi_toolsets.db.m2m_add(session, instance, rel_attr, *related, ignore_conflicts=False) async ¶

fastapi_toolsets.db.m2m_remove(session, instance, rel_attr, *related) async ¶

fastapi_toolsets.db.m2m_set(session, instance, rel_attr, *related) async ¶

fastapi_toolsets.db.wait_for_row_change(session, model, pk_value, *, columns=None, interval=0.5, timeout=None) async ¶

fastapi_toolsets.db.testing.create_database(db_name, *, server_url) async ¶

fastapi_toolsets.db.testing.cleanup_tables(session, base) async ¶

`db`¶

`fastapi_toolsets.db.Database` ¶

`call(request)` `async` ¶

`begin()` `async` ¶

`install(app)` ¶

`lifespan(app)` `async` ¶

`lock_tables(tables, *, mode=LockMode.SHARE_UPDATE_EXCLUSIVE, timeout='5s')` ¶

`session()` `async` ¶

`fastapi_toolsets.db.transaction(session)` `async` ¶

`fastapi_toolsets.db.LockMode` ¶

`fastapi_toolsets.db.lock_tables(session_maker, tables, *, mode=LockMode.SHARE_UPDATE_EXCLUSIVE, timeout='5s')` ¶

`fastapi_toolsets.db.advisory_lock(session, key, *, shared=False, nowait=False, timeout=None)` `async` ¶

`fastapi_toolsets.db.m2m_add(session, instance, rel_attr, *related, ignore_conflicts=False)` `async` ¶

`fastapi_toolsets.db.m2m_remove(session, instance, rel_attr, *related)` `async` ¶

`fastapi_toolsets.db.m2m_set(session, instance, rel_attr, *related)` `async` ¶

`fastapi_toolsets.db.wait_for_row_change(session, model, pk_value, *, columns=None, interval=0.5, timeout=None)` `async` ¶

`fastapi_toolsets.db.testing.create_database(db_name, *, server_url)` `async` ¶

`fastapi_toolsets.db.testing.cleanup_tables(session, base)` `async` ¶