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_notesandstripe_customer_idas return fields becauseresponse_model=UserResponseis 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!
Top comments (0)