Mapping Pydantic models to SQLAlchemy models, other ORM models, and Alembic for migrations
Daniel Herman Varzari
CTO | Full Stack Developer | Architect | Top Expert in Fintech, Blockchain, Digital Assets and Trading Technology
If you want to map a Pydantic model to another model or data structure, you can do so by creating methods to convert between them. This is useful when you need to map Pydantic models to SQLAlchemy models, other ORM models, or even dictionaries with different structures.
Here is an example of how to map a Pydantic model to an SQLAlchemy model and vice versa:
Pydantic Model
from pydantic import BaseModel, Field, EmailStr
class UserCreate(BaseModel):
name: str = Field(..., title="Name", description="The name of the user", example="Alice")
email: EmailStr = Field(..., title="Email", description="The email address of the user", example="[email protected]")
SQLAlchemy Model
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
tablename = 'users'
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
email = Column(String, unique=True, index=True)
Mapping Functions
Now, you can create functions to map between these models:
Pydantic to SQLAlchemy
from sqlalchemy.orm import Session
def create_user(db: Session, user_create: UserCreate) -> User:
db_user = User(name=user_create.name, email=user_create.email)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
SQLAlchemy to Pydantic
from typing import List
def get_user(db: Session, user_id: int) -> UserCreate:
db_user = db.query(User).filter(User.id == user_id).first()
if db_user:
return UserCreate(name=db_user.name, email=db_user.email)
return None
Example Usage
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Create the tables
Base.metadata.create_all(bind=engine)
# Example usage
db = SessionLocal()
# Create a new user
user_create = UserCreate(name="Alice", email="[email protected]")
db_user = create_user(db, user_create)
print(db_user.id, db_user.name, db_user.email)
# Retrieve the user
user = get_user(db, db_user.id)
print(user)
Mapping with More Complex Models
For more complex models, you may need to handle nested structures and relationships. Here's an example with a nested Pydantic model and its corresponding SQLAlchemy model:
Pydantic Model
from pydantic import BaseModel, Field, EmailStr
class UserCreate(BaseModel):
name: str = Field(..., title="Name", description="The name of the user", example="Alice")
email: EmailStr = Field(..., title="Email", description="The email address of the user", example="[email protected]")
class UserResponse(BaseModel):
id: int = Field(..., title="ID", description="The unique identifier of the user")
name: str = Field(..., title="Name", description="The name of the user")
email: EmailStr = Field(..., title="Email", description="The email address of the user")
SQLAlchemy Model
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
tablename = 'users'
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
email = Column(String, unique=True, index=True)
Mapping Functions
Pydantic to SQLAlchemy
from sqlalchemy.orm import Session
def create_user(db: Session, user_create: UserCreate) -> User:
db_user = User(name=user_create.name, email=user_create.email)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
SQLAlchemy to Pydantic
python
Copy code
def get_user(db: Session, user_id: int) -> UserResponse:
db_user = db.query(User).filter(User.id == user_id).first()
if db_user:
return UserResponse(id=db_user.id, name=db_user.name, email=db_user.email)
return None
Example Usage
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Create the tables
Base.metadata.create_all(bind=engine)
# Example usage
db = SessionLocal()
# Create a new user
user_create = UserCreate(name="Alice", email="[email protected]")
db_user = create_user(db, user_create)
print(db_user.id, db_user.name, db_user.email)
# Retrieve the user
user = get_user(db, db_user.id)
print(user)
Mapping with Nested Structures
For completeness, let's expand with a nested model example:
Pydantic Model with Nested Structure
from pydantic import BaseModel, Field, EmailStr
class Address(BaseModel):
street: str = Field(..., title="Street", description="The street address")
city: str = Field(..., title="City", description="The city of the address")
country: str = Field(..., title="Country", description="The country of the address")
class UserCreate(BaseModel):
name: str = Field(..., title="Name", description="The name of the user")
email: EmailStr = Field(..., title="Email", description="The email address of the user")
address: Address = Field(..., title="Address", description="The address of the user")
class UserResponse(BaseModel):
id: int = Field(..., title="ID", description="The unique identifier of the user")
name: str = Field(..., title="Name", description="The name of the user")
email: EmailStr = Field(..., title="Email", description="The email address of the user")
address: Address = Field(..., title="Address", description="The address of the user")
SQLAlchemy Model with Nested Structure
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Address(Base):
tablename = 'addresses'
id = Column(Integer, primary_key=True, index=True)
street = Column(String, index=True)
city = Column(String, index=True)
country = Column(String, index=True)
class User(Base):
tablename = 'users'
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
email = Column(String, unique=True, index=True)
address_id = Column(Integer, ForeignKey('addresses.id'))
address = relationship("Address")
Mapping Functions
Pydantic to SQLAlchemy
def create_user(db: Session, user_create: UserCreate) -> User:
db_address = Address(street=user_create.address.street, city=user_create.address.city, country=user_create.address.country)
db.add(db_address)
db.commit()
db.refresh(db_address)
db_user = User(name=user_create.name, email=user_create.email, address_id=db_address.id)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
SQLAlchemy to Pydantic
def get_user(db: Session, user_id: int) -> UserResponse:
db_user = db.query(User).filter(User.id == user_id).first()
if db_user:
address = Address(street=db_user.address.street, city=db_user.address.city, country=db_user.address.country)
return UserResponse(id=db_user.id, name=db_user.name, email=db_user.email, address=address)
return None
Example Usage
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Create the tables
Base.metadata.create_all(bind=engine)
# Example usage
db = SessionLocal()
# Create a new user with address
user_create = UserCreate(name="Alice", email="[email protected]", address=Address(street="123 Main St", city="Anytown", country="USA"))
db_user = create_user(db, user_create)
print(db_user.id, db_user.name, db_user.email)
# Retrieve the user with address
user = get_user(db, db_user.id)
print(user)
In this example, we have added the id field to both the Pydantic and SQLAlchemy models and demonstrated how to handle nested structures, such as an address. This allows you to create and retrieve complex nested data structures while maintaining the mapping between Pydantic and SQLAlchemy models.
Base.metadata.create_all(bind=engine)
The Base.metadata.create_all(bind=engine) statement is used to create all tables defined by your SQLAlchemy models in the database. This is typically done during the setup or initialization phase of your application.
Here's the complete example code with the Base.metadata.create_all(bind=engine) statement included, demonstrating how to define Pydantic and SQLAlchemy models, map between them, and create the database tables:
Complete Example
Pydantic Models
from pydantic import BaseModel, Field, EmailStr
class Address(BaseModel):
street: str = Field(..., title="Street", description="The street address")
city: str = Field(..., title="City", description="The city of the address")
country: str = Field(..., title="Country", description="The country of the address")
class UserCreate(BaseModel):
name: str = Field(..., title="Name", description="The name of the user")
email: EmailStr = Field(..., title="Email", description="The email address of the user")
address: Address = Field(..., title="Address", description="The address of the user")
class UserResponse(BaseModel):
id: int = Field(..., title="ID", description="The unique identifier of the user")
name: str = Field(..., title="Name", description="The name of the user")
email: EmailStr = Field(..., title="Email", description="The email address of the user")
address: Address = Field(..., title="Address", description="The address of the user")
SQLAlchemy Models
from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Address(Base):
tablename = 'addresses'
id = Column(Integer, primary_key=True, index=True)
street = Column(String, index=True)
city = Column(String, index=True)
country = Column(String, index=True)
class User(Base):
tablename = 'users'
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
email = Column(String, unique=True, index=True)
address_id = Column(Integer, ForeignKey('addresses.id'))
address = relationship("Address")
Database Initialization
python
Copy code
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Create the tables
Base.metadata.create_all(bind=engine)
Mapping Functions
Pydantic to SQLAlchemy
def create_user(db: Session, user_create: UserCreate) -> User:
db_address = Address(street=user_create.address.street, city=user_create.address.city, country=user_create.address.country)
db.add(db_address)
db.commit()
db.refresh(db_address)
db_user = User(name=user_create.name, email=user_create.email, address_id=db_address.id)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
SQLAlchemy to Pydantic
def get_user(db: Session, user_id: int) -> UserResponse:
db_user = db.query(User).filter(User.id == user_id).first()
if db_user:
address = Address(street=db_user.address.street, city=db_user.address.city, country=db_user.address.country)
return UserResponse(id=db_user.id, name=db_user.name, email=db_user.email, address=address)
return None
Example Usage
db = SessionLocal()
# Create a new user with address
user_create = UserCreate(name="Alice", email="[email protected]", address=Address(street="123 Main St", city="Anytown", country="USA"))
db_user = create_user(db, user_create)
print(db_user.id, db_user.name, db_user.email)
# Retrieve the user with address
user = get_user(db, db_user.id)
print(user)
This complete example demonstrates how to define models, initialize the database, and create and retrieve users with addresses using Pydantic and SQLAlchemy. The Base.metadata.create_all(bind=engine) statement ensures that the necessary tables are created in the database before performing any operations.
If the database tables have already been created, running Base.metadata.create_all(bind=engine) again won't cause any harm. SQLAlchemy's create_all method checks if the tables exist before trying to create them, so it won't overwrite or recreate existing tables. However, if you need to modify the schema (e.g., adding new columns, changing data types), you should use a database migration tool like Alembic instead.
Here's how you can ensure the tables are created only if they don't exist and manage the schema changes with Alembic.
Using Alembic for Database Migrations
Install Alembic:
领英推荐
pip install alembic
Initialize Alembic in your project:
alembic init alembic
This will create an alembic directory and an alembic.ini file.
Configure Alembic:
Edit alembic.ini to set the database URL:
sqlalchemy.url = sqlite:///./test.db
Update env.py to import your SQLAlchemy models:
from myapp.models import Base # replace with your actual models import
target_metadata = Base.metadata
Create an initial migration:
alembic revision --autogenerate -m "Initial migration"
This will create a new file in alembic/versions containing the schema of your models.
Apply the migration:
alembic upgrade head
This will apply the migration and create the tables in the database.
Example Usage with Alembic
Here's an example of how you can manage your database schema with Alembic while still using SQLAlchemy and Pydantic for your models:
SQLAlchemy Models
from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Address(Base):
tablename = 'addresses'
id = Column(Integer, primary_key=True, index=True)
street = Column(String, index=True)
city = Column(String, index=True)
country = Column(String, index=True)
class User(Base):
tablename = 'users'
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
email = Column(String, unique=True, index=True)
address_id = Column(Integer, ForeignKey('addresses.id'))
address = relationship("Address")
Pydantic Models
from pydantic import BaseModel, Field, EmailStr
class Address(BaseModel):
street: str = Field(..., title="Street", description="The street address")
city: str = Field(..., title="City", description="The city of the address")
country: str = Field(..., title="Country", description="The country of the address")
class UserCreate(BaseModel):
name: str = Field(..., title="Name", description="The name of the user")
email: EmailStr = Field(..., title="Email", description="The email address of the user")
address: Address = Field(..., title="Address", description="The address of the user")
class UserResponse(BaseModel):
id: int = Field(..., title="ID", description="The unique identifier of the user")
name: str = Field(..., title="Name", description="The name of the user")
email: EmailStr = Field(..., title="Email", description="The email address of the user")
address: Address = Field(..., title="Address", description="The address of the user")
Database Initialization
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Mapping Functions
Pydantic to SQLAlchemy
from sqlalchemy.orm import Session
def create_user(db: Session, user_create: UserCreate) -> User:
db_address = Address(street=user_create.address.street, city=user_create.address.city, country=user_create.address.country)
db.add(db_address)
db.commit()
db.refresh(db_address)
db_user = User(name=user_create.name, email=user_create.email, address_id=db_address.id)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
SQLAlchemy to Pydantic
def get_user(db: Session, user_id: int) -> UserResponse:
db_user = db.query(User).filter(User.id == user_id).first()
if db_user:
address = Address(street=db_user.address.street, city=db_user.address.city, country=db_user.address.country)
return UserResponse(id=db_user.id, name=db_user.name, email=db_user.email, address=address)
return None
Example Usage
db = SessionLocal()
# Create a new user with address
user_create = UserCreate(name="Alice", email="[email protected]", address=Address(street="123 Main St", city="Anytown", country="USA"))
db_user = create_user(db, user_create)
print(db_user.id, db_user.name, db_user.email)
# Retrieve the user with address
user = get_user(db, db_user.id)
print(user)
By using Alembic for migrations, you can manage database schema changes more effectively and avoid issues with re-running Base.metadata.create_all(bind=engine) on an already existing database.
Why alembic?
Here are some reasons why Alembic is beneficial in a project:
Reasons to Use Alembic
Versioning and History:
Alembic allows you to keep a history of all changes made to your database schema. You can apply, revert, or even branch schema changes, providing a clear version control for your database.
Consistency:
Using Alembic ensures that your database schema is consistent across different environments (development, staging, production). All changes are applied in the same order, reducing discrepancies.
Automation:
Alembic can automatically generate migration scripts based on changes detected in your SQLAlchemy models, saving time and reducing errors from manual migration script creation.
Rollback Support:
If a migration causes issues, Alembic allows you to roll back to a previous state, which is crucial for maintaining database integrity.
Integration with SQLAlchemy:
Alembic is specifically designed to work with SQLAlchemy, making it a natural choice for projects using SQLAlchemy for ORM.
How to Use Alembic
Here’s a basic guide on how to get started with Alembic:
1. Installation
pip install alembic
2. Initializing Alembic
alembic init alembic
This creates an alembic directory and configuration files.
3. Configuring Alembic
Edit alembic.ini to set the database URL:
sqlalchemy.url = sqlite:///./test.db # Replace with your actual database URL
Edit alembic/env.py to import your SQLAlchemy models and set target_metadata:
from myapp.models import Base # Replace with your actual models import
target_metadata = Base.metadata
4. Creating an Initial Migration
Generate an initial migration script:
alembic revision --autogenerate -m "Initial migration"
This will create a migration script in alembic/versions.
5. Applying Migrations
Apply the migrations to update the database schema:
alembic upgrade head
Example with SQLAlchemy Models
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
Base = declarative_base()
class Address(Base):
tablename = 'addresses'
id = Column(Integer, primary_key=True, index=True)
street = Column(String, index=True)
city = Column(String, index=True)
country = Column(String, index=True)
class User(Base):
tablename = 'users'
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
email = Column(String, unique=True, index=True)
address_id = Column(Integer, ForeignKey('addresses.id'))
address = relationship("Address")
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Summary
While Base.metadata.create_all(bind=engine) is useful for initial database creation, Alembic provides a more robust solution for managing ongoing database schema changes. It ensures consistency, supports rollback, and integrates well with SQLAlchemy, making it a valuable tool for projects that involve complex schema evolution.