Python SQLAlchemy Guide: ORM, Models, Queries
SQLAlchemy is the most powerful and widely used Object-Relational Mapper (ORM) for Python. It provides two distinct APIs: the Core for low-level SQL expression construction and the ORM for high-level object persistence. Unlike lightweight ORMs that abstract away SQL, SQLAlchemy embraces the relational model while providing Pythonic object access. This comprehensive guide covers SQLAlchemy from model definition through production deployment, including session management, query patterns, relationship loading, Alembic migrations, and advanced optimization techniques.
SQLAlchemy Architecture Overview
SQLAlchemy’s architecture has four layers. At the bottom, the Engine encapsulates database connection pools and dialect-specific behavior. The Connection provides database-agnostic SQL execution. The Core’s SQL Expression Language constructs SQL statements programmatically. At the top, the ORM maps Python classes to database tables and synchronizes object state with the database.
This layered architecture means you can use SQLAlchemy as a query builder without the ORM overhead, or as a full ORM with relationship management and identity maps. Many production applications use both — the ORM for common CRUD operations and the Core for complex reporting queries. The SQLAlchemy documentation provides comprehensive coverage of both APIs.
Installation is straightforward:
pip install sqlalchemy
pip install psycopg2-binary # PostgreSQL driver
# or: pymysql for MySQL, or aiosqlite for async SQLiteSQLAlchemy 2.0, released in early 2023, unified the ORM and Core query interfaces around a single select() pattern and introduced new type-annotation-based mapping. This guide focuses on 2.0-style syntax, which is backward compatible with 1.4 applications. The SQLAlchemy 2.0 migration guide details all changes from 1.4.
Defining Models with SQLAlchemy 2.0
SQLAlchemy 2.0 supports two model declaration styles: the classic declarative_base() style and the newer mapped-annotation style. The mapped-column style uses Python type annotations to define columns:
from sqlalchemy import String, Integer, Float, DateTime, Text, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100), nullable=False)
email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
def __repr__(self):
return f"<User(id={self.id}, name='{self.name}')>"The Mapped type annotation tells SQLAlchemy the Python type of the attribute. The mapped_column() function provides column configuration — primary key, data type, constraints, indexes, and defaults. Common column types include Integer, String(length), Float, Boolean, DateTime, Date, Text, Enum, JSON (PostgreSQL and MySQL), and Uuid (PostgreSQL). The nullable=False argument adds a NOT NULL constraint. The unique=True argument adds a UNIQUE constraint. The index=True argument creates a database index.
Computed columns and server-side defaults:
class Product(Base):
__tablename__ = "products"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(200))
price: Mapped[float] = mapped_column(Float)
tax: Mapped[float] = mapped_column(Float, default=0.08)
total: Mapped[float] = mapped_column(server_default=Func("price * (1 + tax)"))
updated_at: Mapped[datetime] = mapped_column(
DateTime, server_default=Func("now()"), onupdate=Func("now()")
)Relationships
Relationships define how models connect through foreign keys. SQLAlchemy provides one-to-many, many-to-one, one-to-one, and many-to-many relationships. A one-to-many relationship between User and Post:
from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
# One-to-many: one user has many posts
posts: Mapped[list["Post"]] = relationship(back_populates="author")
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200))
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
# Many-to-one: each post has one author
author: Mapped["User"] = relationship(back_populates="posts")The back_populates argument tells both sides of the relationship about each other. Without it, changes to one side are not reflected on the other. The cascade parameter controls how operations propagate:
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
posts: Mapped[list["Post"]] = relationship(
back_populates="author",
cascade="all, delete-orphan"
)With cascade="all, delete-orphan", deleting a User also deletes all owned Posts. The “delete-orphan” option removes Posts removed from the relationship. This cascade behavior is documented in the SQLAlchemy relationship configuration guide.
Many-to-many relationships use an association table:
from sqlalchemy import Table, Column
post_tags = Table(
"post_tags",
Base.metadata,
Column("post_id", ForeignKey("posts.id"), primary_key=True),
Column("tag_id", ForeignKey("tags.id"), primary_key=True),
)
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
tags: Mapped[list["Tag"]] = relationship(secondary=post_tags, back_populates="posts")
class Tag(Base):
__tablename__ = "tags"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50), unique=True)
posts: Mapped[list["Post"]] = relationship(secondary=post_tags, back_populates="tags")Session Management
The Session is the ORM’s unit of work. It tracks object state and flushes changes to the database in transactions.
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# Create engine (connection pool)
engine = create_engine(
"postgresql://user:password@localhost/mydb",
echo=False, # Log SQL statements
pool_size=5, # Connection pool size
max_overflow=10, # Additional connections beyond pool_size
pool_pre_ping=True, # Verify connections before use
)
# Create session factory
SessionLocal = sessionmaker(bind=engine)
# Usage
session = SessionLocal()
try:
new_user = User(name="Alice", email="alice@example.com")
session.add(new_user)
session.commit() # Flush changes and commit transaction
except Exception:
session.rollback() # Rollback on error
raise
finally:
session.close() # Return connection to poolFor web applications, the session lifecycle should match the request. A middleware creates a session at request start, commits or rolls back at request end, and closes the session regardless of outcome. The Flask-SQLAlchemy and FastAPI SQLAlchemy integrations handle this automatically. Session management best practices include: use a single session per request, flush explicitly when you need the generated primary key before commit, roll back on any database error to prevent stale transactions, and avoid holding sessions across async boundaries.
Querying with the Unified select() API
SQLAlchemy 2.0 introduced a unified select() API that works identically with Core and ORM:
from sqlalchemy import select, func, or_, and_
# Get user by ID
stmt = select(User).where(User.id == 1)
user = session.execute(stmt).scalar_one()
# Get all active users
stmt = select(User).where(User.is_active == True)
users = session.execute(stmt).scalars().all()
# With joins
stmt = (
select(Post)
.join(Post.author)
.where(User.name == "Alice")
)
posts = session.execute(stmt).scalars().all()
# Aggregation
stmt = (
select(User.name, func.count(Post.id).label("post_count"))
.join(Post, Post.user_id == User.id)
.group_by(User.id)
)
results = session.execute(stmt).all()
# Subqueries
subq = (
select(Post.user_id, func.count().label("post_count"))
.group_by(Post.user_id)
.subquery()
)
stmt = (
select(User.name, subq.c.post_count)
.join(subq, User.id == subq.c.user_id)
)The unified select() was one of the most significant changes in SQLAlchemy 2.0, as detailed in the What’s New in 2.0 document. It eliminates the old distinction between Query and select() APIs.
Lazy Loading, Eager Loading, and the N+1 Problem
The default lazy loading fetches relationships on first attribute access, potentially causing N+1 query problems:
# N+1 problem: one query for users + one per user for posts
users = session.execute(select(User)).scalars().all()
for user in users:
print(user.posts) # Triggers a query per user
# Solution: eager load with joinedload
from sqlalchemy.orm import joinedload, selectinload
stmt = select(User).options(joinedload(User.posts))
users = session.execute(stmt).scalars().all()joinedload() uses a LEFT OUTER JOIN to fetch related objects in the same query. selectinload() uses a second SELECT with an IN clause — more efficient when the join would produce duplicate rows. The choice between them depends on the data shape: joinedload works well for one-to-one relationships where row duplication is minimal; selectinload is preferred for collections where a joined load could produce large result sets.
Alembic Migrations
Alembic is the database migration tool built for SQLAlchemy. It generates migration scripts by comparing the database schema to the model definitions.
pip install alembic
# Initialize alembic
alembic init alembic
# Configure alembic.ini with database URL
# sqlalchemy.url = postgresql://user:password@localhost/mydb
# In alembic/env.py, set target_metadata:
# target_metadata = Base.metadata
# Auto-generate a migration
alembic revision --autogenerate -m "add user table"
# Apply migrations
alembic upgrade headMigration scripts are Python files in the alembic/versions directory. Each script has upgrade() and downgrade() functions. Review auto-generated migrations carefully — Alembic may miss some changes or misinterpret column renames. The op object provides operations for creating/dropping tables, adding/removing columns, creating indexes, and executing raw SQL. The Alembic documentation covers all available operations.
Typical migration workflow:
# 1. Modify model definitions
# 2. Generate migration
alembic revision --autogenerate -m "add email column"
# 3. Review generated migration
# 4. Apply
alembic upgrade head
# 5. Commit migration script to version controlAdvanced Patterns
Bulk operations bypass the unit-of-work pattern for performance-critical inserts and updates:
# Bulk insert — single INSERT with multiple VALUES
session.execute(
insert(User),
[
{"name": "Alice", "email": "alice@example.com"},
{"name": "Bob", "email": "bob@example.com"},
]
)
session.commit()
# Bulk update
session.execute(
update(User).where(User.is_active == False).values(is_active=True)
)
session.commit()Asynchronous SQLAlchemy with asyncpg or aiosqlite enables non-blocking database access in async applications:
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
async_engine = create_async_engine(
"postgresql+asyncpg://user:password@localhost/mydb"
)
async with AsyncSession(async_engine) as session:
stmt = select(User).where(User.name == "Alice")
result = await session.execute(stmt)
user = result.scalar_one()The asyncio extension provides the same API with async/await syntax. This is essential for FastAPI applications and asyncio-based services. The SQLAlchemy asyncio documentation covers connection pooling, session management, and common patterns.
FAQ
When should I use SQLAlchemy Core vs ORM?
Use Core for complex reporting queries, bulk operations, and when you need explicit SQL control. Use ORM for standard CRUD operations, when object graph manipulation is valuable, and when relationship management simplifies your code.
How do I handle database connection failures?
Use pool_pre_ping=True on the engine to verify connections before use. Implement retry logic at the application level for transient failures. Set appropriate pool_recycle times to prevent connection expiry from firewalls.
What is the N+1 query problem and how do I fix it?
The N+1 problem occurs when lazy loading triggers a separate query for each related object. Fix it with eager loading: use joinedload() or selectinload() in the initial query options.
Can SQLAlchemy work with multiple database types?
Yes. SQLAlchemy supports PostgreSQL, MySQL, SQLite, Oracle, Microsoft SQL Server, and many others through database dialects. The same ORM code works across databases with minimal changes, though database-specific features require dialect-specific configuration.
How do I set up database migrations for an existing database?
Use alembic revision --autogenerate with the existing database as the current state. Alembic will detect the difference between the existing schema and your models and generate the appropriate migration. Review the generated migration before applying it.
Related: Learn Python concurrency for async SQLAlchemy with asyncio. Study Python error handling for database error management patterns. Explore Python design patterns for repository and unit-of-work pattern implementations.