DEV Community: Ali Alf Ostovar

The 'Schema Explosion' Trap in FastAPI: Dynamic Role-Based Field Pruning Without Model Duplication

Ali Alf Ostovar — Fri, 23 Oct 2026 13:00:00 +0000

In Part 1, we established pragmatic boundaries for data access with SQLAlchemy 2.0. In Part 2, we tackled the Pass-Through Service trap using hook-driven lifecycle execution. In Part 3, we resolved route shadowing and architectural coupling in HTTP delivery layers.

Now, we face a friction point that emerges as backend domain models mature: Schema Explosion.

When an API serves multiple user roles (such as Public Visitors, Authenticated Customers, Support Staff, and System Admins), exposing domain entities safely becomes a maintenance bottleneck. In this post, we analyze why static Pydantic DTO proliferation breaks codebases, evaluate the failure modes of manual field pruning, and implement a robust, recursive, and fail-closed field pruning pattern in vanilla Pydantic V2.

1. The Common Doctrine vs. The Frustrating Reality

Standard FastAPI guidance recommends declaring explicit Pydantic response models for every API endpoint:

"Never return ORM models directly! Create distinct Pydantic schemas for public views, user views, and admin views to guarantee clean response boundaries."

While sound in principle, applying this rule naively across an enterprise application with 30 domain entities yields a frustrating reality. Consider a standard User domain model. To satisfy different access levels across CRUD endpoints, developers end up maintaining an array of near-identical classes:

# The "Schema Explosion" anti-pattern
class UserPublicResponse(BaseModel):
    id: UUID
    username: str

class UserOwnerResponse(UserPublicResponse):
    email: EmailStr
    created_at: datetime

class UserAdminResponse(UserOwnerResponse):
    is_active: bool
    stripe_customer_id: str | None
    internal_notes: str | None

class UserUpdateDTO(BaseModel):
    email: EmailStr | None = None

class UserAdminUpdateDTO(UserUpdateDTO):
    is_active: bool | None = None
    internal_notes: str | None = None

When you multiply this pattern across 30 entities, your codebase is overwhelmed by hundreds of DTO variants. Adding a single column to a database table requires updating four to six schema classes, updating mapping utilities, and adjusting endpoint type annotations.

2. Legitimate Criticisms & Failure Modes

When engineering teams grow weary of maintaining duplicate DTOs, they usually fall into one of two dangerous failure modes.

Failure Mode A: The Manual Pruning Anti-Pattern

Developers attempt to reuse a single comprehensive response schema (UserResponse) and manually delete sensitive keys inside the route handler or service layer before returning data:

@router.get("/users/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: UUID,
    current_user: User = Depends(get_current_user),
    service: UserService = Depends()
):
    user = await service.get(user_id)
    data = UserResponse.model_validate(user).model_dump()

    # Manual role-based field deletion in route handler
    if not current_user.is_admin:
        data.pop("internal_notes", None)
        data.pop("stripe_customer_id", None)

    return data

This approach introduces two severe vulnerabilities:

OpenAPI Schema Misrepresentation: The endpoint's generated OpenAPI specification (Swagger UI) continues to declare internal_notes and stripe_customer_id as return fields because response_model=UserResponse is static.
Data Leakage Risk: Security becomes opt-in and decentralized. If a developer forgets to copy-paste the .pop() statements in a newly created endpoint, sensitive internal fields leak silently to unprivileged users.

Failure Mode B: Unhandled Context Defaults (Fail-Open Risks)

If a field-pruning mechanism defaults to an unconstrained state when no context is active, any unhandled endpoint, background task, or asynchronous call stack will default to leaking all fields.

A robust design must enforce a declarative fail-closed baseline: if no execution context is explicitly initialized (or if a request is unauthenticated), the system must prune all fields declared as private by default, rather than assuming unconstrained access.

3. The Real Architectural Problem

The root issue is a coupling conflict: we are using static class definitions to solve a dynamic execution context problem.

Whether a field like stripe_customer_id or internal_notes should be visible in a payload is rarely an intrinsic property of the class structure itself. It is a decision that depends on the requester's active security context (their authenticated identity, assigned roles, or active tenant scopes).

Furthermore, addressing this solely at serialization leaves an open flank: Mass Assignment. If the same schema is used for mutations (PATCH / PUT), an unprivileged user could inject stripe_customer_id into the request body. A comprehensive solution must intercept both the input deserialization boundary and the output serialization boundary.

Instead of duplicating class definitions to represent every role permutation, a pragmatic architecture separates the structural schema contract from the contextual field restriction policy.

4. The Pragmatic Solution: Context-Aware Field Shielding

We can solve this problem in modern Python by combining three foundational primitives:

Python contextvars: To store and propagate request-scoped security restrictions across asynchronous call stacks without passing context parameters explicitly through every function.
Pydantic V2 @model_validator(mode="before"): To intercept incoming payloads and strip unauthorized fields before validation occurs (Mass Assignment protection).
Pydantic V2 @model_serializer(mode="wrap"): To intercept serialization at every model boundary and recursively strip restricted attributes through arbitrary nesting depths and lists, enforcing an explicit declarative fail-closed baseline.

Standalone Implementation (Copy-Pasteable & Tested)

The following self-contained module demonstrates composition-safe, fail-closed context shielding with deep-path recursive pruning in vanilla Pydantic V2:

from contextvars import ContextVar
from typing import Any, ClassVar, FrozenSet
from uuid import UUID, uuid4
from pydantic import BaseModel, ConfigDict, EmailStr, model_serializer, model_validator

# 1. Request-scoped context store for field restrictions
# None signifies an uninitialized / unauthenticated context (triggers fail-closed baseline)
# frozenset() signifies an authenticated session with zero restrictions (e.g. Superuser)
_restricted_fields_ctx: ContextVar[FrozenSet[str] | None] = ContextVar(
    "restricted_fields_ctx", default=None
)

def set_context_restrictions(restricted_paths: set[str] | FrozenSet[str] | None) -> None:
    """Bind restricted field paths to the active asynchronous execution context."""
    if restricted_paths is None:
        _restricted_fields_ctx.set(None)
    else:
        _restricted_fields_ctx.set(frozenset(restricted_paths))

def get_context_restrictions() -> FrozenSet[str] | None:
    """Retrieve active restricted field paths for the current execution context."""
    return _restricted_fields_ctx.get()

