sqlalchemy

name	sqlalchemy
description	Python SQL toolkit and Object Relational Mapper (ORM). Use when working with databases in Python, defining models, building queries, managing sessions, or interacting with SQL databases using Python objects.
license	MIT

SQLAlchemy Skill

SQLAlchemy is the Python SQL toolkit and Object Relational Mapper that provides the full power and flexibility of SQL. It consists of two main components: Core (SQL Expression Language) and ORM (Object Relational Mapper).

When to Use This Skill

Use SQLAlchemy when:

• Working with relational databases in Python
• Defining database models as Python classes
• Building SQL queries programmatically
• Managing database transactions and sessions
• Mapping Python objects to database tables
• Need database-agnostic code that works across PostgreSQL, MySQL, SQLite, etc.

Installation

pip install sqlalchemy

# For async support
pip install sqlalchemy[asyncio]

Architecture Overview

┌─────────────────────────────────────────────────────────────┐
│                    SQLAlchemy ORM                           │
│  (Declarative Mapping, Session, Relationships, Unit of Work)│
├─────────────────────────────────────────────────────────────┤
│                   SQLAlchemy Core                           │
│     (SQL Expression Language, Engine, Connection Pool)      │
├─────────────────────────────────────────────────────────────┤
│                        DBAPI                                │
│         (psycopg2, pymysql, sqlite3, etc.)                  │
└─────────────────────────────────────────────────────────────┘

Engine and Connection

The Engine is the starting point for SQLAlchemy applications:

from sqlalchemy import create_engine

# SQLite (in-memory)
engine = create_engine("sqlite://", echo=True)

# SQLite (file-based)
engine = create_engine("sqlite:///mydatabase.db")

# PostgreSQL
engine = create_engine("postgresql+psycopg2://user:password@localhost/dbname")

# MySQL
engine = create_engine("mysql+pymysql://user:password@localhost/dbname")

# Connection pool settings
engine = create_engine(
    "postgresql+psycopg2://user:password@localhost/dbname",
    pool_size=5,           # Number of connections to keep open
    max_overflow=10,       # Additional connections allowed
    pool_timeout=30,       # Seconds to wait for connection
    pool_recycle=1800,     # Recycle connections after N seconds
)

Using Connections Directly (Core)

from sqlalchemy import text

with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM users WHERE id = :id"), {"id": 1})
    for row in result:
        print(row)

    # For write operations, commit explicitly
    conn.execute(text("INSERT INTO users (name) VALUES (:name)"), {"name": "Alice"})
    conn.commit()

ORM Declarative Mapping

from datetime import datetime
from typing import List, Optional

from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user_account"

    # Primary key with auto-increment
    id: Mapped[int] = mapped_column(primary_key=True)

    # Required string column with max length
    name: Mapped[str] = mapped_column(String(30))

    # Optional column (nullable)
    fullname: Mapped[Optional[str]]

    # Column with default value
    created_at: Mapped[datetime] = mapped_column(default=func.now())

    # One-to-many relationship
    addresses: Mapped[List["Address"]] = relationship(
        back_populates="user",
        cascade="all, delete-orphan"
    )

    def __repr__(self) -> str:
        return f"User(id={self.id!r}, name={self.name!r})"


class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))

    # Many-to-one relationship (back reference)
    user: Mapped["User"] = relationship(back_populates="addresses")

    def __repr__(self) -> str:
        return f"Address(id={self.id!r}, email_address={self.email_address!r})"

Type Annotation Guide

Python Type	SQL Type	Nullable
Mapped[int]	INTEGER	NOT NULL
Mapped[Optional[int]]	INTEGER	NULL
Mapped[str]	VARCHAR	NOT NULL
Mapped[Optional[str]]	VARCHAR	NULL
Mapped[bool]	BOOLEAN	NOT NULL
Mapped[datetime]	DATETIME	NOT NULL
Mapped[float]	FLOAT	NOT NULL
Mapped[bytes]	BLOB/BYTEA	NOT NULL

Creating Tables

# Create all tables defined in Base.metadata
Base.metadata.create_all(engine)

# Drop all tables
Base.metadata.drop_all(engine)

Session and CRUD Operations

Session Basics

from sqlalchemy.orm import Session, sessionmaker

# Option 1: Direct Session usage
with Session(engine) as session:
    # ... operations
    session.commit()

