Sling Academy
Home/SQLAlchemy/How to use Pydantic with SQLAlchemy

How to use Pydantic with SQLAlchemy

Last updated: January 03, 2024

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.

Next Article: Using SQLAlchemy with the ‘with’ statement

Previous Article: How to Implement Pagination in SQLAlchemy

Series: SQLAlchemy Tutorials: From Basic to Advanced

SQLAlchemy

You May Also Like

  • SQLAlchemy: Counting rows for each category (2 approaches)
  • SQLAlchemy: Adding a calculated column to SELECT query
  • SQLAlchemy: Grouping data on multiple columns
  • SQLAlchemy: How to temporarily delete a row
  • SQLAlchemy Composite Indexes: The Complete Guide
  • Full-Text Search in SQLAlchemy: The Ultimate Guide
  • SQLAlchemy: What if you don’t close database connections?
  • SQLAlchemy: How to Remove FOREIGN KEY Constraints (2 Ways)
  • SQLAlchemy: How to Create and Use Temporary Tables
  • SQLAlchemy: Saving Categories and Subcategories in the Same Table
  • SQLAlchemy: How to Automatically Delete Old Records
  • Weighted Random Selection in SQLAlchemy: An In-Depth Guide
  • SQLAlchemy: created_at and updated_at columns
  • How to Use Regular Expressions in SQLAlchemy
  • SQLAlchemy: Ways to Find Results by a Keyword
  • SQLAlchemy: How to Connect to MySQL Database
  • SQLAlchemy: How to Bulk Insert Data into a Table
  • SQLAlchemy: How to Update a Record by ID
  • SQLAlchemy: Singular vs Plural Table Names