Learn Python Series (#50) - FastAPI Advanced - Validation & Dependencies

in StemSocial • 21 hours ago

Learn Python Series (#50) - FastAPI Advanced - Validation & Dependencies

python-logo.png

Repository

What will I learn

  • You will learn how to build complex validation rules beyond basic type checking;
  • how dependency injection works in FastAPI and why it's the framework's most powerful feature;
  • how to compose dependencies for reusable authentication, pagination, and database access;
  • advanced Pydantic v2 features: field validators, model validators, computed fields;
  • how to handle file uploads, form data, and custom error responses.

Requirements

  • A working modern computer running macOS, Windows or Ubuntu;
  • An installed Python 3(.11+) distribution, such as (for example) the Anaconda Distribution;
  • The ambition to learn Python programming.

Difficulty

  • Intermediate, advanced

Curriculum (of the Learn Python Series):

GitHub Account

https://github.com/realScipio

Learn Python Series (#50) - FastAPI Advanced - Validation & Dependencies

Last episode we got FastAPI up and running — routes, type validation, Pydantic models, auto-docs. That was the "Hello World" tier. Now let's build something real.

Real APIs need: passwords that meet complexity rules, email addresses that are actually valid, request data that passes business logic checks (not just type checks), shared authentication across dozens of endpoints, database sessions that open and close cleanly, and error responses that follow a consistent format. FastAPI handles all of this through two features: Pydantic validators and dependency injection.

Nota bene: "Dependency injection" sounds enterprise-Java-scary, but in FastAPI it's remarkably elegant. Think of it as "functions that run automatically before your endpoint, providing whatever it needs." Database connection? Dependency. Current authenticated user? Dependency. Rate limiter? Dependency. Pagination parameters? Dependency. Once this mental model clicks, you'll structure every serious API around it ;-)

Custom validators with Pydantic v2

In episode #49, we used Pydantic's built-in constraints (Field(gt=0, max_length=100)). But real-world validation often needs custom logic — password complexity rules, cross-field validation, data normalization. Pydantic v2 provides field_validator and model_validator for this:

from pydantic import BaseModel, Field, field_validator, model_validator
import re

class UserRegistration(BaseModel):
    username: str = Field(..., min_length=3, max_length=30)
    email: str
    password: str = Field(..., min_length=8)
    password_confirm: str
    age: int = Field(..., ge=13)  # Minimum age 13
    
    @field_validator('username')
    @classmethod
    def username_must_be_alphanumeric(cls, v):
        if not re.match(r'^[a-zA-Z][a-zA-Z0-9_]*$', v):
            raise ValueError(
                'Username must start with a letter and contain '
                'only letters, numbers, and underscores'
            )
        return v.lower()  # Normalize to lowercase
    
    @field_validator('email')
    @classmethod
    def email_must_be_valid(cls, v):
        # Basic email validation (use email-validator package for production)
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(pattern, v):
            raise ValueError('Invalid email address format')
        return v.lower()
    
    @field_validator('password')
    @classmethod
    def password_complexity(cls, v):
        if not re.search(r'[A-Z]', v):
            raise ValueError('Password must contain at least one uppercase letter')
        if not re.search(r'[a-z]', v):
            raise ValueError('Password must contain at least one lowercase letter')
        if not re.search(r'[0-9]', v):
            raise ValueError('Password must contain at least one digit')
        return v
    
    @model_validator(mode='after')
    def passwords_must_match(self):
        if self.password != self.password_confirm:
            raise ValueError('Passwords do not match')
        return self

Several things to notice. @field_validator runs on individual fields — it receives the raw value and returns the (possibly transformed) value. You can normalize data here (lowercasing usernames, stripping whitespace). The @classmethod decorator is required in Pydantic v2.

@model_validator(mode='after') runs after all individual field validators have passed. It receives the fully constructed model instance, so you can check relationships between fields — like ensuring password and password_confirm match, or that end_date is after start_date.

If you need to validate before individual fields are processed, use mode='before':

class FlexibleInput(BaseModel):
    values: list[int]
    
    @model_validator(mode='before')
    @classmethod
    def coerce_single_to_list(cls, data):
        """Accept a single int or a list of ints."""
        if isinstance(data, dict) and isinstance(data.get('values'), int):
            data['values'] = [data['values']]
        return data

This before validator sees the raw input data (a dict) before Pydantic tries to parse it. Useful for accepting flexible input formats.

Reusable validators

When multiple models need the same validation logic, extract it into reusable functions:

from pydantic import field_validator
from typing import Any

def validate_not_empty(v: str) -> str:
    """Strip whitespace and reject empty strings."""
    v = v.strip()
    if not v:
        raise ValueError('Field cannot be empty or whitespace-only')
    return v

def validate_positive(v: int | float) -> int | float:
    """Reject non-positive numbers."""
    if v <= 0:
        raise ValueError('Must be a positive number')
    return v

class Product(BaseModel):
    name: str
    description: str
    price: float
    quantity: int
    
    # Apply reusable validators to specific fields
    _validate_name = field_validator('name')(validate_not_empty)
    _validate_description = field_validator('description')(validate_not_empty)
    _validate_price = field_validator('price')(validate_positive)
    _validate_quantity = field_validator('quantity')(validate_positive)

class Order(BaseModel):
    customer_name: str
    total: float
    
    _validate_name = field_validator('customer_name')(validate_not_empty)
    _validate_total = field_validator('total')(validate_positive)

Same validation logic, used across multiple models, zero code duplication. The functional approach (field_validator('field')(func)) might look unusual, but it's just calling the decorator manually instead of using @ syntax.

Computed fields with Pydantic v2

Sometimes you want fields that are derived from other fields — not stored, but calculated:

from pydantic import BaseModel, computed_field

class CartItem(BaseModel):
    name: str
    unit_price: float
    quantity: int
    tax_rate: float = 0.21  # 21% VAT (Dutch standard)
    
    @computed_field
    @property
    def subtotal(self) -> float:
        return round(self.unit_price * self.quantity, 2)
    
    @computed_field
    @property
    def tax_amount(self) -> float:
        return round(self.subtotal * self.tax_rate, 2)
    
    @computed_field
    @property
    def total(self) -> float:
        return round(self.subtotal + self.tax_amount, 2)

item = CartItem(name="Widget", unit_price=9.99, quantity=3)
print(item.model_dump())
# {'name': 'Widget', 'unit_price': 9.99, 'quantity': 3, 
#  'tax_rate': 0.21, 'subtotal': 29.97, 'tax_amount': 6.29, 'total': 36.26}

Computed fields appear in serialization output (JSON responses, model_dump()) but are never accepted as input. They're pure derived values. This is much cleaner than calculating them in your endpoint code.

Advanced parameter validation: Query, Path, Header, Cookie

FastAPI provides specialized validators for each parameter source:

from fastapi import Query, Path, Header, Cookie

@app.get("/search/")
def search(
    q: str = Query(
        ...,
        min_length=2,
        max_length=100,
        description="Search query",
        examples=["python fastapi", "web framework"],
    ),
    page: int = Query(1, ge=1, le=1000, alias="p"),
    per_page: int = Query(20, ge=1, le=100),
    sort_by: str = Query("relevance", pattern="^(relevance|date|rating)$"),
):
    return {
        "query": q,
        "page": page,
        "per_page": per_page,
        "sort_by": sort_by,
    }

The alias="p" is subtle but powerful — clients send ?p=5 but your code reads page. The pattern parameter applies a regex constraint, so sort_by only accepts "relevance", "date", or "rating". And the examples show up in the OpenAPI docs.

For headers and cookies:

@app.get("/trace/")
def trace_request(
    user_agent: str = Header(..., alias="User-Agent"),
    session_id: str | None = Cookie(None),
    x_request_id: str = Header(...),
):
    return {
        "user_agent": user_agent,
        "session_id": session_id,
        "request_id": x_request_id,
    }

Note: HTTP headers use hyphens (X-Request-Id), but Python variables use underscores (x_request_id). FastAPI converts automatically. The alias override is needed for non-standard casing like User-Agent.

Dependency injection: the core pattern

Dependencies in FastAPI are just functions (or callables) that your endpoint declares it needs. FastAPI calls them automatically before the endpoint runs:

from fastapi import Depends

def get_pagination(
    skip: int = Query(0, ge=0, description="Records to skip"),
    limit: int = Query(20, ge=1, le=100, description="Records per page"),
):
    return {"skip": skip, "limit": limit}

@app.get("/items/")
def list_items(pagination: dict = Depends(get_pagination)):
    # pagination = {"skip": 0, "limit": 20} (from query params)
    all_items = list(range(100))
    return {
        "items": all_items[pagination["skip"]:pagination["skip"] + pagination["limit"]],
        "pagination": pagination,
    }

@app.get("/users/")
def list_users(pagination: dict = Depends(get_pagination)):
    # Same pagination logic, reused
    return {"users": [], "pagination": pagination}

FastAPI sees Depends(get_pagination), calls get_pagination with the request's query parameters, and passes the result to list_items as pagination. The dependency's parameters (skip, limit) are automatically extracted from the request — they appear in the OpenAPI docs, get validated, and have defaults. You get reusable, self-documenting request handling.

Class-based dependencies

For dependencies that need configuration or state, use callable classes:

class Paginator:
    def __init__(self, max_limit: int = 100):
        self.max_limit = max_limit
    
    def __call__(
        self,
        skip: int = Query(0, ge=0),
        limit: int = Query(20, ge=1),
    ):
        # Enforce max_limit regardless of what client requests
        effective_limit = min(limit, self.max_limit)
        return {"skip": skip, "limit": effective_limit}

# Different paginators for different endpoints
standard_paginator = Paginator(max_limit=100)
limited_paginator = Paginator(max_limit=10)

@app.get("/items/")
def list_items(pagination: dict = Depends(standard_paginator)):
    return {"pagination": pagination}

@app.get("/expensive-resource/")
def list_expensive(pagination: dict = Depends(limited_paginator)):
    return {"pagination": pagination}  # Max 10 per page

The __call__ method makes the class instance callable. FastAPI treats it like any other dependency function. The instance holds configuration (max_limit), while each request gets its own __call__ invocation.

Dependency chains: composing complex logic

This is where dependency injection becomes truly powerful. Dependencies can depend on other dependencies, and FastAPI resolves the entire graph:

from fastapi import Header, HTTPException, Depends

# Layer 1: Extract token from header
def get_token(authorization: str = Header(...)):
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid auth scheme")
    return authorization[7:]  # Strip "Bearer "

# Layer 2: Decode token into user data
def get_current_user(token: str = Depends(get_token)):
    # In reality: decode JWT, query database
    users_db = {
        "valid-token-abc": {"id": 1, "username": "scipio", "role": "admin"},
        "valid-token-xyz": {"id": 2, "username": "reader", "role": "viewer"},
    }
    user = users_db.get(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid or expired token")
    return user

# Layer 3: Require specific role
def require_role(required_role: str):
    def dependency(user: dict = Depends(get_current_user)):
        if user["role"] != required_role:
            raise HTTPException(
                status_code=403,
                detail=f"Requires {required_role} role, you have {user['role']}"
            )
        return user
    return dependency

# Usage: endpoints declare exactly what access level they need
@app.get("/profile/")
def read_profile(user: dict = Depends(get_current_user)):
    return {"user": user}

@app.delete("/users/{user_id}")
def delete_user(user_id: int, admin: dict = Depends(require_role("admin"))):
    return {"deleted": user_id, "by": admin["username"]}

@app.get("/dashboard/")
def admin_dashboard(admin: dict = Depends(require_role("admin"))):
    return {"stats": {"users": 42, "items": 100}}

The dependency chain flows: require_role("admin") → get_current_user → get_token → extracts from Authorization header. FastAPI resolves this automatically. If any layer fails (invalid token, wrong role), the appropriate HTTP error is returned and the endpoint never runs.

Notice require_role is a factory function — it returns a dependency function parameterized by the required role. This pattern (a function that returns a dependency) is how you create configurable dependencies.

Dependencies with yield: setup and teardown

For resources that need cleanup (database connections, file handles, external service clients), use yield in your dependency:

def get_db():
    """Provide a database session, ensure cleanup."""
    db = SessionLocal()  # Open connection
    try:
        yield db          # Provide to endpoint
    finally:
        db.close()        # Always close, even on error

def get_http_client():
    """Provide an HTTP client for external API calls."""
    import httpx
    client = httpx.Client(timeout=10.0)
    try:
        yield client
    finally:
        client.close()

@app.get("/users/")
def list_users(db = Depends(get_db)):
    return db.query("SELECT * FROM users")

@app.get("/external-data/")
def get_external(client = Depends(get_http_client)):
    response = client.get("https://api.example.com/data")
    return response.json()

The code before yield is the setup (open connection). The code after yield (in finally) is the teardown (close connection). This runs even if the endpoint raises an exception — exactly like context managers (which we covered in episode #44).

In fact, this IS a context manager pattern integrated into FastAPI's dependency system. The beauty is that your endpoint function doesn't need to know about setup/teardown at all — it just receives db and uses it. Separation of concerns.

Global and router-level dependencies

Apply dependencies to all endpoints at once:

# Apply to entire application
app = FastAPI(dependencies=[Depends(verify_api_key)])

# Apply to a specific router
from fastapi import APIRouter

admin_router = APIRouter(
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(require_role("admin"))],
)

@admin_router.get("/stats")
def admin_stats():
    # require_role("admin") already ran — no need to declare it again
    return {"total_users": 42}

@admin_router.post("/broadcast")
def send_broadcast(message: str):
    return {"sent": True, "message": message}

app.include_router(admin_router)

Every endpoint under /admin/ automatically requires admin authentication. The individual endpoint functions don't even mention authentication — it's handled at the router level.

File uploads

FastAPI handles file uploads through UploadFile, which is memory-efficient (large files are spooled to disk, not loaded entirely into memory):

from fastapi import File, UploadFile, HTTPException

ALLOWED_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
MAX_SIZE = 5 * 1024 * 1024  # 5 MB

@app.post("/upload/avatar")
async def upload_avatar(
    file: UploadFile = File(..., description="Profile picture (max 5MB)"),
    user: dict = Depends(get_current_user),
):
    # Validate content type
    if file.content_type not in ALLOWED_TYPES:
        raise HTTPException(
            status_code=415,
            detail=f"Unsupported file type: {file.content_type}. "
                   f"Allowed: {', '.join(ALLOWED_TYPES)}"
        )
    
    # Read file content
    contents = await file.read()
    
    # Validate file size
    if len(contents) > MAX_SIZE:
        raise HTTPException(
            status_code=413,
            detail=f"File too large: {len(contents)} bytes. Max: {MAX_SIZE}"
        )
    
    # In reality: save to disk or cloud storage
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size_bytes": len(contents),
        "user": user["username"],
    }

@app.post("/upload/documents")
async def upload_documents(files: list[UploadFile] = File(...)):
    results = []
    for f in files:
        content = await f.read()
        results.append({
            "filename": f.filename,
            "size": len(content),
        })
    return {"uploaded": len(results), "files": results}

UploadFile provides async methods (read(), write(), seek(), close()) because file I/O can be slow. The await file.read() doesn't block the event loop while reading large uploads.

Form data

For HTML form submissions (content type application/x-www-form-urlencoded instead of application/json):

from fastapi import Form

@app.post("/login/")
def login(
    username: str = Form(..., min_length=3),
    password: str = Form(..., min_length=8),
    remember_me: bool = Form(False),
):
    # In reality: verify credentials against database
    return {
        "username": username,
        "remember_me": remember_me,
        "message": "Login successful",
    }

Important: you cannot mix Form and JSON Body in the same endpoint. A request is either form-encoded or JSON-encoded — not both. This is an HTTP protocol constraint, not a FastAPI limitation.

You can combine Form and File in the same endpoint though (for multipart form submissions):

@app.post("/submit-report/")
async def submit_report(
    title: str = Form(...),
    description: str = Form(...),
    attachment: UploadFile = File(None),  # Optional file
):
    result = {"title": title, "description": description}
    if attachment:
        content = await attachment.read()
        result["attachment_size"] = len(content)
    return result

Custom error responses

By default, FastAPI returns validation errors in Pydantic's format. You can customize this globally:

from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
    request: Request, exc: RequestValidationError
):
    """Custom format for validation errors."""
    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={
            "error": "validation_error",
            "detail": f"{len(errors)} validation error(s)",
            "errors": errors,
        },
    )

