sqlalchemy · diff
v1.0.0 to v3.0.0
35 added, 1064 removed. Audit A to A.
---
name: sqlalchemy
- description: "Python SQL toolkit and ORM with expressive query API, relationship mapping, async support, and Alembic migrations"
+ description: "Use when working with SQLAlchemy 2.0 in Python - declarative models and relationships, queries and sessions, async engines with FastAPI, Alembic migrations, bulk operations, or PostgreSQL query optimization"
metadata:
author: mte90
- version: "1.0.0"
+ version: "3.0.0"
tags:
- python
- orm
- database
- sql
- alembic
- async
---
# SQLAlchemy
Complete reference for Python SQL toolkit and ORM.
## Overview
SQLAlchemy provides a full suite of well-known enterprise-level persistence patterns, designed for efficient and high-performing database access.
**Key Features:**
- **Core**: SQL expression language for building queries
- **ORM**: Object-relational mapping with declarative models
- **Async**: Full async/await support (SQLAlchemy 2.0+)
- **Migrations**: Alembic integration for schema changes
- **Multiple DBs**: PostgreSQL, MySQL, SQLite, Oracle, SQL Server
### Installation
```bash
pip install sqlalchemy
# With async support
pip install sqlalchemy[asyncio]
# With PostgreSQL async driver
pip install sqlalchemy[asyncio] asyncpg
# With Alembic migrations
pip install alembic
# With specific database drivers
pip install psycopg2-binary # PostgreSQL
pip install pymysql # MySQL
pip install aiosqlite # SQLite async
```
### SQLAlchemy 2.0
This skill covers SQLAlchemy 2.0+ syntax which is the current standard:
```python
# SQLAlchemy 2.0 style (recommended)
from sqlalchemy import select
from sqlalchemy.orm import Session
stmt = select(User).where(User.name == "john")
result = session.execute(stmt)
users = result.scalars().all()
```
## Engine and Connection
### Creating Engine
```python
from sqlalchemy import create_engine
# SQLite
engine = create_engine("sqlite:///database.db")
# PostgreSQL
engine = create_engine("postgresql://user:password@localhost:5432/mydb")
# PostgreSQL with psycopg2
engine = create_engine("postgresql+psycopg2://user:password@localhost/mydb")
# MySQL
engine = create_engine("mysql+pymysql://user:password@localhost:3306/mydb")
# Connection pool settings
engine = create_engine(
"postgresql://user:password@localhost/mydb",
pool_size=10, # Number of connections to keep
max_overflow=20, # Additional connections allowed
pool_timeout=30, # Seconds to wait for connection
pool_recycle=3600, # Recycle connections after 1 hour
echo=True, # Log SQL statements
echo_pool=True, # Log connection pool events
)
```
### Async Engine
```python
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
# Async engine
async_engine = create_async_engine(
"postgresql+asyncpg://user:password@localhost/mydb",
echo=True,
pool_size=10,
)
# Async session factory
AsyncSessionLocal = sessionmaker(
async_engine,
class_=AsyncSession,
expire_on_commit=False,
)
# Usage
async with AsyncSessionLocal() as session:
result = await session.execute(select(User))
users = result.scalars().all()
```
### Connection URL Formats
```python
# SQLite
"sqlite:///database.db" # Relative path
"sqlite:////absolute/path/to/db.db" # Absolute path
"sqlite:///:memory:" # In-memory database
# PostgreSQL
"postgresql://user:password@host:port/database"
"postgresql+asyncpg://user:password@host/database" # Async
# MySQL
"mysql+pymysql://user:password@host:port/database"
"mysql+aiomysql://user:password@host/database" # Async
# Oracle
"oracle+cx_oracle://user:password@host:port/?service_name=myservice"
# SQL Server
"mssql+pyodbc://user:password@host/database?driver=ODBC+Driver+17+for+SQL+Server"
```
## Declarative Models
### Basic Model
```python
from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.sql import func
from datetime import datetime
class Base(DeclarativeBase):
"""Base class for all models."""
pass
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(255), unique=True, nullable=False)
password_hash = Column(String(255), nullable=False)
is_active = Column(Boolean, default=True)
is_superuser = Column(Boolean, default=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
def __repr__(self):
return f"<User(id={self.id}, username='{self.username}')>"
```
### Column Types
```python
from sqlalchemy import (
Column, Integer, BigInteger, SmallInteger,
String, Text, Unicode, UnicodeText,
Float, Double, Numeric, Decimal,
Boolean, Date, Time, DateTime,
JSON, LargeBinary, Enum,
ForeignKey, ForeignKeyConstraint,
Index, UniqueConstraint, CheckConstraint,
)
class Product(Base):
__tablename__ = "products"
# Integer types
id = Column(Integer, primary_key=True)
quantity = Column(Integer, default=0)
big_id = Column(BigInteger)
small_int = Column(SmallInteger)
# String types
name = Column(String(100), nullable=False)
description = Column(Text)
code = Column(Unicode(20)) # Unicode support
# Numeric types
price = Column(Numeric(10, 2)) # Precision, scale
weight = Column(Float)
# Boolean
is_available = Column(Boolean, default=True)
# Date/Time
created_at = Column(DateTime(timezone=True), server_default=func.now())
published_date = Column(Date)
open_time = Column(Time)
# JSON
metadata = Column(JSON)
tags = Column(JSON) # PostgreSQL JSONB
# Binary
image_data = Column(LargeBinary)
# Enum
status = Column(Enum("draft", "published", "archived", name="product_status"))
```
### Table Arguments
```python
class Article(Base):
__tablename__ = "articles"
id = Column(Integer, primary_key=True)
slug = Column(String(100))
title = Column(String(200))
author_id = Column(Integer, ForeignKey("users.id"))
# Table-level constraints
__table_args__ = (
UniqueConstraint("slug", name="uq_article_slug"),
Index("ix_article_author", "author_id"),
CheckConstraint("length(title) > 0", name="ck_article_title"),
{"schema": "blog"}, # Schema name
)
```
+ ### Non-Integer Primary Keys
+
+ See `references/queries.md` for UUID, string PKs, and composite keys.
+
+ ```python
+ import uuid
+ from sqlalchemy.dialects.postgresql import UUID
+
+ # UUID primary key
+ class Email(Base):
+ __tablename__ = "emails"
+ id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+ subject = Column(String(255))
+
+ # String primary key with custom generator
+ import secrets
+
+ def generate_id(prefix: str = "") -> str:
+ """Generate a unique string ID."""
+ return f"{prefix}{secrets.token_hex(8)}"
+
+ class Service(Base):
+ __tablename__ = "services"
+ id = Column(String(64), primary_key=True, default=lambda: generate_id("svc_"))
+ ```
+
## Relationships
### One-to-Many
```python
from sqlalchemy.orm import relationship, Mapped, mapped_column
from typing import List, Optional
class Author(Base):
__tablename__ = "authors"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
# One-to-many relationship
articles: Mapped[List["Article"]] = relationship(
back_populates="author",
cascade="all, delete-orphan",
)
class Article(Base):
__tablename__ = "articles"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200))
author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
# Relationship to parent
author: Mapped["Author"] = relationship(back_populates="articles")
```
### Many-to-Many
```python
# Association table
article_tags = Table(
"article_tags",
Base.metadata,
Column("article_id", Integer, ForeignKey("articles.id", ondelete="CASCADE"), primary_key=True),
Column("tag_id", Integer, ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True),
Column("created_at", DateTime, server_default=func.now()),
)
class Article(Base):
__tablename__ = "articles"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200))
tags: Mapped[List["Tag"]] = relationship(
secondary=article_tags,
back_populates="articles",
)
class Tag(Base):
__tablename__ = "tags"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50), unique=True)
articles: Mapped[List["Article"]] = relationship(
secondary=article_tags,
back_populates="tags",
)
```
- ### Association Object (Many-to-Many with Extra Fields)
-
- ```python
- class OrderItem(Base):
- __tablename__ = "order_items"
-
- order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"), primary_key=True)
- product_id: Mapped[int] = mapped_column(ForeignKey("products.id"), primary_key=True)
- quantity: Mapped[int] = mapped_column(default=1)
- unit_price: Mapped[float] = mapped_column(Numeric(10, 2))
-
- order: Mapped["Order"] = relationship(back_populates="items")
- product: Mapped["Product"] = relationship(back_populates="order_items")
-
- class Order(Base):
- __tablename__ = "orders"
-
- id: Mapped[int] = mapped_column(primary_key=True)
- items: Mapped[List["OrderItem"]] = relationship(back_populates="order", cascade="all, delete-orphan")
-
- class Product(Base):
- __tablename__ = "products"
-
- id: Mapped[int] = mapped_column(primary_key=True)
- order_items: Mapped[List["OrderItem"]] = relationship(back_populates="product")
- ```
-
### One-to-One
```python
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
username: Mapped[str] = mapped_column(String(50), unique=True)
profile: Mapped["UserProfile"] = relationship(
back_populates="user",
uselist=False, # One-to-one
cascade="all, delete-orphan",
)
class UserProfile(Base):
__tablename__ = "user_profiles"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), unique=True)
bio: Mapped[Optional[str]] = mapped_column(Text)
avatar_url: Mapped[Optional[str]] = mapped_column(String(255))
user: Mapped["User"] = relationship(back_populates="profile")
```
### Self-Referential
```python
class Category(Base):
__tablename__ = "categories"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
parent_id: Mapped[Optional[int]] = mapped_column(ForeignKey("categories.id"))
# Self-referential relationship
parent: Mapped[Optional["Category"]] = relationship(
back_populates="children",
remote_side=[id],
)
children: Mapped[List["Category"]] = relationship(
back_populates="parent",
cascade="all, delete-orphan",
)
```
### Relationship Options
```python
class Article(Base):
# Lazy loading options
comments: Mapped[List["Comment"]] = relationship(
lazy="select", # Default: load on access
# lazy="joined", # Eager load with JOIN
# lazy="subquery", # Eager load with separate query
# lazy="dynamic", # Return query object (for filtering)
# lazy="noload", # Don't load
# lazy="raise", # Raise error on access
)
# Cascade options
items: Mapped[List["Item"]] = relationship(
cascade="all", # All operations
cascade="all, delete-orphan", # Delete children when parent deleted
cascade="save-update, merge", # Only these operations
)
# Order by
comments: Mapped[List["Comment"]] = relationship(
order_by="Comment.created_at.desc()",
)
```
- ## Queries (SQLAlchemy 2.0 Style)
-
- ### Basic Queries
-
- ```python
- from sqlalchemy import select
- from sqlalchemy.orm import Session
-
- # Select all
- stmt = select(User)
- users = session.execute(stmt).scalars().all()
-
- # Select with where
- stmt = select(User).where(User.is_active == True)
- active_users = session.execute(stmt).scalars().all()
-
- # Multiple conditions
- stmt = select(User).where(
- User.is_active == True,
- User.created_at > "2024-01-01",
- )
- users = session.execute(stmt).scalars().all()
-
- # Select specific columns
- stmt = select(User.id, User.username, User.email)
- result = session.execute(stmt)
- for row in result:
- print(f"ID: {row.id}, Username: {row.username}")
-
- # Get by primary key
- user = session.get(User, 1) # Returns None if not found
-
- # Get one
- stmt = select(User).where(User.username == "john")
- user = session.execute(stmt).scalar_one_or_none() # None if not found
- user = session.execute(stmt).scalar_one() # Raises if not found
- ```
-
- ### Joins
-
- ```python
- # Simple join
- stmt = select(Article).join(Author)
- articles = session.execute(stmt).scalars().all()
-
- # Join with condition
- stmt = select(Article).join(Author, Article.author_id == Author.id)
-
- # Multiple joins
- stmt = select(Comment).join(Article).join(Author)
-
- # Join with specific columns
- stmt = select(Article.title, Author.name).join(Author)
- result = session.execute(stmt)
- for row in result:
- print(f"{row.title} by {row.name}")
-
- # Left outer join
- from sqlalchemy.orm import outerjoin
- stmt = select(Author, Article).outerjoin(Article)
-
- # Join to alias (self-join)
- from sqlalchemy.orm import aliased
- Manager = aliased(Employee)
- stmt = select(Employee, Manager).join(
- Manager, Employee.manager_id == Manager.id
- )
-
- # Eager loading with joinedload
- from sqlalchemy.orm import joinedload
- stmt = select(Author).options(joinedload(Author.articles))
- authors = session.execute(stmt).unique().scalars().all()
-
- # Selectin load (separate query)
- from sqlalchemy.orm import selectinload
- stmt = select(Author).options(selectinload(Author.articles))
-
- # Load only specific relationships
- stmt = select(Author).options(
- selectinload(Author.articles).selectinload(Article.tags)
- )
- ```
-
- ### Filtering
-
- ```python
- from sqlalchemy import and_, or_, not_, func, desc, asc
-
- # Comparison operators
- stmt = select(User).where(User.age > 18)
- stmt = select(User).where(User.age >= 18)
- stmt = select(User).where(User.age < 65)
- stmt = select(User).where(User.name == "John")
- stmt = select(User).where(User.name != "John")
-
- # LIKE
- stmt = select(User).where(User.name.like("%john%"))
- stmt = select(User).where(User.name.ilike("%JOHN%")) # Case-insensitive
-
- # IN
- stmt = select(User).where(User.id.in_([1, 2, 3]))
- stmt = select(User).where(User.status.in_(["active", "pending"]))
-
- # NOT IN
- stmt = select(User).where(User.id.not_in([1, 2, 3]))
-
- # BETWEEN
- stmt = select(User).where(User.age.between(18, 65))
-
- # IS NULL / IS NOT NULL
- stmt = select(User).where(User.deleted_at.is_(None))
- stmt = select(User).where(User.deleted_at.is_not(None))
-
- # AND / OR / NOT
- stmt = select(User).where(
- and_(User.is_active == True, User.age > 18)
- )
- stmt = select(User).where(
- or_(User.role == "admin", User.role == "moderator")
- )
- stmt = select(User).where(
- not_(User.is_banned)
- )
-
- # Chained filters
- stmt = (
- select(User)
- .where(User.is_active == True)
- .where(User.age >= 18)
- .where(User.country == "US")
- )
- ```
-
- ### Ordering and Limiting
-
- ```python
- # Order by
- stmt = select(User).order_by(User.created_at)
- stmt = select(User).order_by(desc(User.created_at))
- stmt = select(User).order_by(User.last_name, User.first_name)
-
- # Limit and offset
- stmt = select(User).limit(10)
- stmt = select(User).offset(20).limit(10) # Pagination
-
- # Pagination helper
- def paginate(query, page: int, per_page: int = 20):
- return query.offset((page - 1) * per_page).limit(per_page)
-
- stmt = paginate(select(User), page=2)
- ```
-
- ### Aggregation
-
- ```python
- from sqlalchemy import func, count, sum, avg, max, min
-
- # Count
- stmt = select(count()).select_from(User)
- total = session.execute(stmt).scalar()
-
- # Count with filter
- stmt = select(count(User.id)).where(User.is_active == True)
- active_count = session.execute(stmt).scalar()
-
- # Sum, Avg, Min, Max
- stmt = select(sum(Order.total))
- stmt = select(avg(Product.price))
- stmt = select(max(User.age))
- stmt = select(min(Product.price))
-
- # Group by
- stmt = (
- select(Author.name, count(Article.id))
- .join(Article)
- .group_by(Author.id)
- .order_by(desc(count(Article.id)))
- )
-
- # Having
- stmt = (
- select(Author.name, count(Article.id).label("article_count"))
- .join(Article)
- .group_by(Author.id)
- .having(count(Article.id) > 5)
- )
- ```
-
- ### Subqueries
-
- ```python
- # Scalar subquery
- subq = (
- select(func.avg(Product.price))
- .where(Product.category_id == Category.id)
- .scalar_subquery()
- )
- stmt = select(Category.name, subq.label("avg_price"))
-
- # IN subquery
- subq = select(Article.author_id).where(Article.views > 1000)
- stmt = select(Author).where(Author.id.in_(subq))
-
- # EXISTS
- from sqlalchemy import exists
- subq = select(Article.id).where(Article.author_id == Author.id)
- stmt = select(Author).where(exists(subq))
-
- # CTE (Common Table Expression)
- cte = (
- select(Author.name, count(Article.id).label("article_count"))
- .join(Article)
- .group_by(Author.id)
- .cte("author_stats")
- )
- stmt = select(cte).where(cte.c.article_count > 10)
- ```
-
## Sessions
### Session Management
```python
from sqlalchemy.orm import Session, sessionmaker
# Create session factory
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)
# Use session with context manager
with SessionLocal() as session:
user = session.execute(select(User)).scalar()
session.commit()
# Manual management
session = SessionLocal()
try:
user = session.execute(select(User)).scalar()
session.commit()
finally:
session.close()
# Dependency injection (FastAPI style)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
```
### Adding and Updating
```python
# Add single object
user = User(username="john", email="john@example.com")
session.add(user)
session.commit()
session.refresh(user) # Refresh to get server-generated values
# Add multiple objects
users = [
User(username="user1", email="user1@example.com"),
User(username="user2", email="user2@example.com"),
]
session.add_all(users)
session.commit()
# Update object
user = session.get(User, 1)
user.email = "newemail@example.com"
session.commit()
# Update with query
from sqlalchemy import update
stmt = (
update(User)
.where(User.is_active == True)
.values(last_login=func.now())
)
session.execute(stmt)
session.commit()
-
- # Update with returning
- stmt = (
- update(User)
- .where(User.id == 1)
- .values(email="new@example.com")
- .returning(User)
- )
- result = session.execute(stmt)
- updated_user = result.scalar_one()
- session.commit()
```
### Deleting
```python
# Delete object
user = session.get(User, 1)
session.delete(user)
session.commit()
# Delete with query
from sqlalchemy import delete
stmt = delete(User).where(User.is_active == False)
session.execute(stmt)
session.commit()
-
- # Delete with returning
- stmt = (
- delete(User)
- .where(User.id == 1)
- .returning(User.id, User.username)
- )
- result = session.execute(stmt)
- deleted = result.fetchone()
- session.commit()
```
- ### Transactions
+ ### Transaction Management
```python
# Nested transaction (SAVEPOINT)
with session.begin_nested():
session.add(User(username="test"))
# Auto-rollback on exception
# Manual transaction control
session.begin()
try:
session.add(user)
session.commit()
except:
session.rollback()
raise
# Using begin() context manager
with session.begin():
session.add(user)
# Auto-commit or rollback
```
- ### Bulk Operations
-
- ```python
- # Bulk insert (no events, no relationships)
- session.execute(
- insert(User),
- [
- {"username": "user1", "email": "user1@example.com"},
- {"username": "user2", "email": "user2@example.com"},
- ]
- )
- session.commit()
-
- # Bulk update
- session.execute(
- update(User)
- .where(User.is_active == True)
- .values(last_login=func.now())
- )
-
- # ORM bulk operations (with events)
- session.bulk_insert_mappings(User, [
- {"username": "user1", "email": "user1@example.com"},
- {"username": "user2", "email": "user2@example.com"},
- ])
-
- session.bulk_update_mappings(User, [
- {"id": 1, "email": "new1@example.com"},
- {"id": 2, "email": "new2@example.com"},
- ])
- ```
-
- ## Async SQLAlchemy
-
- ### Async Models and Session
-
- ```python
- from sqlalchemy.ext.asyncio import (
- create_async_engine,
- AsyncSession,
- async_sessionmaker,
- )
- from sqlalchemy.orm import Mapped, mapped_column
-
- # Async engine
- engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
-
- # Async session factory
- async_session = async_sessionmaker(
- engine,
- class_=AsyncSession,
- expire_on_commit=False,
- )
-
- # Async context manager
- async with async_session() as session:
- async with session.begin():
- result = await session.execute(select(User))
- users = result.scalars().all()
- ```
-
- ### Async Queries
-
- ```python
- from sqlalchemy import select
-
- async def get_user(user_id: int) -> Optional[User]:
- async with async_session() as session:
- result = await session.execute(
- select(User).where(User.id == user_id)
- )
- return result.scalar_one_or_none()
-
- async def get_users_paginated(page: int, per_page: int) -> List[User]:
- async with async_session() as session:
- stmt = (
- select(User)
- .order_by(User.created_at.desc())
- .offset((page - 1) * per_page)
- .limit(per_page)
- )
- result = await session.execute(stmt)
- return result.scalars().all()
-
- async def create_user(username: str, email: str) -> User:
- async with async_session() as session:
- async with session.begin():
- user = User(username=username, email=email)
- session.add(user)
- await session.flush()
- await session.refresh(user)
- return user
- ```
-
- ### Async with FastAPI
-
- ```python
- from fastapi import FastAPI, Depends, HTTPException
- from sqlalchemy.ext.asyncio import AsyncSession
- from sqlalchemy import select
-
- app = FastAPI()
-
- async def get_db():
- async with async_session() as session:
- try:
- yield session
- finally:
- await session.close()
-
- @app.get("/users/{user_id}")
- async def get_user(
- user_id: int,
- db: AsyncSession = Depends(get_db)
- ):
- result = await db.execute(
- select(User).where(User.id == user_id)
- )
- user = result.scalar_one_or_none()
- if not user:
- raise HTTPException(status_code=404, detail="User not found")
- return user
-
- @app.post("/users")
- async def create_user(
- username: str,
- email: str,
- db: AsyncSession = Depends(get_db)
- ):
- user = User(username=username, email=email)
- db.add(user)
- await db.commit()
- await db.refresh(user)
- return user
- ```
-
- ## Alembic Migrations
-
- ### Setup
-
- ```bash
- # Initialize Alembic
- alembic init alembic
-
- # Edit alembic.ini
- sqlalchemy.url = postgresql://user:password@localhost/mydb
-
- # Or use env.py for dynamic URL
- ```
-
- ```python
- # alembic/env.py
- from logging.config import fileConfig
- from sqlalchemy import engine_from_config
- from sqlalchemy import pool
- from alembic import context
- from myapp.models import Base # Import your models
-
- config = context.config
- fileConfig(config.config_file_name)
- target_metadata = Base.metadata
-
- def run_migrations_offline():
- url = config.get_main_option("sqlalchemy.url")
- context.configure(
- url=url,
- target_metadata=target_metadata,
- literal_binds=True,
- dialect_opts={"paramstyle": "named"},
- )
- with context.begin_transaction():
- context.run_migrations()
-
- def run_migrations_online():
- connectable = engine_from_config(
- config.get_section(config.config_ini_section),
- prefix="sqlalchemy.",
- poolclass=pool.NullPool,
- )
- with connectable.connect() as connection:
- context.configure(
- connection=connection,
- target_metadata=target_metadata,
- )
- with context.begin_transaction():
- context.run_migrations()
- ```
-
- ### Creating Migrations
-
- ```bash
- # Auto-generate migration from model changes
- alembic revision --autogenerate -m "Add user table"
-
- # Create empty migration
- alembic revision -m "Add custom index"
-
- # Apply migrations
- alembic upgrade head
-
- # Rollback one migration
- alembic downgrade -1
-
- # Rollback to specific revision
- alembic downgrade abc123
-
- # View history
- alembic history
-
- # Current revision
- alembic current
- ```
-
- ### Migration File
-
- ```python
- # alembic/versions/abc123_add_user_table.py
- """Add user table
-
- Revision ID: abc123
- Revises:
- Create Date: 2024-01-15 10:00:00.000000
- """
- from alembic import op
- import sqlalchemy as sa
-
- # revision identifiers
- revision = 'abc123'
- down_revision = None
- branch_labels = None
- depends_on = None
-
- def upgrade():
- op.create_table(
- 'users',
- sa.Column('id', sa.Integer(), nullable=False),
- sa.Column('username', sa.String(50), nullable=False),
- sa.Column('email', sa.String(255), nullable=False),
- sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()),
- sa.PrimaryKeyConstraint('id'),
- sa.UniqueConstraint('username'),
- sa.UniqueConstraint('email'),
- )
- op.create_index('ix_users_username', 'users', ['username'])
-
- def downgrade():
- op.drop_index('ix_users_username', table_name='users')
- op.drop_table('users')
- ```
-
- ### Common Migration Operations
-
- ```python
- def upgrade():
- # Create table
- op.create_table(
- 'products',
- sa.Column('id', sa.Integer(), primary_key=True),
- sa.Column('name', sa.String(100), nullable=False),
- )
-
- # Add column
- op.add_column('users', sa.Column('phone', sa.String(20)))
-
- # Drop column
- op.drop_column('users', 'phone')
-
- # Alter column
- op.alter_column(
- 'users',
- 'username',
- existing_type=sa.String(50),
- type_=sa.String(100),
- nullable=True,
- )
-
- # Create index
- op.create_index('ix_users_email', 'users', ['email'], unique=True)
-
- # Drop index
- op.drop_index('ix_users_email', table_name='users')
-
- # Drop foreign key
- op.drop_constraint('fk_articles_author', 'articles', type_='foreignkey')
-
- # Execute raw SQL
- op.execute("UPDATE users SET is_active = TRUE")
- ```
-
- ## Migration Testing
-
- Migrations are production-critical code. Untested migrations cause downtime.
-
- ### Why test migrations
-
- - Migrations run on every deployment
- - Schema changes are irreversible in production
- - Data loss from bad migrations is catastrophic
-
- ### Testing upgrade path
-
- ```python
- # tests/test_migrations/test_001_add_email_column.py
- import pytest
- from alembic import command
- from alembic.config import Config
- from sqlalchemy import create_engine, text
-
- @pytest.fixture
- def alembic_config():
- return Config("alembic.ini")
-
- def test_upgrade_adds_email_column(db_engine, alembic_config):
- """Test that upgrade adds the email column."""
- command.upgrade(alembic_config, "head")
-
- with db_engine.connect() as conn:
- result = conn.execute(text("PRAGMA table_info(users)"))
- columns = [row[1] for row in result]
- assert "email" in columns
-
- def test_downgrade_removes_email_column(db_engine, alembic_config):
- """Test that downgrade removes the email column."""
- command.upgrade(alembic_config, "head")
- command.downgrade(alembic_config, "-1")
-
- with db_engine.connect() as conn:
- result = conn.execute(text("PRAGMA table_info(users)"))
- columns = [row[1] for row in result]
- assert "email" not in columns
- ```
-
- ### Testing both directions
-
- Always test **upgrade AND downgrade** — downgrades are your emergency rollback.
-
- ### Migration test fixtures
-
- ```python
- @pytest.fixture
- def db_engine():
- """Fresh SQLite DB for each migration test."""
- engine = create_engine("sqlite:///:memory:")
- yield engine
- engine.dispose()
- ```
-
- ### Migration coverage
-
- Measuring which migration files have tests:
-
- ```python
- # tests/conftest.py
- import os
- from pathlib import Path
-
- def test_all_migrations_have_tests():
- """Ensure every migration file has a corresponding test."""
- migration_dir = Path("alembic/versions")
- test_dir = Path("tests/test_migrations")
-
- migrations = list(migration_dir.glob("*.py"))
- for migration in migrations:
- # Check if a test file exists
- migration_id = migration.stem.split("_")[0]
- test_files = list(test_dir.glob(f"*{migration_id}*"))
- assert test_files, f"No test for migration {migration.name}"
- ```
-
- ### Common Pitfalls
-
- | Issue | Cause | Solution |
- |-------|-------|----------|
- | Migration tests modify shared DB | No isolation | Use in-memory SQLite per test |
- | Downgrade not tested | Only testing upgrade | Always test both directions |
- | Tests pass locally, fail in CI | SQLite vs PostgreSQL differences | Test against PostgreSQL in CI |
-
- ## PostgreSQL Query Optimization
-
- ### EXPLAIN ANALYZE with SQLAlchemy
-
- ```python
- from sqlalchemy import text
-
- def analyze_query(session, query):
- """Run EXPLAIN ANALYZE on a query."""
- compiled = query.statement.compile(session.bind)
- explain_sql = text(f"EXPLAIN ANALYZE {compiled}")
- result = session.execute(explain_sql).fetchall()
- for row in result:
- print(row[0])
- ```
-
- ### Index strategy
-
- When to add indexes:
-
- ```python
- # Add index for frequently filtered columns
- class User(Base):
- __tablename__ = "users"
- id = Column(Integer, primary_key=True)
- email = Column(String, index=True) # Frequently filtered
- created_at = Column(DateTime, index=True) # Frequently sorted
-
- # Composite index for multi-column filters
- class Event(Base):
- __tablename__ = "events"
- __table_args__ = (
- Index("ix_event_user_date", "user_id", "created_at"), # Composite
- )
- user_id = Column(Integer, ForeignKey("users.id"))
- created_at = Column(DateTime)
- ```
-
- ### Read-only transactions for metrics endpoints
-
- ```python
- from sqlalchemy.orm import Session
-
- def get_metrics(session: Session):
- """Read-only metrics query — no writes, no locks."""
- # Use execution_options for read-only
- result = session.execute(
- text("SELECT status, COUNT(*) FROM services WHERE active = true GROUP BY status"),
- execution_options={"read_only": True}
- )
- return result.fetchall()
- ```
-
- ### Query plan optimization
-
- Detecting and fixing slow queries:
-
- ```python
- # Check for sequential scans (should use index scan)
- def check_query_plan(session, query):
- compiled = query.statement.compile(session.bind)
- plan = session.execute(text(f"EXPLAIN {compiled}")).fetchall()
- plan_text = "\n".join(row[0] for row in plan)
-
- if "Seq Scan" in plan_text:
- logger.warning(f"Sequential scan detected — consider adding index:\n{plan_text}")
- return plan_text
- ```
-
- ### Partitioning large tables
-
- Time-series partitioning pattern:
-
- ```python
- # PostgreSQL native partitioning via raw SQL in migration
- def upgrade():
- op.execute("""
- CREATE TABLE events (
- id SERIAL,
- created_at TIMESTAMP NOT NULL,
- data JSONB
- ) PARTITION BY RANGE (created_at);
- """)
- op.execute("""
- CREATE TABLE events_2026_01
- PARTITION OF events
- FOR VALUES FROM ('2026-01-01') TO ('2026-02-01');
- """)
- ```
-
- ### Common Pitfalls
-
- | Issue | Cause | Solution |
- |-------|-------|----------|
- | Slow /metrics endpoint | Loading all history | Read-only transaction with filtered query |
- | Sequential scan on large table | Missing index | Add index on filtered column |
- | Lock contention | Read-write transaction for read-only query | Use `read_only` execution option |
- | N+1 queries | Lazy loading in loop | Use `joinedload()` or `selectinload()` |
-
- ## Non-Integer Primary Keys
-
- ### UUID primary keys
-
- ```python
- import uuid
- from sqlalchemy.dialects.postgresql import UUID
-
- class Email(Base):
- __tablename__ = "emails"
- id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
- subject = Column(String(255))
- ```
-
- ### String primary keys
-
- Custom ID generation pattern:
-
- ```python
- import secrets
-
- def generate_id(prefix: str = "") -> str:
- """Generate a unique string ID."""
- return f"{prefix}{secrets.token_hex(8)}"
-
- class Email(Base):
- __tablename__ = "emails"
- id = Column(String(64), primary_key=True, default=lambda: generate_id("email_"))
- subject = Column(String(255))
- ```
-
- ### Auto-generation strategies
-
- ```python
- # UUID with server_default (PostgreSQL)
- class User(Base):
- __tablename__ = "users"
- id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()"))
-
- # String ID with Python-side default
- class Service(Base):
- __tablename__ = "services"
- id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
- ```
-
- ### Composite primary keys
-
- ```python
- # Multi-column primary key
- class RolePermission(Base):
- __tablename__ = "role_permissions"
- role_id = Column(Integer, ForeignKey("roles.id"), primary_key=True)
- permission_id = Column(Integer, ForeignKey("permissions.id"), primary_key=True)
- # No separate id column — composite PK
- ```
-
- ### Common Pitfalls
-
- | Issue | Cause | Solution |
- |-------|-------|----------|
- | ID collision with string IDs | Weak random generator | Use `secrets` module, not `random` |
- | UUID not stored efficiently | Storing as String | Use `UUID(as_uuid=True)` with PostgreSQL |
- | Can't auto-increment | Non-integer PK | Set `default=` or `server_default=` |
-
- ## Advanced Query Patterns
-
- ### CTEs (Common Table Expressions)
-
- ```python
- from sqlalchemy import CTE
-
- # Recursive CTE: organizational hierarchy
- org_cte = select(Employee).where(Employee.manager_id.is_(None)).cte(name="org", recursive=True)
- mgr = org_cte.alias("mgr")
- stmt = (
- select(mgr)
- .join(org_cte, mgr.c.manager_id == org_cte.c.id)
- )
- # Non-recursive CTE
- active_cte = (
- select(User.id, User.name)
- .where(User.is_active.is_(True))
- .cte("active_users")
- )
- stmt = select(Order).join(active_cte, Order.user_id == active_cte.c.id)
- ```
-
- ### Window Functions
-
- ```python
- from sqlalchemy import over
-
- # Row number, rank, dense rank
- stmt = (
- select(
- User.name,
- User.salary,
- func.row_number().over(order_by=User.salary.desc()).label("rn"),
- func.rank().over(order_by=User.salary.desc()).label("rank"),
- func.dense_rank().over(order_by=User.salary.desc()).label("drank"),
- func.sum(User.salary).over(partition_by=User.dept).label("dept_total"),
- )
- .order_by(User.salary.desc())
- )
- for row in session.execute(stmt):
- print(f"{row.name}: ${row.salary} (rank: {row.rank})")
- ```
-
- ## Transaction Management
-
- ### Nested Transactions with Savepoints
-
- ```python
- from sqlalchemy import begin_nested
-
- # Using savepoints for partial rollback
- with Session(engine) as session:
- session.begin()
- session.add(User(name="Alice"))
-
- # Create a savepoint
- nested = session.begin_nested()
- try:
- session.add(Order(user_id=1, amount=100))
- # If this fails, only the nested transaction rolls back
- nested.commit()
- except Exception:
- nested.rollback()
- # Alice is still in the session, only Order was rolled back
-
- session.commit()
- ```
-
- ### Read-Only Transactions
-
- ```python
- from sqlalchemy import text
-
- with engine.connect() as conn:
- conn.execute(text("SET TRANSACTION READ ONLY"))
- result = conn.execute(select(User))
- # Any write attempt will raise an error
- ```
-
- ### Session Lifecycle Best Practices
-
- ```python
- # Pattern 1: Context manager (recommended)
- with Session(engine) as session:
- session.add(user)
- session.commit()
- # Session automatically closed
-
- # Pattern 2: Async session
- from sqlalchemy.ext.asyncio import AsyncSession
-
- async with AsyncSession(async_engine) as session:
- async with session.begin():
- session.add(user)
- # Auto-committed and closed
- ```
-
- ## Bulk Operations
-
- ### Bulk Inserts
-
- ```python
- from sqlalchemy import insert
-
- # Core bulk insert (fastest)
- stmt = insert(User).values([
- {"name": "Alice", "email": "alice@example.com"},
- {"name": "Bob", "email": "bob@example.com"},
- {"name": "Charlie", "email": "charlie@example.com"},
- ])
- session.execute(stmt)
- session.commit()
-
- # ORM bulk insert (slower, but triggers events)
- session.add_all([
- User(name="Alice", email="alice@example.com"),
- User(name="Bob", email="bob@example.com"),
- ])
- session.commit()
- ```
-
- ### Bulk Updates
-
- ```python
- from sqlalchemy import update
-
- # Core bulk update
- stmt = (
- update(User)
- .where(User.is_active.is_(True))
- .values(last_login=func.now())
- )
- session.execute(stmt)
- session.commit()
-
- # Bulk update with binding
- stmt = update(User).where(User.name == "old_name").values(name="new_name")
- session.execute(stmt)
- session.commit()
- ```
-
- ### Bulk Deletes
-
- ```python
- from sqlalchemy import delete
-
- stmt = delete(User).where(User.last_login < func.now() - text("interval '90 days'"))
- result = session.execute(stmt)
- session.commit()
- print(f"Deleted {result.rowcount} inactive users")
- ```
-
- ### Performance Tips for Large Datasets
-
- ```python
- # Use yield_per for large result sets
- for user in session.scalars(select(User)).yield_per(100):
- process(user)
-
- # Use server-side cursors with stream()
- for row in session.stream(select(LargeTable)):
- process(row)
-
- # Batch inserts with executemany
- session.execute(insert(User), [
- {"name": f"User {i}", "email": f"user{i}@example.com"}
- for i in range(10000)
- ], execution_options={"max_rows": 1000})
- session.commit()
- ```
-
## Best Practices
### 1. Use Mapped Types (SQLAlchemy 2.0)
```python
# Good: Mapped types with type hints
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
username: Mapped[str] = mapped_column(String(50))
bio: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
articles: Mapped[List["Article"]] = relationship()
```
### 2. Session Scope
```python
# Good: Use context managers
with SessionLocal() as session:
user = session.execute(select(User)).scalar()
session.commit()
# Bad: Forget to close
session = SessionLocal()
user = session.execute(select(User)).scalar()
# session.close() missing!
```
### 3. Eager Loading
```python
# Good: Explicit eager loading
stmt = select(Author).options(selectinload(Author.articles))
# Bad: N+1 query problem
authors = session.execute(select(Author)).scalars().all()
for author in authors:
print(author.articles) # Triggers query for each author!
```
### 4. Use expire_on_commit=False
```python
# Good: Can access attributes after commit
SessionLocal = sessionmaker(
bind=engine,
expire_on_commit=False, # Access attributes after commit
)
with SessionLocal() as session:
user = User(username="john")
session.add(user)
session.commit()
print(user.username) # Works with expire_on_commit=False
```
### 5. Connection Pooling
```python
# Good: Configure pool for production
engine = create_engine(
DATABASE_URL,
pool_size=10,
max_overflow=20,
pool_pre_ping=True, # Check connection health
pool_recycle=3600, # Recycle old connections
)
```
- ## Common Issues & Debugging
-
- ### DetachedInstanceError
-
- ```python
- # Problem: Accessing attributes after session closed
- with Session(engine) as session:
- user = session.scalar(select(User).limit(1))
- # print(user.name) # DetachedInstanceError!
-
- # Fix 1: Expire objects on commit = False
- session = Session(engine, expire_on_commit=False)
-
- # Fix 2: Access within session context
- with Session(engine) as session:
- user = session.scalar(select(User).limit(1))
- name = user.name # OK, still attached
- ```
-
- ### Lazy Loading Outside Sessions
-
- ```python
- # Problem: Accessing relationships after session closed
- with Session(engine) as session:
- user = session.scalar(select(User).limit(1))
- # print(user.posts) # Error! Relationship not loaded
-
- # Fix: Use eager loading
- stmt = select(User).options(selectinload(User.posts))
- user = session.scalar(stmt)
- print(user.posts) # OK, already loaded
- ```
-
- ### N+1 Query Problem
-
- ```python
- # BAD: N+1 queries (1 for users + N for each user's posts)
- users = session.scalars(select(User)).all()
- for user in users:
- print(len(user.posts)) # Triggers 1 query per user!
-
- # GOOD: Single query with joined eager loading
- from sqlalchemy.orm import joinedload
- stmt = select(User).options(joinedload(User.posts))
- users = session.scalars(stmt).unique().all()
-
- # GOOD: Two queries with selective loading
- from sqlalchemy.orm import selectinload
- stmt = select(User).options(selectinload(User.posts))
- users = session.scalars(stmt).all()
- ```
-
- ### Session Leak Detection
-
- ```python
- # Enable session tracking for debugging
- from sqlalchemy import event
-
- @event.listens_for(Session, "after_commit")
- def log_commit(session, context):
- logger.info(f"Session committed: {id(session)}")
-
- @event.listens_for(Session, "after_rollback")
- def log_rollback(session, context):
- logger.warning(f"Session rolled back: {id(session)}")
-
- # Use weak_instance_map to track detached instances
- # Always use context managers to prevent leaks:
- with Session(engine) as session:
- # work...
- session.commit()
- # Guaranteed cleanup
- ```
-
- ## References
+ ## Deep Dives
- - **Official Documentation**: https://docs.sqlalchemy.org/
- - **SQLAlchemy 2.0 Overview**: https://docs.sqlalchemy.org/en/20/changelog/migration_20.html
- - **SQLAlchemy 2.0 Tutorial**: https://docs.sqlalchemy.org/en/20/tutorial/
- - **Alembic Documentation**: https://alembic.sqlalchemy.org/
- - **Alembic Tutorial**: https://alembic.sqlalchemy.org/en/latest/tutorial.html
- - **Async Support**: https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html
- - **ORM Querying Guide**: https://docs.sqlalchemy.org/en/20/orm/queryguide/
- - **Core Expression Language**: https://docs.sqlalchemy.org/en/20/core/expression_api.html
+ Load these reference files on demand for detailed coverage:
+ - **`references/queries.md`** - Queries 2.0 style, CTEs, window functions, bulk operations, PostgreSQL optimization
+ - **`references/async.md`** - Async engines, async sessions, FastAPI integration
+ - **`references/migrations.md`** - Alembic setup, migration creation, testing strategies
+ - **`references/debugging.md`** - DetachedInstanceError, lazy loading issues, N+1 queries, session leak detection