fix: PathDependency and BodyDependency build a bare unconfigured CRUD

This commit is contained in:
2026-08-29 08:39:23 -04:00
parent 716e4f7db7
commit 943115562b
3 changed files with 183 additions and 56 deletions
+30
View File
@@ -62,6 +62,36 @@ async def create_user(body: UserCreateSchema, role: Role = RoleDep):
...
```
## Eager loading
By default both factories fetch through a bare `CrudFactory(model)`, so relationships are not loaded. Pass `load_options` for a one-off, or `crud` to reuse a CRUD class you already configured:
```python
from sqlalchemy.orm import selectinload
from fastapi_toolsets.crud import CrudFactory
from fastapi_toolsets.dependencies import PathDependency
UserDep = PathDependency(
model=User,
field=User.id,
session_dep=get_db,
load_options=[selectinload(User.role)],
)
# Or reuse the app's configured CRUD and its default_load_options
UserCrud = CrudFactory(User, default_load_options=[selectinload(User.role)])
UserDep = PathDependency(model=User, field=User.id, session_dep=get_db, crud=UserCrud)
@router.get("/users/{user_id}")
async def get_user(user: User = UserDep):
return user.role.name # already loaded, no extra query
```
Both parameters work the same way on `BodyDependency`. When given together, the
usual [relationship loading](crud.md#relationship-loading) precedence applies.
---
[:material-api: API Reference](../reference/dependencies.md)
+74 -55
View File
@@ -2,14 +2,15 @@
import inspect
import typing
from collections.abc import Callable
from collections.abc import Callable, Sequence
from typing import Any, cast
from fastapi import Depends
from fastapi.params import Depends as DependsClass
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql.base import ExecutableOption
from .crud import CrudFactory
from .crud import AsyncCrud, CrudFactory
from .types import ModelType, SessionDependency
__all__ = ["BodyDependency", "PathDependency"]
@@ -24,12 +25,59 @@ def _unwrap_session_dep(session_dep: SessionDependency) -> Callable[..., Any]:
return session_dep
def _fetch_dependency(
model: type[ModelType],
field: Any,
*,
session_dep: SessionDependency,
param_name: str,
crud: type[AsyncCrud[ModelType]] | None,
load_options: Sequence[ExecutableOption] | None,
) -> ModelType:
"""Build a Depends() that fetches one row by ``field == <param_name>``."""
session_callable = _unwrap_session_dep(session_dep)
if crud is not None and crud.model is not model:
raise ValueError(
f"crud is bound to {crud.model.__name__}, not {model.__name__}"
)
crud = crud or CrudFactory(model)
# `session` has no default here: the __signature__ override below is what
# FastAPI reads, and it always passes `session` explicitly.
async def dependency(session: AsyncSession, **kwargs: Any) -> ModelType:
return await crud.get(
session,
filters=[field == kwargs[param_name]],
load_options=load_options,
)
dependency.__signature__ = inspect.Signature( # ty:ignore[unresolved-attribute]
parameters=[
inspect.Parameter(
param_name,
inspect.Parameter.KEYWORD_ONLY,
annotation=field.type.python_type,
),
inspect.Parameter(
"session",
inspect.Parameter.KEYWORD_ONLY,
annotation=AsyncSession,
default=Depends(session_callable),
),
]
)
return cast(ModelType, Depends(cast(Callable[..., ModelType], dependency)))
def PathDependency(
model: type[ModelType],
field: Any,
*,
session_dep: SessionDependency,
param_name: str | None = None,
crud: type[AsyncCrud[ModelType]] | None = None,
load_options: Sequence[ExecutableOption] | None = None,
) -> ModelType:
"""Create a dependency that fetches a DB object from a path parameter.
@@ -38,6 +86,10 @@ def PathDependency(
field: Model field to filter by (e.g., User.id)
session_dep: Session dependency function (e.g., get_db)
param_name: Path parameter name (defaults to model_field, e.g., user_id)
crud: Existing CRUD class to fetch with, so its ``default_load_options``
apply. Defaults to a bare ``CrudFactory(model)``.
load_options: SQLAlchemy loader options for the fetch. Overrides the CRUD's
``default_load_options`` entirely rather than merging with them.
Returns:
A Depends() instance that resolves to the model instance
@@ -55,36 +107,14 @@ def PathDependency(
): ...
```
"""
session_callable = _unwrap_session_dep(session_dep)
crud = CrudFactory(model)
name = (
param_name
if param_name is not None
else f"{model.__name__.lower()}_{field.key}"
return _fetch_dependency(
model,
field,
session_dep=session_dep,
param_name=param_name or f"{model.__name__.lower()}_{field.key}",
crud=crud,
load_options=load_options,
)
python_type = field.type.python_type
async def dependency(
session: AsyncSession = Depends(session_callable), **kwargs: Any
) -> ModelType:
value = kwargs[name]
return await crud.get(session, filters=[field == value])
dependency.__signature__ = inspect.Signature( # ty:ignore[unresolved-attribute]
parameters=[
inspect.Parameter(
name, inspect.Parameter.KEYWORD_ONLY, annotation=python_type
),
inspect.Parameter(
"session",
inspect.Parameter.KEYWORD_ONLY,
annotation=AsyncSession,
default=Depends(session_callable),
),
]
)
return cast(ModelType, Depends(cast(Callable[..., ModelType], dependency)))
def BodyDependency(
@@ -93,6 +123,8 @@ def BodyDependency(
*,
session_dep: SessionDependency,
body_field: str,
crud: type[AsyncCrud[ModelType]] | None = None,
load_options: Sequence[ExecutableOption] | None = None,
) -> ModelType:
"""Create a dependency that fetches a DB object from a body field.
@@ -101,6 +133,10 @@ def BodyDependency(
field: Model field to filter by (e.g., User.id)
session_dep: Session dependency function (e.g., get_db)
body_field: Name of the field in the request body
crud: Existing CRUD class to fetch with, so its ``default_load_options``
apply. Defaults to a bare ``CrudFactory(model)``.
load_options: SQLAlchemy loader options for the fetch. Overrides the CRUD's
``default_load_options`` entirely rather than merging with them.
Returns:
A Depends() instance that resolves to the model instance
@@ -120,28 +156,11 @@ def BodyDependency(
): ...
```
"""
session_callable = _unwrap_session_dep(session_dep)
crud = CrudFactory(model)
python_type = field.type.python_type
async def dependency(
session: AsyncSession = Depends(session_callable), **kwargs: Any
) -> ModelType:
value = kwargs[body_field]
return await crud.get(session, filters=[field == value])
dependency.__signature__ = inspect.Signature( # ty:ignore[unresolved-attribute]
parameters=[
inspect.Parameter(
body_field, inspect.Parameter.KEYWORD_ONLY, annotation=python_type
),
inspect.Parameter(
"session",
inspect.Parameter.KEYWORD_ONLY,
annotation=AsyncSession,
default=Depends(session_callable),
),
]
return _fetch_dependency(
model,
field,
session_dep=session_dep,
param_name=body_field,
crud=crud,
load_options=load_options,
)
return cast(ModelType, Depends(cast(Callable[..., ModelType], dependency)))
+79 -1
View File
@@ -7,15 +7,18 @@ from typing import Annotated, Any, cast
import pytest
from fastapi.params import Depends
from sqlalchemy import inspect as sa_inspect
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from fastapi_toolsets.crud import CrudFactory
from fastapi_toolsets.dependencies import (
BodyDependency,
PathDependency,
_unwrap_session_dep,
)
from .conftest import Role, RoleCreate, RoleCrud, User
from .conftest import Role, RoleCreate, RoleCrud, User, UserCreate, UserCrud
async def mock_get_db() -> AsyncGenerator[AsyncSession, None]:
@@ -275,3 +278,78 @@ class TestBodyDependency:
assert result.id == role.id
assert result.name == "body_annotated_role"
class TestDependencyLoadOptions:
"""Both factories can eager-load relations instead of using a bare CRUD."""
@staticmethod
async def _make_user(db_session):
role = await RoleCrud.create(db_session, RoleCreate(name="load_opts_role"))
user = await UserCrud.create(
db_session,
UserCreate(username="load_opts", email="load@opts", role_id=role.id),
)
db_session.expunge_all()
return user
@pytest.mark.anyio
async def test_bare_crud_leaves_relation_unloaded(self, db_session):
"""Baseline: without options the relation is not loaded (what the ticket reports)."""
user = await self._make_user(db_session)
dep = cast(Any, PathDependency(User, User.id, session_dep=mock_get_db))
result = await dep.dependency(session=db_session, user_id=user.id)
assert "role" in sa_inspect(result).unloaded
@pytest.mark.anyio
async def test_load_options_and_crud_eager_load(self, db_session):
"""Every way of asking for eager loading, on both factories, actually loads."""
user = await self._make_user(db_session)
eager = [selectinload(User.role)]
eager_crud = CrudFactory(User, default_load_options=eager)
deps = {
"path/load_options": PathDependency(
User, User.id, session_dep=mock_get_db, load_options=eager
),
"path/crud": PathDependency(
User, User.id, session_dep=mock_get_db, crud=eager_crud
),
"body/load_options": BodyDependency(
User,
User.id,
session_dep=mock_get_db,
body_field="user_id",
load_options=eager,
),
"body/crud": BodyDependency(
User,
User.id,
session_dep=mock_get_db,
body_field="user_id",
crud=eager_crud,
),
}
for label, dep in deps.items():
# Drop the identity map, or the next fetch reuses the already-loaded
# instance and the assertion passes for the wrong reason.
db_session.expunge_all()
result = await cast(Any, dep).dependency(
session=db_session, user_id=user.id
)
assert "role" not in sa_inspect(result).unloaded, label
assert result.role.name == "load_opts_role", label
def test_crud_bound_to_another_model_is_rejected(self):
"""A crud= for a different model would silently query the wrong table.
``ty`` rejects this statically; the runtime guard covers untyped callers.
"""
with pytest.raises(ValueError, match="bound to Role, not User"):
PathDependency(
User, User.id, session_dep=mock_get_db, crud=cast(Any, RoleCrud)
)