Now every validation error across your entire API uses this consistent format. Your API consumers get predictable error responses they can parse programmatically.

Testing with dependency overrides

One of dependency injection's biggest benefits: in tests, you can swap real dependencies for fakes:

from fastapi.testclient import TestClient

# Real dependency: queries actual database
def get_db():
    db = RealDatabase()
    try:
        yield db
    finally:
        db.close()

# Test override: uses in-memory fake
def get_fake_db():
    return FakeDatabase({"users": [{"id": 1, "name": "Test User"}]})

# Override for testing
app.dependency_overrides[get_db] = get_fake_db

client = TestClient(app)

def test_list_users():
    response = client.get("/users/")
    assert response.status_code == 200
    assert len(response.json()["users"]) == 1

# Override auth for testing protected endpoints
def fake_current_user():
    return {"id": 1, "username": "test", "role": "admin"}

app.dependency_overrides[get_current_user] = fake_current_user

def test_admin_endpoint():
    response = client.get("/admin/stats")
    assert response.status_code == 200

# Clean up overrides after testing
app.dependency_overrides.clear()

No mocking libraries needed. No monkey-patching. Just provide a different function with the same return type. This is one of the clearest testing patterns I've ever used — and it works because dependencies are functions with explicit contracts, not hidden global state.

Putting it all together: a mini CRUD API

