FastAPI in Production: Versioning, Pagination, and Error Handling You Won’t Regret

Pictalingo

·

FastAPI in Production: Versioning, Pagination, and Error Handling You Won’t Regret

FastAPI makes it easy to ship an API quickly. The “oops” moment usually happens later: you need to change response shapes, add consistent errors, and paginate lists without breaking clients. This article shows a practical, junior-friendly pattern you can drop into real projects: API versioning with routers, robust pagination, and clean error handling with Problem Details-style responses.

We’ll build a small “tasks” API with:

  • Versioned routes: /api/v1
  • Offset pagination: ?limit=20&offset=40
  • Stable error responses (same shape everywhere)
  • Validation that returns readable messages

Project Setup

Install dependencies:

pip install fastapi uvicorn pydantic

Create main.py and run:

uvicorn main:app --reload

1) Version Your API with Routers (Without Overthinking It)

Versioning gives you freedom to evolve. A simple approach is a versioned prefix and a dedicated router per version. If you ever need breaking changes, add v2 while keeping v1 alive.

from fastapi import FastAPI, APIRouter app = FastAPI(title="Tasks API") api_v1 = APIRouter(prefix="/api/v1", tags=["v1"]) @api_v1.get("/health") def health(): return {"status": "ok"} app.include_router(api_v1)

Tip: Keep “v1” models and endpoints in a v1/ package as the project grows. Versioning is not just URLs—it’s also data contracts.

2) Define Models That Are Stable and Clear

Clients depend on response shapes. Separate input and output models so you can add server-side fields (like id or created_at) safely.

from pydantic import BaseModel, Field from typing import Optional class TaskCreate(BaseModel): title: str = Field(..., min_length=1, max_length=120) description: Optional[str] = Field(None, max_length=500) class TaskOut(BaseModel): id: int title: str description: Optional[str] = None

3) Build a Minimal In-Memory Store (So We Focus on API Patterns)

In production you’d use a database, but we can demonstrate the patterns with a simple list. The API design stays the same when you swap the storage layer.

from typing import List TASKS: List[dict] = [ {"id": 1, "title": "Buy milk", "description": "2% if possible"}, {"id": 2, "title": "Write docs", "description": "Add examples to README"}, {"id": 3, "title": "Ship feature", "description": None}, ] NEXT_ID = 4

4) Pagination That Clients Can Use Reliably

There are many pagination styles. Offset pagination (limit/offset) is easy for most UIs. The key is returning metadata so the client can render “next/previous” controls without guessing.

We’ll return:

  • items: list of tasks
  • total: total number of tasks
  • limit and offset: what the server used
from pydantic import BaseModel from typing import Generic, List, TypeVar T = TypeVar("T") class Page(BaseModel, Generic[T]): items: List[T] total: int limit: int offset: int

Now the endpoint:

from fastapi import Query @api_v1.get("/tasks", response_model=Page[TaskOut]) def list_tasks( limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0), ): total = len(TASKS) slice_ = TASKS[offset : offset + limit] return { "items": slice_, "total": total, "limit": limit, "offset": offset, }

Why validate limit? Setting a hard upper bound (like 100) protects your service from huge responses and accidental abuse.

5) Consistent Error Responses (Problem Details Style)

FastAPI’s default error output is fine, but teams often want a consistent shape across all errors, including custom exceptions. A common format is based on RFC 7807 “Problem Details.” We’ll implement a lightweight version:

from pydantic import BaseModel from typing import Any, Dict, Optional class Problem(BaseModel): type: str = "about:blank" title: str status: int detail: Optional[str] = None instance: Optional[str] = None extra: Optional[Dict[str, Any]] = None

Now define a custom exception that carries a Problem:

class ApiError(Exception): def __init__(self, problem: Problem): self.problem = problem

Add an exception handler so any ApiError becomes a clean JSON response:

from fastapi import Request from fastapi.responses import JSONResponse @app.exception_handler(ApiError) def api_error_handler(request: Request, exc: ApiError): problem = exc.problem # Optionally set instance to request path for debugging payload = problem.model_dump() payload["instance"] = str(request.url.path) return JSONResponse(status_code=problem.status, content=payload)

Use it in an endpoint (for example, when a task doesn’t exist):

@api_v1.get("/tasks/{task_id}", response_model=TaskOut) def get_task(task_id: int): for t in TASKS: if t["id"] == task_id: return t raise ApiError( Problem( title="Task not found", status=404, detail=f"No task with id={task_id}", ) )

This gives clients a predictable error schema they can display or log consistently.

6) Make Validation Errors Readable (Without Hiding Useful Details)

FastAPI returns 422 with a detailed structure for validation errors. That’s great for debugging but sometimes too verbose for clients. You can transform it into something more approachable while preserving field-level info.

Add a handler for validation errors:

from fastapi.exceptions import RequestValidationError from fastapi import status @app.exception_handler(RequestValidationError) def validation_exception_handler(request: Request, exc: RequestValidationError): fields = [] for err in exc.errors(): loc = ".".join(str(p) for p in err.get("loc", [])) fields.append({"field": loc, "message": err.get("msg")}) problem = { "type": "https://example.com/problems/validation-error", "title": "Validation error", "status": status.HTTP_422_UNPROCESSABLE_ENTITY, "detail": "Request failed validation", "instance": str(request.url.path), "extra": {"fields": fields}, } return JSONResponse(status_code=problem["status"], content=problem)

Now when someone posts an empty title, they’ll get a friendly list of which fields failed.