# Option 2: Using sessionmaker (recommended for applications)
SessionFactory = sessionmaker(bind=engine)

with SessionFactory() as session:
    # ... operations
    session.commit()

# Option 3: With explicit begin/commit/rollback
with Session(engine) as session:
    with session.begin():
        # Automatically commits on success, rolls back on exception
        session.add(some_object)

Create (INSERT)

with Session(engine) as session:
    # Create single object
    user = User(name="alice", fullname="Alice Smith")
    session.add(user)

    # Create with related objects
    user_with_addresses = User(
        name="bob",
        fullname="Bob Jones",
        addresses=[
            Address(email_address="[e-mail]"),
            Address(email_address="[e-mail]"),
        ]
    )
    session.add(user_with_addresses)

    # Add multiple objects
    session.add_all([
        User(name="carol"),
        User(name="dave"),
    ])

    session.commit()

Read (SELECT)

from sqlalchemy import select

with Session(engine) as session:
    # Get by primary key
    user = session.get(User, 1)

    # Select all
    stmt = select(User)
    users = session.scalars(stmt).all()

    # Select with filter
    stmt = select(User).where(User.name == "alice")
    alice = session.scalars(stmt).first()

    # Select with multiple conditions
    stmt = select(User).where(
        User.name.like("a%"),
        User.id > 5
    )

    # Select specific columns
    stmt = select(User.name, User.fullname)
    rows = session.execute(stmt).all()
    for name, fullname in rows:
        print(f"{name}: {fullname}")

    # Order by
    stmt = select(User).order_by(User.name.desc())

    # Limit and offset
    stmt = select(User).limit(10).offset(20)

    # Count
    from sqlalchemy import func
    stmt = select(func.count()).select_from(User)
    count = session.scalar(stmt)

Update

with Session(engine) as session:
    # Update via ORM (load then modify)
    user = session.get(User, 1)
    user.fullname = "Alice Johnson"
    session.commit()

    # Bulk update
    from sqlalchemy import update
    stmt = update(User).where(User.name == "alice").values(fullname="Alice Updated")
    session.execute(stmt)
    session.commit()

Delete

with Session(engine) as session:
    # Delete via ORM
    user = session.get(User, 1)
    session.delete(user)
    session.commit()

    # Bulk delete
    from sqlalchemy import delete
    stmt = delete(User).where(User.name == "alice")
    session.execute(stmt)
    session.commit()

Relationships

One-to-Many / Many-to-One

class Parent(Base):
    __tablename__ = "parent"

    id: Mapped[int] = mapped_column(primary_key=True)
    children: Mapped[List["Child"]] = relationship(back_populates="parent")


class Child(Base):
    __tablename__ = "child"

    id: Mapped[int] = mapped_column(primary_key=True)
    parent_id: Mapped[int] = mapped_column(ForeignKey("parent.id"))
    parent: Mapped["Parent"] = relationship(back_populates="children")

One-to-One

class User(Base):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(primary_key=True)
    profile: Mapped["Profile"] = relationship(back_populates="user", uselist=False)


class Profile(Base):
    __tablename__ = "profile"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), unique=True)
    user: Mapped["User"] = relationship(back_populates="profile")

Many-to-Many

from sqlalchemy import Column, Table

# Association table (no ORM class needed)
association_table = Table(
    "association",
    Base.metadata,
    Column("left_id", ForeignKey("left.id"), primary_key=True),
    Column("right_id", ForeignKey("right.id"), primary_key=True),
)


class Left(Base):
    __tablename__ = "left"

    id: Mapped[int] = mapped_column(primary_key=True)
    rights: Mapped[List["Right"]] = relationship(
        secondary=association_table,
        back_populates="lefts"
    )


class Right(Base):
    __tablename__ = "right"

    id: Mapped[int] = mapped_column(primary_key=True)
    lefts: Mapped[List["Left"]] = relationship(
        secondary=association_table,
        back_populates="rights"
    )

Association Object (Many-to-Many with extra data)

class Association(Base):
    __tablename__ = "association"

    left_id: Mapped[int] = mapped_column(ForeignKey("left.id"), primary_key=True)
    right_id: Mapped[int] = mapped_column(ForeignKey("right.id"), primary_key=True)
    extra_data: Mapped[Optional[str]]

    left: Mapped["Left"] = relationship(back_populates="right_associations")
    right: Mapped["Right"] = relationship(back_populates="left_associations")


