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/db.md

@@ -31,6 +31,25 @@ async def list_users(session: AsyncSession = Depends(db)):
 
 The `Database` instance **is** the dependency: use it directly as `Depends(db)`. The whole request runs as a single transaction (CRUD writes use savepoints under it).
 
+The **URL** may be a plain string or a Pydantic [`PostgresDsn`](https://docs.pydantic.dev/latest/api/networks/#pydantic.networks.PostgresDsn). In URL mode you can tune the engine: pass `connect_args` for DBAPI-level options and any other keyword for `create_async_engine` (e.g. `pool_size`, `echo`, `pool_pre_ping`):
+
+```python
+from pydantic_settings import BaseSettings
+from pydantic import PostgresDsn
+
+class Settings(BaseSettings):
+    database_url: PostgresDsn
+
+settings = Settings()
+
+db = Database(
+    settings.database_url,
+    pool_size=20,
+    pool_pre_ping=True,
+    connect_args={"server_settings": {"application_name": "myapp"}},
+)
+```
+
 ## Committing before the response
 
 [`db.install(app)`](../reference/db.md#fastapi_toolsets.db.Database) adds a middleware that commits the request's session when the response starts, after the endpoint returns and before the body is sent. With the middleware installed, the dependency does not commit again.

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/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

pyproject.toml

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

src/fastapi_toolsets/__init__.py

@@ -24,4 +24,4 @@ Example usage:
         return Response(data={"user": user.username}, message="Success")
 """
 
-__version__ = "5.0.0b1"
+__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/db/core.py

@@ -4,6 +4,7 @@ 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,
@@ -84,18 +85,20 @@ class Database:
     untouched).
 
     Args:
-        url: Database connection URL (e.g. ``"postgresql+asyncpg://..."``).
+        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, e.g. ``pool_size``,
-            ``echo``, ``connect_args``).
+            :func:`create_async_engine` (URL mode only).
 
     Raises:
         TypeError: If neither or both of *url* and *engine* are given, or if
-            *engine_options* are passed together with *engine*.
+            *connect_args*/*engine_options* are passed together with *engine*.
 
     Example:
         ```python
@@ -115,12 +118,13 @@ class Database:
 
     def __init__(
         self,
-        url: str | None = None,
+        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):
@@ -128,10 +132,10 @@ class Database:
                 "Database requires exactly one of 'url' or 'engine' "
                 "(got both or neither)."
             )
-        if engine is not None and engine_options:
+        if engine is not None and (engine_options or connect_args is not None):
             raise TypeError(
-                "engine_options are only valid in URL mode; configure the "
-                "engine you pass via 'engine=' yourself."
+                "connect_args/engine_options are only valid in URL mode; "
+                "configure the engine you pass via 'engine=' yourself."
             )
 
         if engine is not None:
@@ -140,7 +144,11 @@ class Database:
         else:
             assert url is not None  # guaranteed by the XOR check above
             self._owns_engine = True
-            self.engine = create_async_engine(url, **engine_options)
+            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,

src/fastapi_toolsets/db/watch.py

@@ -57,34 +57,50 @@ async def wait_for_row_change(
         )
         ```
     """
-    instance = await session.get(model, pk_value)
-    if instance is None:
-        raise NotFoundError(f"{model.__name__} with pk={pk_value!r} not found")
+    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:
 
-    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)
+        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} was deleted")
+            raise NotFoundError(f"{model.__name__} with pk={pk_value!r} not found")
 
-        current = {col: getattr(instance, col) for col in watch_cols}
-        if current != initial:
-            return instance
+        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 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
@@ -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 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 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

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_db.py

@@ -3,12 +3,13 @@
 import asyncio
 import uuid
 from contextlib import asynccontextmanager
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from fastapi import Depends, FastAPI
 from fastapi.responses import StreamingResponse
 from httpx import ASGITransport, AsyncClient
+from pydantic import PostgresDsn
 from sqlalchemy import (
     Column,
     ForeignKey,
@@ -94,6 +95,33 @@ class TestDatabaseConstruction:
         with pytest.raises(TypeError):
             Database(engine=engine, pool_size=5)
 
+    @pytest.mark.anyio
+    async def test_connect_args_with_engine_raises(self, engine):
+        """connect_args are rejected in engine= mode."""
+        with pytest.raises(TypeError):
+            Database(engine=engine, connect_args={"server_settings": {}})
+
+    @pytest.mark.anyio
+    async def test_accepts_postgres_dsn(self):
+        """A Pydantic PostgresDsn is coerced to a string URL."""
+        dsn = PostgresDsn(DATABASE_URL)
+        db = Database(dsn)
+        try:
+            assert db._owns_engine is True
+            assert str(db.engine.url) == str(create_async_engine(DATABASE_URL).url)
+        finally:
+            await db.engine.dispose()
+
+    @pytest.mark.anyio
+    async def test_connect_args_forwarded_to_engine(self):
+        """connect_args are forwarded to create_async_engine in URL mode."""
+        connect_args = {"server_settings": {"application_name": "ft_test"}}
+        with patch("fastapi_toolsets.db.core.create_async_engine") as mocked:
+            mocked.return_value = MagicMock()
+            Database(DATABASE_URL, connect_args=connect_args)
+        _, kwargs = mocked.call_args
+        assert kwargs["connect_args"] == connect_args
+
     @pytest.mark.anyio
     async def test_url_mode_owns_engine(self):
         """URL mode builds and owns the engine."""
@@ -661,6 +689,13 @@ class TestWaitForRowChange:
         with pytest.raises(NotFoundError, match="not found"):
             await wait_for_row_change(db_session, Role, fake_id, interval=0.05)
 
+    @pytest.mark.anyio
+    async def test_unbound_session_raises_type_error(self):
+        """Raises TypeError when the session has no bind to open a watcher on."""
+        unbound = AsyncSession()
+        with pytest.raises(TypeError, match="requires a session bound to an engine"):
+            await wait_for_row_change(unbound, Role, uuid.uuid4())
+
     @pytest.mark.anyio
     async def test_timeout_raises(self, db_session: AsyncSession):
         """Raises TimeoutError when no change is detected within timeout."""
@@ -673,6 +708,66 @@ class TestWaitForRowChange:
                 db_session, Role, role.id, interval=0.05, timeout=0.2
             )
 