# 2. Base Shielded Schema with Recursive Dot-Path Pruning
class ContextShieldSchema(BaseModel):
    """Base Pydantic model enforcing dynamic, recursive field pruning.

    Implements a declarative fail-closed baseline: when context is uninitialized,
    all fields declared in `__private_fields__` are automatically pruned.
    """

    __model_name__: ClassVar[str | None] = None
    __private_fields__: ClassVar[set[str]] = set()

    @classmethod
    def _resolve_relative_paths(cls, restrictions: FrozenSet[str] | None) -> set[str]:
        """Normalize global and namespaced paths into paths relative to this model."""
        if restrictions is None:
            # Declarative fail-closed baseline: suppress explicitly declared private fields
            return set(cls.__private_fields__)

        if not restrictions:
            return set()

        model_name = cls.__model_name__
        model_prefix = f"{model_name}." if model_name else ""
        relative_paths: set[str] = set()

        for path in restrictions:
            if path == "*" or (model_name and path == model_name):
                return {"*"}
            elif model_prefix and path.startswith(model_prefix):
                relative_paths.add(path[len(model_prefix):])
            elif not model_prefix:
                relative_paths.add(path)

        return relative_paths

    @classmethod
    def _prune_data_recursive(cls, data: Any, relative_paths: set[str]) -> Any:
        """Recursively strip restricted attributes through nested dicts and lists."""
        if not isinstance(data, dict) or not relative_paths:
            return data

        if "*" in relative_paths:
            data.clear()
            return data

        nested_restrictions: dict[str, set[str]] = {}
        for path in relative_paths:
            parts = path.split(".", 1)
            if len(parts) == 1:
                data.pop(parts[0], None)
            else:
                key, remaining = parts
                nested_restrictions.setdefault(key, set()).add(remaining)

        if not nested_restrictions:
            return data

        for key in list(data.keys()):
            if key in nested_restrictions:
                rem_paths = nested_restrictions[key]
                val = data[key]
                if isinstance(val, dict):
                    data[key] = cls._prune_data_recursive(val, rem_paths)
                elif isinstance(val, list):
                    data[key] = [
                        cls._prune_data_recursive(item, rem_paths) if isinstance(item, dict) else item
                        for item in val
                    ]
        return data

    @model_validator(mode="before")
    @classmethod
    def _filter_restricted_inputs(cls, data: Any) -> Any:
        """Shield input payloads: strip restricted fields before validation (Mass Assignment guard)."""
        if not isinstance(data, dict):
            return data

        restrictions = get_context_restrictions()
        relative_paths = cls._resolve_relative_paths(restrictions)
        if not relative_paths:
            return data

        data_copy = dict(data)
        return cls._prune_data_recursive(data_copy, relative_paths)

    @model_serializer(mode="wrap")
    def _serialize_with_context_shield(self, handler: Any) -> dict[str, Any]:
        """Intercept serialization: strip restricted fields before returning payload."""
        raw_data: dict[str, Any] = handler(self)
        if not isinstance(raw_data, dict):
            return raw_data

        restrictions = get_context_restrictions()
        relative_paths = self._resolve_relative_paths(restrictions)
        return self._prune_data_recursive(raw_data, relative_paths)

# 3. Domain Model Example (Supports Nested Schema Isolation)
class AddressResponse(ContextShieldSchema):
    __model_name__ = "address"
    __private_fields__ = {"internal_notes"}

    street: str
    city: str
    internal_notes: str | None = None

class UserResponse(ContextShieldSchema):
    __model_name__ = "users"
    __private_fields__ = {"stripe_customer_id", "internal_notes"}

    id: UUID
    username: str
    email: EmailStr
    stripe_customer_id: str | None = None
    internal_notes: str | None = None
    address: AddressResponse

    model_config = ConfigDict(from_attributes=True)

Verification: Fail-Closed Baseline, Deep Nesting, and Mass Assignment Guards

Because each model inheriting from ContextShieldSchema executes its own wrap-mode serializer and input validator during traversal, model namespaces and arbitrary nesting depths remain strictly coordinated:

def test_nested_and_fail_closed_assertions():
    sample_user = UserResponse(
        id=uuid4(),
        username="johndoe",
        email="john@example.com",
        stripe_customer_id="cus_N93284723",
        internal_notes="User top-level secret",
        address=AddressResponse(
            street="123 Main St",
            city="Metropolis",
            internal_notes="Gate access code: 4491"
        )
    )

    # Scenario A: Uninitialized Context (Declarative Fail-Closed Baseline)
    set_context_restrictions(None)
    payload_uninit = sample_user.model_dump()
    assert "stripe_customer_id" not in payload_uninit
    assert "internal_notes" not in payload_uninit
    assert "internal_notes" not in payload_uninit["address"]
    assert payload_uninit["username"] == "johndoe"

    # Scenario B: Superuser Context (Explicit Zero Restrictions)
    set_context_restrictions(set())
    payload_admin = sample_user.model_dump()
    assert payload_admin["stripe_customer_id"] == "cus_N93284723"
    assert payload_admin["address"]["internal_notes"] == "Gate access code: 4491"

    # Scenario C: Deep Nested Path ("users.address.internal_notes")
    # Prunes address-level notes while preserving user-level notes
    set_context_restrictions({"users.address.internal_notes"})
    payload_nested = sample_user.model_dump()
    assert payload_nested["internal_notes"] == "User top-level secret"
    assert "internal_notes" not in payload_nested["address"]

    # Scenario D: Input Shield (Mass Assignment Protection)
    # Attempting to inject restricted fields into input payload
    set_context_restrictions({"users.stripe_customer_id"})
    injected_data = {
        "id": uuid4(),
        "username": "janedoe",
        "email": "jane@example.com",
        "stripe_customer_id": "malicious_override",
        "address": {"street": "456 Side St", "city": "Gotham"}
    }
    validated_model = UserResponse.model_validate(injected_data)
    assert validated_model.stripe_customer_id is None

if __name__ == "__main__":
    test_nested_and_fail_closed_assertions()
    print("All nested, recursive, and fail-closed assertions passed successfully.")

5. What This Pattern Does NOT Try to Do

To preserve architectural clarity, we establish explicit non-goals for context-aware field pruning:

It does NOT replace input validation invariants: Creating records (UserCreate) and updating records (UserUpdate) frequently have distinct schema requirements (such as password validation on registration or optional fields on patch). Context shielding strips restricted fields to prevent Mass Assignment, but business-level validation rules still belong in dedicated input models.
It does NOT eliminate database I/O by itself: Stripping a field during Pydantic serialization prevents sensitive data from reaching the HTTP client, but the column was still queried from the database. For heavy text fields or binary blobs, combine this pattern with SQLAlchemy's load_only() at the repository layer to eliminate I/O overhead.
It does NOT dynamically alter static Swagger UI documentation: In FastAPI, OpenAPI documentation (/docs) is generated once at application startup from static Python type hints. Dynamic serialization filtering strips payload attributes at runtime per request, but it does not rewrite static OpenAPI endpoints per JWT role.