class Left(Base):
    __tablename__ = "left"

    id: Mapped[int] = mapped_column(primary_key=True)
    right_associations: Mapped[List["Association"]] = relationship(back_populates="left")


class Right(Base):
    __tablename__ = "right"

    id: Mapped[int] = mapped_column(primary_key=True)
    left_associations: Mapped[List["Association"]] = relationship(back_populates="right")

Loading Strategies

Lazy Loading (Default)

# Lazy loading - queries database when attribute is accessed
user = session.get(User, 1)
# SELECT ... FROM user WHERE id = 1

addresses = user.addresses  # N+1 query problem!
# SELECT ... FROM address WHERE user_id = 1

Eager Loading with joinedload

from sqlalchemy.orm import joinedload

# Load user and addresses in single query using JOIN
stmt = select(User).options(joinedload(User.addresses)).where(User.id == 1)
user = session.scalars(stmt).unique().first()
# SELECT ... FROM user LEFT OUTER JOIN address ON ...

Eager Loading with selectinload (Recommended)

from sqlalchemy.orm import selectinload

# Load users, then load all addresses with IN clause
stmt = select(User).options(selectinload(User.addresses))
users = session.scalars(stmt).all()
# SELECT ... FROM user
# SELECT ... FROM address WHERE user_id IN (1, 2, 3, ...)

Raise on Lazy Load (Prevent N+1)

from sqlalchemy.orm import raiseload

# Raise error if lazy loading is attempted
stmt = select(User).options(raiseload(User.addresses))
user = session.scalars(stmt).first()
user.addresses  # Raises InvalidRequestError

Setting Default Loading Strategy

class User(Base):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(primary_key=True)
    # Always eager load addresses
    addresses: Mapped[List["Address"]] = relationship(lazy="selection")

Joins and Complex Queries

from sqlalchemy import select, and_, or_, func

# Explicit JOIN
stmt = (
    select(User, Address)
    .join(Address, User.id == Address.user_id)
    .where(User.name == "alice")
)

# JOIN using relationship
stmt = (
    select(Address)
    .join(Address.user)
    .where(User.name == "alice")
)

# LEFT OUTER JOIN
stmt = select(User).outerjoin(User.addresses)

# Subquery
subq = select(func.count(Address.id)).where(Address.user_id == User.id).scalar_subquery()
stmt = select(User.name, subq.label("address_count"))

# GROUP BY and HAVING
stmt = (
    select(User.name, func.count(Address.id).label("count"))
    .join(User.addresses)
    .group_by(User.name)
    .having(func.count(Address.id) > 1)
)

# UNION
stmt1 = select(User.name).where(User.id < 5)
stmt2 = select(User.name).where(User.id > 10)
stmt = stmt1.union(stmt2)

# EXISTS
from sqlalchemy import exists
subq = select(Address).where(Address.user_id == User.id).exists()
stmt = select(User).where(subq)

# IN with subquery
subq = select(Address.user_id).where(Address.email_address.like("[e-mail]"))
stmt = select(User).where(User.id.in_(subq))

Column Operators

# Comparison
User.name == "alice"
User.id != 5
User.id > 10
User.id >= 10
User.id < 10
User.id <= 10
User.id.between(5, 10)

# NULL checks
User.fullname.is_(None)
User.fullname.is_not(None)

# String operations
User.name.like("a%")       # SQL LIKE
User.name.ilike("a%")      # Case-insensitive LIKE
User.name.startswith("a")
User.name.endswith("z")
User.name.contains("bc")

# IN
User.id.in_([1, 2, 3])
User.id.not_in([1, 2, 3])

# Logical operators
and_(User.name == "alice", User.id > 5)
or_(User.name == "alice", User.name == "bob")
~(User.name == "alice")  # NOT

Transactions

# Automatic transaction management with context manager
with Session(engine) as session:
    with session.begin():
        session.add(User(name="alice"))
        session.add(User(name="bob"))
    # Commits automatically, rolls back on exception

# Manual transaction control
session = Session(engine)
try:
    session.add(User(name="alice"))
    session.commit()
except Exception:
    session.rollback()
    raise
finally:
    session.close()

# Nested transactions (savepoints)
with Session(engine) as session:
    session.add(User(name="alice"))

    with session.begin_nested():  # Creates SAVEPOINT
        session.add(User(name="bob"))
        # Can rollback just this savepoint

    session.commit()