Let me show how all these pieces compose into a realistic (if small) application:

from fastapi import FastAPI, Depends, HTTPException, Query
from pydantic import BaseModel, Field, field_validator
from datetime import datetime

app = FastAPI(title="Bookshelf API")

# --- Models ---
class BookCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    author: str = Field(..., min_length=1, max_length=100)
    isbn: str = Field(..., pattern=r'^\d{13}$')
    year: int = Field(..., ge=1450, le=datetime.now().year)
    
    @field_validator('title', 'author')
    @classmethod
    def strip_whitespace(cls, v):
        return v.strip()

class BookResponse(BookCreate):
    id: int
    created_at: datetime

# --- Database (in-memory for demo) ---
books_db: dict[int, dict] = {}
next_id = 1

# --- Dependencies ---
def get_book_or_404(book_id: int):
    if book_id not in books_db:
        raise HTTPException(status_code=404, detail=f"Book {book_id} not found")
    return books_db[book_id]

# --- Endpoints ---
@app.post("/books/", response_model=BookResponse, status_code=201)
def create_book(book: BookCreate):
    global next_id
    db_book = {**book.model_dump(), "id": next_id, "created_at": datetime.now()}
    books_db[next_id] = db_book
    next_id += 1
    return db_book

@app.get("/books/", response_model=list[BookResponse])
def list_books(
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    author: str | None = Query(None, min_length=2),
):
    results = list(books_db.values())
    if author:
        results = [b for b in results if author.lower() in b["author"].lower()]
    return results[skip:skip + limit]

