SQLAlchemy ORM in Python: Relational Databases Step by Step

SQLAlchemy is the most popular Python database library. It offers two APIs: Core (low-level SQL) and ORM (object-relational mapping). This tutorial covers the modern ORM (SQLAlchemy 2.x).

1. Installation and setup

pip install sqlalchemy
pip install alembic           # Migrations
pip install psycopg2-binary   # PostgreSQL
# pip install pymysql         # MySQL

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase

# SQLite (development)
engine = create_engine("sqlite:///app.db", echo=True)

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

class Base(DeclarativeBase):
    pass

2. Define models

from sqlalchemy import String, Integer, Float, Boolean, DateTime, ForeignKey, Text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from datetime import datetime
from typing import Optional, List

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id:         Mapped[int]      = mapped_column(Integer, primary_key=True)
    name:       Mapped[str]      = mapped_column(String(100), nullable=False)
    email:      Mapped[str]      = mapped_column(String(200), unique=True)
    active:     Mapped[bool]     = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)

    orders: Mapped[List["Order"]] = relationship("Order", back_populates="user")

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

class Order(Base):
    __tablename__ = "orders"

    id:         Mapped[int]           = mapped_column(Integer, primary_key=True)
    user_id:    Mapped[int]           = mapped_column(ForeignKey("users.id"))
    total:      Mapped[float]         = mapped_column(Float, nullable=False)
    status:     Mapped[str]           = mapped_column(String(50), default="pending")
    created_at: Mapped[datetime]      = mapped_column(DateTime, default=datetime.now)
    notes:      Mapped[Optional[str]] = mapped_column(Text, nullable=True)

    user:  Mapped["User"]           = relationship("User", back_populates="orders")
    lines: Mapped[List["OrderLine"]] = relationship("OrderLine", back_populates="order")

class OrderLine(Base):
    __tablename__ = "order_lines"

    id:         Mapped[int]   = mapped_column(Integer, primary_key=True)
    order_id:   Mapped[int]   = mapped_column(ForeignKey("orders.id"))
    product:    Mapped[str]   = mapped_column(String(200))
    quantity:   Mapped[int]   = mapped_column(Integer)
    unit_price: Mapped[float] = mapped_column(Float)

    order: Mapped["Order"] = relationship("Order", back_populates="lines")

Base.metadata.create_all(engine)

3. Basic CRUD with Session

from sqlalchemy.orm import Session

# CREATE
with Session(engine) as session:
    user1 = User(name="Alice Smith", email="alice@example.com")
    user2 = User(name="Bob Jones",   email="bob@example.com")
    session.add_all([user1, user2])
    session.commit()
    print(f"Created user with id={user1.id}")

# READ
with Session(engine) as session:
    user = session.get(User, 1)          # By primary key
    print(user)

    users = session.query(User).all()    # All records
    for u in users:
        print(f"{u.id}: {u.name}")

# UPDATE
with Session(engine) as session:
    user = session.get(User, 1)
    user.name = "Alice Johnson"
    session.commit()

# DELETE
with Session(engine) as session:
    user = session.get(User, 2)
    session.delete(user)
    session.commit()

4. Advanced queries

from sqlalchemy import select, func, and_, or_, desc
from sqlalchemy.orm import Session

with Session(engine) as session:
    # WHERE
    stmt = select(User).where(User.active == True)
    active_users = session.scalars(stmt).all()

    # Multiple conditions
    stmt = select(User).where(
        and_(User.active == True, User.name.like("Alice%"))
    )

    # OR
    stmt = select(User).where(
        or_(User.email.endswith("@gmail.com"), User.active == False)
    )

    # Sort and paginate
    stmt = (
        select(User)
        .where(User.active == True)
        .order_by(desc(User.created_at))
        .limit(10)
        .offset(20)
    )

    # Count
    total = session.scalar(select(func.count(User.id)))
    print(f"Total users: {total}")

    # Aggregations
    stmt = select(
        func.count(Order.id).label("total_orders"),
        func.sum(Order.total).label("total_sum"),
        func.avg(Order.total).label("avg_total"),
    )
    result = session.execute(stmt).one()
    print(f"Orders: {result.total_orders}, Sum: {result.total_sum:.2f}")

