From 97ab10edcdb45b479b5677ac1225502624408713 Mon Sep 17 00:00:00 2001 From: d3vyce <44915747+d3vyce@users.noreply.github.com> Date: Wed, 4 Feb 2026 16:38:08 +0100 Subject: [PATCH] feat/dependencies module (#30) PathDependency and BodyDependency --- src/fastapi_toolsets/dependencies/__init__.py | 5 + src/fastapi_toolsets/dependencies/factory.py | 139 +++++++++++++ tests/test_dependencies.py | 186 ++++++++++++++++++ 3 files changed, 330 insertions(+) create mode 100644 src/fastapi_toolsets/dependencies/__init__.py create mode 100644 src/fastapi_toolsets/dependencies/factory.py create mode 100644 tests/test_dependencies.py diff --git a/src/fastapi_toolsets/dependencies/__init__.py b/src/fastapi_toolsets/dependencies/__init__.py new file mode 100644 index 0000000..b31c5c5 --- /dev/null +++ b/src/fastapi_toolsets/dependencies/__init__.py @@ -0,0 +1,5 @@ +"""FastAPI dependency factories for database objects.""" + +from .factory import BodyDependency, PathDependency + +__all__ = ["BodyDependency", "PathDependency"] diff --git a/src/fastapi_toolsets/dependencies/factory.py b/src/fastapi_toolsets/dependencies/factory.py new file mode 100644 index 0000000..10f4c4c --- /dev/null +++ b/src/fastapi_toolsets/dependencies/factory.py @@ -0,0 +1,139 @@ +"""Dependency factories for FastAPI routes.""" + +import inspect +from collections.abc import AsyncGenerator, Callable +from typing import Any, TypeVar, cast + +from fastapi import Depends +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import DeclarativeBase + +from ..crud import CrudFactory + +ModelType = TypeVar("ModelType", bound=DeclarativeBase) +SessionDependency = Callable[[], AsyncGenerator[AsyncSession, None]] + + +def PathDependency( + model: type[ModelType], + field: Any, + *, + session_dep: SessionDependency, + param_name: str | None = None, +) -> ModelType: + """Create a dependency that fetches a DB object from a path parameter. + + Args: + model: SQLAlchemy model class + 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) + + Returns: + A Depends() instance that resolves to the model instance + + Raises: + NotFoundError: If no matching record is found + + Example: + UserDep = PathDependency(User, User.id, session_dep=get_db) + + @router.get("/user/{id}") + async def get( + user: User = UserDep, + ): ... + """ + crud = CrudFactory(model) + name = ( + param_name + if param_name is not None + else "{}_{}".format(model.__name__.lower(), field.key) + ) + python_type = field.type.python_type + + async def dependency( + session: AsyncSession = Depends(session_dep), **kwargs: Any + ) -> ModelType: + value = kwargs[name] + return await crud.get(session, filters=[field == value]) + + setattr( + dependency, + "__signature__", + inspect.Signature( + parameters=[ + inspect.Parameter( + name, inspect.Parameter.KEYWORD_ONLY, annotation=python_type + ), + inspect.Parameter( + "session", + inspect.Parameter.KEYWORD_ONLY, + annotation=AsyncSession, + default=Depends(session_dep), + ), + ] + ), + ) + + return cast(ModelType, Depends(cast(Callable[..., ModelType], dependency))) + + +def BodyDependency( + model: type[ModelType], + field: Any, + *, + session_dep: SessionDependency, + body_field: str, +) -> ModelType: + """Create a dependency that fetches a DB object from a body field. + + Args: + model: SQLAlchemy model class + 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 + + Returns: + A Depends() instance that resolves to the model instance + + Raises: + NotFoundError: If no matching record is found + + Example: + UserDep = BodyDependency( + User, User.ctfd_id, session_dep=get_db, body_field="user_id" + ) + + @router.post("/assign") + async def assign( + user: User = UserDep, + ): ... + """ + crud = CrudFactory(model) + python_type = field.type.python_type + + async def dependency( + session: AsyncSession = Depends(session_dep), **kwargs: Any + ) -> ModelType: + value = kwargs[body_field] + return await crud.get(session, filters=[field == value]) + + setattr( + dependency, + "__signature__", + inspect.Signature( + parameters=[ + inspect.Parameter( + body_field, inspect.Parameter.KEYWORD_ONLY, annotation=python_type + ), + inspect.Parameter( + "session", + inspect.Parameter.KEYWORD_ONLY, + annotation=AsyncSession, + default=Depends(session_dep), + ), + ] + ), + ) + + return cast(ModelType, Depends(cast(Callable[..., ModelType], dependency))) diff --git a/tests/test_dependencies.py b/tests/test_dependencies.py new file mode 100644 index 0000000..2a89b81 --- /dev/null +++ b/tests/test_dependencies.py @@ -0,0 +1,186 @@ +"""Tests for fastapi_toolsets.dependencies module.""" + +import inspect +import uuid +from collections.abc import AsyncGenerator +from typing import Any, cast + +import pytest +from fastapi.params import Depends +from sqlalchemy.ext.asyncio import AsyncSession + +from fastapi_toolsets.dependencies import BodyDependency, PathDependency + +from .conftest import Role, RoleCreate, RoleCrud, User + + +async def mock_get_db() -> AsyncGenerator[AsyncSession, None]: + """Mock session dependency for testing.""" + yield None + + +class TestPathDependency: + """Tests for PathDependency factory.""" + + def test_returns_depends_instance(self): + """PathDependency returns a Depends instance.""" + dep = PathDependency(Role, Role.id, session_dep=mock_get_db) + assert isinstance(dep, Depends) + + def test_signature_has_default_param_name(self): + """PathDependency uses model_field as default param name.""" + dep = cast(Any, PathDependency(Role, Role.id, session_dep=mock_get_db)) + func = dep.dependency + + sig = inspect.signature(func) + params = list(sig.parameters.keys()) + + assert "role_id" in params + assert "session" in params + + def test_signature_has_correct_type_annotation(self): + """PathDependency uses field's python type for annotation.""" + dep = cast(Any, PathDependency(Role, Role.id, session_dep=mock_get_db)) + func = dep.dependency + + sig = inspect.signature(func) + + assert sig.parameters["role_id"].annotation == uuid.UUID + assert sig.parameters["session"].annotation == AsyncSession + + def test_signature_session_has_depends_default(self): + """PathDependency session param has Depends as default.""" + dep = cast(Any, PathDependency(Role, Role.id, session_dep=mock_get_db)) + func = dep.dependency + + sig = inspect.signature(func) + + assert isinstance(sig.parameters["session"].default, Depends) + + def test_custom_param_name_in_signature(self): + """PathDependency uses custom param_name in signature.""" + dep = cast( + Any, + PathDependency( + Role, Role.id, session_dep=mock_get_db, param_name="role_uuid" + ), + ) + func = dep.dependency + + sig = inspect.signature(func) + params = list(sig.parameters.keys()) + + assert "role_uuid" in params + assert "id" not in params + + def test_string_field_type(self): + """PathDependency handles string field types.""" + dep = cast(Any, PathDependency(User, User.username, session_dep=mock_get_db)) + func = dep.dependency + + sig = inspect.signature(func) + + assert sig.parameters["user_username"].annotation is str + + @pytest.mark.anyio + async def test_dependency_fetches_object(self, db_session): + """PathDependency inner function fetches object from database.""" + role = await RoleCrud.create(db_session, RoleCreate(name="test_role")) + + dep = cast(Any, PathDependency(Role, Role.id, session_dep=mock_get_db)) + func = dep.dependency + + result = await func(session=db_session, role_id=role.id) + + assert result.id == role.id + assert result.name == "test_role" + + +class TestBodyDependency: + """Tests for BodyDependency factory.""" + + def test_returns_depends_instance(self): + """BodyDependency returns a Depends instance.""" + dep = BodyDependency( + Role, Role.id, session_dep=mock_get_db, body_field="role_id" + ) + assert isinstance(dep, Depends) + + def test_signature_has_body_field_as_param(self): + """BodyDependency uses body_field as param name.""" + dep = cast( + Any, + BodyDependency( + Role, Role.id, session_dep=mock_get_db, body_field="role_id" + ), + ) + func = dep.dependency + + sig = inspect.signature(func) + params = list(sig.parameters.keys()) + + assert "role_id" in params + assert "session" in params + + def test_signature_has_correct_type_annotation(self): + """BodyDependency uses field's python type for annotation.""" + dep = cast( + Any, + BodyDependency( + Role, Role.id, session_dep=mock_get_db, body_field="role_id" + ), + ) + func = dep.dependency + + sig = inspect.signature(func) + + assert sig.parameters["role_id"].annotation == uuid.UUID + assert sig.parameters["session"].annotation == AsyncSession + + def test_signature_session_has_depends_default(self): + """BodyDependency session param has Depends as default.""" + dep = cast( + Any, + BodyDependency( + Role, Role.id, session_dep=mock_get_db, body_field="role_id" + ), + ) + func = dep.dependency + + sig = inspect.signature(func) + + assert isinstance(sig.parameters["session"].default, Depends) + + def test_different_body_field_name(self): + """BodyDependency can use any body_field name.""" + dep = cast( + Any, + BodyDependency( + User, User.id, session_dep=mock_get_db, body_field="user_uuid" + ), + ) + func = dep.dependency + + sig = inspect.signature(func) + params = list(sig.parameters.keys()) + + assert "user_uuid" in params + assert "id" not in params + + @pytest.mark.anyio + async def test_dependency_fetches_object(self, db_session): + """BodyDependency inner function fetches object from database.""" + role = await RoleCrud.create(db_session, RoleCreate(name="body_test_role")) + + dep = cast( + Any, + BodyDependency( + Role, Role.id, session_dep=mock_get_db, body_field="role_id" + ), + ) + func = dep.dependency + + result = await func(session=db_session, role_id=role.id) + + assert result.id == role.id + assert result.name == "body_test_role"