Pydantic in Action: Integrating with FastAPI and SQLAlchemy

In the previous post, we mastered custom validators, field aliases, and model configuration to tailor Pydantic’s behavior for complex data. Now, let’s put Pydantic to work in real-world applications by integrating it with FastAPI and SQLAlchemy. Pydantic’s type safety and validation make it a natural fit for FastAPI, a high-performance web framework, and it bridges the gap between API payloads and database models with SQLAlchemy. However, syncing API models with database schemas can be tricky. This post explores how to use Pydantic for API request/response handling, integrate it with SQLAlchemy for database operations, and manage data flows effectively.

We’ll build a simple blog API to demonstrate these concepts, covering request validation, response shaping, and ORM integration. Let’s get started!

Using Pydantic Models for FastAPI Request and Response

FastAPI leverages Pydantic to define and validate request and response models, automatically generating OpenAPI documentation and handling serialization. Let’s define models for a blog post and use them in a FastAPI route.

from fastapi import FastAPI
from pydantic import BaseModel
from datetime import datetime

app = FastAPI()

class BlogCreate(BaseModel):
    title: str
    content: str

class BlogResponse(BaseModel):
    id: int
    title: str
    content: str
    created_at: datetime

@app.post("/blogs/", response_model=BlogResponse)
async def create_blog(blog: BlogCreate):
    # Simulate saving to DB and returning a response
    return {"id": 1, "title": blog.title, "content": blog.content, "created_at": datetime.now()}

Here, BlogCreate validates the incoming request body, ensuring title and content are strings. FastAPI uses BlogResponse to shape the response, serializing the output to JSON. If the request data is invalid (e.g., missing title), FastAPI returns a 422 error with detailed validation messages.

Request Body vs Query Parameters

FastAPI, with Pydantic, supports various input types: request bodies, query parameters, and path parameters. Pydantic ensures these inputs are validated against type hints.

Here’s an example with a query parameter for filtering blogs and a path parameter for retrieving a specific blog:

from fastapi import Query, Path

@app.get("/blogs/", response_model=list[BlogResponse])
async def get_blogs(category: str | None = Query(default=None)):
    # Simulate fetching blogs by category
    return [
        {"id": 1, "title": "First Post", "content": "Content", "created_at": datetime.now()}
    ]

@app.get("/blogs/{blog_id}", response_model=BlogResponse)
async def get_blog(blog_id: int = Path(ge=1)):
    # Simulate fetching a blog by ID
    return {"id": blog_id, "title": "Sample Post", "content": "Content", "created_at": datetime.now()}

The Query and Path helpers allow you to specify constraints (e.g., ge=1 ensures blog_id is positive). Pydantic validates these inputs seamlessly, and defaults (like category: str | None) handle optional parameters.

Controlling API Output with Response Models

FastAPI’s response_model parameter lets you control what fields are returned, using Pydantic’s serialization features to include or exclude fields. This is critical for hiding sensitive data or reducing payload size.

class User(BaseModel):
    username: str
    email: str
    password: str  # Sensitive field

class UserResponse(BaseModel):
    username: str
    email: str

@app.post("/users/", response_model=UserResponse, response_model_exclude_unset=True)
async def create_user(user: User):
    # Simulate saving user, exclude password from response
    return user

Here, UserResponse omits the password field, and response_model_exclude_unset=True ensures only explicitly set fields are included. You can also use response_model_include={"field1", "field2"} to select specific fields dynamically.

Integrating with SQLAlchemy

SQLAlchemy defines database models, while Pydantic handles API models. To bridge them, Pydantic supports ORM mode (via orm_mode=True in V1 or from_attributes=True in V2) to map database objects to Pydantic models.

Here’s an example with a SQLAlchemy Blog model and a corresponding Pydantic model:

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from pydantic import BaseModel, ConfigDict
from datetime import datetime

Base = declarative_base()

class BlogDB(Base):
    __tablename__ = "blogs"
    id = Column(Integer, primary_key=True)
    title = Column(String)
    content = Column(String)
    created_at = Column(DateTime, default=datetime.now)

class BlogResponse(BaseModel):
    id: int
    title: str
    content: str
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)

With from_attributes=True, you can convert a SQLAlchemy object to a Pydantic model:

# Simulate fetching a blog from the database
db_blog = BlogDB(id=1, title="ORM Post", content="Content", created_at=datetime.now())
pydantic_blog = BlogResponse.from_orm(db_blog)  # V1: orm_mode=True
print(pydantic_blog.dict())  # {'id': 1, 'title': 'ORM Post', ...}

This ensures type-safe API responses while keeping database logic separate.

Handling Data Flow: Create, Read, Update Models

To maintain clarity, define separate Pydantic models for create, read, and update operations. This avoids exposing auto-generated fields (like id or created_at) in input models.

class BlogBase(BaseModel):
    title: str
    content: str

class BlogCreate(BlogBase):
    pass  # No additional fields needed

class BlogUpdate(BlogBase):
    title: str | None = None  # Allow partial updates
    content: str | None = None

class BlogResponse(BlogBase):
    id: int
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)

@app.post("/blogs/", response_model=BlogResponse)
async def create_blog(blog: BlogCreate):
    # Simulate saving to DB
    return {"id": 1, "title": blog.title, "content": blog.content, "created_at": datetime.now()}

@app.patch("/blogs/{blog_id}", response_model=BlogResponse)
async def update_blog(blog_id: int, blog: BlogUpdate):
    # Simulate updating DB
    return {"id": blog_id, "title": blog.title or "Unchanged", "content": blog.content or "Unchanged", "created_at": datetime.now()}

Using a shared BlogBase class ensures consistency, while BlogUpdate allows partial updates by making fields optional.

Error Handling and Validation in APIs

Pydantic’s validation errors are automatically converted to FastAPI’s HTTP 422 responses with detailed messages. You can customize error handling using HTTPException:

from fastapi import HTTPException

@app.post("/blogs/custom/")
async def create_blog(blog: BlogCreate):
    if len(blog.title) < 5:
        raise HTTPException(status_code=400, detail="Title must be at least 5 characters")
    return {"id": 1, "title": blog.title, "content": blog.content, "created_at": datetime.now()}

FastAPI formats Pydantic errors consistently, but custom exceptions let you enforce business rules with specific status codes and messages.

Recap and Takeaways

Pydantic’s integration with FastAPI and SQLAlchemy streamlines web development:

• FastAPI uses Pydantic for request validation and response serialization, with automatic OpenAPI docs.
• Separate Pydantic models for create, read, and update operations keep APIs clean.
• SQLAlchemy integration via from_attributes=True bridges database and API layers.
• Custom error handling enhances user-facing APIs.

These patterns ensure type safety, maintainability, and scalability in production systems.