FastAPI Request Validation & Dependency Injection: Build Clean Endpoints with Pydantic v2

Pictalingo

·

FastAPI Request Validation & Dependency Injection: Build Clean Endpoints with Pydantic v2

If you’re building APIs with FastAPI, you’ll quickly run into two recurring questions:

  • How do I validate and normalize incoming data reliably (beyond basic types)?
  • How do I keep endpoints clean while still handling auth, DB sessions, pagination, and shared logic?

This hands-on guide shows a practical pattern for junior/mid developers: use Pydantic v2 models for strong validation/serialization, and FastAPI dependencies to keep your endpoints short and testable.

What we’re building

We’ll build a tiny “Tasks API” with:

  • Request validation (including trimming strings, bounds checks, and enums)
  • Query params normalized into a single “filters” object
  • Dependencies for auth (get_current_user) and pagination (get_pagination)
  • Consistent response models
  • Working code you can run locally

Project setup

Install dependencies:

pip install fastapi uvicorn pydantic

Create main.py with the code below. Then run:

uvicorn main:app --reload

Define models that validate and normalize data

Pydantic v2 lets you enforce rules and normalize inputs at the model boundary. That means your business logic can assume data is already “clean”.

from enum import Enum from typing import Annotated, Optional, List from fastapi import FastAPI, Depends, HTTPException, Query, Header, status from pydantic import BaseModel, Field, field_validator, ConfigDict app = FastAPI(title="Tasks API") class TaskStatus(str, Enum): todo = "todo" doing = "doing" done = "done" class TaskCreate(BaseModel): model_config = ConfigDict(extra="forbid") title: str = Field(..., min_length=3, max_length=80) description: Optional[str] = Field(None, max_length=500) status: TaskStatus = Field(default=TaskStatus.todo) @field_validator("title") @classmethod def normalize_title(cls, v: str) -> str: v = v.strip() # Normalize multiple spaces inside the title v = " ".join(v.split()) return v class TaskOut(BaseModel): id: int title: str description: Optional[str] status: TaskStatus owner_id: int class ErrorOut(BaseModel): detail: str

Notes:

  • extra="forbid" rejects unknown fields (helps catch client bugs early).
  • Field(...) constraints protect your API from invalid data.
  • @field_validator normalizes the title (trim + collapse spaces).

Build dependencies for auth, pagination, and filters

Dependencies let you build reusable “middleware-like” functions that your endpoints can opt into. The endpoint signature stays readable and testable.

1) Auth dependency: get_current_user

In real projects you’d verify JWTs or sessions. Here we’ll accept a simple header X-User-Id and treat it as authenticated.

class User(BaseModel): id: int def get_current_user(x_user_id: Annotated[Optional[str], Header()] = None) -> User: if not x_user_id: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing X-User-Id header", ) try: user_id = int(x_user_id) except ValueError: raise HTTPException(status_code=400, detail="X-User-Id must be an integer") if user_id <= 0: raise HTTPException(status_code=400, detail="X-User-Id must be positive") return User(id=user_id)

2) Pagination dependency: get_pagination

Instead of repeating limit and offset on every endpoint, centralize it.

class Pagination(BaseModel): limit: int offset: int def get_pagination( limit: Annotated[int, Query(20, ge=1, le=100)] = 20, offset: Annotated[int, Query(0, ge=0)] = 0, ) -> Pagination: return Pagination(limit=limit, offset=offset)

3) Query filters dependency: parse query params into an object

This is a big quality-of-life improvement: convert several query params into one validated model.

class TaskFilters(BaseModel): status: Optional[TaskStatus] = None q: Optional[str] = None @field_validator("q") @classmethod def normalize_query(cls, v: Optional[str]) -> Optional[str]: if v is None: return None v = v.strip() return v if v else None # turn empty strings into None def get_task_filters( status: Annotated[Optional[TaskStatus], Query(None)] = None, q: Annotated[Optional[str], Query(None, max_length=80)] = None, ) -> TaskFilters: return TaskFilters(status=status, q=q)

In-memory “database” (for a runnable demo)

To keep things runnable without a real DB, we’ll store tasks in a list. The dependency patterns stay the same when you later swap in SQLAlchemy.

TASKS: List[TaskOut] = [] NEXT_ID = 1

Create endpoint: clean handler, validated input, consistent output