5. Joins and eager loading

from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload

with Session(engine) as session:
    # Eager load related data
    stmt = (
        select(User)
        .options(selectinload(User.orders))
        .where(User.active == True)
    )
    users = session.scalars(stmt).all()
    for u in users:
        print(f"{u.name}: {len(u.orders)} orders")

    # Explicit JOIN
    stmt = (
        select(User, Order)
        .join(Order, Order.user_id == User.id)
        .where(Order.status == "pending")
    )
    for user, order in session.execute(stmt):
        print(f"{user.name} → Order #{order.id}: ${order.total:.2f}")

    # Subquery: users without orders
    subq = select(Order.user_id).distinct()
    stmt = select(User).where(User.id.not_in(subq))
    no_orders = session.scalars(stmt).all()

6. Many-to-many relationship

from sqlalchemy import Table, Column, Integer, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship

tag_product = Table(
    "tag_product",
    Base.metadata,
    Column("tag_id",     Integer, ForeignKey("tags.id"),     primary_key=True),
    Column("product_id", Integer, ForeignKey("products.id"), primary_key=True),
)

class Tag(Base):
    __tablename__ = "tags"
    id:       Mapped[int] = mapped_column(Integer, primary_key=True)
    name:     Mapped[str] = mapped_column(String(50), unique=True)
    products: Mapped[list["Product"]] = relationship(
        "Product", secondary=tag_product, back_populates="tags"
    )

class Product(Base):
    __tablename__ = "products"
    id:    Mapped[int]   = mapped_column(Integer, primary_key=True)
    name:  Mapped[str]   = mapped_column(String(200))
    price: Mapped[float] = mapped_column(Float)
    tags:  Mapped[list["Tag"]] = relationship(
        "Tag", secondary=tag_product, back_populates="products"
    )

with Session(engine) as session:
    py_tag  = Tag(name="python")
    web_tag = Tag(name="web")
    prod = Product(name="FastAPI Book", price=39.99, tags=[py_tag, web_tag])
    session.add(prod)
    session.commit()

7. Repository pattern

from sqlalchemy.orm import Session
from sqlalchemy import select
from typing import Optional, List

class UserRepository:
    def __init__(self, session: Session):
        self.session = session

    def create(self, name: str, email: str) -> User:
        user = User(name=name, email=email)
        self.session.add(user)
        self.session.flush()
        return user

    def get_by_id(self, id: int) -> Optional[User]:
        return self.session.get(User, id)

    def get_by_email(self, email: str) -> Optional[User]:
        return self.session.scalar(select(User).where(User.email == email))

    def list_active(self, limit: int = 100) -> List[User]:
        return list(self.session.scalars(
            select(User).where(User.active == True).limit(limit)
        ))

    def deactivate(self, id: int) -> bool:
        user = self.get_by_id(id)
        if user:
            user.active = False
            return True
        return False

with Session(engine) as session:
    repo = UserRepository(session)
    new_user = repo.create("Carol", "carol@example.com")
    session.commit()
    print(f"Created: {new_user}")

8. Alembic migrations

alembic init alembic

# Edit alembic/env.py:
# from my_app.models import Base
# target_metadata = Base.metadata

alembic revision --autogenerate -m "create initial tables"
alembic upgrade head
alembic history
alembic downgrade -1

Summary

Feature	SQLAlchemy ORM
Modern ORM	Mapped[type] + mapped_column()
Session	Session(engine) as context manager
Queries	select(Model).where(condition)
Relationships	relationship() with back_populates
Migrations	Alembic autogenerate
Async	AsyncSession + create_async_engine