@app.get("/books/{book_id}", response_model=BookResponse)
def get_book(book: dict = Depends(get_book_or_404)):
    return book

@app.delete("/books/{book_id}", status_code=204)
def delete_book(book_id: int, book: dict = Depends(get_book_or_404)):
    del books_db[book_id]

Notice: get_book_or_404 is a dependency that both validates the book exists AND returns it. The get_book and delete_book endpoints both use it, so the "not found" logic is written once. The BookCreate model handles all input validation. The BookResponse model controls what gets returned. And all of this generates complete OpenAPI documentation at /docs.

Oké, samengevat

In this episode, we explored advanced FastAPI validation and dependency injection:

  • @field_validator enables custom validation logic per field with data normalization
  • @model_validator validates relationships between fields (password confirmation, date ranges)
  • Reusable validators can be shared across multiple models as plain functions
  • @computed_field adds derived read-only fields to Pydantic models
  • Query, Path, Header, Cookie provide source-specific validation with aliases, patterns, and constraints
  • Dependency injection provides reusable, composable pre-endpoint logic
  • Dependency chains resolve automatically — auth → user → role → endpoint
  • Factory dependencies (require_role("admin")) create parameterized dependencies
  • yield dependencies handle setup/teardown like context managers
  • Global and router-level dependencies apply logic to all endpoints at once
  • UploadFile handles file uploads with async, memory-efficient streaming
  • dependency_overrides enables clean testing without mocking

Dependencies are FastAPI's secret weapon. Once you start thinking in dependencies, you stop writing monolithic endpoint functions with 200 lines of setup logic. Your endpoints become what they should be — thin handlers that declare what they need and focus on business logic. Everything else is composed from reusable parts ;-)

Graag gedaan. Tot de volgende!

@scipio

#stem #stemsocial #steemstem #python #programming
21 hours ago in StemSocial by
$2.33
  • Pending payout amount: $2.33
  • Breakdown: 0.00 HBD, 18.77 HIVE, 18.77 HP
  • Payout in 6 days
443 votes
Reply 1
Sort:  
  • Trending
    • [-]
     6 hours ago  

    Thanks for your contribution to the STEMsocial community. Feel free to join us on discord to get to know the rest of us!

    Please consider delegating to the @stemsocial account (85% of the curation rewards are returned).

    Consider setting @stemsocial as a beneficiary of this post's rewards if you would like to support the community and contribute to its mission of promoting science and education on Hive. 
     

    $0.00
      Reply