Async Support

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

# Create async engine
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")

# Create async session factory
AsyncSessionFactory = async_sessionmaker(engine, expire_on_commit=False)

async def get_user(user_id: int) -> User | None:
    async with AsyncSessionFactory() as session:
        stmt = select(User).where(User.id == user_id)
        result = await session.execute(stmt)
        return result.scalar_one_or_none()

async def create_user(name: str) -> User:
    async with AsyncSessionFactory() as session:
        user = User(name=name)
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user

# Important: Use selectinload for eager loading in async
async def get_user_with_addresses(user_id: int) -> User | None:
    async with AsyncSessionFactory() as session:
        stmt = (
            select(User)
            .options(selectinload(User.addresses))
            .where(User.id == user_id)
        )
        result = await session.execute(stmt)
        return result.scalar_one_or_none()

Common Patterns

Repository Pattern

from typing import Generic, TypeVar
from sqlalchemy import select
from sqlalchemy.orm import Session

T = TypeVar("T", bound=Base)


class Repository(Generic[T]):
    def __init__(self, session: Session, model: type[T]):
        self.session = session
        self.model = model

    def get(self, id: int) -> T | None:
        return self.session.get(self.model, id)

    def get_all(self) -> list[T]:
        return list(self.session.scalars(select(self.model)).all())

    def add(self, entity: T) -> T:
        self.session.add(entity)
        return entity

    def delete(self, entity: T) -> None:
        self.session.delete(entity)


# Usage
with Session(engine) as session:
    user_repo = Repository(session, User)
    user = user_repo.get(1)
    all_users = user_repo.get_all()

Soft Delete

from datetime import datetime


class SoftDeleteMixin:
    deleted_at: Mapped[Optional[datetime]] = mapped_column(default=None)

    @property
    def is_deleted(self) -> bool:
        return self.deleted_at is not None

    def soft_delete(self) -> None:
        self.deleted_at = datetime.utcnow()


class User(SoftDeleteMixin, Base):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]


# Query only non-deleted
stmt = select(User).where(User.deleted_at.is_(None))

Timestamp Mixin

from datetime import datetime
from sqlalchemy import func


class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        default=func.now(),
        onupdate=func.now()
    )


class User(TimestampMixin, Base):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]

Database Reflection

from sqlalchemy import MetaData, Table

# Reflect existing database schema
metadata = MetaData()
metadata.reflect(bind=engine)

# Access reflected table
users_table = metadata.tables["users"]

# Query reflected table
with engine.connect() as conn:
    result = conn.execute(users_table.select())

Best Practices

1. Always use context managers for Session to ensure proper cleanup
2. Prefer selectinload over joinedload for collection relationships
3. Use raiseload during development to catch N+1 query problems
4. Keep transactions short - commit as soon as the logical unit of work is done
5. Use expire_on_commit=False in async contexts and when passing objects outside session scope
6. Define __repr__ methods on models for easier debugging
7. Use type annotations with Mapped for better IDE support and type checking
8. Index foreign keys and columns used in WHERE clauses
9. Use bulk operations (insert().values([...])) for large datasets
10. Handle sessions per-request in web applications, not globally

Common Column Types

from sqlalchemy import (
    String, Text, Integer, BigInteger, SmallInteger,
    Float, Numeric, Boolean, Date, DateTime, Time,
    LargeBinary, JSON, Enum, UUID
)
from sqlalchemy.dialects.postgresql import ARRAY, JSONB

# Examples
name: Mapped[str] = mapped_column(String(100))
description: Mapped[str] = mapped_column(Text)
price: Mapped[float] = mapped_column(Numeric(10, 2))
data: Mapped[dict] = mapped_column(JSON)
tags: Mapped[list] = mapped_column(ARRAY(String))  # PostgreSQL only

Constraints and Indexes

from sqlalchemy import CheckConstraint, UniqueConstraint, Index


class Product(Base):
    __tablename__ = "product"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)
    price: Mapped[float]
    category: Mapped[str]

    __table_args__ = (
        CheckConstraint("price > 0", name="positive_price"),
        UniqueConstraint("name", "category", name="unique_name_category"),
        Index("idx_category", "category"),
        Index("idx_name_price", "name", "price"),
    )