Project 10 (Final Project): Production Validation Framework
Project 10 (Final Project): Production Validation Framework
Build a complete production application demonstrating all Pydantic features—API validation, settings management, database models, LLM integration, custom types, and comprehensive error handling. This capstone project proves mastery of data validation architecture in Python.
Learning Objectives
By completing this project, you will:
- Design layered validation architectures - Understand how validation flows through API, service, and repository layers
- Create reusable custom type libraries - Build domain-specific types that enforce business rules consistently
- Implement comprehensive error handling - Aggregate and present errors in user-friendly formats across all layers
- Optimize validation performance - Use model_construct, caching, and lazy validation for production workloads
- Integrate all Pydantic features - Combine settings, validators, generics, discriminated unions, and ORM integration
- Build production-ready systems - Apply testing pyramids, deployment patterns, and observability for validated data
Deep Theoretical Foundation
Layered Architecture: API to Service to Repository
Production applications require clear separation of concerns. Each layer has distinct validation responsibilities:
┌─────────────────────────────────────────────────────────────────────────────┐
│ CLIENT / EXTERNAL WORLD │
│ (Mobile apps, web browsers, other APIs) │
└─────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ API LAYER │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Responsibilities: │ │
│ │ - Parse HTTP requests (JSON, form data, query params) │ │
│ │ - Validate INPUT MODELS (UserCreate, OrderRequest) │ │
│ │ - Handle authentication/authorization │ │
│ │ - Transform domain objects to OUTPUT MODELS (UserResponse) │ │
│ │ - Return structured error responses │ │
│ │ │ │
│ │ Pydantic Features: │ │
│ │ - BaseModel for request/response schemas │ │
│ │ - Field constraints (min_length, ge, le) │ │
│ │ - Custom validators for input normalization │ │
│ │ - response_model for serialization │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ SERVICE LAYER │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Responsibilities: │ │
│ │ - Implement business logic and rules │ │
│ │ - Orchestrate between repositories │ │
│ │ - Validate DOMAIN MODELS (User, Order with business rules) │ │
│ │ - Handle cross-entity validation │ │
│ │ - Emit domain events │ │
│ │ │ │
│ │ Pydantic Features: │ │
│ │ - @model_validator for cross-field validation │ │
│ │ - Custom types for domain concepts (Money, PhoneNumber) │ │
│ │ - Discriminated unions for polymorphic domain objects │ │
│ │ - Result types for operation outcomes │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ REPOSITORY LAYER │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Responsibilities: │ │
│ │ - Persist and retrieve data │ │
│ │ - Translate between domain and DB models │ │
│ │ - Handle database constraints │ │
│ │ - Manage transactions │ │
│ │ │ │
│ │ Pydantic Features: │ │
│ │ - SQLModel for unified Pydantic + SQLAlchemy models │ │
│ │ - from_attributes for ORM object conversion │ │
│ │ - model_construct for trusted internal data (skip validation) │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ DATABASE / EXTERNAL SERVICES │
│ (PostgreSQL, Redis, LLM APIs, etc.) │
└─────────────────────────────────────────────────────────────────────────────┘
Key Principle: Validate at boundaries. Data crossing a boundary (external to internal, layer to layer) must be validated at that point.
Model Layering: Input to Domain to Output
A single concept (e.g., “User”) requires multiple models at different layers:
┌─────────────────────────────────────────────────────────────────────────────┐
│ MODEL HIERARCHY │
│ │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │
│ │ INPUT MODEL │ │ DOMAIN MODEL │ │ OUTPUT MODEL │ │
│ │ (API Request) │ │ (Business Core) │ │ (API Response) │ │
│ └────────┬────────┘ └────────┬────────┘ └────────┬────────┘ │
│ │ │ │ │
│ UserCreate User (DB) UserResponse │
│ ├── email: EmailStr ├── id: int ├── id: int │
│ ├── name: str ├── email: str ├── email: str │
│ ├── password: str ├── name: str ├── name: str │
│ └── (required fields) ├── hashed_password ├── created_at │
│ ├── created_at ├── task_count │
│ └── (relationships) └── (computed fields) │
│ │
│ UserUpdate UserListResponse │
│ ├── email: EmailStr? ├── items: list[User] │
│ ├── name: str? ├── total: int │
│ └── (all optional) ├── page: int │
│ └── per_page: int │
└─────────────────────────────────────────────────────────────────────────────┘
Model Flow:
# 1. API receives input model
@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate): # Validated input
# 2. Service layer creates domain model
domain_user = user_service.create(user)
# 3. Response model serializes output
return domain_user # Automatically converted to UserResponse
Custom Type Libraries for Domains
Reusable custom types enforce consistency across your entire application:
┌─────────────────────────────────────────────────────────────────────────────┐
│ DOMAIN TYPE LIBRARY │
│ │
│ ┌───────────────────────────────────────────────────────────────────┐ │
│ │ types/money.py │ │
│ │ │ │
│ │ class Money: │ │
│ │ amount: Decimal (always 2 decimal places) │ │
│ │ currency: str (ISO 4217 code) │ │
│ │ │ │
│ │ - Arithmetic operations (+, -, *) │ │
│ │ - Currency conversion │ │
│ │ - Comparison operators │ │
│ │ - JSON serialization: {"amount": "99.99", "currency": "USD"} │ │
│ └───────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────────┐ │
│ │ types/phone.py │ │
│ │ │ │
│ │ PhoneNumber = Annotated[str, BeforeValidator(normalize_phone)] │ │
│ │ │ │
│ │ Input: "(555) 123-4567", "5551234567", "+1-555-123-4567" │ │
│ │ Output: "+15551234567" (E.164 format) │ │
│ └───────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────────┐ │
│ │ types/address.py │ │
│ │ │ │
│ │ class Address(BaseModel): │ │
│ │ street: str │ │
│ │ city: str │ │
│ │ state: USStateCode │ │
│ │ postal_code: PostalCode │ │
│ │ country: CountryCode = "US" │ │
│ │ │ │
│ │ @model_validator: Validate postal code matches state │ │
│ └───────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────────┐ │
│ │ types/identifiers.py │ │
│ │ │ │
│ │ UserId = Annotated[int, Field(ge=1)] │ │
│ │ OrderId = Annotated[str, Field(pattern=r'^ORD-[A-Z0-9]{10}$')] │ │
│ │ SKU = Annotated[str, Field(pattern=r'^[A-Z]{3}-\d{6}$')] │ │
│ └───────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
Benefits of Custom Type Libraries:
- Single source of truth - Phone validation logic in one place
- Self-documenting - Types describe domain concepts
- Type safety - IDE understands
Moneyis not justfloat - Testable - Unit test types independently of models
Comprehensive Error Handling Strategy
Production applications need errors that help developers and users:
┌─────────────────────────────────────────────────────────────────────────────┐
│ ERROR HANDLING ARCHITECTURE │
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Layer 1: Pydantic ValidationError │ │
│ │ │ │
│ │ - Field-level errors with location (path) │ │
│ │ - Error type codes (string_too_short, missing, etc.) │ │
│ │ - Input value that failed │ │
│ │ - Context for constraint violations │ │
│ └───────────────────────────────────────────────────────────────────── │ │
│ │ │ │
│ ▼ │ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Layer 2: Application Error Aggregation │ │
│ │ │ │
│ │ class AppError: │ │
│ │ code: str # "USER_EMAIL_TAKEN" │ │
│ │ message: str # "Email already registered" │ │
│ │ field: str? # "email" │ │
│ │ details: dict? # {"existing_user_id": 123} │ │
│ │ │ │
│ │ class ValidationResult: │ │
│ │ success: bool │ │
│ │ data: T? │ │
│ │ errors: list[AppError] │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │ │
│ ▼ │ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Layer 3: HTTP Response Formatting │ │
│ │ │ │
│ │ { │ │
│ │ "success": false, │ │
│ │ "error": { │ │
│ │ "code": "VALIDATION_ERROR", │ │
│ │ "message": "Request validation failed", │ │
│ │ "errors": [ │ │
│ │ { │ │
│ │ "field": "email", │ │
│ │ "code": "USER_EMAIL_TAKEN", │ │
│ │ "message": "Email already registered" │ │
│ │ }, │ │
│ │ { │ │
│ │ "field": "order.items[0].quantity", │ │
│ │ "code": "VALUE_TOO_LOW", │ │
│ │ "message": "Quantity must be at least 1" │ │
│ │ } │ │
│ │ ] │ │
│ │ }, │ │
│ │ "request_id": "req_abc123" │ │
│ │ } │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
Error Handling Pattern:
from pydantic import ValidationError
class AppError(BaseModel):
code: str
message: str
field: str | None = None
details: dict | None = None
class ErrorResponse(BaseModel):
success: bool = False
error: dict
request_id: str
def pydantic_to_app_errors(exc: ValidationError) -> list[AppError]:
"""Convert Pydantic errors to application errors."""
errors = []
for error in exc.errors():
field_path = ".".join(str(loc) for loc in error["loc"])
errors.append(AppError(
code=error["type"].upper().replace("_", "_"),
message=error["msg"],
field=field_path,
details={"input": error.get("input"), "ctx": error.get("ctx")}
))
return errors
Performance Optimization Techniques
Validation has overhead. Production systems need optimization strategies:
┌─────────────────────────────────────────────────────────────────────────────┐
│ PERFORMANCE OPTIMIZATION │
│ │
│ 1. model_construct() - Skip Validation for Trusted Data │
│ ─────────────────────────────────────────────────────────────────────────│
│ │
│ # SLOW: Full validation (from database) │
│ user = User.model_validate(db_row) │
│ │
│ # FAST: Skip validation (data already validated when inserted) │
│ user = User.model_construct(**db_row) │
│ │
│ When to use model_construct: │
│ - Loading from database (already validated on insert) │
│ - Internal service-to-service communication │
│ - Cached data retrieval │
│ - NEVER with external input! │
│ │
│ 2. Lazy Validation with __get_pydantic_core_schema__ │
│ ─────────────────────────────────────────────────────────────────────────│
│ │
│ # Expensive validation (e.g., database lookup) only when needed │
│ class LazyUser: │
│ def __get_pydantic_core_schema__(cls, ...): │
│ # Defer expensive checks until actually validated │
│ ... │
│ │
│ 3. Caching Validated Objects │
│ ─────────────────────────────────────────────────────────────────────────│
│ │
│ from functools import lru_cache │
│ │
│ @lru_cache(maxsize=1000) │
│ def get_validated_config(config_key: str) -> AppSettings: │
│ # Validate once, cache result │
│ return AppSettings.model_validate(load_config(config_key)) │
│ │
│ 4. model_validate_json() vs model_validate() │
│ ─────────────────────────────────────────────────────────────────────────│
│ │
│ # SLOWER: Parse JSON to dict, then validate │
│ data = json.loads(json_string) │
│ user = User.model_validate(data) │
│ │
│ # FASTER: Direct JSON parsing in Rust core │
│ user = User.model_validate_json(json_string) │
│ │
│ 5. Selective Validation with computed_field │
│ ─────────────────────────────────────────────────────────────────────────│
│ │
│ class Order(BaseModel): │
│ items: list[OrderItem] │
│ │
│ @computed_field │
│ @cached_property │
│ def total(self) -> Money: │
│ # Computed only when accessed │
│ return sum(item.subtotal for item in self.items) │
└─────────────────────────────────────────────────────────────────────────────┘
Testing Pyramid for Validation
A comprehensive testing strategy ensures validation correctness:
┌───────────┐
╱ ╲
╱ E2E Tests ╲
╱ (Few, Slow) ╲
╱ ╲
╱ - Full API flows ╲
╱ - Real database ╲
╱ - External services ╲
├───────────────────────────┤
╱ Integration Tests ╲
╱ (Medium Speed) ╲
╱ ╲
╱ - API endpoint validation ╲
╱ - Service layer with mocked repos ╲
╱ - Database constraint testing ╲
├─────────────────────────────────────────┤
╱ Unit Tests ╲
╱ (Many, Fast) ╲
╱ ╲
╱ - Individual model validation ╲
╱ - Custom type behavior ╲
╱ - Validator functions ╲
╱ - Error message formatting ╲
└─────────────────────────────────────────────────────────┘
Unit Test Examples:
───────────────────
def test_money_type_parses_string():
assert Money("99.99", "USD").amount == Decimal("99.99")
def test_phone_normalizes_formats():
assert normalize_phone("(555) 123-4567") == "+15551234567"
def test_user_create_requires_password():
with pytest.raises(ValidationError):
UserCreate(email="test@example.com", name="Test")
Integration Test Examples:
──────────────────────────
def test_create_user_endpoint_validates_email():
response = client.post("/users", json={"email": "invalid"})
assert response.status_code == 422
assert "email" in response.json()["errors"][0]["field"]
def test_order_service_validates_inventory():
with pytest.raises(InsufficientStockError):
order_service.create_order(items=[{"sku": "ABC", "qty": 1000}])
E2E Test Examples:
──────────────────
def test_complete_checkout_flow():
# Create user
user = api.create_user(...)
# Add items to cart
cart = api.add_to_cart(user.id, items=[...])
# Checkout with payment
order = api.checkout(cart.id, payment_method="card")
# Verify order in database
assert db.get_order(order.id).status == "confirmed"
Production Deployment Considerations
Validation in production requires additional infrastructure:
┌─────────────────────────────────────────────────────────────────────────────┐
│ PRODUCTION INFRASTRUCTURE │
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Configuration & Secrets │ │
│ │ │ │
│ │ class Settings(BaseSettings): │ │
│ │ # Core settings │ │
│ │ debug: bool = False │ │
│ │ environment: Literal["dev", "staging", "prod"] │ │
│ │ │ │
│ │ # Database │ │
│ │ database_url: PostgresDsn │ │
│ │ database_pool_size: int = Field(ge=1, le=100) │ │
│ │ │ │
│ │ # Secrets │ │
│ │ secret_key: SecretStr │ │
│ │ api_key: SecretStr │ │
│ │ llm_api_key: SecretStr │ │
│ │ │ │
│ │ model_config = SettingsConfigDict( │ │
│ │ env_file=".env", │ │
│ │ env_prefix="APP_", │ │
│ │ ) │ │
│ │ │ │
│ │ @model_validator(mode='after') │ │
│ │ def validate_production_settings(self): │ │
│ │ if self.environment == "prod" and self.debug: │ │
│ │ raise ValueError("Debug mode not allowed in prod") │ │
│ │ return self │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Observability │ │
│ │ │ │
│ │ - Validation error metrics (count by error type) │ │
│ │ - Validation latency histograms │ │
│ │ - Structured logging with field paths │ │
│ │ - Error sampling for debugging │ │
│ │ │ │
│ │ logger.info("validation_complete", │ │
│ │ model="UserCreate", │ │
│ │ duration_ms=12.5, │ │
│ │ valid=True, │ │
│ │ request_id="req_abc123" │ │
│ │ ) │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Health Checks │ │
│ │ │ │
│ │ @app.get("/health") │ │
│ │ async def health(): │ │
│ │ return { │ │
│ │ "status": "healthy", │ │
│ │ "version": settings.version, │ │
│ │ "environment": settings.environment, │ │
│ │ "database": await check_db_connection(), │ │
│ │ "cache": await check_redis_connection(), │ │
│ │ } │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
Project Specification
Functional Requirements
Build a production-ready e-commerce order management system that demonstrates mastery of all Pydantic features:
Core Features:
- User Management - Registration, authentication, profile management
- Product Catalog - Products with variants, pricing, inventory
- Order Processing - Cart, checkout, order status tracking
- Payment Integration - Payment validation and webhook handling
- AI-Powered Features - Product description generation, order summarization
- Configuration Management - Environment-based settings with secrets
Validation Requirements:
- All external input validated at API layer
- Business rules enforced at service layer
- Database constraints at repository layer
- Custom types for domain concepts (Money, SKU, Address)
- Discriminated unions for polymorphic data (PaymentMethod, WebhookEvent)
- Generic response wrappers for consistent API responses
API Endpoints
Authentication:
POST /auth/register Register new user
POST /auth/login Login and get tokens
POST /auth/refresh Refresh access token
Users:
GET /users/me Get current user profile
PATCH /users/me Update profile
GET /users/me/orders List user's orders
Products:
GET /products List products (paginated, filtered)
GET /products/{sku} Get product details
POST /products Create product (admin)
PATCH /products/{sku} Update product (admin)
Cart:
GET /cart Get current cart
POST /cart/items Add item to cart
PATCH /cart/items/{item_id} Update cart item quantity
DELETE /cart/items/{item_id} Remove item from cart
Orders:
POST /orders Create order from cart
GET /orders/{order_id} Get order details
POST /orders/{order_id}/cancel Cancel order
Payments:
POST /payments Process payment
GET /payments/{payment_id} Get payment status
Webhooks:
POST /webhooks/payment Receive payment webhooks
AI Features:
POST /ai/product-description Generate product description
POST /ai/order-summary Generate order summary
Domain Models
# Domain concepts that appear throughout the system
# Custom Types
Money # amount + currency with arithmetic
SKU # Product identifier (pattern: ABC-123456)
OrderId # Order identifier (pattern: ORD-XXXXXXXXXX)
PhoneNumber # E.164 normalized phone
Address # Structured address with validation
PaymentMethod # Discriminated union (card, bank, wallet)
# Core Entities
User # id, email, name, phone, addresses
Product # sku, name, description, price, inventory
ProductVariant # size, color, additional price
CartItem # product_sku, variant, quantity, unit_price
Cart # user_id, items, totals
Order # id, user, items, shipping, payment, status
Payment # id, order_id, method, amount, status
# API Models (per entity)
{Entity}Create # Required fields for creation
{Entity}Update # Optional fields for updates
{Entity}Response # Fields for API responses
{Entity}List # Paginated list wrapper
Solution Architecture
Project Structure
production-app/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI application entry point
│ ├── dependencies.py # Shared dependencies (auth, db)
│ │
│ ├── config/
│ │ ├── __init__.py
│ │ └── settings.py # Pydantic Settings
│ │
│ ├── types/ # Custom domain types
│ │ ├── __init__.py # Export all types
│ │ ├── money.py # Money class
│ │ ├── phone.py # PhoneNumber type
│ │ ├── address.py # Address model
│ │ ├── identifiers.py # SKU, OrderId, etc.
│ │ └── validators.py # Shared validators
│ │
│ ├── models/ # Pydantic models per domain
│ │ ├── __init__.py
│ │ ├── base.py # Base models, mixins
│ │ ├── user.py # User models (all layers)
│ │ ├── product.py # Product models
│ │ ├── cart.py # Cart models
│ │ ├── order.py # Order models
│ │ ├── payment.py # Payment models (unions!)
│ │ ├── webhook.py # Webhook event models
│ │ ├── ai.py # LLM structured outputs
│ │ ├── responses.py # Generic response wrappers
│ │ └── errors.py # Error response models
│ │
│ ├── api/
│ │ ├── __init__.py
│ │ ├── routes/
│ │ │ ├── __init__.py
│ │ │ ├── auth.py
│ │ │ ├── users.py
│ │ │ ├── products.py
│ │ │ ├── cart.py
│ │ │ ├── orders.py
│ │ │ ├── payments.py
│ │ │ ├── webhooks.py
│ │ │ └── ai.py
│ │ │
│ │ └── error_handlers.py # Exception handlers
│ │
│ ├── services/ # Business logic
│ │ ├── __init__.py
│ │ ├── user_service.py
│ │ ├── product_service.py
│ │ ├── cart_service.py
│ │ ├── order_service.py
│ │ ├── payment_service.py
│ │ └── ai_service.py # LLM integration
│ │
│ ├── db/
│ │ ├── __init__.py
│ │ ├── session.py # Database connection
│ │ ├── models.py # SQLModel table models
│ │ └── repositories/
│ │ ├── __init__.py
│ │ ├── base.py # Generic repository
│ │ ├── user_repo.py
│ │ ├── product_repo.py
│ │ ├── cart_repo.py
│ │ └── order_repo.py
│ │
│ └── utils/
│ ├── __init__.py
│ ├── security.py # Password hashing, JWT
│ └── pagination.py # Pagination helpers
│
├── tests/
│ ├── __init__.py
│ ├── conftest.py # Fixtures
│ ├── unit/
│ │ ├── test_types.py
│ │ ├── test_models.py
│ │ └── test_validators.py
│ ├── integration/
│ │ ├── test_api_users.py
│ │ ├── test_api_orders.py
│ │ └── test_services.py
│ └── e2e/
│ └── test_checkout_flow.py
│
├── alembic/ # Database migrations
│ ├── versions/
│ └── env.py
│
├── pyproject.toml
├── docker-compose.yml
├── Dockerfile
└── README.md
Component Interaction Diagram
┌─────────────────────────────────────────────────────────────────────────────┐
│ EXTERNAL │
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ Mobile │ │ Web │ │ Webhook │ │ LLM API │ │
│ │ App │ │ Client │ │ Events │ │ (OpenAI) │ │
│ └─────┬─────┘ └─────┬─────┘ └─────┬─────┘ └─────┬─────┘ │
└─────────┼──────────────┼──────────────┼──────────────┼────────────────────┘
│ │ │ │
▼ ▼ ▼ ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ API LAYER (FastAPI) │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ /auth │ │ /users │ │ /orders │ │/webhooks │ │ /ai │ │
│ │ │ │ │ │ │ │ │ │ │ │
│ │ UserAuth │ │UserCreate│ │OrderReq │ │ Webhook │ │ AIQuery │ │
│ │ Response │ │UserResp │ │OrderResp │ │ Event │ │AIOutput │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │ │ │
│ └─────────────┼─────────────┼─────────────┼─────────────┘ │
│ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────┐ │
│ │ Error Handler │ │
│ │ ValidationError → ErrorResponse │ │
│ └─────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ SERVICE LAYER │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ UserService │ │ OrderService │ │PaymentService│ │ AIService │ │
│ │ │ │ │ │ │ │ │ │
│ │ - validate │ │ - validate │ │ - validate │ │ - generate │ │
│ │ business │ │ inventory │ │ payment │ │ content │ │
│ │ rules │ │ - calculate │ │ - process │ │ - parse LLM │ │
│ │ │ │ totals │ │ webhooks │ │ output │ │
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │ │ │
│ │ Domain Models (with custom types) │ │
│ │ ┌─────────────────────────────────┐ │ │
│ ├───►│ Money, Address, PaymentMethod │◄─────────────┤ │
│ │ └─────────────────────────────────┘ │ │
└──────────┼─────────────────────────────────────────────────────┼──────────┘
│ │
▼ ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ REPOSITORY LAYER │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ UserRepo │ │ OrderRepo │ │ ProductRepo │ │
│ │ │ │ │ │ │ │
│ │ - SQLModel │ │ - SQLModel │ │ - SQLModel │ │
│ │ - CRUD ops │ │ - CRUD ops │ │ - CRUD ops │ │
│ │ - Relations │ │ - Relations │ │ - Relations │ │
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │ │
│ └─────────────────┼─────────────────┘ │
│ ▼ │
│ ┌───────────────┐ │
│ │ Session │ │
│ └───────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ INFRASTRUCTURE │
│ │
│ ┌───────────────┐ ┌───────────────┐ ┌───────────────┐ │
│ │ PostgreSQL │ │ Redis │ │ OpenAI │ │
│ │ (Primary DB) │ │ (Cache) │ │ (LLM API) │ │
│ └───────────────┘ └───────────────┘ └───────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
Phased Implementation Guide
Phase 1: Project Foundation (Day 1)
Goal: Set up project structure with settings and basic types.
- Create project structure with all directories
- Install dependencies:
pip install fastapi uvicorn pydantic pydantic-settings sqlmodel \ alembic asyncpg redis instructor openai python-jose passlib - Implement settings management:
# app/config/settings.py from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic import Field, SecretStr, PostgresDsn, model_validator from typing import Literal class Settings(BaseSettings): # Application app_name: str = "Production App" environment: Literal["dev", "staging", "prod"] = "dev" debug: bool = False version: str = "1.0.0" # Database database_url: PostgresDsn database_pool_size: int = Field(5, ge=1, le=50) # Redis redis_url: str = "redis://localhost:6379" # Security secret_key: SecretStr access_token_expire_minutes: int = 30 refresh_token_expire_days: int = 7 # LLM openai_api_key: SecretStr openai_model: str = "gpt-4" model_config = SettingsConfigDict( env_file=".env", env_prefix="APP_", case_sensitive=False, ) @model_validator(mode='after') def validate_production(self): if self.environment == "prod": if self.debug: raise ValueError("Debug must be False in production") return self settings = Settings() - Create base custom types:
# app/types/money.py from decimal import Decimal from pydantic import GetCoreSchemaHandler from pydantic_core import CoreSchema, core_schema from typing import Any class Money: __slots__ = ('amount', 'currency') def __init__(self, amount: Decimal | str | float, currency: str = "USD"): self.amount = Decimal(str(amount)).quantize(Decimal("0.01")) self.currency = currency.upper() def __repr__(self): return f"{self.currency} {self.amount}" def __add__(self, other: "Money") -> "Money": if self.currency != other.currency: raise ValueError("Cannot add different currencies") return Money(self.amount + other.amount, self.currency) @classmethod def __get_pydantic_core_schema__( cls, _source: type, handler: GetCoreSchemaHandler ) -> CoreSchema: return core_schema.no_info_after_validator_function( cls._validate, core_schema.union_schema([ core_schema.is_instance_schema(Money), core_schema.dict_schema(), core_schema.str_schema(), core_schema.float_schema(), ]) ) @classmethod def _validate(cls, value: Any) -> "Money": if isinstance(value, Money): return value if isinstance(value, dict): return cls(value["amount"], value.get("currency", "USD")) return cls(value)
Checkpoint: Settings load from .env and custom types work.
Phase 2: Core Models and Types (Day 2-3)
Goal: Create all domain models and custom types.
- Create custom types:
PhoneNumber(Annotated type with normalization)Address(BaseModel with state/postal validation)SKU,OrderId(Annotated pattern types)
- Create base models and mixins:
# app/models/base.py from pydantic import BaseModel, ConfigDict from datetime import datetime class TimestampMixin(BaseModel): created_at: datetime updated_at: datetime | None = None class APIModelConfig: """Standard config for API models""" model_config = ConfigDict( from_attributes=True, populate_by_name=True, str_strip_whitespace=True, ) -
Implement all domain models (User, Product, Cart, Order, Payment)
- Create generic response wrappers:
# app/models/responses.py from pydantic import BaseModel from typing import Generic, TypeVar T = TypeVar('T') class APIResponse(BaseModel, Generic[T]): success: bool = True data: T | None = None message: str | None = None class PaginatedResponse(BaseModel, Generic[T]): items: list[T] total: int page: int per_page: int pages: int
Checkpoint: All models import and validate correctly.
Phase 3: Database Layer (Day 4-5)
Goal: Implement SQLModel tables and repositories.
- Create SQLModel table definitions:
# app/db/models.py from sqlmodel import SQLModel, Field, Relationship from typing import Optional from datetime import datetime class UserDB(SQLModel, table=True): __tablename__ = "users" id: int | None = Field(default=None, primary_key=True) email: str = Field(unique=True, index=True) name: str hashed_password: str phone: str | None = None created_at: datetime = Field(default_factory=datetime.utcnow) orders: list["OrderDB"] = Relationship(back_populates="user") - Set up database session:
# app/db/session.py from sqlmodel import create_engine, Session from app.config.settings import settings engine = create_engine(str(settings.database_url)) def get_session(): with Session(engine) as session: yield session - Implement repository pattern:
# app/db/repositories/base.py from typing import Generic, TypeVar, Type from sqlmodel import Session, SQLModel, select ModelType = TypeVar("ModelType", bound=SQLModel) class BaseRepository(Generic[ModelType]): def __init__(self, session: Session, model: Type[ModelType]): self.session = session self.model = model def get(self, id: int) -> ModelType | None: return self.session.get(self.model, id) def list(self, skip: int = 0, limit: int = 100) -> list[ModelType]: stmt = select(self.model).offset(skip).limit(limit) return self.session.exec(stmt).all() def create(self, obj: ModelType) -> ModelType: self.session.add(obj) self.session.commit() self.session.refresh(obj) return obj - Set up Alembic migrations
Checkpoint: Database operations work with migrations.
Phase 4: Service Layer (Day 6-7)
Goal: Implement business logic with validation.
- User service with authentication:
# app/services/user_service.py from app.models.user import UserCreate, UserResponse from app.db.repositories.user_repo import UserRepository from app.utils.security import hash_password, verify_password class UserService: def __init__(self, repo: UserRepository): self.repo = repo def create_user(self, user_data: UserCreate) -> UserResponse: # Check email uniqueness if self.repo.get_by_email(user_data.email): raise EmailAlreadyExistsError(user_data.email) # Create user hashed_pw = hash_password(user_data.password) user = self.repo.create( email=user_data.email, name=user_data.name, hashed_password=hashed_pw, ) return UserResponse.model_validate(user) - Order service with inventory validation:
# app/services/order_service.py class OrderService: def create_order(self, user_id: int, cart: Cart) -> Order: # Validate inventory for all items for item in cart.items: product = self.product_repo.get(item.sku) if product.inventory < item.quantity: raise InsufficientInventoryError(item.sku, item.quantity) # Calculate totals subtotal = sum(item.subtotal for item in cart.items) tax = self.calculate_tax(subtotal, cart.shipping_address) total = subtotal + tax # Create order order = Order( user_id=user_id, items=cart.items, subtotal=subtotal, tax=tax, total=total, status="pending" ) return self.order_repo.create(order) - Payment service with webhook handling (discriminated unions)
Checkpoint: Services enforce business rules.
Phase 5: API Layer (Day 8-9)
Goal: Implement all API endpoints with validation.
- Create error handlers:
# app/api/error_handlers.py from fastapi import FastAPI, Request from fastapi.exceptions import RequestValidationError from pydantic import ValidationError from app.models.errors import ErrorResponse def register_handlers(app: FastAPI): @app.exception_handler(RequestValidationError) async def validation_handler(request: Request, exc: RequestValidationError): errors = [] for error in exc.errors(): errors.append({ "field": ".".join(str(loc) for loc in error["loc"]), "message": error["msg"], "type": error["type"], }) return JSONResponse( status_code=422, content=ErrorResponse( success=False, code="VALIDATION_ERROR", message="Request validation failed", errors=errors, ).model_dump() ) -
Implement authentication middleware
-
Create all route handlers with proper models
- Add request ID tracking:
@app.middleware("http") async def add_request_id(request: Request, call_next): request_id = request.headers.get("X-Request-ID", str(uuid4())) response = await call_next(request) response.headers["X-Request-ID"] = request_id return response
Checkpoint: All endpoints work with proper validation.
Phase 6: Payment Webhooks with Discriminated Unions (Day 10)
Goal: Handle polymorphic webhook events.
- Define webhook event models:
# app/models/webhook.py from pydantic import BaseModel, Field from typing import Literal, Union, Annotated class PaymentSucceededEvent(BaseModel): type: Literal["payment.succeeded"] payment_id: str amount: Money order_id: str class PaymentFailedEvent(BaseModel): type: Literal["payment.failed"] payment_id: str error_code: str error_message: str class RefundCreatedEvent(BaseModel): type: Literal["refund.created"] refund_id: str payment_id: str amount: Money WebhookEvent = Annotated[ Union[PaymentSucceededEvent, PaymentFailedEvent, RefundCreatedEvent], Field(discriminator="type") ] - Implement webhook handler:
@router.post("/webhooks/payment") async def handle_payment_webhook( event: WebhookEvent, x_signature: str = Header(...) ): # Verify webhook signature if not verify_signature(event, x_signature): raise HTTPException(401, "Invalid signature") # Dispatch based on event type match event: case PaymentSucceededEvent(): await process_payment_success(event) case PaymentFailedEvent(): await process_payment_failure(event) case RefundCreatedEvent(): await process_refund(event) return {"status": "processed"}
Checkpoint: Webhook events parse correctly by type.
Phase 7: LLM Integration (Day 11-12)
Goal: Add AI-powered features with structured outputs.
- Define structured output models:
# app/models/ai.py from pydantic import BaseModel, Field class ProductDescription(BaseModel): """Structured product description from LLM""" headline: str = Field(max_length=100) short_description: str = Field(max_length=250) features: list[str] = Field(min_length=3, max_length=5) target_audience: str seo_keywords: list[str] class OrderSummary(BaseModel): """Natural language order summary""" summary: str = Field(max_length=500) highlights: list[str] estimated_delivery: str next_steps: list[str] - Implement AI service:
# app/services/ai_service.py import instructor from openai import OpenAI from app.config.settings import settings class AIService: def __init__(self): self.client = instructor.from_openai( OpenAI(api_key=settings.openai_api_key.get_secret_value()) ) def generate_product_description( self, product_name: str, category: str, features: list[str] ) -> ProductDescription: return self.client.chat.completions.create( model=settings.openai_model, response_model=ProductDescription, messages=[{ "role": "user", "content": f"""Generate a product description for: Name: {product_name} Category: {category} Features: {', '.join(features)} """ }] )
Checkpoint: LLM returns validated structured data.
Phase 8: Testing and Production Readiness (Day 13-14)
Goal: Comprehensive tests and production configuration.
- Write unit tests for types and models:
# tests/unit/test_types.py def test_money_arithmetic(): m1 = Money("10.00", "USD") m2 = Money("5.50", "USD") assert (m1 + m2).amount == Decimal("15.50") def test_money_different_currencies_raises(): m1 = Money("10.00", "USD") m2 = Money("10.00", "EUR") with pytest.raises(ValueError): m1 + m2 - Write integration tests for API:
# tests/integration/test_api_orders.py def test_create_order_validates_inventory(): # Add product with limited inventory product = create_product(inventory=5) # Try to order more than available response = client.post("/orders", json={ "items": [{"sku": product.sku, "quantity": 10}] }) assert response.status_code == 422 assert "inventory" in response.json()["errors"][0]["message"].lower() - Add observability:
- Structured logging with request IDs
- Validation error metrics
- Performance tracing
- Create Docker configuration:
FROM python:3.11-slim WORKDIR /app COPY pyproject.toml . RUN pip install . COPY app/ app/ CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0"] - Create docker-compose for local development
Checkpoint: Tests pass, application runs in Docker.
Testing Strategy
Unit Tests
# tests/unit/test_models.py
import pytest
from pydantic import ValidationError
from app.models.user import UserCreate, UserUpdate
from app.types.money import Money
from app.types.phone import normalize_phone
class TestUserModels:
def test_user_create_valid(self):
user = UserCreate(
email="test@example.com",
name="Test User",
password="securepassword123"
)
assert user.email == "test@example.com"
def test_user_create_invalid_email(self):
with pytest.raises(ValidationError) as exc:
UserCreate(email="invalid", name="Test", password="password123")
assert "email" in str(exc.value)
def test_user_update_all_optional(self):
update = UserUpdate()
assert update.email is None
assert update.name is None
class TestMoneyType:
def test_parse_string(self):
money = Money("99.99")
assert money.amount == Decimal("99.99")
assert money.currency == "USD"
def test_parse_dict(self):
money = Money._validate({"amount": "50.00", "currency": "EUR"})
assert money.amount == Decimal("50.00")
assert money.currency == "EUR"
def test_quantize_to_cents(self):
money = Money("99.999")
assert money.amount == Decimal("100.00")
class TestPhoneNumber:
@pytest.mark.parametrize("input,expected", [
("(555) 123-4567", "+15551234567"),
("5551234567", "+15551234567"),
("+1-555-123-4567", "+15551234567"),
("1-555-123-4567", "+15551234567"),
])
def test_normalize_formats(self, input, expected):
assert normalize_phone(input) == expected
def test_invalid_phone_raises(self):
with pytest.raises(ValueError):
normalize_phone("123") # Too short
Integration Tests
# tests/integration/test_api_users.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
class TestUserAPI:
def test_register_user_success(self):
response = client.post("/auth/register", json={
"email": "new@example.com",
"name": "New User",
"password": "securepassword123"
})
assert response.status_code == 201
data = response.json()["data"]
assert data["email"] == "new@example.com"
assert "password" not in data
assert "id" in data
def test_register_duplicate_email(self):
# First registration
client.post("/auth/register", json={
"email": "dupe@example.com",
"name": "First",
"password": "password123"
})
# Duplicate
response = client.post("/auth/register", json={
"email": "dupe@example.com",
"name": "Second",
"password": "password123"
})
assert response.status_code == 409
def test_validation_error_format(self):
response = client.post("/auth/register", json={
"email": "invalid",
"name": "",
"password": "short"
})
assert response.status_code == 422
errors = response.json()["errors"]
assert len(errors) >= 3
# Check error structure
email_error = next(e for e in errors if "email" in e["field"])
assert "message" in email_error
assert "type" in email_error
# tests/integration/test_webhooks.py
class TestPaymentWebhooks:
def test_payment_succeeded_event(self):
response = client.post("/webhooks/payment", json={
"type": "payment.succeeded",
"payment_id": "pay_123",
"amount": {"amount": "99.99", "currency": "USD"},
"order_id": "ORD-ABC1234567"
}, headers={"X-Signature": "valid_sig"})
assert response.status_code == 200
def test_payment_failed_event(self):
response = client.post("/webhooks/payment", json={
"type": "payment.failed",
"payment_id": "pay_456",
"error_code": "card_declined",
"error_message": "Card was declined"
}, headers={"X-Signature": "valid_sig"})
assert response.status_code == 200
def test_unknown_event_type(self):
response = client.post("/webhooks/payment", json={
"type": "unknown.event",
"data": {}
}, headers={"X-Signature": "valid_sig"})
assert response.status_code == 422 # Validation error
End-to-End Tests
# tests/e2e/test_checkout_flow.py
import pytest
from tests.fixtures import create_test_user, create_test_product
class TestCheckoutFlow:
@pytest.fixture(autouse=True)
def setup(self, db_session):
self.user = create_test_user(db_session)
self.product = create_test_product(db_session, inventory=10)
def test_complete_checkout(self, client):
# 1. Login
login_resp = client.post("/auth/login", json={
"email": self.user.email,
"password": "testpassword"
})
token = login_resp.json()["data"]["access_token"]
headers = {"Authorization": f"Bearer {token}"}
# 2. Add to cart
cart_resp = client.post("/cart/items", json={
"sku": self.product.sku,
"quantity": 2
}, headers=headers)
assert cart_resp.status_code == 200
# 3. Get cart
cart = client.get("/cart", headers=headers).json()["data"]
assert len(cart["items"]) == 1
# 4. Create order
order_resp = client.post("/orders", json={
"shipping_address": {
"street": "123 Main St",
"city": "New York",
"state": "NY",
"postal_code": "10001",
"country": "US"
}
}, headers=headers)
assert order_resp.status_code == 201
order = order_resp.json()["data"]
# 5. Process payment
payment_resp = client.post("/payments", json={
"order_id": order["id"],
"method": {"type": "card", "token": "tok_visa"}
}, headers=headers)
assert payment_resp.status_code == 200
# 6. Verify order status
order = client.get(f"/orders/{order['id']}", headers=headers).json()["data"]
assert order["status"] == "confirmed"
# 7. Verify inventory decreased
product = client.get(f"/products/{self.product.sku}").json()["data"]
assert product["inventory"] == 8
Common Pitfalls and Debugging
Pitfall 1: Circular Import with Model Dependencies
Problem: User model references Order, Order references User.
Symptom:
ImportError: cannot import name 'UserResponse' from partially initialized module
Solution:
# Use TYPE_CHECKING for type hints only
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from app.models.order import OrderResponse
class UserResponse(BaseModel):
orders: list["OrderResponse"] = [] # Forward reference as string
# Rebuild models after all imports
UserResponse.model_rebuild()
Pitfall 2: model_construct Bypasses Validation
Problem: Using model_construct with untrusted data.
Symptom: Invalid data gets into your system without errors.
Solution:
# NEVER use model_construct with external data
user = User.model_construct(**request.json()) # BAD!
# ONLY use with internal/trusted data
user = User.model_construct(**db_row) # OK - already validated on insert
# When in doubt, use model_validate
user = User.model_validate(db_row) # Safe
Pitfall 3: Discriminated Union Order Matters
Problem: Most specific types must come first in union.
Symptom: Always matches the first type in union.
Solution:
# BAD: Generic type first catches everything
PaymentMethod = Union[
GenericPayment, # Catches all!
CardPayment,
BankPayment,
]
# GOOD: Specific types first
PaymentMethod = Annotated[
Union[
CardPayment,
BankPayment,
GenericPayment, # Fallback last
],
Field(discriminator="type")
]
Pitfall 4: Settings Not Loading in Tests
Problem: Tests use different environment than expected.
Symptom: Tests fail with missing environment variables.
Solution:
# conftest.py
import pytest
from app.config.settings import Settings
@pytest.fixture(autouse=True)
def test_settings(monkeypatch):
"""Override settings for tests"""
monkeypatch.setenv("APP_DATABASE_URL", "postgresql://test:test@localhost/test")
monkeypatch.setenv("APP_SECRET_KEY", "test-secret-key")
monkeypatch.setenv("APP_OPENAI_API_KEY", "test-key")
# Force settings reload
from app.config import settings
settings.__init__()
Pitfall 5: SQLModel and Pydantic Model Confusion
Problem: Using SQLModel table classes as API response models.
Symptom: Relationships not loaded, or sensitive fields exposed.
Solution:
# SEPARATE your API models from DB models
# db/models.py
class UserDB(SQLModel, table=True):
id: int | None = Field(primary_key=True)
email: str
hashed_password: str # Sensitive!
orders: list["OrderDB"] = Relationship(...)
# models/user.py
class UserResponse(BaseModel):
id: int
email: str
# NO hashed_password!
# Relationships converted to simple types
model_config = ConfigDict(from_attributes=True)
Pitfall 6: Money Type Precision Loss
Problem: Floating point used for money calculations.
Symptom: 0.1 + 0.2 != 0.3 issues in totals.
Solution:
# ALWAYS use Decimal for money
from decimal import Decimal, ROUND_HALF_UP
class Money:
def __init__(self, amount: str | float | Decimal):
# Convert to string first to avoid float precision issues
self.amount = Decimal(str(amount)).quantize(
Decimal("0.01"),
rounding=ROUND_HALF_UP
)
# In models
total: Money # Not float!
Extensions and Challenges
Extension 1: Multi-Currency Support
Add currency conversion to the Money type:
class Money:
async def convert_to(self, target_currency: str) -> "Money":
rate = await get_exchange_rate(self.currency, target_currency)
return Money(self.amount * rate, target_currency)
class Order(BaseModel):
currency: str = "USD"
items: list[OrderItem]
@computed_field
def total_in_currency(self) -> Money:
total = sum(item.subtotal for item in self.items)
if item.currency != self.currency:
return total.convert_to(self.currency)
return total
Extension 2: Schema Versioning
Support multiple API versions with model evolution:
# models/v1/user.py
class UserResponseV1(BaseModel):
id: int
name: str
# models/v2/user.py
class UserResponseV2(BaseModel):
id: int
full_name: str # Renamed
email_verified: bool # New field
# Adapter
def v1_to_v2(v1: UserResponseV1) -> UserResponseV2:
return UserResponseV2(
id=v1.id,
full_name=v1.name,
email_verified=False # Default for migrated users
)
Extension 3: Async Validation
Implement validators that require async operations:
from pydantic import model_validator
class OrderCreate(BaseModel):
items: list[OrderItem]
@model_validator(mode='after')
async def validate_inventory_async(self):
for item in self.items:
available = await check_inventory_async(item.sku)
if available < item.quantity:
raise ValueError(f"Insufficient inventory for {item.sku}")
return self
Extension 4: Feature Flags with Pydantic
Type-safe feature flags:
class FeatureFlags(BaseSettings):
enable_ai_descriptions: bool = False
enable_multi_currency: bool = False
max_cart_items: int = Field(50, ge=1)
model_config = SettingsConfigDict(env_prefix="FEATURE_")
# Usage
if settings.feature_flags.enable_ai_descriptions:
description = ai_service.generate(product)
Extension 5: GraphQL Integration
Use Pydantic models with Strawberry GraphQL:
import strawberry
from app.models.user import UserResponse
@strawberry.experimental.pydantic.type(model=UserResponse)
class UserType:
pass
@strawberry.type
class Query:
@strawberry.field
async def user(self, id: int) -> UserType:
user = await user_service.get(id)
return UserType.from_pydantic(user)
Real-World Connections
Industry Patterns This Project Demonstrates
- Clean Architecture - Separation of concerns across layers
- Domain-Driven Design - Rich domain models with behavior
- API-First Development - OpenAPI-driven contracts
- Event-Driven Architecture - Webhook event handling
- Repository Pattern - Decoupled data access
Companies Using These Patterns
- Stripe - Webhook events with discriminated unions
- Shopify - Order management with complex validation
- Netflix - Pydantic for configuration management
- Anthropic/OpenAI - Structured outputs from LLMs
Skills Developed
| Skill | Application |
|---|---|
| API Design | Building RESTful APIs with proper validation |
| Type Systems | Advanced Python typing for safety |
| Data Modeling | Separating concerns across layers |
| Error Handling | User-friendly, consistent errors |
| Testing | Pyramid strategy with proper coverage |
| DevOps | Containerization and configuration |
Self-Assessment Checklist
Architecture Understanding
- Can I explain why we separate API, Service, and Repository layers?
- Can I describe the flow of data validation through all layers?
- Can I explain when to use each model type (Input, Domain, Output)?
- Can I describe the benefits of custom domain types?
Implementation Skills
- Can I create custom Pydantic types with
__get_pydantic_core_schema__? - Can I implement discriminated unions for polymorphic data?
- Can I integrate Pydantic Settings for configuration?
- Can I use SQLModel for database models?
- Can I implement structured LLM outputs with Instructor?
Error Handling
- Can I create custom exception handlers for FastAPI?
- Can I aggregate and format errors across validation layers?
- Can I provide helpful error messages for API consumers?
Performance
- Do I know when to use
model_constructvsmodel_validate? - Can I implement caching for validated objects?
- Do I understand
model_validate_jsonperformance benefits?
Testing
- Can I write unit tests for custom types and validators?
- Can I write integration tests for API endpoints?
- Can I write E2E tests for complete user flows?
- Do I have >80% code coverage?
Production Readiness
- Does my application run in Docker?
- Are secrets properly managed with
SecretStr? - Do I have proper logging and observability?
- Are database migrations working?
Resources
Documentation
- Pydantic V2 Documentation
- FastAPI Documentation
- SQLModel Documentation
- Instructor (LLM Structured Outputs)
- Pydantic Settings
Books
- “Architecture Patterns with Python” by Harry Percival & Bob Gregory
- “Fluent Python” by Luciano Ramalho
- “Robust Python” by Patrick Viafore
- “Building Data Science Applications with FastAPI” by Francois Voron
Related Projects
Video Resources
This capstone project integrates everything learned in the Pydantic Data Validation Deep Dive. Completing it demonstrates mastery of data validation architecture in Python.