@app.post( "/tasks", response_model=TaskOut, responses={401: {"model": ErrorOut}, 400: {"model": ErrorOut}}, ) def create_task( payload: TaskCreate, user: User = Depends(get_current_user), ): global NEXT_ID task = TaskOut( id=NEXT_ID, title=payload.title, description=payload.description, status=payload.status, owner_id=user.id, ) NEXT_ID += 1 TASKS.append(task) return task

What’s nice here:

  • payload is already normalized (trimmed title, constrained lengths).
  • user is guaranteed to exist (or a 401 was raised earlier).
  • The handler reads like straightforward business logic.

List endpoint: filters + pagination + auth without clutter

@app.get( "/tasks", response_model=list[TaskOut], responses={401: {"model": ErrorOut}}, ) def list_tasks( user: User = Depends(get_current_user), filters: TaskFilters = Depends(get_task_filters), page: Pagination = Depends(get_pagination), ): # Start with tasks owned by the current user items = [t for t in TASKS if t.owner_id == user.id] # Apply filters if filters.status: items = [t for t in items if t.status == filters.status] if filters.q: q_lower = filters.q.lower() items = [t for t in items if q_lower in t.title.lower()] # Apply pagination return items[page.offset : page.offset + page.limit]

Instead of 5+ query params and custom parsing inside the endpoint, we inject:

  • filters as a validated model
  • page as a validated model

Update endpoint: partial updates with a dedicated model

For patch-like updates, create a separate model with optional fields, still validated.

class TaskUpdate(BaseModel): model_config = ConfigDict(extra="forbid") title: Optional[str] = Field(None, min_length=3, max_length=80) description: Optional[str] = Field(None, max_length=500) status: Optional[TaskStatus] = None @field_validator("title") @classmethod def normalize_title(cls, v: Optional[str]) -> Optional[str]: if v is None: return None v = " ".join(v.strip().split()) return v @app.patch( "/tasks/{task_id}", response_model=TaskOut, responses={401: {"model": ErrorOut}, 404: {"model": ErrorOut}}, ) def update_task( task_id: int, payload: TaskUpdate, user: User = Depends(get_current_user), ): # Find task for idx, t in enumerate(TASKS): if t.id == task_id and t.owner_id == user.id: updated = t.model_copy(update=payload.model_dump(exclude_unset=True)) TASKS[idx] = updated return updated raise HTTPException(status_code=404, detail="Task not found")

Key idea: exclude_unset=True ensures we only update fields actually provided by the client, not overwrite with null unintentionally.

Try it out with curl

Create a task (note the header X-User-Id):

curl -X POST http://127.0.0.1:8000/tasks \ -H "Content-Type: application/json" \ -H "X-User-Id: 7" \ -d '{"title":" Buy milk ","description":"2% if possible","status":"todo"}'

List tasks (with pagination and filters):

curl "http://127.0.0.1:8000/tasks?limit=10&offset=0&status=todo&q=milk" \ -H "X-User-Id: 7"

Update a task:

curl -X PATCH http://127.0.0.1:8000/tasks/1 \ -H "Content-Type: application/json" \ -H "X-User-Id: 7" \ -d '{"status":"done","title":"Buy milk and eggs"}'

Practical tips (the stuff you’ll use in real projects)

  • Use dependencies to enforce invariants: Auth, tenant context, “must be admin”, rate limiting—these belong in dependencies so your handlers stay business-focused.

  • Forbid extras on inbound models (extra="forbid"): It prevents silent client mistakes (like sending titel instead of title).

  • Normalize strings at the boundary: Trim and collapse whitespace in validators so you don’t chase formatting bugs in DB queries later.

  • Separate “create” vs “update” models: A POST typically needs required fields; a PATCH should use optional fields and exclude_unset.

  • Response models are your contract: Always set response_model so you don’t accidentally leak internal fields.

Where to go next

This pattern scales well. When you introduce a real database (SQLAlchemy/SQLModel), your endpoints can remain almost unchanged:

  • Replace the in-memory list with repository functions
  • Add a get_db() dependency that yields a session
  • Keep validation, filters, and pagination exactly the same

Once you’re comfortable with this style, you’ll notice your API code becomes easier to test, easier to read, and harder to break—especially as the number of endpoints grows.

Leave a Reply

Your email address will not be published. Required fields are marked *