How to use Pydantic with SQLAlchemy

This tutorial provides insight on integrating Pydantic for data validation with SQLAlchemy for database operations, enhancing your Python applications with robust data handling capabilities.

Introduction

In modern web development, ensuring data validity and integrity is critical. Pydantic and SQLAlchemy are two powerful Python libraries that help achieve this. Pydantic is employed for data validation by defining the shape of your data using Python classes. SQLAlchemy, on the other hand, is an Object-Relational Mapping (ORM) tool that facilitates database operations without having to write raw SQL queries. Utilizing them in tandem can significantly streamline your data handling processes. In this guide, we’ll explore how to effectively combine Pydantic with SQLAlchemy.

Getting Started

Before delving into integration, ensure that you have both Pydantic and SQLAlchemy installed. If not, you can install them with:

pip install pydantic sqlalchemy

Now, let’s create the basic models to represent our data. We’ll start by defining a SQLAlchemy model:

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

# Setting up the database connection and session
engine = create_engine('sqlite:///./test.db')
Session = sessionmaker(bind=engine)
Base.metadata.create_all(engine)

With our SQLAlchemy model in place, we can now create a Pydantic model to use for validation:

from pydantic import BaseModel

class UserSchema(BaseModel):
    name: str
    email: str

Now that we have both models established, single operations that perform CRUD functions will interact with both Pydantic and SQLAlchemy models. Assuming we’re trying to create a new user, we ensure data validation via Pydantic before manipulating the database with SQLAlchemy:

# some hypothetical route in a web application:
from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.post('/users/')
def create_user(user: UserSchema):
    db_session = Session()
    db_user = User(name=user.name, email=user.email)
    db_session.add(db_user)
    db_session.commit()
    db_session.refresh(db_user)
    return db_user

This route receives data, validates it against UserSchema, and then commits it to the database as a new User instance.

Advanced Usage

As we delve into more complex scenarios, such as using Pydantic with SQLAlchemy for reading data and automatic conversion between models, more advanced techniques such as custom type decorators become useful:

This involves defining custom pydantic ‘converter’ that can be used to translate SQLAlchemy instances into Pydantic schemas:

from typing import Type, TypeVar
from pydantic import BaseModel

t = TypeVar('T', bound=BaseModel)

# Utility function to convert SQLAlchemy objects to Pydantic models.
def to_pydantic(db_object: Base, pydantic_model: Type[T]) -> T:
    return pydantic_model(**db_object.__dict__)

Equipped with this converter, querying and translating data back and forth between your web application and the database becomes more seamless:

# Continuing from the hypothetical web application example:

@app.get('/users/{user_id}')
def read_user(user_id: int):
    db_session = Session()
    db_user = db_sessio.query(User).filter_by(id=user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return to_pydantic(db_user, UserSchema)

With this pattern, you’re able to single-handedly deal with all the steps of your data’s lifecycle – validation, database operations, and serialization for client responses, all in a highly maintainable way.

Integrating with Async SQL

Moving forward, you may wish to use the asynchronous version of SQL (with async and await keywords). SQLAlchemy supports this, and Pydantic can be used async-compatible:

# ...Import declarations and User model definition omitted for brevity...

@app.post('/users/')
async def create_user_async(user: UserSchema):
    async with async_session() as session:
        async with session.begin():
            db_user = User(name=user.name, email=user.email)
            session.add(db_user)
        await session.commit()
    return await to_pydantic(db_user, UserSchema)

It’s important to update your session handling to the asynchronous version using asyncio and the create_async_engine method from SQLAlchemy 1.4 or newer.

Conclusion

In this guide, we’ve learned to integrate Pydantic for data validation with SQLAlchemy for database interactions. Starting with basic CRUD operations, we escalated to creating custom converter functions for translating between Pydantic models and SQLAlchemy records. Adapting these techniques to an async context further prepares your application for scalable, performant, and clean data handling. With this knowledge in hand, you’re ready to build robust and efficient Python web applications using these powerful libraries.