Version 5.0.0-beta2

* feat: fixture refresh DB-generated values onto returned instances

* refactor: replace utils.get_obj_by_attr/get_field_by_attr with registry.obj/field lookups

* refactor: log fixture command output instead of print

* chore: clean up fixture module

docs/module/crud.md

@@ -167,7 +167,7 @@ user = await UserCrud.update(session, UserUpdate(credits=10), [User.id == user_i
 ```
 
 !!! warning
-    `with_for_update` requires an open transaction. Wrap your call in `async with session.begin()` or use the `get_transaction` helper if you are not already inside one.
+    `with_for_update` requires an open transaction. Wrap your call in `async with session.begin()` or use the `transaction` helper if you are not already inside one.
 
 !!! note
     `NOWAIT` raises `sqlalchemy.exc.OperationalError` immediately if the row is locked rather than waiting.

docs/module/db.md

@@ -7,96 +7,156 @@ SQLAlchemy async session management with transactions, table locking, advisory l
 
 ## Overview
 
-The `db` module provides helpers to create FastAPI dependencies and context managers for `AsyncSession`, along with utilities for nested transactions, table locks, advisory locks, and polling for row changes.
+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.
 
-## Session dependency
+## Setup
 
-Use [`create_db_dependency`](../reference/db.md#fastapi_toolsets.db.create_db_dependency) to create a FastAPI dependency that yields a session and auto-commits on success:
+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 sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
-from fastapi_toolsets.db import create_db_dependency
+from fastapi import Depends, FastAPI
+from sqlalchemy.ext.asyncio import AsyncSession
 
-engine = create_async_engine(url="postgresql+asyncpg://...", future=True)
-session_maker = async_sessionmaker(bind=engine, expire_on_commit=False)
+from fastapi_toolsets.db import Database
 
-get_db = create_db_dependency(session_maker=session_maker)
+db = Database("postgresql+asyncpg://postgres:postgres@localhost/app")
 
-@router.get("/users")
-async def list_users(session: AsyncSession = Depends(get_db)):
+app = FastAPI()
+db.install(app)  # commit middleware + engine disposal on shutdown
+
+@app.get("/users")
+async def list_users(session: AsyncSession = Depends(db)):
     ...
 ```
 
-## Session context manager
+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).
 
-Use [`create_db_context`](../reference/db.md#fastapi_toolsets.db.create_db_context) for sessions outside request handlers (e.g. background tasks, CLI commands):
+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 fastapi_toolsets.db import create_db_context
+from pydantic_settings import BaseSettings
+from pydantic import PostgresDsn
 
-db_context = create_db_context(session_maker=session_maker)
+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. 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.
+
+!!! 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_context() as session:
+    async with db.session() as session:
         ...
 ```
 
-## Nested transactions
+## Transactions
 
-[`get_transaction`](../reference/db.md#fastapi_toolsets.db.get_transaction) handles savepoints automatically, allowing safe nesting:
+[`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 get_transaction
+from fastapi_toolsets.db import transaction
 
-async def create_user_with_role(session=session):
-    async with get_transaction(session=session):
+async def create_user_with_role(session):
+    async with transaction(session):
         ...
-        async with get_transaction(session=session):  # uses savepoint
+        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
 
-[`lock_tables`](../reference/db.md#fastapi_toolsets.db.lock_tables) acquires PostgreSQL table-level locks before executing critical sections. It opens a **dedicated session** internally and yields it to the caller, so the lock is guaranteed to be released when the context exits:
+[`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 lock_tables, LockMode
+from fastapi_toolsets.db import LockMode
 
-async with lock_tables(session_maker=session_maker, tables=[User], mode=LockMode.EXCLUSIVE) as session:
+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 before giving up. On timeout, a [`LockTimeoutError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.LockTimeoutError) is raised instead of a raw database error:
+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 lock_tables(session_maker, [Order], timeout="2s") as session:
+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. The lock is released explicitly when the context exits, regardless of whether the transaction has committed.
+[`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
+# 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
+# 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
+# Blocking with a timeout: raises LockTimeoutError if not acquired in time
 async with advisory_lock(session=session, key=42, timeout="5s"):
     ...
 
-# Shared — multiple readers allowed simultaneously, blocks exclusive writers
+# Shared lock: multiple readers allowed simultaneously, blocks exclusive writers
 async with advisory_lock(session=session, key=42, shared=True):
     ...
 
@@ -106,11 +166,11 @@ 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 — it is released when the context exits, even if the surrounding transaction is still open.
+    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, useful for waiting on async side effects:
+[`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
@@ -120,7 +180,7 @@ await wait_for_row_change(
     session=session,
     model=Order,
     pk_value=order_id,
-    columns=[Order.status],
+    columns=["status"],
     interval=1.0,
     timeout=30.0,
 )
@@ -128,28 +188,24 @@ await wait_for_row_change(
 
 ## Creating a database
 
-!!! info "Added in `v2.1`"
-
-[`create_database`](../reference/db.md#fastapi_toolsets.db.create_database) creates a database at a given URL. It connects to *server_url* and issues a `CREATE DATABASE` statement:
+[`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 import create_database
+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 instead — it handles drop-before, create, and drop-after automatically.
+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
 
-!!! info "Added in `v2.1`"
-
-[`cleanup_tables`](../reference/db.md#fastapi_toolsets.db.cleanup_tables) truncates all 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 import cleanup_tables
+from fastapi_toolsets.db.testing import cleanup_tables
 
 @pytest.fixture(autouse=True)
 async def clean(db_session):
@@ -159,50 +215,50 @@ async def clean(db_session):
 
 ## Many-to-Many helpers
 
-SQLAlchemy's ORM collection API triggers lazy-loads when you append to a relationship inside a savepoint (e.g. inside `lock_tables` or a nested `get_transaction`). The three `m2m_*` helpers bypass the ORM collection entirely and issue direct SQL against the association table.
+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`: insert associations
 
-[`m2m_add`](../reference/db.md#fastapi_toolsets.db.m2m_add) inserts one or more rows into a secondary table without touching the ORM collection:
+[`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 lock_tables, m2m_add
+from fastapi_toolsets.db import m2m_add
 
-async with lock_tables(session_maker, [Tag]) as session:
+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 silently skip associations that already exist:
+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`: 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 get_transaction, m2m_remove
+from fastapi_toolsets.db import m2m_remove, transaction
 
-async with get_transaction(session):
+async with transaction(session):
     await m2m_remove(session, post, Post.tags, tag1, tag2)
 ```
 
-### `m2m_set` — replace the full set
+### `m2m_set`: replace the full set
 
-[`m2m_set`](../reference/db.md#fastapi_toolsets.db.m2m_set) atomically replaces all associations: it deletes every existing row for the owner instance then inserts the new set. Passing no related instances clears the association entirely:
+[`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 get_transaction, m2m_set
+from fastapi_toolsets.db import m2m_set, transaction
 
 # Replace all tags
-async with get_transaction(session):
+async with transaction(session):
     await m2m_set(session, post, Post.tags, tag_a, tag_b)
 
 # Clear all tags
-async with get_transaction(session):
+async with transaction(session):
     await m2m_set(session, post, Post.tags)
 ```
 

docs/module/fixtures.md

@@ -65,6 +65,13 @@ Both functions return a `dict[str, list[...]]` mapping each fixture name to the
 
 A fixture with no `contexts` defined takes `Context.BASE` by default.
 
+`Context.BASE` fixtures are always included alongside whatever context you load or list — there's no way to load a non-base context in isolation:
+
+```python
+# also loads any Context.BASE fixtures, even though only TESTING is requested
+await load_fixtures_by_context(session, fixtures, Context.TESTING)
+```
+
 ### Custom contexts
 
 Plain strings and any `Enum` subclass are accepted wherever a `Context` enum is expected.
@@ -80,6 +87,7 @@ class AppContext(str, Enum):
 def staging_data():
     return [Config(key="feature_x", enabled=True)]
 
+# loads staging_data plus any Context.BASE fixtures
 await load_fixtures_by_context(session, fixtures, AppContext.STAGING)
 ```
 
@@ -108,8 +116,8 @@ def users():
 def users():
     return [User(id=2, username="tester")]
 
-# loads both admin and tester
-await load_fixtures_by_context(session, fixtures, Context.BASE, Context.TESTING)
+# loads both admin and tester (Context.BASE is included automatically)
+await load_fixtures_by_context(session, fixtures, Context.TESTING)
 ```
 
 Registering two variants with overlapping context sets raises `ValueError`.
@@ -147,18 +155,26 @@ Fixtures with the same name are allowed as long as their context sets do not ove
 
 ## Looking up fixture instances
 
-[`get_obj_by_attr`](../reference/fixtures.md#fastapi_toolsets.fixtures.utils.get_obj_by_attr) retrieves a specific instance from a fixture function by attribute value — useful when building cross-fixture `depends_on` relationships:
+[`FixtureRegistry.obj`](../reference/fixtures.md#fastapi_toolsets.fixtures.registry.FixtureRegistry.obj) retrieves a specific instance from a registered fixture by attribute value, looked up by name on the registry — useful when building cross-fixture `depends_on` relationships:
 
 ```python
-from fastapi_toolsets.fixtures import get_obj_by_attr
-
 @fixtures.register(depends_on=["roles"])
 def users():
-    admin_role = get_obj_by_attr(roles, "name", "admin")
+    admin_role = fixtures.obj("roles", "name", "admin")
     return [User(id=1, username="alice", role_id=admin_role.id)]
 ```
 
-Raises `StopIteration` if no matching instance is found.
+Looking the fixture up by name (instead of importing the `roles` function directly) means fixture modules never need to import each other, which avoids circular imports in larger projects split across multiple files — the same reason `depends_on` takes fixture names rather than the functions themselves. The registry passed in must be the one that actually contains the fixture by load time; with a single shared registry this is automatic, but if you merge registries with `include_registry`, call `obj`/`field` on the merged registry.
+
+[`FixtureRegistry.field`](../reference/fixtures.md#fastapi_toolsets.fixtures.registry.FixtureRegistry.field) is shorthand for pulling a single attribute (`id` by default):
+
+```python
+@fixtures.register(depends_on=["roles"])
+def users():
+    return [User(id=1, username="alice", role_id=fixtures.field("roles", "name", "admin"))]
+```
+
+Both raise `StopIteration` if no matching instance is found, and `KeyError` if the fixture name isn't registered.
 
 ## Pytest integration
 

docs/module/models.md

@@ -134,7 +134,7 @@ SessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=EventSe
 ```
 
 !!! info "Callbacks fire on `session.commit()` only — not on savepoints."
-    Savepoints created by [`get_transaction`](db.md) or `begin_nested()` do **not**
+    Savepoints created by [`transaction`](db.md) or `begin_nested()` do **not**
     trigger callbacks. All events accumulated across flushes are dispatched once
     when the outermost `commit()` is called.
 

docs/module/pytest.md

@@ -107,10 +107,10 @@ url = worker_database_url("postgresql+asyncpg://user:pass@localhost/myapp", defa
 
 ## Manual table cleanup
 
-[`cleanup_tables`](../reference/db.md#fastapi_toolsets.db.cleanup_tables) truncates all tables in a single statement and can be called directly when you need more control:
+[`cleanup_tables`](../reference/db.md#fastapi_toolsets.db.testing.cleanup_tables) truncates all tables in a single statement and can be called directly when you need more control:
 
 ```python
-from fastapi_toolsets.db import cleanup_tables
+from fastapi_toolsets.pytest import cleanup_tables
 
 @pytest.fixture(autouse=True)
 async def clean(db_session):

docs/reference/db.md

@@ -1,46 +1,48 @@
 # `db`
 
-Here's the reference for all database session utilities, transaction helpers, and locking functions.
+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`:
 
 ```python
 from fastapi_toolsets.db import (
+    Database,
     LockMode,
     advisory_lock,
-    cleanup_tables,
-    create_database,
-    create_db_dependency,
-    create_db_context,
-    get_transaction,
     lock_tables,
     m2m_add,
     m2m_remove,
     m2m_set,
+    transaction,
     wait_for_row_change,
 )
 ```
 
+## ::: fastapi_toolsets.db.Database
+
+## ::: fastapi_toolsets.db.transaction
+
 ## ::: fastapi_toolsets.db.LockMode
 
-## ::: fastapi_toolsets.db.create_db_dependency
-
-## ::: fastapi_toolsets.db.create_db_context
-
-## ::: fastapi_toolsets.db.get_transaction
-
 ## ::: fastapi_toolsets.db.lock_tables
 
 ## ::: fastapi_toolsets.db.advisory_lock
 
-## ::: fastapi_toolsets.db.wait_for_row_change
-
-## ::: fastapi_toolsets.db.create_database
-
-## ::: fastapi_toolsets.db.cleanup_tables
-
 ## ::: fastapi_toolsets.db.m2m_add
 
 ## ::: fastapi_toolsets.db.m2m_remove
 
 ## ::: fastapi_toolsets.db.m2m_set
+
+## ::: fastapi_toolsets.db.wait_for_row_change
+
+Admin and test helpers live in `fastapi_toolsets.db.testing`:
+
+```python
+from fastapi_toolsets.db.testing import cleanup_tables, create_database
+```
+
+## ::: fastapi_toolsets.db.testing.create_database
+
+## ::: fastapi_toolsets.db.testing.cleanup_tables

docs/reference/fixtures.md

@@ -12,7 +12,6 @@ from fastapi_toolsets.fixtures import (
     FixtureRegistry,
     load_fixtures,
     load_fixtures_by_context,
-    get_obj_by_attr,
 )
 ```
 
@@ -27,5 +26,3 @@ from fastapi_toolsets.fixtures import (
 ## ::: fastapi_toolsets.fixtures.utils.load_fixtures
 
 ## ::: fastapi_toolsets.fixtures.utils.load_fixtures_by_context
-
-## ::: fastapi_toolsets.fixtures.utils.get_obj_by_attr

docs_src/examples/pagination_search/app.py

@@ -2,8 +2,10 @@ from fastapi import FastAPI
 
 from fastapi_toolsets.exceptions import init_exceptions_handlers
 
+from .db import db
 from .routes import router
 
 app = FastAPI()
+db.install(app=app)
 init_exceptions_handlers(app=app)
 app.include_router(router=router)

docs_src/examples/pagination_search/db.py

@@ -1,17 +1,14 @@
 from typing import Annotated
 
 from fastapi import Depends
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.ext.asyncio import AsyncSession
 
-from fastapi_toolsets.db import create_db_context, create_db_dependency
+from fastapi_toolsets.db import Database
 
 DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/postgres"
 
-engine = create_async_engine(url=DATABASE_URL, future=True)
-async_session_maker = async_sessionmaker(bind=engine, expire_on_commit=False)
+db = Database(url=DATABASE_URL)
 
-get_db = create_db_dependency(session_maker=async_session_maker)
-get_db_context = create_db_context(session_maker=async_session_maker)
+get_db = db
 
-
-SessionDep = Annotated[AsyncSession, Depends(get_db)]
+SessionDep = Annotated[AsyncSession, Depends(db)]

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fastapi-toolsets"
-version = "4.1.3"
+version = "5.0.0b2"
 description = "Production-ready utilities for FastAPI applications"
 readme = "README.md"
 license = "MIT"

src/fastapi_toolsets/__init__.py

@@ -7,18 +7,21 @@ Example usage:
     from fastapi import FastAPI, Depends
     from fastapi_toolsets.exceptions import init_exceptions_handlers
     from fastapi_toolsets.crud import CrudFactory
-    from fastapi_toolsets.db import create_db_dependency
+    from fastapi_toolsets.db import Database
     from fastapi_toolsets.schemas import Response
 
+    db = Database("postgresql+asyncpg://postgres:postgres@localhost/app")
+
     app = FastAPI()
+    db.install(app)
     init_exceptions_handlers(app)
 
     UserCrud = CrudFactory(User)
 
     @app.get("/users/{user_id}", response_model=Response[dict])
-    async def get_user(user_id: int, session = Depends(get_db)):
+    async def get_user(user_id: int, session = Depends(db)):
         user = await UserCrud.get(session, [User.id == user_id])
         return Response(data={"user": user.username}, message="Success")
 """
 
-__version__ = "4.1.3"
+__version__ = "5.0.0b2"

src/fastapi_toolsets/cli/commands/fixtures.py

@@ -6,7 +6,8 @@ import typer
 from rich.console import Console
 from rich.table import Table
 
-from ...fixtures import Context, LoadStrategy, load_fixtures_by_context
+from ...fixtures import Context, LoadStrategy
+from ...logger import get_logger
 from ..config import get_db_context, get_fixtures_registry
 from ..utils import async_command
 
@@ -16,13 +17,14 @@ fixture_cli = typer.Typer(
     no_args_is_help=True,
 )
 console = Console()
+logger = get_logger()
 
 
 @fixture_cli.command("list")
 def list_fixtures(
     ctx: typer.Context,
     context: Annotated[
-        Context | None,
+        str | None,
         typer.Option(
             "--context",
             "-c",
@@ -32,10 +34,10 @@ def list_fixtures(
 ) -> None:
     """List all registered fixtures."""
     registry = get_fixtures_registry()
-    fixtures = registry.get_by_context(context.value) if context else registry.get_all()
+    fixtures = registry.get_by_context(context) if context else registry.get_all()
 
     if not fixtures:
-        print("No fixtures found.")
+        logger.info("No fixtures found.")
         return
 
     table = Table("Name", "Contexts", "Dependencies")
@@ -46,7 +48,7 @@ def list_fixtures(
         table.add_row(fixture.name, contexts, deps)
 
     console.print(table)
-    print(f"\nTotal: {len(fixtures)} fixture(s)")
+    logger.info("Total: %d fixture(s)", len(fixtures))
 
 
 @fixture_cli.command("load")
@@ -54,7 +56,7 @@ def list_fixtures(
 async def load(
     ctx: typer.Context,
     contexts: Annotated[
-        list[Context] | None,
+        list[str] | None,
         typer.Argument(help="Contexts to load."),
     ] = None,
     strategy: Annotated[
@@ -69,26 +71,27 @@ async def load(
     ] = False,
 ) -> None:
     """Load fixtures into the database."""
+    from ...fixtures import load_fixtures_by_context
+
     registry = get_fixtures_registry()
     db_context = get_db_context()
 
-    context_list = list(contexts) if contexts else [Context.BASE]
+    context_list = contexts or [Context.BASE.value]
 
     ordered = registry.resolve_context_dependencies(*context_list)
 
     if not ordered:
-        print("No fixtures to load for the specified context(s).")
+        logger.info("No fixtures to load for the specified context(s).")
         return
 
-    print(f"\nFixtures to load ({strategy.value} strategy):")
-    for name in ordered:
-        fixture = registry.get(name)
-        instances = list(fixture.func())
-        model_name = type(instances[0]).__name__ if instances else "?"
-        print(f"  - {name}: {len(instances)} {model_name}(s)")
-
     if dry_run:
-        print("\n[Dry run - no changes made]")
+        logger.info("Fixtures to load (%s strategy):", strategy.value)
+        for name in ordered:
+            variants = registry.get_load_variants(name, *context_list)
+            instances = [inst for v in variants for inst in v.func()]
+            model_name = type(instances[0]).__name__ if instances else "?"
+            logger.info("  - %s: %d %s(s)", name, len(instances), model_name)
+        logger.info("[Dry run - no changes made]")
         return
 
     async with db_context() as session:
@@ -97,4 +100,4 @@ async def load(
         )
 
     total = sum(len(items) for items in result.values())
-    print(f"\nLoaded {total} record(s) successfully.")
+    logger.info("Loaded %d record(s) successfully.", total)

src/fastapi_toolsets/cli/config.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import importlib
 import sys
-from typing import TYPE_CHECKING, Any, Literal, overload
+from typing import TYPE_CHECKING, Any, Literal, TypeVar, overload
 
 import typer
 
@@ -13,6 +13,8 @@ from .pyproject import find_pyproject, load_pyproject
 if TYPE_CHECKING:
     from ..fixtures import FixtureRegistry
 
+T = TypeVar("T")
+
 
 def _ensure_project_in_path():
     """Add project root to sys.path if not installed in editable mode."""
@@ -88,19 +90,39 @@ def get_config_value(key: str, required: bool = False) -> Any | None:
     return value
 
 
+@overload
+def _import_typed(
+    key: str, expected_type: type[T], *, required: Literal[True]
+) -> T: ...  # pragma: no cover
+@overload
+def _import_typed(
+    key: str, expected_type: type[T], *, required: bool
+) -> T | None: ...  # pragma: no cover
+def _import_typed(key: str, expected_type: type[T], *, required: bool) -> T | None:
+    """Import a config value by key and validate its type.
+
+    Raises:
+        typer.BadParameter: If required and missing, or if the imported
+            value isn't an instance of *expected_type*.
+    """
+    import_path = get_config_value(key, required=required)
+    if not import_path:
+        return None
+
+    obj = import_from_string(import_path)
+    if not isinstance(obj, expected_type):
+        raise typer.BadParameter(
+            f"'{key}' must be a {expected_type.__name__} instance, got {type(obj).__name__}"
+        )
+
+    return obj
+
+
 def get_fixtures_registry() -> FixtureRegistry:
     """Import and return the fixtures registry from config."""
     from ..fixtures import FixtureRegistry
 
-    import_path = get_config_value("fixtures", required=True)
-    registry = import_from_string(import_path)
-
-    if not isinstance(registry, FixtureRegistry):
-        raise typer.BadParameter(
-            f"'fixtures' must be a FixtureRegistry instance, got {type(registry).__name__}"
-        )
-
-    return registry
+    return _import_typed("fixtures", FixtureRegistry, required=True)
 
 
 def get_db_context() -> Any:
@@ -111,15 +133,4 @@ def get_db_context() -> Any:
 
 def get_custom_cli() -> typer.Typer | None:
     """Import and return the custom CLI Typer instance from config."""
-    import_path = get_config_value("custom_cli")
-    if not import_path:
-        return None
-
-    custom = import_from_string(import_path)
-
-    if not isinstance(custom, typer.Typer):
-        raise typer.BadParameter(
-            f"'custom_cli' must be a Typer instance, got {type(custom).__name__}"
-        )
-
-    return custom
+    return _import_typed("custom_cli", typer.Typer, required=False)

src/fastapi_toolsets/cli/utils.py

@@ -1,6 +1,5 @@
 """CLI utility functions."""
 
-import asyncio
 import functools
 from collections.abc import Callable, Coroutine
 from typing import Any, ParamSpec, TypeVar
@@ -24,6 +23,8 @@ def async_command(func: Callable[P, Coroutine[Any, Any, T]]) -> Callable[P, T]:
 
     @functools.wraps(func)
     def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
+        import asyncio
+
         return asyncio.run(func(*args, **kwargs))
 
     return wrapper

src/fastapi_toolsets/crud/factory.py

@@ -22,7 +22,7 @@ from sqlalchemy.orm import DeclarativeBase, QueryableAttribute, selectinload
 from sqlalchemy.sql.base import ExecutableOption
 from sqlalchemy.sql.roles import WhereHavingRole
 
-from ..db import get_transaction
+from ..db import transaction
 from ..exceptions import InvalidOrderFieldError, NotFoundError
 from ..schemas import (
     CursorPaginatedResponse,
@@ -716,7 +716,7 @@ class AsyncCrud(Generic[ModelType]):
         Returns:
             Created model instance, or ``Response[schema]`` when ``schema`` is given.
         """
-        async with get_transaction(session):
+        async with transaction(session):
             m2m_exclude = cls._m2m_schema_fields()
             data = (
                 obj.model_dump(exclude=m2m_exclude) if m2m_exclude else obj.model_dump()
@@ -1067,7 +1067,7 @@ class AsyncCrud(Generic[ModelType]):
         Raises:
             NotFoundError: If no record found
         """
-        async with get_transaction(session):
+        async with transaction(session):
             m2m_exclude = cls._m2m_schema_fields()
 
             # Eagerly load M2M relationships that will be updated so that
@@ -1127,7 +1127,7 @@ class AsyncCrud(Generic[ModelType]):
         Returns:
             Model instance
         """
-        async with get_transaction(session):
+        async with transaction(session):
             values = obj.model_dump(exclude_unset=True)
             q = insert(cls.model).values(**values)
             if set_:
@@ -1189,7 +1189,7 @@ class AsyncCrud(Generic[ModelType]):
         Returns:
             ``None``, or ``Response[None]`` when ``return_response=True``.
         """
-        async with get_transaction(session):
+        async with transaction(session):
             result = await session.execute(select(cls.model).where(and_(*filters)))
             objects = result.scalars().all()
             for obj in objects:

src/fastapi_toolsets/db.py

@@ -1,591 +0,0 @@
-"""Database utilities: sessions, transactions, and locks."""
-
-import asyncio
-from collections.abc import AsyncGenerator, Callable
-from contextlib import AbstractAsyncContextManager, asynccontextmanager
-from enum import Enum
-from typing import Any, TypeVar, cast
-
-import asyncpg
-from sqlalchemy import Table, delete, text, tuple_
-from sqlalchemy import exc as sa_exc
-from sqlalchemy.dialects.postgresql import insert as pg_insert
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
-from sqlalchemy.orm import DeclarativeBase, QueryableAttribute
-from sqlalchemy.orm.relationships import RelationshipProperty
-
-from .exceptions import LockTimeoutError, NotFoundError, PoolExhaustedError
-
-
-def _is_lock_not_available(e: sa_exc.DBAPIError) -> bool:
-    return e.orig is not None and isinstance(
-        e.orig.__cause__, asyncpg.exceptions.LockNotAvailableError
-    )
-
-
-__all__ = [
-    "LockMode",
-    "advisory_lock",
-    "cleanup_tables",
-    "create_database",
-    "create_db_context",
-    "create_db_dependency",
-    "get_transaction",
-    "lock_tables",
-    "m2m_add",
-    "m2m_remove",
-    "m2m_set",
-    "wait_for_row_change",
-]
-
-
-_SessionT = TypeVar("_SessionT", bound=AsyncSession)
-
-
-def create_db_dependency(
-    session_maker: async_sessionmaker[_SessionT],
-) -> Callable[[], AsyncGenerator[_SessionT, None]]:
-    """Create a FastAPI dependency for database sessions.
-
-    Creates a dependency function that yields a session and auto-commits
-    if a transaction is active when the request completes.
-
-    Args:
-        session_maker: Async session factory from create_session_factory()
-
-    Returns:
-        An async generator function usable with FastAPI's Depends()
-
-    Example:
-        ```python
-        from fastapi import Depends
-        from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
-        from fastapi_toolsets.db import create_db_dependency
-
-        engine = create_async_engine("postgresql+asyncpg://...")
-        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
-        get_db = create_db_dependency(SessionLocal)
-
-        @app.get("/users")
-        async def list_users(session: AsyncSession = Depends(get_db)):
-            ...
-        ```
-    """
-
-    async def get_db() -> AsyncGenerator[_SessionT, None]:
-        async with session_maker() as session:
-            try:
-                await session.connection()
-            except sa_exc.TimeoutError as e:
-                raise PoolExhaustedError() from e
-            yield session
-            if session.in_transaction():
-                await session.commit()
-
-    return get_db
-
-
-def create_db_context(
-    session_maker: async_sessionmaker[_SessionT],
-) -> Callable[[], AbstractAsyncContextManager[_SessionT]]:
-    """Create a context manager for database sessions.
-
-    Creates a context manager for use outside of FastAPI request handlers,
-    such as in background tasks, CLI commands, or tests.
-
-    Args:
-        session_maker: Async session factory from create_session_factory()
-
-    Returns:
-        An async context manager function
-
-    Example:
-        ```python
-        from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
-        from fastapi_toolsets.db import create_db_context
-
-        engine = create_async_engine("postgresql+asyncpg://...")
-        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
-        get_db_context = create_db_context(SessionLocal)
-
-        async def background_task():
-            async with get_db_context() as session:
-                user = await UserCrud.get(session, [User.id == 1])
-                ...
-        ```
-    """
-    get_db = create_db_dependency(session_maker)
-    return asynccontextmanager(get_db)
-
-
-@asynccontextmanager
-async def get_transaction(
-    session: AsyncSession,
-) -> AsyncGenerator[AsyncSession, None]:
-    """Get a transaction context, handling nested transactions.
-
-    If already in a transaction, creates a savepoint (nested transaction).
-    Otherwise, starts a new transaction.
-
-    Args:
-        session: AsyncSession instance
-
-    Yields:
-        The session within the transaction context
-
-    Example:
-        ```python
-        async with get_transaction(session):
-            session.add(model)
-            # Auto-commits on exit, rolls back on exception
-        ```
-    """
-    if session.in_transaction():
-        async with session.begin_nested():
-            yield session
-    else:
-        async with session.begin():
-            yield session
-
-
-class LockMode(str, Enum):
-    """PostgreSQL table lock modes.
-
-    See: https://www.postgresql.org/docs/current/explicit-locking.html
-    """
-
-    ACCESS_SHARE = "ACCESS SHARE"
-    ROW_SHARE = "ROW SHARE"
-    ROW_EXCLUSIVE = "ROW EXCLUSIVE"
-    SHARE_UPDATE_EXCLUSIVE = "SHARE UPDATE EXCLUSIVE"
-    SHARE = "SHARE"
-    SHARE_ROW_EXCLUSIVE = "SHARE ROW EXCLUSIVE"
-    EXCLUSIVE = "EXCLUSIVE"
-    ACCESS_EXCLUSIVE = "ACCESS EXCLUSIVE"
-
-
-def lock_tables(
-    session_maker: async_sessionmaker[_SessionT],
-    tables: list[type[DeclarativeBase]],
-    *,
-    mode: LockMode = LockMode.SHARE_UPDATE_EXCLUSIVE,
-    timeout: str = "5s",
-) -> AbstractAsyncContextManager[_SessionT]:
-    """Lock PostgreSQL tables for the duration of a transaction.
-
-    Args:
-        session_maker: Async session factory used to create the dedicated
-            session.
-        tables: List of SQLAlchemy model classes to lock.
-        mode: Lock mode (default: SHARE UPDATE EXCLUSIVE).
-        timeout: Lock timeout (default: "5s").
-
-    Yields:
-        The dedicated session, open within the locked transaction.
-
-    Raises:
-        SQLAlchemyError: If the lock cannot be acquired within *timeout*.
-
-    Example:
-        ```python
-        from fastapi_toolsets.db import lock_tables, LockMode
-
-        async with lock_tables(session_maker, [User, Account]) as session:
-            # Tables are locked; changes are committed when the context exits.
-            user = await UserCrud.get(session, [User.id == 1])
-            user.balance += 100
-
-        # With custom lock mode
-        async with lock_tables(session_maker, [Order], mode=LockMode.EXCLUSIVE) as session:
-            await process_order(session, order_id)
-        ```
-    """
-    table_names = ",".join(table.__tablename__ for table in tables)
-
-    @asynccontextmanager
-    async def _lock() -> AsyncGenerator[_SessionT, None]:
-        async with session_maker() as session:
-            try:
-                await session.execute(text(f"SET LOCAL lock_timeout='{timeout}'"))
-                await session.execute(text(f"LOCK {table_names} IN {mode.value} MODE"))
-                yield session
-                await session.commit()
-            except sa_exc.TimeoutError as e:
-                await session.rollback()
-                raise PoolExhaustedError(
-                    f"Connection pool exhausted while locking '{table_names}'. "
-                ) from e
-            except sa_exc.DBAPIError as e:
-                await session.rollback()
-                if _is_lock_not_available(e):
-                    raise LockTimeoutError(
-                        f"Lock on '{table_names}' could not be acquired within {timeout}."
-                    ) from e
-                raise  # pragma: no cover
-            except BaseException:
-                await session.rollback()
-                raise
-
-    return _lock()
-
-
-@asynccontextmanager
-async def advisory_lock(
-    session: AsyncSession,
-    key: int | tuple[int, int],
-    *,
-    shared: bool = False,
-    nowait: bool = False,
-    timeout: str | None = None,
-) -> AsyncGenerator[bool, None]:
-    """Acquire a PostgreSQL session-level advisory lock.
-
-    Args:
-        session: AsyncSession instance.
-        key: Lock key — a single ``int`` (bigint) or a ``(int, int)`` pair for namespacing.
-        shared: Acquire a shared lock (multiple holders allowed). Default is exclusive.
-        nowait: Return ``False`` immediately if the lock is unavailable instead of waiting.
-        timeout: Maximum wait time (e.g. ``"5s"``, ``"500ms"``). Raises ``DBAPIError``
-            if exceeded. Ignored when *nowait* is ``True``.
-
-    Yields:
-        ``True`` if the lock was acquired, ``False`` if *nowait* is ``True`` and the lock
-        is already held.
-
-    Raises:
-        LockTimeoutError: If *timeout* is set and the lock cannot be acquired in time.
-
-    Example:
-        ```python
-        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):
-            ...
-        ```
-    """
-    suffix = "_shared" if shared else ""
-    acquire_fn = f"{'pg_try_advisory_lock' if nowait else 'pg_advisory_lock'}{suffix}"
-    release_fn = f"pg_advisory_unlock{suffix}"
-
-    if isinstance(key, tuple):
-        k1, k2 = key
-        args = "CAST(:k1 AS integer), CAST(:k2 AS integer)"
-        params: dict[str, int] = {"k1": k1, "k2": k2}
-    else:
-        args = ":k"
-        params = {"k": key}
-
-    acquire_sql = text(f"SELECT {acquire_fn}({args})")
-    release_sql = text(f"SELECT {release_fn}({args})")
-
-    if timeout is not None and not nowait:
-        await session.execute(text(f"SET LOCAL lock_timeout='{timeout}'"))
-
-    try:
-        result = await session.execute(acquire_sql, params)
-    except sa_exc.DBAPIError as e:
-        if _is_lock_not_available(e):
-            raise LockTimeoutError(
-                f"Advisory lock {key!r} could not be acquired within {timeout}."
-            ) from e
-        raise  # pragma: no cover
-    acquired = result.scalar() if nowait else True
-    try:
-        yield acquired
-    finally:
-        if acquired:
-            await session.execute(release_sql, params)
-
-
-async def create_database(
-    db_name: str,
-    *,
-    server_url: str,
-) -> None:
-    """Create a database.
-
-    Connects to *server_url* using ``AUTOCOMMIT`` isolation and issues a
-    ``CREATE DATABASE`` statement for *db_name*.
-
-    Args:
-        db_name: Name of the database to create.
-        server_url: URL used for server-level DDL (must point to an existing
-            database on the same server).
-
-    Example:
-        ```python
-        from fastapi_toolsets.db import create_database
-
-        SERVER_URL = "postgresql+asyncpg://postgres:postgres@localhost/postgres"
-        await create_database("myapp_test", server_url=SERVER_URL)
-        ```
-    """
-    engine = create_async_engine(server_url, isolation_level="AUTOCOMMIT")
-    try:
-        async with engine.connect() as conn:
-            await conn.execute(text(f"CREATE DATABASE {db_name}"))
-    finally:
-        await engine.dispose()
-
-
-async def cleanup_tables(
-    session: AsyncSession,
-    base: type[DeclarativeBase],
-) -> None:
-    """Truncate all tables for fast between-test cleanup.
-
-    Executes a single ``TRUNCATE … RESTART IDENTITY CASCADE`` statement
-    across every table in *base*'s metadata, which is significantly faster
-    than dropping and re-creating tables between tests.
-
-    This is a no-op when the metadata contains no tables.
-
-    Args:
-        session: An active async database session.
-        base: SQLAlchemy DeclarativeBase class containing model metadata.
-
-    Example:
-        ```python
-        @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)
-        ```
-    """
-    tables = base.metadata.sorted_tables
-    if not tables:
-        return
-
-    table_names = ", ".join(f'"{t.name}"' for t in tables)
-    await session.execute(text(f"TRUNCATE {table_names} RESTART IDENTITY CASCADE"))
-    await session.commit()
-
-
-_M = TypeVar("_M", bound=DeclarativeBase)
-
-
-async def wait_for_row_change(
-    session: AsyncSession,
-    model: type[_M],
-    pk_value: Any,
-    *,
-    columns: list[str] | None = None,
-    interval: float = 0.5,
-    timeout: float | None = None,
-) -> _M:
-    """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``).
-
-    Args:
-        session: AsyncSession instance
-        model: SQLAlchemy model class
-        pk_value: Primary key value of the row to watch
-        columns: Optional list of column names to watch. If None, all columns
-                 are watched.
-        interval: Polling interval in seconds (default: 0.5)
-        timeout: Maximum time to wait in seconds. None means wait forever.
-
-    Returns:
-        The refreshed model instance with updated values
-
-    Raises:
-        NotFoundError: If the row does not exist or is deleted during polling
-        TimeoutError: If timeout expires before a change is detected
-
-    Example:
-        ```python
-        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,
-        )
-        ```
-    """
-    instance = await session.get(model, pk_value)
-    if instance is None:
-        raise NotFoundError(f"{model.__name__} with pk={pk_value!r} not found")
-
-    if columns is not None:
-        watch_cols = columns
-    else:
-        watch_cols = [attr.key for attr in model.__mapper__.column_attrs]
-
-    initial = {col: getattr(instance, col) for col in watch_cols}
-
-    elapsed = 0.0
-    while True:
-        await asyncio.sleep(interval)
-        elapsed += interval
-
-        if timeout is not None and elapsed >= timeout:
-            raise TimeoutError(
-                f"No change detected on {model.__name__} "
-                f"with pk={pk_value!r} within {timeout}s"
-            )
-
-        session.expunge(instance)
-        instance = await session.get(model, pk_value)
-
-        if instance is None:
-            raise NotFoundError(f"{model.__name__} with pk={pk_value!r} was deleted")
-
-        current = {col: getattr(instance, col) for col in watch_cols}
-        if current != initial:
-            return instance
-
-
-def _m2m_prop(rel_attr: QueryableAttribute) -> RelationshipProperty:  # type: ignore[type-arg]
-    """Return the validated M2M RelationshipProperty for *rel_attr*.
-
-    Raises TypeError if *rel_attr* is not a Many-to-Many relationship.
-    """
-    prop = rel_attr.property
-    if not isinstance(prop, RelationshipProperty) or prop.secondary is None:
-        raise TypeError(
-            f"m2m helpers require a Many-to-Many relationship attribute, "
-            f"got {rel_attr!r}. Use a relationship with a secondary table."
-        )
-    return prop
-
-
-async def m2m_add(
-    session: AsyncSession,
-    instance: DeclarativeBase,
-    rel_attr: QueryableAttribute,
-    *related: DeclarativeBase,
-    ignore_conflicts: bool = False,
-) -> None:
-    """Insert rows into a Many-to-Many association table without loading the ORM collection.
-
-    Args:
-        session: DB async session.
-        instance: The "owner" side model instance (e.g. the ``A`` in ``A.b_list``).
-        rel_attr: The M2M relationship attribute on the model class (e.g. ``A.b_list``).
-        *related: One or more related instances to associate with ``instance``.
-        ignore_conflicts: When ``True``, silently skip rows that already exist
-            in the association table (``ON CONFLICT DO NOTHING``).
-
-    Raises:
-        TypeError: If ``rel_attr`` is not a Many-to-Many relationship.
-    """
-    prop = _m2m_prop(rel_attr)
-    if not related:
-        return
-
-    secondary = cast(Table, prop.secondary)
-    assert secondary is not None  # guaranteed by _m2m_prop
-    sync_pairs = prop.secondary_synchronize_pairs
-    assert sync_pairs is not None  # set whenever secondary is set
-
-    # synchronize_pairs:           [(parent_col, assoc_col), ...]
-    # secondary_synchronize_pairs: [(related_col, assoc_col), ...]
-    rows: list[dict[str, Any]] = []
-    for rel_instance in related:
-        row: dict[str, Any] = {}
-        for parent_col, assoc_col in prop.synchronize_pairs:
-            row[assoc_col.name] = getattr(instance, cast(str, parent_col.key))
-        for related_col, assoc_col in sync_pairs:
-            row[assoc_col.name] = getattr(rel_instance, cast(str, related_col.key))
-        rows.append(row)
-
-    stmt = pg_insert(secondary).values(rows)
-    if ignore_conflicts:
-        stmt = stmt.on_conflict_do_nothing()
-    await session.execute(stmt)
-
-
-async def m2m_remove(
-    session: AsyncSession,
-    instance: DeclarativeBase,
-    rel_attr: QueryableAttribute,
-    *related: DeclarativeBase,
-) -> None:
-    """Remove rows from a Many-to-Many association table without loading the ORM collection.
-
-    Args:
-        session: DB async session.
-        instance: The "owner" side model instance (e.g. the ``A`` in ``A.b_list``).
-        rel_attr: The M2M relationship attribute on the model class (e.g. ``A.b_list``).
-        *related: One or more related instances to disassociate from ``instance``.
-
-    Raises:
-        TypeError: If ``rel_attr`` is not a Many-to-Many relationship.
-    """
-    prop = _m2m_prop(rel_attr)
-    if not related:
-        return
-
-    secondary = cast(Table, prop.secondary)
-    assert secondary is not None  # guaranteed by _m2m_prop
-    related_pairs = prop.secondary_synchronize_pairs
-    assert related_pairs is not None  # set whenever secondary is set
-
-    parent_where = [
-        assoc_col == getattr(instance, cast(str, parent_col.key))
-        for parent_col, assoc_col in prop.synchronize_pairs
-    ]
-
-    if len(related_pairs) == 1:
-        related_col, assoc_col = related_pairs[0]
-        related_values = [getattr(r, cast(str, related_col.key)) for r in related]
-        related_where = assoc_col.in_(related_values)
-    else:
-        assoc_cols = [ac for _, ac in related_pairs]
-        rel_cols = [rc for rc, _ in related_pairs]
-        related_values_t = [
-            tuple(getattr(r, cast(str, rc.key)) for rc in rel_cols) for r in related
-        ]
-        related_where = tuple_(*assoc_cols).in_(related_values_t)
-
-    await session.execute(delete(secondary).where(*parent_where, related_where))
-
-
-async def m2m_set(
-    session: AsyncSession,
-    instance: DeclarativeBase,
-    rel_attr: QueryableAttribute,
-    *related: DeclarativeBase,
-) -> None:
-    """Replace the entire Many-to-Many association set atomically.
-
-    Args:
-        session: DB async session.
-        instance: The "owner" side model instance (e.g. the ``A`` in ``A.b_list``).
-        rel_attr: The M2M relationship attribute on the model class (e.g. ``A.b_list``).
-        *related: The new complete set of related instances.
-
-    Raises:
-        TypeError: If ``rel_attr`` is not a Many-to-Many relationship.
-    """
-    prop = _m2m_prop(rel_attr)
-    secondary = cast(Table, prop.secondary)
-    assert secondary is not None  # guaranteed by _m2m_prop
-
-    parent_where = [
-        assoc_col == getattr(instance, cast(str, parent_col.key))
-        for parent_col, assoc_col in prop.synchronize_pairs
-    ]
-    await session.execute(delete(secondary).where(*parent_where))
-
-    if related:
-        await m2m_add(session, instance, rel_attr, *related)

src/fastapi_toolsets/db/__init__.py

@@ -0,0 +1,18 @@
+"""Database package: the ``Database`` facade plus PostgreSQL power-tools."""
+
+from .core import Database, transaction
+from .locks import LockMode, advisory_lock, lock_tables
+from .m2m import m2m_add, m2m_remove, m2m_set
+from .watch import wait_for_row_change
+
+__all__ = [
+    "Database",
+    "LockMode",
+    "advisory_lock",
+    "lock_tables",
+    "m2m_add",
+    "m2m_remove",
+    "m2m_set",
+    "transaction",
+    "wait_for_row_change",
+]

src/fastapi_toolsets/db/core.py

@@ -0,0 +1,323 @@
+"""The ``Database`` facade: session lifecycle, dependency, middleware, transactions."""
+
+from collections.abc import AsyncGenerator
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from typing import Any
+
+from pydantic import PostgresDsn
+from sqlalchemy import exc as sa_exc
+from sqlalchemy.ext.asyncio import (
+    AsyncEngine,
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
+from sqlalchemy.orm import DeclarativeBase
+from starlette.requests import Request
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
+
+from ..exceptions import PoolExhaustedError
+from .locks import LockMode, lock_tables
+
+
+@asynccontextmanager
+async def transaction(
+    session: AsyncSession,
+) -> AsyncGenerator[AsyncSession, None]:
+    """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.
+
+    Args:
+        session: AsyncSession instance.
+
+    Yields:
+        The session within the transaction context.
+
+    Example:
+        ```python
+        from fastapi_toolsets.db import transaction
+
+        async with transaction(session):
+            session.add(model)
+        ```
+    """
+    if session.in_transaction():
+        async with session.begin_nested():
+            yield session
+    else:
+        async with session.begin():
+            yield session
+
+
+class _CommitOnResponseMiddleware:
+    """Commit the request's DB session before the response is sent."""
+
+    def __init__(self, app: ASGIApp, *, state_attr: str) -> None:
+        self.app = app
+        self.state_attr = state_attr
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        async def send_wrapper(message: Message) -> None:
+            if message["type"] == "http.response.start":
+                # ``scope["state"]`` is the same dict ``request.state`` writes
+                # to, so this is the session stashed by the dependency.
+                state = scope.get("state")
+                session = state.get(self.state_attr) if state else None
+                if session is not None and session.in_transaction():
+                    await session.commit()
+            await send(message)
+
+        await self.app(scope, receive, send_wrapper)
+
+
+class 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).
+
+    Args:
+        url: Database connection URL. Accepts a plain string or a Pydantic
+            :class:`~pydantic.PostgresDsn`.
+        engine: An existing :class:`AsyncEngine` to reuse instead of *url*.
+        session_class: Session class for the sessionmaker (e.g. ``EventSession``).
+        expire_on_commit: Expire attributes after commit. Defaults to ``False``.
+        autoflush: Autoflush the session before queries. Defaults to ``True``.
+        connect_args: DBAPI-level connection arguments forwarded to
+            :func:`create_async_engine` (URL mode only).
+        **engine_options: Extra keyword arguments forwarded to
+            :func:`create_async_engine` (URL mode only).
+
+    Raises:
+        TypeError: If neither or both of *url* and *engine* are given, or if
+            *connect_args*/*engine_options* are passed together with *engine*.
+
+    Example:
+        ```python
+        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])
+        ```
+    """
+
+    def __init__(
+        self,
+        url: str | PostgresDsn | None = None,
+        *,
+        engine: AsyncEngine | None = None,
+        session_class: type[AsyncSession] = AsyncSession,
+        expire_on_commit: bool = False,
+        autoflush: bool = True,
+        connect_args: dict[str, Any] | None = None,
+        **engine_options: Any,
+    ) -> None:
+        if (url is None) == (engine is None):
+            raise TypeError(
+                "Database requires exactly one of 'url' or 'engine' "
+                "(got both or neither)."
+            )
+        if engine is not None and (engine_options or connect_args is not None):
+            raise TypeError(
+                "connect_args/engine_options are only valid in URL mode; "
+                "configure the engine you pass via 'engine=' yourself."
+            )
+
+        if engine is not None:
+            self._owns_engine = False
+            self.engine: AsyncEngine = engine
+        else:
+            assert url is not None  # guaranteed by the XOR check above
+            self._owns_engine = True
+            if connect_args is not None:
+                engine_options["connect_args"] = connect_args
+            # ``PostgresDsn`` (and other URL objects) are not str subclasses, so
+            # coerce to the string form SQLAlchemy expects.
+            self.engine = create_async_engine(str(url), **engine_options)
+        self._sessionmaker: async_sessionmaker[AsyncSession] = async_sessionmaker(
+            self.engine,
+            class_=session_class,
+            expire_on_commit=expire_on_commit,
+            autoflush=autoflush,
+        )
+        # Private, per-instance state attribute; cannot collide with another
+        # Database or be mismatched against the middleware.
+        self._state_attr = f"_ft_db_session_{id(self):x}"
+        self._middleware_installed = False
+        self._disposed = False
+
+    async def _dispose(self) -> None:
+        """Dispose the engine once, only if we own it (idempotent)."""
+        if self._owns_engine and not self._disposed:
+            self._disposed = True
+            await self.engine.dispose()
+
+    @asynccontextmanager
+    async def lifespan(self, app: Any) -> AsyncGenerator[None, None]:
+        """Dispose the engine on shutdown; use as ``FastAPI(lifespan=db.lifespan)``.
+
+        Args:
+            app: The ASGI application (unused; required by the lifespan protocol).
+
+        Yields:
+            Control to the application for its lifetime.
+
+        Example:
+            ```python
+            app = FastAPI(lifespan=db.lifespan)
+            ```
+        """
+        try:
+            yield
+        finally:
+            await self._dispose()
+
+    def install(self, app: Any) -> None:
+        """Wire the commit middleware and engine disposal onto *app*.
+
+        Args:
+            app: The FastAPI/Starlette application to wire.
+
+        Example:
+            ```python
+            @asynccontextmanager
+            async def lifespan(app):
+                ...  # your startup
+                yield
+                ...  # your shutdown
+
+            app = FastAPI(lifespan=lifespan)
+            db.install(app)
+            ```
+        """
+        app.add_middleware(_CommitOnResponseMiddleware, state_attr=self._state_attr)
+        self._middleware_installed = True
+
+        inner_lifespan = app.router.lifespan_context
+
+        @asynccontextmanager
+        async def _composed(app_: Any) -> AsyncGenerator[None, None]:
+            async with self.lifespan(app_):
+                async with inner_lifespan(app_):
+                    yield
+
+        app.router.lifespan_context = _composed
+
+    @asynccontextmanager
+    async def _open(self) -> AsyncGenerator[AsyncSession, None]:
+        """Open a session and eagerly acquire a connection (fail-fast on pool)."""
+        async with self._sessionmaker() as session:
+            try:
+                await session.connection()
+            except sa_exc.TimeoutError as e:
+                raise PoolExhaustedError() from e
+            yield session
+
+    async def __call__(self, request: Request) -> AsyncGenerator[AsyncSession, None]:
+        """FastAPI dependency: yield a session and commit once at the right time.
+
+        Args:
+            request: The incoming request (injected by FastAPI).
+
+        Yields:
+            An AsyncSession for the duration of the request.
+
+        Example:
+            ```python
+            @app.get("/users/{user_id}")
+            async def get_user(user_id: int, session=Depends(db)):
+                return await UserCrud.get(session, [User.id == user_id])
+            ```
+        """
+        async with self._open() as session:
+            setattr(request.state, self._state_attr, session)
+            yield session
+            if not self._middleware_installed and session.in_transaction():
+                await session.commit()
+
+    @asynccontextmanager
+    async def session(self) -> AsyncGenerator[AsyncSession, None]:
+        """Open a session outside request handlers (background tasks, CLI, tests).
+
+        Commits on clean exit, rolls back on exception.
+
+        Yields:
+            An AsyncSession ready for database operations.
+
+        Example:
+            ```python
+            async with db.session() as session:
+                user = await UserCrud.get(session, [User.id == 1])
+            ```
+        """
+        async with self._open() as session:
+            yield session
+            if session.in_transaction():
+                await session.commit()
+
+    @asynccontextmanager
+    async def begin(self) -> AsyncGenerator[AsyncSession, None]:
+        """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:
+            An AsyncSession open within a transaction.
+
+        Example:
+            ```python
+            async with db.begin() as session:
+                session.add(User(name="ada"))
+            ```
+        """
+        async with self.session() as session, transaction(session):
+            yield session
+
+    def lock_tables(
+        self,
+        tables: list[type[DeclarativeBase]],
+        *,
+        mode: LockMode = LockMode.SHARE_UPDATE_EXCLUSIVE,
+        timeout: str = "5s",
+    ) -> AbstractAsyncContextManager[AsyncSession]:
+        """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.
+
+        Args:
+            tables: List of SQLAlchemy model classes to lock.
+            mode: Lock mode (default: ``SHARE UPDATE EXCLUSIVE``).
+            timeout: Lock timeout (default: ``"5s"``).
+
+        Yields:
+            The dedicated session, open within the locked transaction.
+
+        Raises:
+            LockTimeoutError: If the lock cannot be acquired within *timeout*.
+            PoolExhaustedError: If the connection pool is exhausted.
+
+        Example:
+            ```python
+            async with db.lock_tables([User, Account]) as session:
+                user = await UserCrud.get(session, [User.id == 1])
+                user.balance += 100
+            ```
+        """
+        return lock_tables(self._sessionmaker, tables, mode=mode, timeout=timeout)

src/fastapi_toolsets/db/locks.py

@@ -0,0 +1,185 @@
+"""PostgreSQL locking helpers: table locks and advisory locks."""
+
+from collections.abc import AsyncGenerator
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from enum import Enum
+from typing import TypeVar
+
+import asyncpg
+from sqlalchemy import exc as sa_exc
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+from sqlalchemy.orm import DeclarativeBase
+
+from ..exceptions import LockTimeoutError, PoolExhaustedError
+
+_SessionT = TypeVar("_SessionT", bound=AsyncSession)
+
+
+def _is_lock_not_available(e: sa_exc.DBAPIError) -> bool:
+    return e.orig is not None and isinstance(
+        e.orig.__cause__, asyncpg.exceptions.LockNotAvailableError
+    )
+
+
+class LockMode(str, Enum):
+    """PostgreSQL table lock modes.
+
+    See: https://www.postgresql.org/docs/current/explicit-locking.html
+    """
+
+    ACCESS_SHARE = "ACCESS SHARE"
+    ROW_SHARE = "ROW SHARE"
+    ROW_EXCLUSIVE = "ROW EXCLUSIVE"
+    SHARE_UPDATE_EXCLUSIVE = "SHARE UPDATE EXCLUSIVE"
+    SHARE = "SHARE"
+    SHARE_ROW_EXCLUSIVE = "SHARE ROW EXCLUSIVE"
+    EXCLUSIVE = "EXCLUSIVE"
+    ACCESS_EXCLUSIVE = "ACCESS EXCLUSIVE"
+
+
+def lock_tables(
+    session_maker: async_sessionmaker[_SessionT],
+    tables: list[type[DeclarativeBase]],
+    *,
+    mode: LockMode = LockMode.SHARE_UPDATE_EXCLUSIVE,
+    timeout: str = "5s",
+) -> AbstractAsyncContextManager[_SessionT]:
+    """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.
+
+    Args:
+        session_maker: Async session factory used to create the dedicated
+            session.
+        tables: List of SQLAlchemy model classes to lock.
+        mode: Lock mode (default: SHARE UPDATE EXCLUSIVE).
+        timeout: Lock timeout (default: "5s").
+
+    Yields:
+        The dedicated session, open within the locked transaction.
+
+    Raises:
+        LockTimeoutError: If the lock cannot be acquired within *timeout*.
+        PoolExhaustedError: If the connection pool is exhausted.
+
+    Example:
+        ```python
+        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
+        ```
+    """
+    table_names = ",".join(table.__tablename__ for table in tables)
+
+    @asynccontextmanager
+    async def _lock() -> AsyncGenerator[_SessionT, None]:
+        async with session_maker() as session:
+            try:
+                await session.execute(text(f"SET LOCAL lock_timeout='{timeout}'"))
+                await session.execute(text(f"LOCK {table_names} IN {mode.value} MODE"))
+                yield session
+                await session.commit()
+            except sa_exc.TimeoutError as e:
+                await session.rollback()
+                raise PoolExhaustedError(
+                    f"Connection pool exhausted while locking '{table_names}'. "
+                ) from e
+            except sa_exc.DBAPIError as e:
+                await session.rollback()
+                if _is_lock_not_available(e):
+                    raise LockTimeoutError(
+                        f"Lock on '{table_names}' could not be acquired within {timeout}."
+                    ) from e
+                raise  # pragma: no cover
+            except BaseException:
+                await session.rollback()
+                raise
+
+    return _lock()
+
+
+@asynccontextmanager
+async def advisory_lock(
+    session: AsyncSession,
+    key: int | tuple[int, int],
+    *,
+    shared: bool = False,
+    nowait: bool = False,
+    timeout: str | None = None,
+) -> AsyncGenerator[bool, None]:
+    """Acquire a PostgreSQL session-level advisory lock.
+
+    Args:
+        session: AsyncSession instance.
+        key: Lock key, either a single ``int`` (bigint) or a ``(int, int)`` pair for namespacing.
+        shared: Acquire a shared lock (multiple holders allowed). Default is exclusive.
+        nowait: Return ``False`` immediately if the lock is unavailable instead of waiting.
+        timeout: Maximum wait time (e.g. ``"5s"``, ``"500ms"``). Raises ``DBAPIError``
+            if exceeded. Ignored when *nowait* is ``True``.
+
+    Yields:
+        ``True`` if the lock was acquired, ``False`` if *nowait* is ``True`` and the lock
+        is already held.
+
+    Raises:
+        LockTimeoutError: If *timeout* is set and the lock cannot be acquired in time.
+
+    Example:
+        ```python
+        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):
+            ...
+        ```
+    """
+    suffix = "_shared" if shared else ""
+    acquire_fn = f"{'pg_try_advisory_lock' if nowait else 'pg_advisory_lock'}{suffix}"
+    release_fn = f"pg_advisory_unlock{suffix}"
+
+    if isinstance(key, tuple):
+        k1, k2 = key
+        args = "CAST(:k1 AS integer), CAST(:k2 AS integer)"
+        params: dict[str, int] = {"k1": k1, "k2": k2}
+    else:
+        args = ":k"
+        params = {"k": key}
+
+    acquire_sql = text(f"SELECT {acquire_fn}({args})")
+    release_sql = text(f"SELECT {release_fn}({args})")
+
+    # Lock management runs raw SQL on the caller's session. Guard it with
+    # ``no_autoflush`` so acquiring or releasing the lock never flushes the
+    # caller's pending ORM changes; SQLAlchemy 2.1 autoflushes on raw
+    # ``text()`` too, where 2.0 did not.
+    try:
+        with session.no_autoflush:
+            if timeout is not None and not nowait:
+                await session.execute(text(f"SET LOCAL lock_timeout='{timeout}'"))
+            result = await session.execute(acquire_sql, params)
+    except sa_exc.DBAPIError as e:
+        if _is_lock_not_available(e):
+            raise LockTimeoutError(
+                f"Advisory lock {key!r} could not be acquired within {timeout}."
+            ) from e
+        raise  # pragma: no cover
+    acquired = result.scalar() if nowait else True
+    try:
+        yield acquired
+    finally:
+        if acquired:
+            with session.no_autoflush:
+                await session.execute(release_sql, params)

src/fastapi_toolsets/db/m2m.py

@@ -0,0 +1,170 @@
+"""Many-to-Many association-table helpers (direct, without loading collections)."""
+
+from typing import Any, TypeVar, cast
+
+from sqlalchemy import ColumnElement, Table, delete, tuple_
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import DeclarativeBase, QueryableAttribute
+from sqlalchemy.orm.relationships import RelationshipProperty
+
+_M = TypeVar("_M", bound=DeclarativeBase)
+
+
+def _m2m_prop(rel_attr: QueryableAttribute) -> tuple[RelationshipProperty, Table]:  # type: ignore[type-arg]
+    """Return the validated M2M RelationshipProperty and its secondary table.
+
+    Raises TypeError if *rel_attr* is not a Many-to-Many relationship.
+    """
+    prop = rel_attr.property
+    if not isinstance(prop, RelationshipProperty) or prop.secondary is None:
+        raise TypeError(
+            f"m2m helpers require a Many-to-Many relationship attribute, "
+            f"got {rel_attr!r}. Use a relationship with a secondary table."
+        )
+    return prop, cast(Table, prop.secondary)
+
+
+def _parent_where(
+    prop: RelationshipProperty,  # type: ignore[type-arg]
+    instance: DeclarativeBase,
+) -> list[ColumnElement[bool]]:
+    """Build the WHERE clauses matching the owner side of *instance*."""
+    return [
+        assoc_col == getattr(instance, cast(str, parent_col.key))
+        for parent_col, assoc_col in prop.synchronize_pairs
+    ]
+
+
+async def m2m_add(
+    session: AsyncSession,
+    instance: DeclarativeBase,
+    rel_attr: QueryableAttribute,
+    *related: DeclarativeBase,
+    ignore_conflicts: bool = False,
+) -> None:
+    """Insert rows into a Many-to-Many association table without loading the ORM collection.
+
+    Args:
+        session: DB async session.
+        instance: The "owner" side model instance (e.g. the ``A`` in ``A.b_list``).
+        rel_attr: The M2M relationship attribute on the model class (e.g. ``A.b_list``).
+        *related: One or more related instances to associate with ``instance``.
+        ignore_conflicts: When ``True``, silently skip rows that already exist
+            in the association table (``ON CONFLICT DO NOTHING``).
+
+    Raises:
+        TypeError: If ``rel_attr`` is not a Many-to-Many relationship.
+
+    Example:
+        ```python
+        from fastapi_toolsets.db import m2m_add, transaction
+
+        async with transaction(session):
+            await m2m_add(session, post, Post.tags, tag1, tag2)
+        ```
+    """
+    prop, secondary = _m2m_prop(rel_attr)
+    if not related:
+        return
+
+    sync_pairs = prop.secondary_synchronize_pairs
+    assert sync_pairs is not None  # set whenever secondary is set
+
+    # synchronize_pairs:           [(parent_col, assoc_col), ...]
+    # secondary_synchronize_pairs: [(related_col, assoc_col), ...]
+    rows: list[dict[str, Any]] = []
+    for rel_instance in related:
+        row: dict[str, Any] = {}
+        for parent_col, assoc_col in prop.synchronize_pairs:
+            row[assoc_col.name] = getattr(instance, cast(str, parent_col.key))
+        for related_col, assoc_col in sync_pairs:
+            row[assoc_col.name] = getattr(rel_instance, cast(str, related_col.key))
+        rows.append(row)
+
+    stmt = pg_insert(secondary).values(rows)
+    if ignore_conflicts:
+        stmt = stmt.on_conflict_do_nothing()
+    await session.execute(stmt)
+
+
+async def m2m_remove(
+    session: AsyncSession,
+    instance: DeclarativeBase,
+    rel_attr: QueryableAttribute,
+    *related: DeclarativeBase,
+) -> None:
+    """Remove rows from a Many-to-Many association table without loading the ORM collection.
+
+    Args:
+        session: DB async session.
+        instance: The "owner" side model instance (e.g. the ``A`` in ``A.b_list``).
+        rel_attr: The M2M relationship attribute on the model class (e.g. ``A.b_list``).
+        *related: One or more related instances to disassociate from ``instance``.
+
+    Raises:
+        TypeError: If ``rel_attr`` is not a Many-to-Many relationship.
+
+    Example:
+        ```python
+        from fastapi_toolsets.db import m2m_remove, transaction
+
+        async with transaction(session):
+            await m2m_remove(session, post, Post.tags, tag1)
+        ```
+    """
+    prop, secondary = _m2m_prop(rel_attr)
+    if not related:
+        return
+
+    related_pairs = prop.secondary_synchronize_pairs
+    assert related_pairs is not None  # set whenever secondary is set
+
+    parent_where = _parent_where(prop, instance)
+
+    if len(related_pairs) == 1:
+        related_col, assoc_col = related_pairs[0]
+        related_values = [getattr(r, cast(str, related_col.key)) for r in related]
+        related_where = assoc_col.in_(related_values)
+    else:
+        assoc_cols = [ac for _, ac in related_pairs]
+        rel_cols = [rc for rc, _ in related_pairs]
+        related_values_t = [
+            tuple(getattr(r, cast(str, rc.key)) for rc in rel_cols) for r in related
+        ]
+        related_where = tuple_(*assoc_cols).in_(related_values_t)
+
+    await session.execute(delete(secondary).where(*parent_where, related_where))
+
+
+async def m2m_set(
+    session: AsyncSession,
+    instance: DeclarativeBase,
+    rel_attr: QueryableAttribute,
+    *related: DeclarativeBase,
+) -> None:
+    """Replace the entire Many-to-Many association set atomically.
+
+    Args:
+        session: DB async session.
+        instance: The "owner" side model instance (e.g. the ``A`` in ``A.b_list``).
+        rel_attr: The M2M relationship attribute on the model class (e.g. ``A.b_list``).
+        *related: The new complete set of related instances.
+
+    Raises:
+        TypeError: If ``rel_attr`` is not a Many-to-Many relationship.
+
+    Example:
+        ```python
+        from fastapi_toolsets.db import m2m_set, transaction
+
+        async with transaction(session):
+            await m2m_set(session, post, Post.tags, tag1, tag2)  # replaces all
+        ```
+    """
+    prop, secondary = _m2m_prop(rel_attr)
+
+    await session.execute(delete(secondary).where(*_parent_where(prop, instance)))
+
+    if related:
+        await m2m_add(session, instance, rel_attr, *related)

src/fastapi_toolsets/db/testing.py

@@ -0,0 +1,69 @@
+"""Database admin and test helpers: DDL and truncation."""
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+from sqlalchemy.orm import DeclarativeBase
+
+
+async def create_database(
+    db_name: str,
+    *,
+    server_url: str,
+) -> None:
+    """Create a database.
+
+    Connects to *server_url* using ``AUTOCOMMIT`` isolation and issues a
+    ``CREATE DATABASE`` statement for *db_name*.
+
+    Args:
+        db_name: Name of the database to create.
+        server_url: URL used for server-level DDL (must point to an existing
+            database on the same server).
+
+    Example:
+        ```python
+        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)
+        ```
+    """
+    engine = create_async_engine(server_url, isolation_level="AUTOCOMMIT")
+    try:
+        async with engine.connect() as conn:
+            await conn.execute(text(f"CREATE DATABASE {db_name}"))
+    finally:
+        await engine.dispose()
+
+
+async def cleanup_tables(
+    session: AsyncSession,
+    base: type[DeclarativeBase],
+) -> None:
+    """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.
+
+    Args:
+        session: An active async database session.
+        base: SQLAlchemy DeclarativeBase class containing model metadata.
+
+    Example:
+        ```python
+        @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)
+        ```
+    """
+    tables = base.metadata.sorted_tables
+    if not tables:
+        return
+
+    table_names = ", ".join(f'"{t.name}"' for t in tables)
+    await session.execute(text(f"TRUNCATE {table_names} RESTART IDENTITY CASCADE"))
+    await session.commit()

src/fastapi_toolsets/db/watch.py

@@ -0,0 +1,106 @@
+"""Row-watching helpers: poll a database row until it changes."""
+
+import asyncio
+from typing import Any, TypeVar
+
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import DeclarativeBase
+
+from ..exceptions import NotFoundError
+
+_M = TypeVar("_M", bound=DeclarativeBase)
+
+
+async def wait_for_row_change(
+    session: AsyncSession,
+    model: type[_M],
+    pk_value: Any,
+    *,
+    columns: list[str] | None = None,
+    interval: float = 0.5,
+    timeout: float | None = None,
+) -> _M:
+    """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``).
+
+    Args:
+        session: AsyncSession instance.
+        model: SQLAlchemy model class.
+        pk_value: Primary key value of the row to watch.
+        columns: Optional list of column names to watch. If None, all columns
+            are watched.
+        interval: Polling interval in seconds (default: 0.5).
+        timeout: Maximum time to wait in seconds. None means wait forever.
+
+    Returns:
+        The refreshed model instance with updated values.
+
+    Raises:
+        NotFoundError: If the row does not exist or is deleted during polling.
+        TimeoutError: If timeout expires before a change is detected.
+
+    Example:
+        ```python
+        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,
+        )
+        ```
+    """
+    bind = getattr(session, "bind", None)
+    if bind is None:
+        raise TypeError(
+            "wait_for_row_change requires a session bound to an engine "
+            "(session.bind is None)"
+        )
+    watcher = AsyncSession(bind=bind)
+    try:
+
+        async def _reload() -> _M | None:
+            await watcher.rollback()
+            return await watcher.get(model, pk_value, populate_existing=True)
+
+        instance = await _reload()
+        if instance is None:
+            raise NotFoundError(f"{model.__name__} with pk={pk_value!r} not found")
+
+        if columns is not None:
+            watch_cols = columns
+        else:
+            watch_cols = [attr.key for attr in model.__mapper__.column_attrs]
+
+        initial = {col: getattr(instance, col) for col in watch_cols}
+
+        elapsed = 0.0
+        while True:
+            await asyncio.sleep(interval)
+            elapsed += interval
+
+            if timeout is not None and elapsed >= timeout:
+                raise TimeoutError(
+                    f"No change detected on {model.__name__} "
+                    f"with pk={pk_value!r} within {timeout}s"
+                )
+
+            instance = await _reload()
+
+            if instance is None:
+                raise NotFoundError(
+                    f"{model.__name__} with pk={pk_value!r} was deleted"
+                )
+
+            current = {col: getattr(instance, col) for col in watch_cols}
+            if current != initial:
+                return instance
+    finally:
+        await watcher.close()

src/fastapi_toolsets/fixtures/__init__.py

@@ -1,21 +1,28 @@
 """Fixture system for seeding databases with dependency resolution."""
 
-from .enum import LoadStrategy
-from .registry import Context, FixtureRegistry
-from .utils import (
-    get_field_by_attr,
-    get_obj_by_attr,
-    load_fixtures,
-    load_fixtures_by_context,
-)
+from .enum import Context, LoadStrategy
 
 __all__ = [
     "Context",
     "FixtureRegistry",
     "LoadStrategy",
-    "get_field_by_attr",
-    "get_obj_by_attr",
     "load_fixtures",
     "load_fixtures_by_context",
-    "register_fixtures",
 ]
+
+_LAZY = {
+    "FixtureRegistry": ".registry",
+    "load_fixtures": ".utils",
+    "load_fixtures_by_context": ".utils",
+}
+
+
+def __getattr__(name: str):
+    module_name = _LAZY.get(name)
+    if module_name is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+    import importlib
+
+    module = importlib.import_module(module_name, __name__)
+    return getattr(module, name)

src/fastapi_toolsets/fixtures/registry.py

@@ -7,11 +7,8 @@ from typing import Any, cast
 
 from sqlalchemy.orm import DeclarativeBase
 
-from ..logger import get_logger
 from .enum import Context
 
-logger = get_logger()
-
 
 def _normalize_contexts(
     contexts: list[str | Enum] | tuple[str | Enum, ...],
@@ -20,6 +17,11 @@ def _normalize_contexts(
     return [c.value if isinstance(c, Enum) else c for c in contexts]
 
 
+def _context_filter_values(contexts: tuple[str | Enum, ...]) -> set[str]:
+    """Normalize *contexts* for filtering, always including Context.BASE."""
+    return set(_normalize_contexts(contexts)) | {Context.BASE.value}
+
+
 @dataclass
 class Fixture:
     """A fixture definition with metadata."""
@@ -70,8 +72,6 @@ class FixtureRegistry:
         @fixtures.register(contexts=[Context.TESTING])
         def users():
             return [User(id=2, username="tester")]
-        # load_fixtures_by_context(..., Context.BASE, Context.TESTING)
-        # → loads both User(admin) and User(tester) under the "users" name
         ```
     """
 
@@ -189,9 +189,7 @@ class FixtureRegistry:
             ValueError: If the fixture has multiple context variants — use
                 :meth:`get_variants` in that case.
         """
-        if name not in self._fixtures:
-            raise KeyError(f"Fixture '{name}' not found")
-        variants = self._fixtures[name]
+        variants = self.get_variants(name)
         if len(variants) > 1:
             raise ValueError(
                 f"Fixture '{name}' has {len(variants)} context variants. "
@@ -205,8 +203,9 @@ class FixtureRegistry:
         Args:
             name: Fixture name.
             *contexts: If given, only return variants whose context set
-                intersects with these values.  Both :class:`Context` enum
-                values and plain strings are accepted.
+                intersects with these values (:class:`Context.BASE` variants
+                are always included).  Both :class:`Context` enum values and
+                plain strings are accepted.
 
         Returns:
             List of matching :class:`Fixture` objects (may be empty when a
@@ -220,16 +219,89 @@ class FixtureRegistry:
         variants = self._fixtures[name]
         if not contexts:
             return list(variants)
-        context_values = set(_normalize_contexts(contexts))
+        context_values = _context_filter_values(contexts)
         return [v for v in variants if set(v.contexts) & context_values]
 
+    def get_load_variants(self, name: str, *contexts: str | Enum) -> list[Fixture]:
+        """Return variants for *name* filtered by *contexts*.
+
+        Raises:
+            KeyError: If no fixture with *name* is registered.
+        """
+        variants = self.get_variants(name, *contexts)
+        if contexts and not variants:
+            return self.get_variants(name)
+        return variants
+
     def get_all(self) -> list[Fixture]:
         """Get all registered fixtures (all variants of all names)."""
         return [f for variants in self._fixtures.values() for f in variants]
 
+    def get_dependencies(self, name: str) -> list[str]:
+        """Get the union of ``depends_on`` across all variants of *name*.
+
+        Raises:
+            KeyError: If no fixture named *name* is registered.
+        """
+        variants = self._fixtures.get(name)
+        if variants is None:
+            raise KeyError(f"Fixture '{name}' not found")
+
+        seen: set[str] = set()
+        deps: list[str] = []
+        for variant in variants:
+            for dep in variant.depends_on:
+                if dep not in seen:
+                    deps.append(dep)
+                    seen.add(dep)
+        return deps
+
+    def obj(self, name: str, attr_name: str, value: Any) -> DeclarativeBase:
+        """Get a model instance from a registered fixture by attribute value.
+
+        Args:
+            name: Fixture name to look up.
+            attr_name: Name of the attribute to match against.
+            value: Value to match.
+
+        Returns:
+            The first model instance where the attribute matches the given value.
+
+        Raises:
+            KeyError: If no fixture named *name* is registered.
+            StopIteration: If no matching object is found.
+        """
+        instances = (
+            obj for variant in self.get_variants(name) for obj in variant.func()
+        )
+        try:
+            return next(obj for obj in instances if getattr(obj, attr_name) == value)
+        except StopIteration:
+            raise StopIteration(
+                f"No object with {attr_name}={value} found in fixture '{name}'"
+            ) from None
+
+    def field(self, name: str, attr_name: str, value: Any, *, field: str = "id") -> Any:
+        """Get a single field value from a fixture object matched by an attribute.
+
+        Args:
+            name: Fixture name to look up.
+            attr_name: Name of the attribute to match against.
+            value: Value to match.
+            field: Attribute name to return from the matched object (default: ``"id"``).
+
+        Returns:
+            The value of ``field`` on the first matching model instance.
+
+        Raises:
+            KeyError: If no fixture named *name* is registered.
+            StopIteration: If no matching object is found.
+        """
+        return getattr(self.obj(name, attr_name, value), field)
+
     def get_by_context(self, *contexts: str | Enum) -> list[Fixture]:
         """Get fixtures for specific contexts."""
-        context_values = set(_normalize_contexts(contexts))
+        context_values = _context_filter_values(contexts)
         return [
             f
             for variants in self._fixtures.values()
@@ -254,7 +326,6 @@ class FixtureRegistry:
             ValueError: If circular dependency detected
         """
         resolved: list[str] = []
-        seen: set[str] = set()
         visiting: set[str] = set()
 
         def visit(name: str) -> None:
@@ -264,25 +335,11 @@ class FixtureRegistry:
                 raise ValueError(f"Circular dependency detected: {name}")
 
             visiting.add(name)
-            variants = self._fixtures.get(name)
-            if variants is None:
-                raise KeyError(f"Fixture '{name}' not found")
-
-            # Union of depends_on across all variants, preserving first-seen order.
-            seen_deps: set[str] = set()
-            all_deps: list[str] = []
-            for variant in variants:
-                for dep in variant.depends_on:
-                    if dep not in seen_deps:
-                        all_deps.append(dep)
-                        seen_deps.add(dep)
-
-            for dep in all_deps:
+            for dep in self.get_dependencies(name):
                 visit(dep)
 
             visiting.remove(name)
             resolved.append(name)
-            seen.add(name)
 
         for name in names:
             visit(name)
@@ -303,9 +360,4 @@ class FixtureRegistry:
         # appear multiple times if it has variants in different contexts).
         names = list(dict.fromkeys(f.name for f in context_fixtures))
 
-        all_deps: set[str] = set()
-        for name in names:
-            deps = self.resolve_dependencies(name)
-            all_deps.update(deps)
-
-        return self.resolve_dependencies(*all_deps)
+        return self.resolve_dependencies(*names)

src/fastapi_toolsets/fixtures/utils.py

@@ -1,17 +1,18 @@
 """Fixture loading utilities for database seeding."""
 
-from collections.abc import Callable, Sequence
+from collections.abc import Iterator
 from enum import Enum
-from typing import Any
+from typing import Any, cast
 
+from sqlalchemy import Table, select
 from sqlalchemy import inspect as sa_inspect
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm import DeclarativeBase, selectinload
+from sqlalchemy.orm.interfaces import ExecutableOption, ORMOption
 
-from ..db import get_transaction
+from ..db import transaction
 from ..logger import get_logger
-from ..types import ModelType
 from .enum import LoadStrategy
 from .registry import FixtureRegistry, _normalize_contexts
 
@@ -93,17 +94,42 @@ def _group_by_column_set(
     return list(groups.values())
 
 
+def _grouped_table_dicts(
+    model_cls: type[DeclarativeBase], instances: list[DeclarativeBase]
+) -> Iterator[
+    tuple[type[DeclarativeBase], list[dict[str, Any]], list[DeclarativeBase]]
+]:
+    """Yield (cls, group_dicts, group_instances) per table in the inheritance
+    chain and per column-set group, skipping empty groups.
+    """
+    for cls in _get_table_chain(model_cls):
+        dicts = [_instance_to_dict_for_cls(i, cls) for i in instances]
+        for group_dicts, group_instances in _group_by_column_set(dicts, instances):
+            if group_dicts and group_dicts[0]:  # pragma: no branch
+                yield cls, group_dicts, group_instances
+
+
 async def _batch_insert(
     session: AsyncSession,
     model_cls: type[DeclarativeBase],
     instances: list[DeclarativeBase],
 ) -> None:
-    """INSERT all instances — raises on conflict (no duplicate handling)."""
-    for cls in _get_table_chain(model_cls):
-        dicts = [_instance_to_dict_for_cls(i, cls) for i in instances]
-        for group_dicts, _ in _group_by_column_set(dicts, instances):
-            if group_dicts and group_dicts[0]:  # pragma: no branch
-                await session.execute(pg_insert(cls).values(group_dicts))
+    """INSERT all instances, raises on conflict."""
+    for cls, group_dicts, group_instances in _grouped_table_dicts(model_cls, instances):
+        table = cast(Table, cls.__table__)
+        missing_pk_cols = [
+            col for col in table.primary_key.columns if col.key not in group_dicts[0]
+        ]
+        if not missing_pk_cols:
+            await session.execute(pg_insert(table), group_dicts)
+            continue
+        stmt = pg_insert(table).returning(
+            *missing_pk_cols, sort_by_parameter_order=True
+        )
+        result = await session.execute(stmt, group_dicts)
+        for inst, row in zip(group_instances, result):
+            for col, val in zip(missing_pk_cols, row):
+                setattr(inst, col.key, val)
 
 
 async def _batch_merge(
@@ -112,30 +138,26 @@ async def _batch_merge(
     instances: list[DeclarativeBase],
 ) -> None:
     """UPSERT: insert new rows, update existing ones with the provided values."""
-    for cls in _get_table_chain(model_cls):
+    for cls, group_dicts, _ in _grouped_table_dicts(model_cls, instances):
         pk_names = [col.name for col in cls.__table__.primary_key]
         pk_names_set = set(pk_names)
         own_col_keys = {col.key for col in cls.__table__.columns}
         non_pk_cols = [k for k in own_col_keys if k not in pk_names_set]
 
-        dicts = [_instance_to_dict_for_cls(i, cls) for i in instances]
-        for group_dicts, _ in _group_by_column_set(dicts, instances):
-            if not group_dicts or not group_dicts[0]:  # pragma: no cover
-                continue
-            stmt = pg_insert(cls).values(group_dicts)
+        stmt = pg_insert(cls).values(group_dicts)
 
-            inserted_keys = set(group_dicts[0])
-            update_cols = [col for col in non_pk_cols if col in inserted_keys]
+        inserted_keys = set(group_dicts[0])
+        update_cols = [col for col in non_pk_cols if col in inserted_keys]
 
-            if update_cols:
-                stmt = stmt.on_conflict_do_update(
-                    index_elements=pk_names,
-                    set_={col: stmt.excluded[col] for col in update_cols},
-                )
-            else:
-                stmt = stmt.on_conflict_do_nothing(index_elements=pk_names)
+        if update_cols:
+            stmt = stmt.on_conflict_do_update(
+                index_elements=pk_names,
+                set_={col: stmt.excluded[col] for col in update_cols},
+            )
+        else:
+            stmt = stmt.on_conflict_do_nothing(index_elements=pk_names)
 
-            await session.execute(stmt)
+        await session.execute(stmt)
 
 
 async def _batch_skip_existing(
@@ -169,8 +191,14 @@ async def _batch_skip_existing(
     loaded = list(no_pk)
     if no_pk:
         no_pk_dicts = [_instance_to_dict(i) for i in no_pk]
-        for group_dicts, _ in _group_by_column_set(no_pk_dicts, no_pk):
-            await session.execute(pg_insert(model_cls).values(group_dicts))
+        for group_dicts, group_instances in _group_by_column_set(no_pk_dicts, no_pk):
+            stmt = pg_insert(cast(Table, model_cls.__table__)).returning(
+                *mapper.primary_key, sort_by_parameter_order=True
+            )
+            result = await session.execute(stmt, group_dicts)
+            for inst, row in zip(group_instances, result):
+                for col, val in zip(mapper.primary_key, row):
+                    setattr(inst, col.key, val)
 
     if with_pk_pairs:
         with_pk = [i for i, _ in with_pk_pairs]
@@ -196,6 +224,64 @@ async def _batch_skip_existing(
     return loaded
 
 
+def _relationship_load_options(model: type[DeclarativeBase]) -> list[ExecutableOption]:
+    """Build selectinload options for all direct relationships on a model."""
+    return [
+        selectinload(getattr(model, rel.key)) for rel in model.__mapper__.relationships
+    ]
+
+
+async def _reload_with_relationships(
+    session: AsyncSession,
+    instances: list[DeclarativeBase],
+    load_options: list[ExecutableOption],
+) -> list[DeclarativeBase]:
+    """Reload instances in a single bulk query with relationship eager-loading."""
+    model = type(instances[0])
+    mapper = model.__mapper__
+    pk_cols = mapper.primary_key
+
+    if len(pk_cols) == 1:
+        pk_attr = getattr(model, pk_cols[0].key)
+        pks = [getattr(inst, pk_cols[0].key) for inst in instances]
+        result = await session.execute(
+            select(model).where(pk_attr.in_(pks)).options(*load_options)
+        )
+        by_pk = {getattr(row, pk_cols[0].key): row for row in result.unique().scalars()}
+        return [by_pk[pk] for pk in pks]
+
+    # Composite PK: fall back to per-instance reload
+    reloaded: list[DeclarativeBase] = []
+    for instance in instances:
+        pk = _get_primary_key(instance)
+        refreshed = await session.get(
+            model,
+            pk,
+            options=cast(list[ORMOption], load_options),
+            populate_existing=True,
+        )
+        if refreshed is not None:  # pragma: no branch
+            reloaded.append(refreshed)
+    return reloaded
+
+
+async def _refresh_loaded(
+    session: AsyncSession, instances: list[DeclarativeBase]
+) -> list[DeclarativeBase]:
+    """Re-select freshly written rows, eager-loading relationships."""
+    if not instances:
+        return []
+    refreshed: list[DeclarativeBase | None] = [None] * len(instances)
+    for model_cls, group in _group_by_type(instances):
+        positions = [i for i, inst in enumerate(instances) if type(inst) is model_cls]
+        load_options = _relationship_load_options(model_cls)
+        for pos, new in zip(
+            positions, await _reload_with_relationships(session, group, load_options)
+        ):
+            refreshed[pos] = new
+    return cast(list[DeclarativeBase], refreshed)
+
+
 async def _load_ordered(
     session: AsyncSession,
     registry: FixtureRegistry,
@@ -208,14 +294,11 @@ async def _load_ordered(
 
     for name in ordered_names:
         variants = (
-            registry.get_variants(name, *contexts)
+            registry.get_load_variants(name, *contexts)
             if contexts is not None
             else registry.get_variants(name)
         )
 
-        if contexts is not None and not variants:
-            variants = registry.get_variants(name)
-
         if not variants:  # pragma: no cover
             results[name] = []
             continue
@@ -229,7 +312,7 @@ async def _load_ordered(
         model_name = type(instances[0]).__name__
         loaded: list[DeclarativeBase] = []
 
-        async with get_transaction(session):
+        async with transaction(session):
             for model_cls, group in _group_by_type(instances):
                 match strategy:
                     case LoadStrategy.INSERT:
@@ -244,8 +327,10 @@ async def _load_ordered(
                     case _:  # pragma: no cover
                         pass
 
+            loaded = await _refresh_loaded(session, loaded)
+
         results[name] = loaded
-        logger.info(f"Loaded fixture '{name}': {len(loaded)} {model_name}(s)")
+        logger.info("Loaded fixture '%s': %d %s(s)", name, len(loaded), model_name)
 
     return results
 
@@ -264,56 +349,6 @@ def _get_primary_key(instance: DeclarativeBase) -> Any | None:
     return None
 
 
-def get_obj_by_attr(
-    fixtures: Callable[[], Sequence[ModelType]], attr_name: str, value: Any
-) -> ModelType:
-    """Get a SQLAlchemy model instance by matching an attribute value.
-
-    Args:
-        fixtures: A fixture function registered via ``@registry.register``
-            that returns a sequence of SQLAlchemy model instances.
-        attr_name: Name of the attribute to match against.
-        value: Value to match.
-
-    Returns:
-        The first model instance where the attribute matches the given value.
-
-    Raises:
-        StopIteration: If no matching object is found in the fixture group.
-    """
-    try:
-        return next(obj for obj in fixtures() if getattr(obj, attr_name) == value)
-    except StopIteration:
-        raise StopIteration(
-            f"No object with {attr_name}={value} found in fixture '{getattr(fixtures, '__name__', repr(fixtures))}'"
-        ) from None
-
-
-def get_field_by_attr(
-    fixtures: Callable[[], Sequence[ModelType]],
-    attr_name: str,
-    value: Any,
-    *,
-    field: str = "id",
-) -> Any:
-    """Get a single field value from a fixture object matched by an attribute.
-
-    Args:
-        fixtures: A fixture function registered via ``@registry.register``
-            that returns a sequence of SQLAlchemy model instances.
-        attr_name: Name of the attribute to match against.
-        value: Value to match.
-        field: Attribute name to return from the matched object (default: ``"id"``).
-
-    Returns:
-        The value of ``field`` on the first matching model instance.
-
-    Raises:
-        StopIteration: If no matching object is found in the fixture group.
-    """
-    return getattr(get_obj_by_attr(fixtures, attr_name, value), field)
-
-
 async def load_fixtures(
     session: AsyncSession,
     registry: FixtureRegistry,
@@ -348,8 +383,8 @@ async def load_fixtures_by_context(
     Args:
         session: Database session
         registry: Fixture registry
-        *contexts: Contexts to load (e.g., ``Context.BASE``, ``Context.TESTING``,
-            or plain strings for custom contexts)
+        *contexts: Contexts to load (e.g., ``Context.TESTING``, or plain
+            strings for custom contexts)
         strategy: How to handle existing records
 
     Returns:

src/fastapi_toolsets/models/watched.py

@@ -204,6 +204,11 @@ async def _invoke_callback(
         await result
 
 
+async def _reload_if_present(session: AsyncSession, obj: Any, state: Any) -> None:
+    """Re-populate *obj* from the DB if its row still exists."""
+    await session.get(type(obj), state.key[1], populate_existing=True)
+
+
 class EventSession(AsyncSession):
     """AsyncSession subclass that dispatches lifecycle callbacks after commit."""
 
@@ -253,7 +258,7 @@ class EventSession(AsyncSession):
                     state is None or state.detached or state.transient
                 ):  # pragma: no cover
                     continue
-                await self.refresh(obj)
+                await _reload_if_present(self, obj, state)
                 for handler in _get_handlers(type(obj), ModelEvent.CREATE):
                     await _invoke_callback(handler, obj, ModelEvent.CREATE, None)
             except Exception as exc:
@@ -277,7 +282,7 @@ class EventSession(AsyncSession):
                     state is None or state.detached or state.transient
                 ):  # pragma: no cover
                     continue
-                await self.refresh(obj)
+                await _reload_if_present(self, obj, state)
                 for handler in _get_handlers(type(obj), ModelEvent.UPDATE):
                     await _invoke_callback(handler, obj, ModelEvent.UPDATE, changes)
             except Exception as exc:

src/fastapi_toolsets/pytest/plugin.py

@@ -1,16 +1,14 @@
 """Pytest plugin for using FixtureRegistry fixtures in tests."""
 
-from collections.abc import Callable, Sequence
-from typing import Any, cast
+from collections.abc import Sequence
+from typing import Any
 
 import pytest
-from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.orm import DeclarativeBase, selectinload
-from sqlalchemy.orm.interfaces import ExecutableOption, ORMOption
+from sqlalchemy.orm import DeclarativeBase
 
-from ..db import get_transaction
 from ..fixtures import FixtureRegistry, LoadStrategy
+from ..fixtures.utils import _get_primary_key, _load_ordered, _refresh_loaded
 
 
 def register_fixtures(
@@ -57,7 +55,7 @@ def register_fixtures(
 
         # Build list of pytest fixture dependencies
         pytest_deps = [session_fixture]
-        for dep in fixture.depends_on:
+        for dep in registry.get_dependencies(fixture.name):
             pytest_deps.append(f"{prefix}{dep}")
 
         # Create the fixture function
@@ -83,56 +81,38 @@ def _create_fixture_function(
     fixture_name: str,
     dependencies: list[str],
     strategy: LoadStrategy,
-) -> Callable[..., Any]:
+) -> Any:
     """Create a fixture function with the correct signature.
 
     The function signature must include all dependencies as parameters
-    for pytest to resolve them correctly.
+    for pytest (and pytest-anyio's fixture chaining) to resolve them
+    correctly — dynamic resolution via ``request.getfixturevalue`` deadlocks
+    when called from inside an already-running async fixture.
     """
-    # Get the fixture definition
     fixture_def = registry.get(fixture_name)
 
-    # Build the function dynamically with correct parameters
-    # We need the session as first param, then all dependencies
     async def fixture_func(**kwargs: Any) -> Sequence[DeclarativeBase]:
-        # Get session from kwargs (first dependency)
         session: AsyncSession = kwargs[dependencies[0]]
+        result = (await _load_ordered(session, registry, [fixture_name], strategy))[
+            fixture_name
+        ]
 
-        # Load the fixture data
-        instances = list(fixture_def.func())
+        if strategy is LoadStrategy.SKIP_EXISTING:
+            # _load_ordered only returns newly-inserted rows for this
+            # strategy (the CLI seeding contract). A test fixture should
+            # still hand back the full, usable set including rows that
+            # were already present, so top up with those.
+            declared = list(fixture_def.func())
+            result_pks = {_get_primary_key(r) for r in result}
+            missing = [
+                d
+                for d in declared
+                if (pk := _get_primary_key(d)) is not None and pk not in result_pks
+            ]
+            if missing:
+                result = result + await _refresh_loaded(session, missing)
 
-        if not instances:
-            return []
-
-        loaded: list[DeclarativeBase] = []
-
-        async with get_transaction(session):
-            for instance in instances:
-                if strategy == LoadStrategy.INSERT:
-                    session.add(instance)
-                    loaded.append(instance)
-                elif strategy == LoadStrategy.MERGE:
-                    merged = await session.merge(instance)
-                    loaded.append(merged)
-                elif strategy == LoadStrategy.SKIP_EXISTING:  # pragma: no branch
-                    pk = _get_primary_key(instance)
-                    if pk is not None:
-                        existing = await session.get(type(instance), pk)
-                        if existing is None:
-                            session.add(instance)
-                            loaded.append(instance)
-                        else:
-                            loaded.append(existing)
-                    else:
-                        session.add(instance)
-                        loaded.append(instance)
-
-        if loaded:  # pragma: no branch
-            load_options = _relationship_load_options(type(loaded[0]))
-            if load_options:
-                return await _reload_with_relationships(session, loaded, load_options)
-
-        return loaded
+        return result
 
     # Update function signature to include dependencies
     # This is needed for pytest to inject the right fixtures
@@ -146,65 +126,3 @@ def _create_fixture_function(
     created_func.__doc__ = f"Load {fixture_name} fixture data."
 
     return created_func
-
-
-def _relationship_load_options(model: type[DeclarativeBase]) -> list[ExecutableOption]:
-    """Build selectinload options for all direct relationships on a model."""
-    return [
-        selectinload(getattr(model, rel.key)) for rel in model.__mapper__.relationships
-    ]
-
-
-async def _reload_with_relationships(
-    session: AsyncSession,
-    instances: list[DeclarativeBase],
-    load_options: list[ExecutableOption],
-) -> list[DeclarativeBase]:
-    """Reload instances in a single bulk query with relationship eager-loading.
-
-    Uses one SELECT … WHERE pk IN (…) so selectinload can batch all relationship
-    queries — 1 + N_relationships round-trips regardless of how many instances
-    there are, instead of one session.get() per instance.
-
-    Preserves the original insertion order.
-    """
-    model = type(instances[0])
-    mapper = model.__mapper__
-    pk_cols = mapper.primary_key
-
-    if len(pk_cols) == 1:
-        pk_attr = getattr(model, pk_cols[0].key)
-        pks = [getattr(inst, pk_cols[0].key) for inst in instances]
-        result = await session.execute(
-            select(model).where(pk_attr.in_(pks)).options(*load_options)
-        )
-        by_pk = {getattr(row, pk_cols[0].key): row for row in result.unique().scalars()}
-        return [by_pk[pk] for pk in pks]
-
-    # Composite PK: fall back to per-instance reload
-    reloaded: list[DeclarativeBase] = []
-    for instance in instances:
-        pk = _get_primary_key(instance)
-        refreshed = await session.get(
-            model,
-            pk,
-            options=cast(list[ORMOption], load_options),
-            populate_existing=True,
-        )
-        if refreshed is not None:  # pragma: no branch
-            reloaded.append(refreshed)
-    return reloaded
-
-
-def _get_primary_key(instance: DeclarativeBase) -> Any | None:
-    """Get the primary key value of a model instance."""
-    mapper = instance.__class__.__mapper__
-    pk_cols = mapper.primary_key
-
-    if len(pk_cols) == 1:
-        return getattr(instance, pk_cols[0].name, None)
-
-    pk_values = tuple(getattr(instance, col.name, None) for col in pk_cols)
-    if all(v is not None for v in pk_values):
-        return pk_values
-    return None

src/fastapi_toolsets/pytest/utils.py

@@ -15,7 +15,7 @@ from sqlalchemy.ext.asyncio import (
 )
 from sqlalchemy.orm import DeclarativeBase
 
-from ..db import cleanup_tables, create_database
+from ..db.testing import cleanup_tables, create_database
 from ..models.watched import EventSession
 
 

tests/test_cli.py

@@ -277,6 +277,10 @@ class TestFixturesCli:
             '@registry.register(depends_on=["roles"], contexts=[Context.TESTING])\n'
             "def users():\n"
             '    return [{"id": 1, "name": "alice", "role_id": 1}]\n'
+            "\n"
+            '@registry.register(contexts=["staging"])\n'
+            "def staging_only():\n"
+            '    return [{"id": 3, "name": "staging-user"}]\n'
         )
 
         # Create db module
@@ -316,7 +320,7 @@ class TestFixturesCli:
         assert result.exit_code == 0
         assert "roles" in result.output
         assert "users" in result.output
-        assert "Total: 2 fixture(s)" in result.output
+        assert "Total: 3 fixture(s)" in result.output
 
     def test_fixtures_list_with_context(self, cli_env):
         """fixtures list --context filters by context."""
@@ -338,6 +342,27 @@ class TestFixturesCli:
         assert "roles" in result.output
         assert "[Dry run - no changes made]" in result.output
 
+    def test_fixtures_list_with_custom_context(self, cli_env):
+        """fixtures list --context accepts contexts outside the Context enum, and
+        always includes base fixtures alongside the requested context."""
+        tmp_path, cli = cli_env
+        result = runner.invoke(cli, ["fixtures", "list", "--context", "staging"])
+
+        assert result.exit_code == 0
+        assert "staging_only" in result.output
+        assert "roles" in result.output
+        assert "Total: 2 fixture(s)" in result.output
+
+    def test_fixtures_load_custom_context_dry_run(self, cli_env):
+        """fixtures load accepts a custom context argument outside the Context enum,
+        and always loads base fixtures alongside it."""
+        tmp_path, cli = cli_env
+        result = runner.invoke(cli, ["fixtures", "load", "staging", "--dry-run"])
+
+        assert result.exit_code == 0
+        assert "staging_only" in result.output
+        assert "roles" in result.output
+
     def test_fixtures_load_invalid_strategy(self, cli_env):
         """fixtures load with invalid strategy shows error."""
         tmp_path, cli = cli_env

tests/test_example_pagination_search.py

@@ -91,13 +91,19 @@ async def seed(session: AsyncSession):
 class TestAppSessionDep:
     @pytest.mark.anyio
     async def test_get_db_yields_async_session(self):
-        """get_db yields a real AsyncSession when called directly."""
-        from docs_src.examples.pagination_search.db import get_db
+        """The Database dependency yields a real AsyncSession when called directly."""
+        from starlette.requests import Request
 
-        gen = get_db()
-        session = await gen.__anext__()
-        assert isinstance(session, AsyncSession)
-        await gen.aclose()
+        from fastapi_toolsets.db import Database
+
+        db = Database(DATABASE_URL)
+        try:
+            gen = db(Request({"type": "http", "headers": []}))
+            session = await gen.__anext__()
+            assert isinstance(session, AsyncSession)
+            await gen.aclose()
+        finally:
+            await db.engine.dispose()
 
 
 class TestOffsetPagination:

tests/test_fixtures.py

@@ -2,6 +2,7 @@
 
 import uuid
 from enum import Enum
+from typing import cast
 
 import pytest
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -10,8 +11,6 @@ from fastapi_toolsets.fixtures import (
     Context,
     FixtureRegistry,
     LoadStrategy,
-    get_field_by_attr,
-    get_obj_by_attr,
     load_fixtures,
     load_fixtures_by_context,
 )
@@ -267,7 +266,34 @@ class TestFixtureRegistry:
 
         testing_fixtures = registry.get_by_context(Context.TESTING)
         names = {f.name for f in testing_fixtures}
-        assert names == {"test_data"}
+        assert names == {"test_data", "base_data"}
+
+    def test_get_by_context_always_includes_base(self):
+        """Context.BASE fixtures load even for a fully custom context."""
+        registry = FixtureRegistry()
+
+        @registry.register(contexts=[Context.BASE])
+        def base_data():
+            return []
+
+        @registry.register(contexts=["staging"])
+        def staging_data():
+            return []
+
+        names = {f.name for f in registry.get_by_context("staging")}
+        assert names == {"staging_data", "base_data"}
+
+    def test_get_load_variants_falls_back_to_all_when_context_has_no_match(self):
+        """get_load_variants returns every variant if none match the requested
+        context (and none are Context.BASE either)."""
+        registry = FixtureRegistry()
+
+        @registry.register(contexts=["staging"])
+        def env_data():
+            return []
+
+        variants = registry.get_load_variants("env_data", "production")
+        assert [v.contexts for v in variants] == [["staging"]]
 
 
 class TestIncludeRegistry:
@@ -812,6 +838,45 @@ class TestLoadFixtures:
             db_session, registry, "int_roles", strategy=LoadStrategy.SKIP_EXISTING
         )
         assert len(result["int_roles"]) == 1
+        # The generated autoincrement PK must be written back onto the
+        # returned instance, not just visible via a fresh DB query.
+        assert cast(IntRole, result["int_roles"][0]).id is not None
+
+    @pytest.mark.anyio
+    async def test_insert_refreshes_autoincrement_pk_on_returned_instance(
+        self, db_session: AsyncSession
+    ):
+        """INSERT strategy writes the generated PK back onto the returned instance."""
+        registry = FixtureRegistry()
+
+        @registry.register
+        def int_roles():
+            return [IntRole(name="auto")]
+
+        result = await load_fixtures(
+            db_session, registry, "int_roles", strategy=LoadStrategy.INSERT
+        )
+        assert cast(IntRole, result["int_roles"][0]).id is not None
+
+    @pytest.mark.anyio
+    async def test_merge_refreshes_server_default_on_returned_instance(
+        self, db_session: AsyncSession
+    ):
+        """MERGE strategy refreshes the returned instance with server-generated values."""
+        registry = FixtureRegistry()
+
+        @registry.register
+        def challenges():
+            return [
+                Challenge(id=uuid.uuid4(), title="Solo", challenge_type="challenge")
+            ]
+
+        result = await load_fixtures(
+            db_session, registry, "challenges", strategy=LoadStrategy.MERGE
+        )
+        # `points` has a column default of 0 applied by the DB, never set on
+        # the in-memory instance — the returned object must reflect it.
+        assert cast(Challenge, result["challenges"][0]).points == 0
 
 
 class TestLoadFixturesByContext:
@@ -891,8 +956,8 @@ class TestLoadFixturesByContext:
         assert await UserCrud.count(db_session) == 1
 
 
-class TestGetObjByAttr:
-    """Tests for get_obj_by_attr helper function."""
+class TestRegistryObj:
+    """Tests for FixtureRegistry.obj."""
 
     def setup_method(self):
         """Set up test fixtures for each test."""
@@ -934,23 +999,20 @@ class TestGetObjByAttr:
                 ),
             ]
 
-        self.roles = roles
-        self.users = users
-
     def test_get_by_id(self):
         """Get an object by its id attribute."""
-        role = get_obj_by_attr(self.roles, "id", self.role_id_1)
-        assert role.name == "admin"
+        role = self.registry.obj("roles", "id", self.role_id_1)
+        assert cast(Role, role).name == "admin"
 
     def test_get_user_by_username(self):
         """Get a user by username."""
-        user = get_obj_by_attr(self.users, "username", "bob")
+        user = cast(User, self.registry.obj("users", "username", "bob"))
         assert user.id == self.user_id_2
         assert user.email == "bob@example.com"
 
     def test_returns_first_match(self):
         """Returns the first matching object when multiple could match."""
-        user = get_obj_by_attr(self.users, "role_id", self.role_id_1)
+        user = cast(User, self.registry.obj("users", "role_id", self.role_id_1))
         assert user.username == "alice"
 
     def test_no_match_raises_stop_iteration(self):
@@ -959,16 +1021,37 @@ class TestGetObjByAttr:
             StopIteration,
             match="No object with name=nonexistent found in fixture 'roles'",
         ):
-            get_obj_by_attr(self.roles, "name", "nonexistent")
+            self.registry.obj("roles", "name", "nonexistent")
 
     def test_no_match_on_wrong_value_type(self):
         """Raises StopIteration when value type doesn't match."""
         with pytest.raises(StopIteration):
-            get_obj_by_attr(self.roles, "id", "not-a-uuid")
+            self.registry.obj("roles", "id", "not-a-uuid")
+
+    def test_unknown_fixture_raises_key_error(self):
+        """Raises KeyError when the fixture name isn't registered."""
+        with pytest.raises(KeyError):
+            self.registry.obj("unknown", "id", self.role_id_1)
+
+    def test_searches_across_context_variants(self):
+        """obj() finds matches across all context variants of a fixture name, not just one."""
+        registry = FixtureRegistry()
+        tester_id = uuid.uuid4()
+
+        @registry.register(contexts=[Context.BASE])
+        def variant_users() -> list[User]:
+            return [User(id=uuid.uuid4(), username="admin", email="admin@x.com")]
+
+        @registry.register(contexts=[Context.TESTING])
+        def variant_users() -> list[User]:  # noqa: F811
+            return [User(id=tester_id, username="tester", email="tester@x.com")]
+
+        user = cast(User, registry.obj("variant_users", "username", "tester"))
+        assert user.id == tester_id
 
 
-class TestGetFieldByAttr:
-    """Tests for get_field_by_attr helper function."""
+class TestRegistryField:
+    """Tests for FixtureRegistry.field."""
 
     def setup_method(self):
         self.registry = FixtureRegistry()
@@ -984,22 +1067,20 @@ class TestGetFieldByAttr:
                 Role(id=role_id_2, name="user"),
             ]
 
-        self.roles = roles
-
     def test_returns_id_by_default(self):
         """Returns the id field when no field is specified."""
-        result = get_field_by_attr(self.roles, "name", "admin")
+        result = self.registry.field("roles", "name", "admin")
         assert result == self.role_id_1
 
     def test_returns_specified_field(self):
         """Returns the requested field instead of id."""
-        result = get_field_by_attr(self.roles, "id", self.role_id_2, field="name")
+        result = self.registry.field("roles", "id", self.role_id_2, field="name")
         assert result == "user"
 
     def test_no_match_raises_stop_iteration(self):
-        """Propagates StopIteration from get_obj_by_attr when no match found."""
+        """Propagates StopIteration from obj() when no match found."""
         with pytest.raises(StopIteration, match="No object with name=missing"):
-            get_field_by_attr(self.roles, "name", "missing")
+            self.registry.field("roles", "name", "missing")
 
 
 class TestGetPrimaryKey:

tests/test_models.py

@@ -21,12 +21,12 @@ from fastapi_toolsets.models import (
     listens_for,
 )
 from fastapi_toolsets.models.watched import (
-    EventSession,
     _EVENT_HANDLERS,
     _SESSION_CREATES,
     _SESSION_DELETES,
     _SESSION_UPDATES,
     _WATCHED_MODELS,
+    EventSession,
     _after_flush,
     _after_rollback,
     _get_watched_fields,
@@ -1001,6 +1001,57 @@ class TestEventCallbacks:
 
         assert _test_events == []
 
+    @pytest.mark.anyio
+    async def test_create_survives_row_deleted_before_reload(self, mixin_session):
+        """A row deleted by another transaction right after commit still fires CREATE."""
+        keep = WatchedModel(status="active", other="x")
+        doomed = WatchedModel(status="active", other="x")
+        mixin_session.add_all([keep, doomed])
+        await mixin_session.flush()
+        doomed_id = doomed.id
+
+        raced = {"done": False}
+
+        async def kill_doomed_row_once():
+            if raced["done"]:
+                return
+            raced["done"] = True
+            engine = create_async_engine(DATABASE_URL, echo=False)
+            async with async_sessionmaker(engine)() as other:
+                row = await other.get(WatchedModel, doomed_id)
+                await other.delete(row)
+                await other.commit()
+            await engine.dispose()
+
+        real_get = mixin_session.get
+        real_refresh = mixin_session.refresh
+
+        def _matches_doomed(pk):
+            return pk == doomed_id or (isinstance(pk, tuple) and pk[0] == doomed_id)
+
+        async def racing_get(model, pk, *args, **kwargs):
+            if _matches_doomed(pk):
+                await kill_doomed_row_once()
+            return await real_get(model, pk, *args, **kwargs)
+
+        async def racing_refresh(obj, *args, **kwargs):
+            if getattr(obj, "id", None) == doomed_id:
+                await kill_doomed_row_once()
+            return await real_refresh(obj, *args, **kwargs)
+
+        # Patch both possible reload mechanisms (session.get / session.refresh)
+        # so this test still exercises the race regardless of which one
+        # EventSession.commit() uses internally to pick up server defaults.
+        mixin_session.get = racing_get
+        mixin_session.refresh = racing_refresh
+        with patch.object(_watched_module._logger, "error") as mock_error:
+            await mixin_session.commit()
+            mock_error.assert_not_called()
+
+        assert raced["done"]
+        created_ids = {e["obj_id"] for e in _test_events if e["event"] == "create"}
+        assert created_ids == {keep.id, doomed_id}
+
 
 class TestTransientObject:
     """Create + delete within the same transaction should fire no events."""
@@ -1506,8 +1557,8 @@ class TestListensFor:
         assert all(e["event"] == "change" for e in _listener_events)
 
 
-class TestEventSessionWithGetTransaction:
-    """Verify callbacks fire correctly when using get_transaction / lock_tables."""
+class TestEventSessionWithTransaction:
+    """Verify callbacks fire correctly when using transaction / lock_tables."""
 
     @pytest.fixture(autouse=True)
     def clear_events(self):
@@ -1517,10 +1568,10 @@ class TestEventSessionWithGetTransaction:
 
     @pytest.mark.anyio
     async def test_callbacks_fire_after_outer_commit_not_savepoint(self, mixin_session):
-        """get_transaction creates a savepoint; callbacks fire only on outer commit."""
-        from fastapi_toolsets.db import get_transaction
+        """transaction creates a savepoint; callbacks fire only on outer commit."""
+        from fastapi_toolsets.db import transaction
 
-        async with get_transaction(mixin_session):
+        async with transaction(mixin_session):
             obj = WatchedModel(status="active", other="x")
             mixin_session.add(obj)
 
@@ -1535,14 +1586,14 @@ class TestEventSessionWithGetTransaction:
 
     @pytest.mark.anyio
     async def test_nested_transactions_accumulate_events(self, mixin_session):
-        """Multiple get_transaction blocks accumulate events for a single commit."""
-        from fastapi_toolsets.db import get_transaction
+        """Multiple transaction blocks accumulate events for a single commit."""
+        from fastapi_toolsets.db import transaction
 
-        async with get_transaction(mixin_session):
+        async with transaction(mixin_session):
             obj1 = WatchedModel(status="first", other="x")
             mixin_session.add(obj1)
 
-        async with get_transaction(mixin_session):
+        async with transaction(mixin_session):
             obj2 = WatchedModel(status="second", other="y")
             mixin_session.add(obj2)
 
@@ -1556,14 +1607,14 @@ class TestEventSessionWithGetTransaction:
     @pytest.mark.anyio
     async def test_savepoint_rollback_suppresses_events(self, mixin_session):
         """Objects from a rolled-back savepoint don't fire callbacks."""
-        from fastapi_toolsets.db import get_transaction
+        from fastapi_toolsets.db import transaction
 
         survivor = WatchedModel(status="kept", other="x")
         mixin_session.add(survivor)
         await mixin_session.flush()
 
         try:
-            async with get_transaction(mixin_session):
+            async with transaction(mixin_session):
                 doomed = WatchedModel(status="doomed", other="y")
                 mixin_session.add(doomed)
                 await mixin_session.flush()
@@ -1590,9 +1641,9 @@ class TestEventSessionWithGetTransaction:
         assert len(creates) == 1
 
     @pytest.mark.anyio
-    async def test_update_inside_get_transaction(self, mixin_session):
-        """UPDATE events fire with correct changes after get_transaction commit."""
-        from fastapi_toolsets.db import get_transaction
+    async def test_update_inside_transaction(self, mixin_session):
+        """UPDATE events fire with correct changes after transaction commit."""
+        from fastapi_toolsets.db import transaction
 
         obj = WatchedModel(status="initial", other="x")
         mixin_session.add(obj)
@@ -1600,7 +1651,7 @@ class TestEventSessionWithGetTransaction:
 
         _test_events.clear()
 
-        async with get_transaction(mixin_session):
+        async with transaction(mixin_session):
             obj.status = "updated"
 
         await mixin_session.commit()
@@ -1696,7 +1747,7 @@ class TestEventSessionWithNullableFields:
 
 
 class TestEventSessionWithFastAPIDependency:
-    """Verify EventSession works when session comes from create_db_dependency."""
+    """Verify EventSession works when session comes from the Database dependency."""
 
     @pytest.fixture(autouse=True)
     def clear_events(self):
@@ -1706,31 +1757,24 @@ class TestEventSessionWithFastAPIDependency:
 
     @pytest.mark.anyio
     async def test_create_event_fires_via_dependency(self):
-        """CREATE callback fires when session is provided by create_db_dependency."""
+        """CREATE callback fires when session is provided by the Database dependency."""
         from fastapi import Depends, FastAPI
         from httpx import ASGITransport, AsyncClient
-        from sqlalchemy.ext.asyncio import (
-            AsyncSession,
-            async_sessionmaker,
-            create_async_engine,
-        )
+        from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 
-        from fastapi_toolsets.db import create_db_dependency
+        from fastapi_toolsets.db import Database
         from fastapi_toolsets.models import EventSession
 
         engine = create_async_engine(DATABASE_URL, echo=False)
-        session_factory = async_sessionmaker(
-            engine, expire_on_commit=False, class_=EventSession
-        )
 
         async with engine.begin() as conn:
             await conn.run_sync(MixinBase.metadata.create_all)
 
-        get_db = create_db_dependency(session_factory)
+        db = Database(engine=engine, session_class=EventSession)
         app = FastAPI()
 
         @app.post("/watched")
-        async def create_watched(session: AsyncSession = Depends(get_db)):
+        async def create_watched(session: AsyncSession = Depends(db)):
             obj = WatchedModel(status="from-api", other="x")
             session.add(obj)
             return {"id": str(obj.id)}
@@ -1753,40 +1797,33 @@ class TestEventSessionWithFastAPIDependency:
 
     @pytest.mark.anyio
     async def test_update_event_fires_via_dependency(self):
-        """UPDATE callback fires when session is provided by create_db_dependency."""
+        """UPDATE callback fires when session is provided by the Database dependency."""
         from fastapi import Depends, FastAPI
         from httpx import ASGITransport, AsyncClient
-        from sqlalchemy.ext.asyncio import (
-            AsyncSession,
-            async_sessionmaker,
-            create_async_engine,
-        )
+        from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 
-        from fastapi_toolsets.db import create_db_dependency
+        from fastapi_toolsets.db import Database
         from fastapi_toolsets.models import EventSession
 
         engine = create_async_engine(DATABASE_URL, echo=False)
-        session_factory = async_sessionmaker(
-            engine, expire_on_commit=False, class_=EventSession
-        )
 
         async with engine.begin() as conn:
             await conn.run_sync(MixinBase.metadata.create_all)
 
-        get_db = create_db_dependency(session_factory)
+        db = Database(engine=engine, session_class=EventSession)
         app = FastAPI()
 
         # Pre-seed an object.
-        async with session_factory() as seed_session:
+        async with db.session() as seed_session:
             obj = WatchedModel(status="initial", other="x")
             seed_session.add(obj)
-            await seed_session.commit()
+            await seed_session.flush()
             obj_id = obj.id
 
         _test_events.clear()
 
         @app.put("/watched/{item_id}")
-        async def update_watched(item_id: str, session: AsyncSession = Depends(get_db)):
+        async def update_watched(item_id: str, session: AsyncSession = Depends(db)):
             from sqlalchemy import select
 
             stmt = select(WatchedModel).where(WatchedModel.id == item_id)

tests/test_pytest.py

@@ -11,7 +11,7 @@ from sqlalchemy.engine import make_url
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
 
-from fastapi_toolsets.db import get_transaction
+from fastapi_toolsets.db import transaction
 from fastapi_toolsets.fixtures import Context, FixtureRegistry, LoadStrategy
 from fastapi_toolsets.pytest import (
     create_async_client,
@@ -20,7 +20,7 @@ from fastapi_toolsets.pytest import (
     register_fixtures,
     worker_database_url,
 )
-from fastapi_toolsets.pytest.plugin import (
+from fastapi_toolsets.fixtures.utils import (
     _get_primary_key,
     _relationship_load_options,
     _reload_with_relationships,
@@ -387,14 +387,14 @@ class TestCreateDbSession:
             assert session.autoflush is False
 
     @pytest.mark.anyio
-    async def test_get_transaction_commits_visible_to_separate_session(self):
-        """Data written via get_transaction() is committed and visible to other sessions."""
+    async def test_transaction_commits_visible_to_separate_session(self):
+        """Data written via transaction() is committed and visible to other sessions."""
         role_id = uuid.uuid4()
 
         async with create_db_session(DATABASE_URL, Base, drop_tables=False) as session:
-            # Simulate what _create_fixture_function does: insert via get_transaction
+            # Simulate what _create_fixture_function does: insert via transaction()
             # with no explicit commit afterward.
-            async with get_transaction(session):
+            async with transaction(session):
                 role = Role(id=role_id, name="visible_to_other_session")
                 session.add(role)
 
@@ -409,9 +409,9 @@ class TestCreateDbSession:
                     result = await other.execute(select(Role).where(Role.id == role_id))
                     fetched = result.scalar_one_or_none()
                     assert fetched is not None, (
-                        "Fixture data inserted via get_transaction() must be committed "
+                        "Fixture data inserted via transaction() must be committed "
                         "and visible to a separate session. If create_db_session uses "
-                        "create_db_context, auto-begin forces get_transaction() into "
+                        "db.session(), auto-begin forces transaction() into "
                         "savepoints instead of real commits."
                     )
                     assert fetched.name == "visible_to_other_session"

uv.lock

@@ -330,7 +330,7 @@ wheels = [
 
 [[package]]
 name = "fastapi-toolsets"
-version = "4.1.3"
+version = "5.0.0b2"
 source = { editable = "." }
 dependencies = [
     { name = "asyncpg" },
@@ -1262,15 +1262,15 @@ asyncio = [
 
 [[package]]
 name = "starlette"
-version = "1.0.1"
+version = "1.3.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "anyio" },
     { name = "typing-extensions", marker = "python_full_version < '3.13'" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/08/a3/84e821cc54b4ab50ae6dbc6ac3800a651b65ec35f045cc73785380654057/starlette-1.0.1.tar.gz", hash = "sha256:512399c5f1de7fac99c88572212ded9ddeddef2fb32afa82d724000e88b38f4f", size = 2659596, upload-time = "2026-05-21T21:58:58.433Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/eb/e3/7c1dc7381d9f8ab7d854328ebfa884e62cb3f3d8549ddfd37c7814f42afa/starlette-1.3.1.tar.gz", hash = "sha256:05d0213193f2fbaae60e2ecb593b4add4262ad4e46536b54abe36f11a71724e0", size = 2703240, upload-time = "2026-06-12T09:23:11.602Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/ec/e1/b2df4bc09a1e51ff664c1e17018a4274b42e5e9352e4a478ea540512dc88/starlette-1.0.1-py3-none-any.whl", hash = "sha256:7c0e69b2ee1c848bd54669d908500117a3ee13de603a21427e5c6fc1adf98dcd", size = 72802, upload-time = "2026-05-21T21:58:56.551Z" },
+    { url = "https://files.pythonhosted.org/packages/ec/bb/2799cc2ede3ed41131f8975621e7213dfc7ef4acbbaadfa440f32500c370/starlette-1.3.1-py3-none-any.whl", hash = "sha256:c7372aae11c3c3f26a42df7bd626cec2f47d03483d261d369516a615a53714c6", size = 73632, upload-time = "2026-06-12T09:23:10.017Z" },
 ]
 
 [[package]]