6. Production Realization: How ZCore Implements Zchema

When designing the ZCore Framework, we integrated this exact context-shielding strategy directly into our base schema component, Zchema.

Zchema automatically coordinates Pydantic V2 serialization with ZContext (our request-scoped context manager). To solve the challenge where static OpenAPI docs fall short for dynamic frontend rendering, ZCore hooks into Pydantic's __get_pydantic_json_schema__ and exposes a dedicated GET /_schema runtime introspection endpoint via BaseRouter:

// GET /users/_schema (Queried at runtime by an authenticated customer)
{
  "success": true,
  "data": {
    "title": "UserResponse",
    "type": "object",
    "properties": {
      "id": { "type": "string", "format": "uuid" },
      "username": { "type": "string" },
      "email": { "type": "string", "format": "email" }
      // 'stripe_customer_id' and 'internal_notes' are omitted based on requester permissions
    }
  }
}

This bridges the gap between static API specifications and dynamic runtime permissions. Administrative UI platforms (such as Retool or dynamic form engines) can inspect runtime field permissions dynamically, rendering input fields and table columns accurately without hardcoding permission rules on the frontend.

Summary

Managing multiple user roles does not require duplicating Pydantic response models across dozens of files. By decoupling structural field definitions from runtime security policies using contextvars and Pydantic V2 serialization hooks, you maintain clean, DRY schemas while maintaining robust field-level output filtering and Mass Assignment protection.

You certainly don't need to adopt an entire framework to benefit from this; the snippets above are intentionally designed to work standalone in any vanilla FastAPI application. However, if you are exploring modular systems with FastAPI, feel free to check out ZCore.

Next in the Series: In Part 5, we will examine The Unit of Work Dilemma in FastAPI: Is Wrapping SQLAlchemy's Session an Anti-Pattern?

How do you handle role-based field visibility in your FastAPI services? Do you maintain separate DTOs or use dynamic filters? Share your approach in the comments!

Why Generic CRUD Routers Break in FastAPI: Route Shadowing, Architecture Bypassing, and Pragmatic Fixes

Ali Alf Ostovar — Mon, 19 Oct 2026 13:00:00 +0000

In Part 1, we explored how to draw clean data boundaries with SQLAlchemy 2.0 without fighting the ORM. In Part 2, we tackled the Pass-Through Service trap, showing how reusable base services with lifecycle hooks eliminate anemic forwarder methods while keeping transactions atomic.

Now, we arrive at the HTTP delivery layer.

Once your application manages 20 or 30 domain entities, writing route handlers becomes an exercise in mind-numbing repetition:

POST / (Create)
GET /{id} (Retrieve by ID)
GET / (Paginated list)
PUT /{id} (Full update)
PATCH /{id} (Partial update)
DELETE /{id} (Deletion)
POST /search (Dynamic filtering)

Every entity requires essentially the same baseline endpoints, HTTP status codes, and dependency wirings.

Naturally, developers turn to generic CRUD routers. Open-source libraries like fastapi-crudrouter and FastCRUD emerged to solve this, and for rapid prototyping, internal tooling, or lightweight CRUD microservices, they work admirably well.

However, as applications scale and adopt layered architectures, generic routers often introduce architectural friction. In this post, we’ll analyze where traditional CRUD generators fall short, look at how the subtle Route Shadowing problem causes unexpected HTTP 422 errors, and explore a pragmatic way to scaffold endpoints that respect a dedicated Service layer.

1. The Friction Points of Generic Routers

1. Bypassing the Service Layer

Most generic CRUD routers couple HTTP endpoints directly to database sessions:

[Direct Coupling]
HTTP Endpoint ──► Direct ORM Queries / AsyncSession

For basic administrative tables or throwaway prototypes, this direct link is convenient. But in domain-driven systems, business operations rarely map 1:1 to raw database queries.

If creating a user requires hashing credentials, verifying tenant quotas, and publishing a domain event, bypassing the Service layer forces you into an uncomfortable trade-off: abandon the generic router and rewrite the endpoint manually, or rely on awkward monkey-patching and route hooks.

2. Coarse-Grained Authorization

Enterprise APIs frequently require fine-grained permissions evaluated per operational action:

A support role might hold users:view but not users:delete.
A manager role holds orders:update but not orders:delete.

Many route generators provide only an all-or-nothing router dependency (e.g., dependencies=[Depends(verify_token)]). Configuring discrete permission scopes on a per-route basis within a generated class often requires more lines of configuration than writing the endpoints by hand.

3. The "Route Shadowing" Pitfall

FastAPI (via Starlette) evaluates registered routes sequentially in the order they are added to the routing table. When an incoming request matches a route pattern, Starlette executes that handler and halts further route evaluation.

Consider a generic router that registers standard endpoints in this common order:

GET /{id} (where the endpoint parameter id is typed in the function signature as uuid.UUID)
GET /summary (a custom endpoint returning aggregate metrics)

Now, consider what happens when a client sends a request to /summary:

GET /products/summary HTTP/1.1
Host: api.example.com

Starlette evaluates the routes matching the GET method sequentially:

It encounters GET /{id} first.
If the route was registered without a native path converter (i.e. declared as /{id} rather than /{id:uuid}), Starlette's regex matches any string segment. It captures "summary" as the id path parameter.
FastAPI then passes "summary" to Pydantic for validation against the declared uuid.UUID parameter type.
Validation fails, and the framework returns an unexpected HTTP 422 Unprocessable Entity:

{
  "detail": [
    {
      "loc": ["path", "id"],
      "msg": "Input should be a valid UUID",
      "type": "uuid_parsing"
    }
  ]
}

The request never reaches the /summary handler because it was shadowed by the wildcard route.

While using Starlette's native path converters (/{id:uuid}) allows Starlette to reject non-matching formats during route matching and fall through to static paths, converters alone do not prevent collisions when dealing with string slug primary keys, untyped integer paths, or custom sub-routes declared across independent plugins.

In hand-written routers, developers manually place static paths above parameterized wildcards. But in generic, dynamically generated routers, manual ordering becomes fragile and error-prone.

2. A Layered Architectural Approach

To scaffold endpoints without sacrificing architectural separation, a generic router should satisfy three core criteria:

Delegate to the Service Layer: The router should handle HTTP mechanics, serialization, and status codes—delegating actual business workflows to an injected Service instance.
Action-Based Security Hooks: The router should automatically assign discrete action identifiers (e.g., model:view, model:create) to routes and wire permission dependencies cleanly.
Route Reordering at Initialization: The router should reorder its internal route table upon setup so that static tokens evaluate before dynamic wildcards.