+    @pytest.mark.anyio
+    async def test_detects_update_under_repeatable_read(self, engine):
+        """Detects external commits even when the watcher pins a snapshot."""
+        async with engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+        rr_engine = engine.execution_options(isolation_level="REPEATABLE READ")
+        factory = async_sessionmaker(rr_engine, expire_on_commit=False)
+        try:
+            async with factory() as setup:
+                role = Role(name="rr_role")
+                setup.add(role)
+                await setup.commit()
+                role_id = role.id
+
+            async def update_later():
+                await asyncio.sleep(0.15)
+                async with factory() as other:
+                    r = await other.get(Role, role_id)
+                    assert r is not None
+                    r.name = "rr_updated"
+                    await other.commit()
+
+            watcher = factory()
+            try:
+                # Pin a snapshot before the update lands.
+                await watcher.get(Role, role_id)
+                update_task = asyncio.create_task(update_later())
+                result = await wait_for_row_change(
+                    watcher, Role, role_id, interval=0.05, timeout=2.0
+                )
+                await update_task
+                assert result.name == "rr_updated"
+            finally:
+                await watcher.close()
+        finally:
+            async with engine.begin() as conn:
+                await conn.run_sync(Base.metadata.drop_all)
+
+    @pytest.mark.anyio
+    async def test_stale_then_deleted_instance_raises_not_found(
+        self, db_session: AsyncSession, engine
+    ):
+        """A stale expired instance in the identity map yields NotFoundError."""
+        role = Role(name="stale_role")
+        db_session.add(role)
+        await db_session.commit()
+        role_id = role.id
+
+        # db_session still holds `role`; delete it from another committed session.
+        factory = async_sessionmaker(engine, expire_on_commit=False)
+        async with factory() as other:
+            r = await other.get(Role, role_id)
+            await other.delete(r)
+            await other.commit()
+
+        with pytest.raises(NotFoundError):
+            await wait_for_row_change(
+                db_session, Role, role_id, interval=0.05, timeout=0.5
+            )
+
     @pytest.mark.anyio
     async def test_deleted_row_raises(self, db_session: AsyncSession, engine):
         """Raises NotFoundError when the row is deleted during polling."""
@@ -693,6 +788,43 @@ class TestWaitForRowChange:
             await wait_for_row_change(db_session, Role, role.id, interval=0.05)
         await delete_task
 
+    @pytest.mark.anyio
+    async def test_does_not_disturb_ambient_transaction(
+        self, db_session: AsyncSession, engine
+    ):
+        """A read-only ambient transaction around the call survives untouched."""
+        role = Role(name="ambient_role")
+        db_session.add(role)
+        await db_session.commit()
+
+        async def update_later():
+            await asyncio.sleep(0.15)
+            factory = async_sessionmaker(engine, expire_on_commit=False)
+            async with factory() as other:
+                r = await other.get(Role, role.id)
+                assert r is not None
+                r.name = "ambient_updated"
+                await other.commit()
+
+        update_task = asyncio.create_task(update_later())
+        async with transaction(db_session):
+            # A read before the watch, establishing an ambient transaction
+            # that must remain usable once wait_for_row_change returns.
+            await db_session.get(Role, role.id)
+            result = await wait_for_row_change(
+                db_session, Role, role.id, interval=0.05, timeout=2.0
+            )
+            await update_task
+            assert result.name == "ambient_updated"
+            # The ambient transaction must still be open and usable here.
+            assert db_session.in_transaction()
+            other_role = Role(name="added_within_ambient_tx")
+            db_session.add(other_role)
+
+        # transaction() committed cleanly on exit; the write above landed.
+        check = await db_session.get(Role, other_role.id)
+        assert check is not None
+
 
 class TestCreateDatabase:
     """Tests for create_database."""

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

tests/test_pytest.py

@@ -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,

uv.lock

@@ -330,7 +330,7 @@ wheels = [
 
 [[package]]
 name = "fastapi-toolsets"
-version = "5.0.0b1"
+version = "5.0.0b2"
 source = { editable = "." }
 dependencies = [
     { name = "asyncpg" },