7) Create Endpoints That Use the Same Patterns

Let’s add create and delete with the same style. Create returns the created task, delete returns 204.

from fastapi import Response @api_v1.post("/tasks", response_model=TaskOut, status_code=201) def create_task(payload: TaskCreate): global NEXT_ID task = {"id": NEXT_ID, "title": payload.title, "description": payload.description} NEXT_ID += 1 TASKS.append(task) return task @api_v1.delete("/tasks/{task_id}", status_code=204) def delete_task(task_id: int): for i, t in enumerate(TASKS): if t["id"] == task_id: TASKS.pop(i) return Response(status_code=204) raise ApiError( Problem( title="Task not found", status=404, detail=f"No task with id={task_id}", ) )

8) Quick Manual Tests (Copy/Paste)

List tasks:

curl "http://127.0.0.1:8000/api/v1/tasks?limit=2&offset=0"

Get a task:

curl "http://127.0.0.1:8000/api/v1/tasks/1"

Create a task:

curl -X POST "http://127.0.0.1:8000/api/v1/tasks" \ -H "Content-Type: application/json" \ -d '{"title":"Learn FastAPI patterns","description":"versioning + pagination + errors"}'

Trigger validation error (empty title):

curl -X POST "http://127.0.0.1:8000/api/v1/tasks" \ -H "Content-Type: application/json" \ -d '{"title":"","description":"nope"}'

Delete a task:

curl -X DELETE "http://127.0.0.1:8000/api/v1/tasks/999"

Production Notes (Small Things That Save Big Headaches)

  • Keep contracts stable: Add fields freely, but treat field removals or type changes as breaking changes—use /api/v2 for those.

  • Cap pagination: A max limit is a simple safety measure.

  • Consistent errors: Your frontend and logs will thank you when every error includes title, status, and detail.

  • Use OpenAPI: FastAPI’s docs at /docs become dramatically more useful when response models are accurate and errors are predictable.

Complete Example: main.py

from fastapi import FastAPI, APIRouter, Query, Request, Response, status from fastapi.responses import JSONResponse from fastapi.exceptions import RequestValidationError from pydantic import BaseModel, Field from typing import Any, Dict, Generic, List, Optional, TypeVar app = FastAPI(title="Tasks API") api_v1 = APIRouter(prefix="/api/v1", tags=["v1"]) # ----------------------- # Models # ----------------------- class TaskCreate(BaseModel): title: str = Field(..., min_length=1, max_length=120) description: Optional[str] = Field(None, max_length=500) class TaskOut(BaseModel): id: int title: str description: Optional[str] = None T = TypeVar("T") class Page(BaseModel, Generic[T]): items: List[T] total: int limit: int offset: int class Problem(BaseModel): type: str = "about:blank" title: str status: int detail: Optional[str] = None instance: Optional[str] = None extra: Optional[Dict[str, Any]] = None class ApiError(Exception): def __init__(self, problem: Problem): self.problem = problem # ----------------------- # Data (in-memory) # ----------------------- TASKS: List[dict] = [ {"id": 1, "title": "Buy milk", "description": "2% if possible"}, {"id": 2, "title": "Write docs", "description": "Add examples to README"}, {"id": 3, "title": "Ship feature", "description": None}, ] NEXT_ID = 4 # ----------------------- # Exception handlers # ----------------------- @app.exception_handler(ApiError) def api_error_handler(request: Request, exc: ApiError): payload = exc.problem.model_dump() payload["instance"] = str(request.url.path) return JSONResponse(status_code=exc.problem.status, content=payload) @app.exception_handler(RequestValidationError) def validation_exception_handler(request: Request, exc: RequestValidationError): fields = [] for err in exc.errors(): loc = ".".join(str(p) for p in err.get("loc", [])) fields.append({"field": loc, "message": err.get("msg")}) problem = { "type": "https://example.com/problems/validation-error", "title": "Validation error", "status": status.HTTP_422_UNPROCESSABLE_ENTITY, "detail": "Request failed validation", "instance": str(request.url.path), "extra": {"fields": fields}, } return JSONResponse(status_code=problem["status"], content=problem) # ----------------------- # Routes # ----------------------- @api_v1.get("/health") def health(): return {"status": "ok"} @api_v1.get("/tasks", response_model=Page[TaskOut]) def list_tasks( limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0), ): total = len(TASKS) slice_ = TASKS[offset : offset + limit] return {"items": slice_, "total": total, "limit": limit, "offset": offset} @api_v1.get("/tasks/{task_id}", response_model=TaskOut) def get_task(task_id: int): for t in TASKS: if t["id"] == task_id: return t raise ApiError(Problem(title="Task not found", status=404, detail=f"No task with id={task_id}")) @api_v1.post("/tasks", response_model=TaskOut, status_code=201) def create_task(payload: TaskCreate): global NEXT_ID task = {"id": NEXT_ID, "title": payload.title, "description": payload.description} NEXT_ID += 1 TASKS.append(task) return task @api_v1.delete("/tasks/{task_id}", status_code=204) def delete_task(task_id: int): for i, t in enumerate(TASKS): if t["id"] == task_id: TASKS.pop(i) return Response(status_code=204) raise ApiError(Problem(title="Task not found", status=404, detail=f"No task with id={task_id}")) app.include_router(api_v1)

If you adopt just three things from this article—APIRouter versioning, capped pagination with metadata, and consistent error responses—you’ll avoid a big chunk of the “we shipped fast and now everything is painful” phase.

Leave a Reply

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