3. Resolving Route Precedence via Specificity Sorting

To prevent static routes from being shadowed by dynamic parameters, we sort the router's internal table during router initialization (__init__).

We evaluate the structural specificity of each path segment:

def _sort_routes(self) -> None:
    """Sort routes using hierarchical specificity scoring to prioritize static paths."""

    def _route_specificity_key(route: Any) -> tuple[int, int, list[int], int]:
        path = getattr(route, "path", "")
        segments = [seg for seg in path.strip("/").split("/") if seg]

        segment_scores: list[int] = []
        for seg in segments:
            # Dynamic path parameters have higher penalty scores
            if seg.startswith("{") and seg.endswith("}"):
                if ":path}" in seg:
                    segment_scores.append(2)  # Catch-all paths: lowest precedence
                else:
                    segment_scores.append(1)  # Parameterized paths: lower precedence
            else:
                segment_scores.append(0)      # Static segments: highest precedence

        has_dynamic = 1 if any(s > 0 for s in segment_scores) else 0
        total_dynamic = sum(segment_scores)

        # Sort tuple: (has_dynamic, total_dynamic_score, segment_breakdown, -total_length)
        return (has_dynamic, total_dynamic, segment_scores, -len(segments))

    self.router.routes.sort(key=_route_specificity_key)

How this heuristic works:

Static routes like /lookup or /search evaluate with 0 dynamic penalties.
Wildcards like /{id} evaluate with dynamic penalties.
Catch-all routes like /{path:path} evaluate with the lowest priority.

Because Python compares tuples lexicographically:

has_dynamic ensures purely static paths (0) always evaluate before paths containing dynamic parameters (1).
total_dynamic ensures routes with a lower cumulative dynamic score evaluate earlier.
segment_scores guarantees that static segments take precedence over parameters at the same position (e.g., [0, 1] precedes [1, 0]).

Important Boundary: This sorting operates as a setup-time heuristic inside the router's constructor. Because it runs once during application startup, it adds negligible initialization overhead and zero per-request runtime penalties. However, developers must be mindful that if custom routes are appended directly to the underlying APIRouter after initialization, manual ordering still applies.

4. Connecting the Router to the Service Layer

Rather than executing database queries inline, our BaseRouter dynamically injects the domain service and resolves the primary key type via model reflection:

from typing import Any, Generic, TypeVar
from fastapi import APIRouter, Depends, status
from pydantic import BaseModel
from sqlalchemy import inspect
import uuid

CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)

class BaseRouter(Generic[CreateSchemaType, UpdateSchemaType]):
    model: type[Any]
    create_schema: type[CreateSchemaType] | None = None
    update_schema: type[UpdateSchemaType] | None = None
    schema_out: type[BaseModel] | None = None
    service: Any = None
    pk_type: type[Any] | None = None
    prefix: str = ""
    tags: list[str] | None = None

    def __init__(self) -> None:
        if not self.service:
            raise ValueError(f"Service class must be defined in '{self.__class__.__name__}'.")

        self.router = APIRouter(prefix=self.prefix, tags=self.tags or [])
        self._register_routes()
        self._sort_routes()  # Order routes to ensure static paths take precedence

    def _resolve_pk_type(self) -> type[Any]:
        """Detect primary key type via model reflection for accurate URL converters."""
        if getattr(self, "pk_type", None) is not None:
            return self.pk_type

        try:
            mapper = inspect(self.model)
            if mapper.primary_key:
                pk_col = mapper.primary_key[0]
                return getattr(pk_col.type, "python_type", uuid.UUID)
        except Exception:
            pass
        return uuid.UUID

    def _get_pk_path(self, pk_type: type[Any]) -> str:
        if pk_type is int:
            return "/{id:int}"
        if pk_type is uuid.UUID:
            return "/{id:uuid}"
        return "/{id}"

Discrete Scope Dependencies

When endpoints are registered, the router maps operational keys to standardized action strings:

    def get_route_action(self, route_key: RouteKey) -> str:
        action_map = {
            RouteKey.POST: self.model.actions().CREATE,       # e.g., "products:create"
            RouteKey.GET: self.model.actions().VIEW,          # e.g., "products:view"
            RouteKey.GET_ALL: self.model.actions().LISTVIEW,  # e.g., "products:listview"
            RouteKey.UPDATE: self.model.actions().UPDATE,     # e.g., "products:update"
            RouteKey.DELETE: self.model.actions().DELETE,     # e.g., "products:delete"
        }
        return action_map[route_key]

    def get_route_dependencies(self, route_key: RouteKey, action: str) -> list[Any]:
        # Returns a callable dependency checking required scopes for the active user
        return [HasScopes(action)] if action else []

Here, HasScopes acts as a standard FastAPI dependency that verifies whether the authenticated identity holds the declared scope.

Action-Level vs. Object-Level Authorization:

