Database Schemas¶
Pydantic schemas handle three things in this codebase: input validation, output serialization, and API contracts that frontend and backend can rely on. Schemas are separate from the SQLAlchemy models — keeping the two layers split lets you control exactly what each endpoint accepts and returns.
Where Schemas Live¶
Each module owns its schemas, colocated with the model and CRUD:
backend/src/modules/
├── user/schemas.py # UserCreate, UserRead, UserUpdate, UserAnonymize, ...
├── tier/schemas.py # TierCreate, TierRead, TierUpdate
├── rate_limit/schemas.py # RateLimitCreate, RateLimitRead, RateLimitUpdate
└── api_keys/schemas.py # APIKeyCreate, APIKeyRead, APIKeyUpdate, KeyUsageRead
Cross-module shared schemas (timestamp/soft-delete mixins, common error shapes) live in backend/src/modules/common/schemas.py.
Common Mixin Schemas¶
modules/common/schemas.py provides two reusable Pydantic mixins matching the SQLAlchemy mixins:
# modules/common/schemas.py
class TimestampSchema(BaseModel):
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC).replace(tzinfo=None))
updated_at: datetime | None = Field(default=None)
# serializers cast both to ISO strings
class PersistentDeletion(BaseModel):
deleted_at: datetime | None = Field(default=None)
is_deleted: bool = False
Compose them onto your full-record schema where applicable.
The User Schemas¶
modules/user/schemas.py is the most extensive example. The pattern is one schema per role — each operation gets its own shape:
from datetime import datetime
from typing import Annotated
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from ..common.schemas import PersistentDeletion, TimestampSchema
# Common fields shared by create/update/full-record
class UserBase(BaseModel):
name: Annotated[str, Field(min_length=2, max_length=30, examples=["User Userson"])]
username: Annotated[
str,
Field(min_length=2, max_length=20, pattern=r"^[a-z0-9]+$", examples=["userson"]),
]
email: Annotated[EmailStr, Field(examples=["user.userson@example.com"])]
# Full record (used internally — never returned to clients)
class User(TimestampSchema, UserBase, PersistentDeletion):
hashed_password: str
is_superuser: bool = False
profile_image_url: str = "https://www.profileimageurl.com"
tier_id: int | None = None
# OAuth
google_id: str | None = None
github_id: str | None = None
oauth_provider: str | None = None
email_verified: bool = False
# API response — explicitly excludes sensitive fields
class UserRead(BaseModel):
id: int
name: Annotated[str, Field(min_length=2, max_length=30)]
username: Annotated[str, Field(min_length=2, max_length=20, pattern=r"^[a-z0-9]+$")]
email: EmailStr
profile_image_url: str
is_deleted: bool = False
tier_id: int | None
is_superuser: bool = False
email_verified: bool = False
oauth_provider: str | None = None
# API request body for POST /users/
class UserCreate(UserBase):
model_config = ConfigDict(extra="forbid") # reject unknown fields
password: Annotated[
str,
Field(
min_length=8,
description=(
"Password must be at least 8 characters and include a number, "
"uppercase letter, lowercase letter, and special character"
),
examples=["Str1ngst!"],
pattern=r"^.{8,}|[0-9]+|[A-Z]+|[a-z]+|[^a-zA-Z0-9]+$",
),
]
# OAuth fields — populated when user signs up via Google/GitHub
google_id: str | None = None
github_id: str | None = None
oauth_provider: str | None = None
# What the service writes to the DB (raw password replaced with hash)
class UserCreateInternal(UserBase):
hashed_password: str
google_id: str | None = None
github_id: str | None = None
oauth_provider: str | None = None
email_verified: bool = False
# Partial update — every field optional
class UserUpdate(BaseModel):
model_config = ConfigDict(extra="forbid")
name: Annotated[str | None, Field(min_length=2, max_length=30, default=None)]
username: Annotated[
str | None,
Field(min_length=2, max_length=20, pattern=r"^[a-z0-9]+$", default=None),
]
email: Annotated[EmailStr | None, Field(default=None)]
profile_image_url: Annotated[
str | None,
Field(pattern=r"^(https?|ftp)://[^\s/$.?#].[^\s]*$", default=None),
]
class UserUpdateInternal(UserUpdate):
updated_at: datetime # service stamps this before persisting
class UserTierUpdate(BaseModel):
tier_id: int
class UserDelete(BaseModel):
model_config = ConfigDict(extra="forbid")
is_deleted: bool
deleted_at: datetime
# GDPR/LGPD anonymization payload
class UserAnonymize(BaseModel):
model_config = ConfigDict(extra="forbid")
name: str
username: str
hashed_password: str | None = None
# ...other PII-clearing fields...
Naming Conventions¶
The schemas follow a consistent vocabulary across modules:
| Suffix | Use |
|---|---|
Base |
Common fields shared across create/update/full schemas |
(none — class name = User) |
Full-record schema (every column, mostly internal) |
Read |
API response — drops sensitive/internal fields |
Create |
API request body for POST |
CreateInternal |
What the service stores (raw password → hashed_password) |
Update |
Partial update body for PATCH (all fields optional) |
UpdateInternal |
What the service stores on update (e.g. with stamped updated_at) |
TierUpdate, Anonymize, Delete, … |
Operation-specific narrow schemas |
Why Internal vs External¶
The split between Create and CreateInternal (and likewise for updates) keeps the API surface honest:
UserCreateacceptspassword: strfrom the client.- The service hashes the password and constructs a
UserCreateInternalwithhashed_passwordinstead. crud_users.create(db=db, object=user_internal)is what actually hits the database.
The client can never set hashed_password directly, and the model never sees a plaintext password.
Field Validation¶
Annotated + Field¶
The codebase uses Annotated[T, Field(...)] for validation rules:
Field parameter |
Effect |
|---|---|
min_length / max_length |
String length bounds |
pattern |
Regex validation (e.g. r"^[a-z0-9]+$" for usernames) |
gt / ge / lt / le |
Numeric bounds |
default |
Default value |
examples |
OpenAPI example values shown in /docs |
description |
Doc string visible in OpenAPI |
EmailStr¶
Pydantic's EmailStr validates the email format and normalizes the casing.
ConfigDict(extra="forbid")¶
Set on UserCreate, UserUpdate, etc. — anything the client sends beyond the declared fields raises a 422. This matters most for create/update payloads where stray fields could otherwise sneak through.
from_attributes¶
Use ConfigDict(from_attributes=True) when you need to build a Pydantic schema from a SQLAlchemy model instance directly. The boilerplate's services mostly work with dicts (FastCRUD's default return shape), so this is rarely needed — but it's the right setting if you do UserRead.model_validate(orm_user).
Schema Patterns¶
Optional Fields in Updates¶
The convention is all fields optional in *Update schemas:
class UserUpdate(BaseModel):
model_config = ConfigDict(extra="forbid")
name: Annotated[str | None, Field(min_length=2, max_length=30, default=None)]
email: Annotated[EmailStr | None, Field(default=None)]
# ...
The service then writes only the fields the client actually provided.
Custom Validators¶
For cross-field rules or transforms:
from pydantic import field_validator, model_validator
class WidgetCreate(BaseModel):
name: str
color: str
quantity: int = 1
@field_validator("name")
@classmethod
def normalize_name(cls, v: str) -> str:
if v.lower() in {"admin", "system"}:
raise ValueError("Reserved name")
return v.strip().lower()
@model_validator(mode="after")
def check_quantity(self) -> "WidgetCreate":
if self.color == "rare" and self.quantity > 1:
raise ValueError("Rare widgets are limited to one per request")
return self
field_validator validates one field; model_validator(mode="after") runs after all fields are set and can validate combinations.
Computed Fields¶
For values derived at serialization time (not stored):
from pydantic import computed_field
class UserReadWithStats(UserRead):
created_at: datetime # add this if your read schema doesn't already have it
@computed_field
@property
def display_name(self) -> str:
return f"@{self.username}"
@computed_field
@property
def age_days(self) -> int:
return (datetime.now(UTC) - self.created_at).days
Multi-Record Responses¶
The boilerplate uses FastCRUD's PaginatedListResponse for paginated list endpoints:
from fastcrud import PaginatedListResponse, compute_offset, paginated_response
@router.get("/", response_model=PaginatedListResponse[UserRead])
async def get_users(
db: Annotated[AsyncSession, Depends(async_session)],
user_service: Annotated[UserService, Depends(get_user_service)],
page: int = 1,
items_per_page: int = 10,
) -> dict[str, Any]:
result = await user_service.get_paginated(
skip=compute_offset(page, items_per_page),
limit=items_per_page,
db=db,
)
return paginated_response(crud_data=result, page=page, items_per_page=items_per_page)
The response shape:
{
"data": [{ "id": 1, "name": "...", "username": "..." }],
"total_count": 150,
"has_more": true,
"page": 1,
"items_per_page": 10
}
For single-record endpoints, return the schema directly:
@router.get("/me", response_model=UserRead)
async def me(current_user: Annotated[dict[str, Any], Depends(get_current_user)]):
return current_user
Adding Schemas for a New Module¶
- Create the schema file:
backend/src/modules/widgets/schemas.py - Define a
WidgetBasewith the fields shared by create/update/read - Add
WidgetCreate,WidgetRead,WidgetUpdate(and any internal variants you need) - Wire them up in the module's
routes.pyandservice.py
# backend/src/modules/widgets/schemas.py
from datetime import datetime
from typing import Annotated
from pydantic import BaseModel, ConfigDict, Field
from ..common.schemas import TimestampSchema
class WidgetBase(BaseModel):
name: Annotated[str, Field(min_length=1, max_length=50)]
description: Annotated[str | None, Field(max_length=255, default=None)]
class WidgetCreate(WidgetBase):
model_config = ConfigDict(extra="forbid")
class WidgetRead(WidgetBase):
model_config = ConfigDict(from_attributes=True)
id: int
owner_id: int
created_at: datetime
class WidgetUpdate(BaseModel):
model_config = ConfigDict(extra="forbid")
name: Annotated[str | None, Field(min_length=1, max_length=50, default=None)]
description: Annotated[str | None, Field(max_length=255, default=None)]
Common Pitfalls¶
Don't expose sensitive fields¶
# BAD — leaks the password hash
class UserRead(BaseModel):
hashed_password: str
# GOOD — read-only public shape
class UserRead(BaseModel):
id: int
name: str
username: str
email: EmailStr
profile_image_url: str
Don't query the database in validators¶
# BAD — every request hits the DB twice
@field_validator("email")
@classmethod
def email_must_be_unique(cls, v):
if crud_users.exists(email=v): # I/O in a validator
raise ValueError("Email already exists")
# GOOD — let the DB unique constraint and service-layer logic handle it
The boilerplate's UserService.create already checks for duplicates before insert. The DB unique constraint is the final guardrail.
Don't reuse the same schema for create and update¶
A Create schema requires fields that an Update schema should be able to omit. Splitting them avoids accidental "this field defaulted because the client forgot it" bugs.
Next Steps¶
- CRUD Operations - How schemas plug into FastCRUD
- Migrations - Manage the underlying database changes
- API Endpoints - Use schemas in route handlers