Route dependencies enforce Action-Level permissions (e.g., Is this user permitted to delete products?). In contrast, Object-Level authorization (e.g., Does this user own product #42 or belong to the correct tenant?) must remain encapsulated inside the Service or Repository queries. Conflating the two in the routing layer breaks domain isolation.

Endpoint Delegation via Extensible Action Hooks

Notice how route binding is decoupled from endpoint execution. Instead of embedding inline handler logic directly within the registration loop, the router delegates to dedicated endpoint methods (self.create_endpoint, self.get_endpoint):

    def _register_routes(self) -> None:
        """Dynamically generate and bind endpoints to the APIRouter."""
        service_callable = self.service
        service_dependency = Depends(Injector(service_callable))
        target_pk_type = self._resolve_pk_type()
        pk_path = self._get_pk_path(target_pk_type)

        if RouteKey.POST not in self.exclude:
            c_schema = self.create_schema

            async def _create_endpoint(
                data_in: c_schema,
                service_inst: BaseService = service_dependency,
            ) -> ResponseWrapper:
                return await self.create_endpoint(data_in, service_inst)

            self.router.add_api_route(
                path="/",
                endpoint=_create_endpoint,
                methods=["POST"],
                dependencies=self._get_route_dependencies(RouteKey.POST),
                status_code=status.HTTP_201_CREATED,
                response_model=ResponseWrapper[self.schema_out]
                if self.schema_out
                else ResponseWrapper,
                openapi_extra=self._get_openapi_extra(RouteKey.POST),
            )

        if RouteKey.GET not in self.exclude:

            async def _get_endpoint(
                id: target_pk_type,
                service_inst: BaseService = service_dependency,
            ) -> ResponseWrapper:
                return await self.get_endpoint(id, service_inst)

            self.router.add_api_route(
                path=pk_path,
                endpoint=_get_endpoint,
                methods=["GET"],
                dependencies=self._get_route_dependencies(RouteKey.GET),
                status_code=status.HTTP_200_OK,
                response_model=ResponseWrapper[self.schema_out]
                if self.schema_out
                else ResponseWrapper,
                openapi_extra=self._get_openapi_extra(RouteKey.GET),
            )

The corresponding action hooks handle delegating to the injected service:

    async def create_endpoint(self, data_in: Any, service: BaseService) -> ResponseWrapper:
        data = await service.create(data_in)
        return ResponseWrapper(data=data)

    async def get_endpoint(self, id: Any, service: BaseService) -> ResponseWrapper:
        data = await service.get(id=id)
        return ResponseWrapper(data=data)

This design provides a crucial extension point: if a specific domain requires modifying HTTP response headers, setting cookies, or triggering an asynchronous worker upon entity creation, developers can override create_endpoint directly without needing to re-register routes or duplicate dependency bindings.

Because every request flows through the Service layer, all pre_create, post_create, and transaction hooks defined on the service execute consistently.

5. Beyond Basic CRUD: The `/lookup` Dropdown Pattern

A recurring architectural issue in client-server systems is over-fetching full entities merely to populate UI selection components:

A web form needs a dropdown of Active Categories (ID + Title). Calling GET /categories serializes audit logs, descriptions, timestamps, and nested relationship counts.

To solve this, a mature generic router can pair standard CRUD with a specialized, field-projected /lookup endpoint:

    async def lookup_endpoint(
        self, search_in: SearchRequest, service: Any
    ) -> ResponseWrapper:
        # 1. Whitelist filterable and sortable fields from the lookup schema
        allowed_fields = self._get_effective_lookup_fields()
        self._validate_lookup_request(search_in, allowed_fields)

        # 2. Instruct ORM to apply a load_only() loader option to the query
        load_fields, loader_options = self._resolve_lookup_projections()

        result = await service.search(
            search_in,
            fields=load_fields,
            options=loader_options,
        )
        return ResponseWrapper(data=result)

By applying a load_only() loader option to the database query, only the primary key and the specific columns declared in lookup_schema are retrieved from the database, eliminating unnecessary serialization and network payload bloat.

6. Declarative Routing in Practice

Defining a complete, production-ready router requires only configuring the schema, model, and service bindings:

from zcore import BaseRouter, RouteKey
from .models import Product
from .schemas import ProductCreate, ProductUpdate, ProductResponse, ProductLookup
from .services import ProductService

class ProductRouter(BaseRouter):
    model = Product
    create_schema = ProductCreate
    update_schema = ProductUpdate
    schema_out = ProductResponse
    lookup_schema = ProductLookup
    service = ProductService

    prefix = "/products"
    tags = ["Products"]

    # Exclude endpoints you don't wish to expose
    exclude = {RouteKey.DELETE}

# Attach router to FastAPI
app.include_router(ProductRouter().router)

With this configuration:

Endpoints (POST /products/, GET /products/, GET /products/{id}, POST /products/lookup) are scaffolded with accurate status codes and typed path parameters.
Operations route through ProductService, ensuring lifecycle hooks, password hashing, and domain events execute.
Every endpoint is protected by discrete scope checks (products:create, products:view).
Static routes are sorted ahead of dynamic wildcards during initialization to prevent route shadowing.

7. What This Router Does NOT Try to Do

To keep generic routing sustainable, defining non-goals is essential:

It does not replace bespoke endpoints: Specialized workflows—such as checkout orchestrations, multi-file uploads, or reporting pipelines—should be written as standard FastAPI route handlers.
It is not an all-encompassing routing engine: It builds standard Starlette APIRoute instances under the hood. It provides structural scaffolding, not a replacement for FastAPI's routing system.
It does not eliminate all route design responsibility: While specificity sorting mitigates common static-vs-wildcard collisions, developers remain responsible for coherent API path design.

8. The Big Picture: A Cohesive 3-Tier Flow

Over this 3-part series, we have constructed a clean, layered architecture:

HTTP Request
     ↓
[BaseRouter]        → Precedence-sorted, auto-scoped endpoints (HTTP concerns)
     ↓
[BaseService]       → Hook-driven lifecycle: pre_create, post_create, _safe_commit (Business concerns)
     ↓
[BaseRepository]    → Scoped queries, non-committing persistence, eager loading (Data concerns)
     ↓
SQLAlchemy 2.0      → Relational Database

By maintaining clear boundaries, we eliminate repetitive boilerplate while preserving testability, security, and extensibility.

In ZCore, we designed BaseRouter, BaseService, and BaseRepository to operate together as a pragmatic architectural chassis for FastAPI.

How do you handle repetitive CRUD endpoints in your FastAPI projects? Do you write them by hand or use generator patterns? Let us know in the comments below!

The "Pass-Through" Service Trap in FastAPI: Why CRUD Service Layers Often Feel Like Useless Boilerplate

Ali Alf Ostovar — Wed, 14 Oct 2026 13:00:00 +0000

In Part 1 (The Repository Pattern in FastAPI), we explored how to draw clean data boundaries with SQLAlchemy 2.0 without fighting the ORM.

Once your repositories are in place, standard architectural advice immediately points to the next step: introduce a Service layer.

A common architectural recommendation is:

"Keep your route handlers thin! Routes should only handle HTTP status codes and serialization. All business logic belongs in the Service layer."

It sounds disciplined. But after a few weeks of building standard application modules, many developers look at their service files and experience a familiar frustration:

# The "Pass-Through" (Forwarding) Service
class ProductService:
    def __init__(self, repository: ProductRepository):
        self.repository = repository

    async def get(self, id: UUID) -> Product:
        return await self.repository.get(id=id)

    async def list(self, pagination: Any) -> list[Product]:
        return await self.repository.get_list(pagination=pagination)

    async def delete(self, id: UUID) -> None:
        await self.repository.delete(target=id)

In many routine CRUD operations, the service function does little more than forward arguments directly to the repository.

You are maintaining three separate files (Router, Service, Repository) to execute a single read or delete operation.

This is the Pass-Through Service Trap (or Forwarding Service). In this post, we’ll analyze why this happens, why simply deleting the service layer creates worse problems, and how we solved this using Lifecycle Hooks and safe transaction awareness.

1. The Pendulum Swing: Why Deleting the Service Layer Backfires

When developers get exhausted by writing pass-through forwarders, they often swing to the opposite extreme: they delete the service layer entirely.

They call repositories (or raw queries) directly inside the FastAPI route handler:

@router.post("/orders")
async def create_order(
    payload: OrderCreate,
    repo: OrderRepository = Depends(),
    db: AsyncSession = Depends(get_db),
):
    # Route handler now owns business logic:
    order = await repo.create(payload)
    await audit_logger.record(order.id)
    await db.commit()
    return order

This feels fast at first. But the moment real business requirements arrive, this approach collapses:

Hidden Side Effects: If an order requires deducting inventory, computing discounts, and dispatching events, your route handler quickly balloons to 150 lines of mixed concerns.
Coupled Testing: Testing the workflow becomes tightly coupled to HTTP handling and dependency wiring, rather than exercising a reusable application service directly.
Loss of Reusability: If an asynchronous background worker, a CLI task, or an internal event subscriber needs to create an order, it cannot invoke the route handler. You are forced to copy-paste the logic or refactor back into a service under deadline pressure.

We don't want fat route handlers. But we also don't want anemic forwarders.

2. The Core Problem: Re-inventing the CRUD Lifecycle

The root cause of the Pass-Through trap is that developers treat every single domain service as a bespoke, hand-crafted pipeline—even when the underlying lifecycle is identical:

Validated Payload → Pre-persistence Mutation & Domain Checks → Persistence → Post-processing → Transaction Finalization

Whether you are creating a User, a Company, or a Product, this sequence rarely changes. What changes are the specific domain rules (e.g., hashing a password before creating a user, or attaching dynamic tenant metadata).

Instead of forcing developers to write boilerplate forwarders over and over, a mature service layer should implement the Template Method Pattern: provide the end-to-end orchestration flow out of the box, and expose granular Lifecycle Hooks for domain-specific logic.

3. The Implementation: Orchestration with Composable Hooks

In our architecture, we divided the business layer into focused capabilities matching the repository: ReadServiceMixin, WriteServiceMixin, and SearchServiceMixin.

The following is a simplified excerpt of the write-service implementation. The surrounding module defines AbstractService; unrelated imports and methods are omitted for clarity:

from typing import Any, Generic, TypeVar

from pydantic import BaseModel
from zcore.db.setup import Base

ModelType = TypeVar("ModelType", bound=Base)

class WriteServiceMixin(Generic[ModelType], AbstractService[ModelType]):
    """Standardized mutation, persistence, and deletion orchestration."""

    async def pre_create(self, schema: BaseModel) -> dict[str, Any] | None:
        """Triggered prior to database insertion. Return trusted fields to merge."""
        return None

    async def post_create(self, model: ModelType) -> None:
        """Triggered after insertion, before commit (in-session post-processing)."""
        pass

    async def pre_update(
        self, target: ModelType | Any, schema: BaseModel, partial: bool
    ) -> dict[str, Any] | None:
        return None

    async def post_update(self, model: ModelType) -> None:
        pass

    async def pre_delete(self, id: Any, force: bool = False) -> None:
        pass

    async def post_delete(self, model: ModelType, force: bool = False) -> None:
        pass

The Orchestrator Method

The create method orchestrates the entire pipeline. Developers never have to rewrite this boilerplate:

    async def create(self, schema: BaseModel, **extra_data: Any) -> ModelType:
        # Step A: Run domain pre-flight logic; explicit extra_data overrides hook data
        hook_data = await self.pre_create(schema) or {}
        combined_extra = {**hook_data, **extra_data}

        # Step B: Execute persistence via repository (issues INSERT and flush)
        result = await self.on_create(schema, **combined_extra)

        # Step C: Run in-session post-processing
        await self.post_create(result)

        # Step D: Safely finalize the transaction
        await self._safe_commit()
        return result

Notice two critical architectural traits here:

Protocol-Agnostic Design: The service raises application-level, framework-neutral exceptions (such as EntityNotFound) instead of Starlette's HTTPException. The HTTP layer handles translation at the API boundary, keeping the service completely independent of web protocols.
Hook-driven Extension: For standard CRUD behavior, a service can require only minimal boilerplate to bind the model and repository. When an entity needs business rules, you can override the relevant lifecycle hook without rewriting the persistence pipeline.

4. Understanding Hook Timing: In-Session vs. Post-Commit

A critical architectural distinction that developers must understand when using lifecycle hooks is the timing of post_create:

[on_create (flush)] ──→ [post_create (in-session)] ──→ [_safe_commit (commit)]

post_create is an in-session hook (Post-Insert, Pre-Commit):

What it is for: Inspecting auto-generated primary keys or timestamps, updating sibling entities within the same database transaction, or staging deferred events.
What it is NOT for: Irreversible external network calls (such as directly invoking a third-party payment gateway or sending an SMS).

If you fire an external network call directly inside post_create and the subsequent _safe_commit() fails due to a database constraint or serialization conflict, your external side-effect has already occurred while your database rolled back—causing a phantom side-effect.

For external side-effects:

In-memory event queues: Suitable for transient, best-effort notifications where losing an event during an abrupt process termination is acceptable.
Transactional Outbox Table: Essential for mission-critical operations. Storing the event row in an outbox table within the same atomic database transaction guarantees event durability (eliminating dual-write inconsistencies). However, note the distinction: persisting the event atomically only guarantees it is safely stored; guaranteed delivery still requires an asynchronous outbox relay worker, retry policies, and idempotent consumers to handle eventual dispatch.

5. Real-World Example: Domain Logic Without the Boilerplate

Look at how concise a domain service becomes when you only write the code unique to your business rules:

# 1. The Schema (schemas.py)
class UserCreate(BaseModel):
    email: EmailStr
    # Validated on HTTP input, but excluded during model_dump()
    password: str = Field(min_length=8, max_length=128, exclude=True)


# 2. The Service (services.py)
class UserService(BaseService[User]):
    def __init__(self, repository: UserRepository):
        super().__init__(model=User, repository=repository)

    async def pre_create(self, schema: UserCreate) -> dict[str, Any]:
        # Hash password and inject into extra_data
        hashed = Security.hash_password(schema.password)
        return {
            "hashed_password": hashed,
            "tenant_id": ctx.get("tenant_id"),
        }

    async def post_create(self, user: User) -> None:
        # Update user activity log in the same atomic session
        await self.repository.log_audit_event(user.id, action="user_registered")

Because password is marked with exclude=True, Pydantic validates it on the incoming request, but omits it when dumping into the repository. The hook cleanly intercepts the raw value, generates the hash, and injects it directly into the database payload.

All standard persistence calls, dialect-aware queries, error handling, and transaction finalization remain entirely encapsulated. (And with constructor auto-wiring, you don't have to write manual dependency injection factories to instantiate it).

6. The Transaction Dilemma: The `_safe_commit` Pattern

One of the thorniest architectural questions in a layered FastAPI system is: Who owns the transaction commit?

If the service always commits, it cannot be safely composed. If an OrderService needs to coordinate with an InventoryService.deduct() and a LedgerService.record_entry(), the inner service calls will commit prematurely, destroying transaction atomicity across those database tables if a subsequent step fails.
If the service never commits, developers must remember to manually commit in every single endpoint, re-introducing boilerplate.

We solved this through conditional transaction awareness (_safe_commit):

    async def _safe_commit(self) -> None:
        """Execute a session commit only if transaction state is not managed externally."""
        session_info = self.repository.db.info
        if not session_info.get("uow_managed", False):
            try:
                await self.repository.db.commit()
            except Exception:
                await self.repository.db.rollback()
                raise

How the Rollback Guarantee Operates:

Standalone Execution: When an endpoint calls user_service.create() directly, uow_managed is False. The service cleanly commits the transaction upon completion. If an exception occurs during pre_create, on_create, or post_create, it propagates outward to the surrounding transaction or request-session scope, which is responsible for executing the rollback.
Coordinated Workflows: When multiple services participate in a higher-level transaction, an explicit Unit of Work marks uow_managed = True. The service respects this boundary: it flushes queries so generated IDs are available to sibling services, but skips the physical commit—leaving the atomic commit or rollback decision entirely to the outer Unit of Work boundary.

The Cooperative Contract (and Its Boundaries)

It is important to understand the mechanics and limits of this check: session.info is SQLAlchemy's mutable dictionary scoped to a single AsyncSession instance.

_safe_commit() operates on an in-process, cooperative contract:

Shared Session Requirement: All participating services must receive the same AsyncSession instance. In the standard request-scoped setup, resolving them within the same scope provides that shared instance. Services instantiated manually or resolved through separate scopes may receive different sessions and therefore will not share the same transaction boundary.
Cooperative Opt-In: It is not an engine-level distributed transaction coordinator. If a developer coordinates multiple mutating services without wrapping them in an outer Unit of Work context (or without explicitly setting uow_managed = True), each service assumes standalone ownership and will commit independently.

This contract trades heavy, rigid transaction management for a pragmatic convention that keeps individual CRUD operations self-contained while allowing multi-service workflows to remain atomic when coordinated.

7. What This Pattern Does NOT Try to Do

To keep architecture clean and prevent excessive inheritance hierarchies, we establish strict non-goals:

It does not force every workflow into hooks: If an endpoint executes a complex read-only report aggregating data from external APIs without database mutations, you don't force it into pre_create. Write an explicit, named service method (generate_monthly_report()).
It does not replace domain models: Business validation and state transition rules that enforce invariants belong in your Domain Model—not necessarily every SQLAlchemy ORM model that exists primarily for table mapping and persistence. The service layer acts as the application orchestrator, coordinating persistence, hooks, and transaction boundaries.

8. Summary: Stop Writing Forwarders

A Service layer in FastAPI should not be an anemic proxy that merely mirrors repository methods.

By utilizing reusable base services equipped with standard CRUD lifecycles and explicit hooks (pre_create, post_create, _safe_commit), you eliminate hundreds of lines of useless boilerplate while ensuring that your business rules and transaction boundaries remain clean, testable, and safe.

In ZCore, we designed BaseService around this exact hook-driven philosophy.

Up Next: Part 3

Now that we have clean data access in the Repository and hook-driven business logic in the Service layer, how do we expose these operations to HTTP without writing repetitive FastAPI endpoints for every single entity?

In Part 3, we break down Why Generic CRUD Routers Fail in FastAPI: Route Shadowing, Architectural Coupling, and How to Fix Them.

How do you structure your Service layer in FastAPI? Do you prefer explicit service methods for everything, or hook-driven base pipelines? Share your thoughts in the comments below!

The Repository Pattern in FastAPI: Abstraction or Overengineering?

Ali Alf Ostovar — Mon, 12 Oct 2026 13:00:00 +0000

If you search for architectural patterns in modern Python, you will inevitably encounter the Repository pattern.

On one side, advocates argue that repositories are essential for clean architecture, decoupling business logic from databases, and enabling testability. On the other side, an equally vocal group of engineers argues that placing a repository layer on top of a modern Object-Relational Mapper (ORM) like SQLAlchemy is redundant boilerplate at best—and a harmful anti-pattern at worst.

Both sides have valid points. The issue is rarely whether the Repository pattern is inherently "good" or "bad." The real issue is: what are you expecting the repository to abstract?

In this post, we’ll examine both sides of the argument and look at the practical boundaries we established for data access while building and structuring our FastAPI applications.

1. The Classic Pitch for the Repository Pattern

The textbook case for the Repository pattern stems from Domain-Driven Design (DDD) and Martin Fowler’s enterprise patterns. In a standard FastAPI setup, it is typically presented like this:

HTTP Request → FastAPI Route Handler → Service Layer → Repository → Database

The arguments in favor are familiar:

Separation of Concerns: Route handlers shouldn't construct database queries, and service functions shouldn't juggle raw database sessions directly.
Centralized Data Access: Queries are kept in one place instead of being scattered and copy-pasted across multiple route files.
Testability: You can mock the repository interface in unit tests without standing up a live database connection.

This sounds reasonable on paper. But in practice, many implementations quickly turn into a maintenance burden.

2. Why Critics Call It an Anti-Pattern

Critics of the pattern—especially those working with modern ORMs like SQLAlchemy 2.0—point out several legitimate failure modes:

The "Database Portability" Myth

The most common justification in introductory tutorials is: "If you use a repository, you can switch from PostgreSQL to MongoDB tomorrow!"

In real-world engineering, teams rarely swap relational databases for document stores midway through a project. Even if they do, relational concepts like joins, foreign keys, and atomic multi-table transactions cannot be cleanly abstracted away by a generic interface. Pretending storage engines are interchangeable usually produces an anemic, lowest-common-denominator API.

Leaky Abstractions & Performance Penalties

Real applications need eager loading (selectinload, joinedload), partial column selection (load_only), complex joins, and window functions.

When a repository attempts to hide the ORM entirely, it usually fails in one of two ways:

God Methods: You end up writing bespoke methods for every query combination: get_by_id_with_user_and_orders(), get_active_users_page(), etc.
Leaking the ORM Anyway: You pass SQLAlchemy binary expressions or options into the repository—meaning the caller is still coupled to the ORM, defeating the abstraction.

A repository is not automatically a good abstraction simply because it is called a repository.

3. The Real Problem: What Are You Abstracting?

The breakdown happens when developers treat the repository as a Storage Engine Replacement:

[Bad Abstraction]
Service → GenericRepository (Tries to hide SQL/ORM) → SQLAlchemy

If your goal is to pretend that SQLAlchemy does not exist, you will fight the ORM at every turn.

Instead, a practical repository should act as an Application-Level Data Access Boundary:

[Useful Abstraction]
Service → Application Data Boundary → SQLAlchemy 2.0

The goal here is not database independence. The goal is architectural consistency.

4. What We Actually Needed from Data Access

While structuring our application modules, we explicitly decided what we did not care about:

We did not care about switching to MongoDB.
We did not want to hide SQLAlchemy's query capabilities.
We did not want anemic interfaces that broke when complex joins were needed.

Instead, we focused on the concrete requirements we encountered:

Eliminate Repetitive CRUD Boilerplate: Writing identical query filters, primary-key lookups, and dialect-aware batch operations across multiple domain models was a waste of engineering time.
Enforce System-Wide Conventions Consistently: Concerns like soft-deletion (deleted_at is None) and model-level query scoping needed to be applied reliably without depending on developers remembering to add .where(...) to every query.
Keep Route Handlers Clean: Endpoints should handle HTTP mechanics; they should not assemble database queries directly.
First-Class Eager Loading & Field Pruning: The repository had to accept execution options (load_only, joinedload, selectinload) natively without awkward workarounds.

5. A Pragmatic Implementation: Composable Mixins

To achieve this without creating a rigid base class, we divided data access into focused, composable responsibilities using mixins:

BaseRepository[ModelType]
├── ReadRepositoryMixin   (Lookups, pagination, existence checks)
├── WriteRepositoryMixin  (Create, batch inserts, dialect returning, soft-delete)
└── SearchRepositoryMixin (Structured, policy-checked dynamic queries)

Here is the exact implementation structure we arrived at:

1. Base Abstraction & Scoping

The base class binds the session and model, providing a single query-entrypoint that automatically respects model-level scopes:

from collections.abc import Sequence
from typing import Any, Generic, TypeVar
from sqlalchemy import Select, inspect, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import load_only
from sqlalchemy.orm.interfaces import ExecutableOption

ModelType = TypeVar("ModelType", bound=Any)

class AbstractRepository(Generic[ModelType]):
    db: AsyncSession
    model: type[ModelType]
    pk: Any
    pk_name: str
    cursor_field: str

    def _get_base_query(self) -> Select:
        query = select(self.model)
        scoper = getattr(self.model, "scope_query", None)
        if scoper:
            return scoper(query)
        return query

    def _apply_filters(self, query: Select, *criterion: Any, **filters: Any) -> Select:
        if criterion:
            query = query.where(*criterion)
        if filters:
            query = query.filter_by(**filters)
        return query

    def _supports_soft_delete(self) -> bool:
        return hasattr(self.model, "deleted_at") and hasattr(self.model, "soft_delete")

2. Reading Without Hiding ORM Capabilities

Notice that get() does not attempt to hide SQLAlchemy. It accepts SQLAlchemy ExecutableOption instances (such as joinedload or selectinload) and field lists (load_only) directly:

class ReadRepositoryMixin(AbstractRepository[ModelType]):
    async def get(
        self,
        *criterion: Any,
        fields: list[Any] | None = None,
        options: list[ExecutableOption] | None = None,
        **filters: Any,
    ) -> ModelType | None:
        query = self._get_base_query()
        query = self._apply_filters(query, *criterion, **filters)
        if fields:
            query = query.options(load_only(*fields))
        if options:
            query = query.options(*options) if isinstance(options, list) else query.options(options)
        result = await self.db.execute(query)
        return result.scalars().first()

3. Mutations Without Hijacking Transactions

A common anti-pattern in naive repositories is calling await session.commit() inside write methods. This prematurely closes the transaction boundary, preventing the higher-level caller (such as a service layer or transaction manager) from composing multiple repository calls into one atomic operation.

In our write mixin, the repository handles persistence mechanics—it flushes changes to populate IDs, but never commits the transaction:

class WriteRepositoryMixin(Generic[ModelType], AbstractRepository[ModelType]):
    async def create(self, schema: BaseModel, **extra_data: Any) -> ModelType:
        data = schema.model_dump()
        data.update(extra_data)
        record = self.model(**data)
        self.db.add(record)
        await self.db.flush()
        await self.db.refresh(record)
        return record

    async def delete(self, target: ModelType | Any, force: bool = False) -> ModelType | None:
        if isinstance(target, self.model):
            record = target
        else:
            if force:
                query = select(self.model).where(getattr(self.model, self.pk_name) == target)
                result = await self.db.execute(query)
                record = result.scalars().first()
            else:
                record = await self.get(**{self.pk_name: target})

            if not record:
                return None

        if not force and self._supports_soft_delete():
            record.soft_delete()
            await self.db.flush()
            await self.db.refresh(record)
            return record

        await self.db.delete(record)
        await self.db.flush()
        return record

Transaction ownership remains with the caller. The repository issues operations against the session; commit decisions belong at a higher layer.

6. What This Repository Does NOT Try to Do

To keep the pattern healthy, establishing clear boundaries on what the repository should avoid is just as important as what it implements:

It does not replace SQLAlchemy: We do not define artificial query syntax to hide SQLAlchemy expressions. If a domain query requires an advanced CTE or custom window function, we write SQLAlchemy queries directly.
It does not promise database portability: The repository is explicitly designed around relational databases and SQLAlchemy 2.0 semantics.
It does not own business logic: Validations, cross-entity coordination, and external side-effects belong in services, not repositories.
It does not own transactions: The repository executes operations against the session; it never triggers physical COMMIT calls.

7. So, Is the Repository Pattern an Anti-Pattern?

Sometimes.

If you are writing a repository layer solely because an architecture textbook said you must, or if you are wrapping basic ORM calls in an attempt to make PostgreSQL swappable with MongoDB, it is almost certainly an anti-pattern that creates unnecessary indirection.

However, if you treat the repository as a consistent application data boundary—one that automates repetitive CRUD mechanics, guarantees scoping and soft-deletion policies, and respects your ORM's native querying power—it brings structure and stability to growing codebases.

An Open-Source Note

After refining these composable mixins across several applications, I eventually structured this data layer—along with an atomic Unit of Work and scoped dependency injection—into an architectural chassis for FastAPI called ZCore.

You certainly don’t need to adopt an entire framework to benefit from this; the snippets above are intentionally designed to work standalone in vanilla SQLAlchemy 2.0.

However, if you are building modular systems with FastAPI and want to explore the complete repository implementation (or borrow the mixins directly), feel free to check out the project on GitHub.