Models, sessions, and queries

# PostgreSQL, synchronous
pip install sqlalchemy "psycopg[binary]"

# PostgreSQL, asynchronous
pip install sqlalchemy asyncpg

# SQLite is included with Python
pip install sqlalchemy

CREATE TABLE user_account (
  id INTEGER PRIMARY KEY,
  name VARCHAR(50) NOT NULL,
  email VARCHAR(255) UNIQUE NOT NULL
);

from sqlalchemy import URL, create_engine

url = URL.create(
    "postgresql+psycopg",
    username="app", password="p@ss/word",
    host="db.internal", port=5432, database="shop",
)
engine = create_engine(
    url,
    echo=False,           # SQL logging while learning
    pool_size=10, max_overflow=20,
    pool_pre_ping=True,   # test stale connection on checkout
    pool_recycle=1800,
)

sqlite = create_engine("sqlite:///app.db")

-- Engine opens/reuses a connection. It does not query by itself.
BEGIN;
SELECT 1;
COMMIT;

from datetime import datetime
from sqlalchemy import DateTime, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), index=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    nickname: Mapped[str | None] = mapped_column(nullable=True)

Base.metadata.create_all(engine)  # development/tests only

CREATE TABLE user_account (
  id SERIAL PRIMARY KEY,
  name VARCHAR(50) NOT NULL,
  email VARCHAR(255) NOT NULL UNIQUE,
  is_active BOOLEAN NOT NULL,
  created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(),
  nickname VARCHAR
);
CREATE INDEX ix_user_account_name ON user_account (name);

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)

# Commits on normal exit; rolls back if the block raises.
with SessionLocal.begin() as session:
    user = User(name="Ada", email="[email protected]")
    session.add(user)
    # flush happens before commit; user.id is available after flush

# Explicit alternative
with SessionLocal() as session:
    try:
        session.add(User(name="Lin", email="[email protected]"))
        session.commit()
    except:
        session.rollback()
        raise

BEGIN;
INSERT INTO user_account (name, email, is_active)
VALUES ('Ada', '[email protected]', true);
COMMIT;

-- Any exception before COMMIT:
ROLLBACK;

session.flush()               # sends pending INSERT/UPDATE/DELETE
session.refresh(user)         # SELECT current row
session.expire(user)          # reload attributes on next access
attached = session.merge(detached_user)  # copy state into this Session

with session.begin_nested():  # SAVEPOINT
    session.add(optional_audit_record())

SELECT * FROM user_account WHERE id = :id;
SAVEPOINT sa_savepoint_1;
INSERT INTO audit_log (message) VALUES (:message);
RELEASE SAVEPOINT sa_savepoint_1;

from sqlalchemy import select

with SessionLocal.begin() as session:
    # CREATE
    user = User(name="Ada", email="[email protected]")
    session.add(user)

    # READ by primary key (identity-map aware)
    same_user = session.get(User, user.id)

    # UPDATE
    same_user.name = "Ada Lovelace"

    # DELETE
    session.delete(same_user)

INSERT INTO user_account (name, email, is_active)
VALUES ('Ada', '[email protected]', true);
SELECT * FROM user_account WHERE id = 1;
UPDATE user_account SET name = 'Ada Lovelace' WHERE id = 1;
DELETE FROM user_account WHERE id = 1;

stmt = select(User).where(User.email == "[email protected]")

first_or_none = session.scalar(stmt)             # first scalar or None
exactly_one = session.scalars(stmt).one()        # raises for 0 or 2+
zero_or_one = session.scalars(stmt).one_or_none() # raises for 2+
all_users = session.scalars(select(User)).all()  # list[User]

SELECT id, name, email, is_active, created_at, nickname
FROM user_account WHERE email = '[email protected]';

from sqlalchemy import select

users = session.scalars(
    select(User).where(User.is_active)
).all()

stmt = select(User.id, User.name).order_by(User.name)
for row in session.execute(stmt):
    print(row.id, row.name)        # attribute
    print(row[0], row[1])          # tuple
    print(row._mapping["name"])    # mapping

rows = session.execute(stmt).mappings().all()
# [{"id": 1, "name": "Ada"}, ...]

SELECT id, name, email, is_active, created_at, nickname
FROM user_account WHERE is_active = true;

SELECT id, name FROM user_account ORDER BY name ASC;

from sqlalchemy import func, select

user = session.get(User, 42)  # primary key lookup; may use identity map
user = session.execute(
    select(User).where(User.email == "[email protected]")
).scalar_one_or_none()

rows = session.execute(
    select(User.id, User.name).order_by(User.id).limit(10)
).mappings().all()  # keys: "id", "name"; values are not User objects

total = session.scalar(select(func.count()).select_from(User))
assert total is not None

SELECT * FROM user_account WHERE id = 42;
SELECT * FROM user_account WHERE email = :email;
SELECT id, name FROM user_account ORDER BY id LIMIT 10;
SELECT count(*) FROM user_account;

from sqlalchemy import func, select

# Users who have an address, ordered by that address.
stmt = (select(User).join(User.addresses)
        .where(Address.email.like("%@example.com"))
        .order_by(Address.email))
users = session.scalars(stmt).unique().all()

# Keep users without addresses too; select both objects.
stmt = select(User, Address).outerjoin(User.addresses)
for user, address in session.execute(stmt):
    print(user.name, address.email if address else None)

# Count distinct parents after joining a collection.
count = session.scalar(select(func.count(func.distinct(User.id)))
                       .join(User.addresses))

SELECT u.* FROM user_account AS u
JOIN address AS a ON a.user_id = u.id
WHERE a.email LIKE '%@example.com' ORDER BY a.email;

SELECT u.*, a.* FROM user_account AS u
LEFT OUTER JOIN address AS a ON a.user_id = u.id;

SELECT count(DISTINCT u.id) FROM user_account AS u
JOIN address AS a ON a.user_id = u.id;

from sqlalchemy.orm import aliased

manager = aliased(Employee, name="manager")
stmt = (select(Employee.name, manager.name.label("manager_name"))
        .join(manager, Employee.manager_id == manager.id))

home = aliased(Address, name="home")
stmt = (select(User, home)
        .join(User.addresses.of_type(home))
        .where(home.email.like("%@home.test")))

# With multiple FROMs, make the join's starting point explicit.
stmt = select(User.name, Address.email).join_from(User, Address)

SELECT employee.name, manager.name AS manager_name
FROM employee JOIN employee AS manager
ON employee.manager_id = manager.id;

SELECT u.*, home.* FROM user_account AS u
JOIN address AS home ON u.id = home.user_id
WHERE home.email LIKE '%@home.test';

SELECT u.name, a.email FROM user_account AS u
JOIN address AS a ON u.id = a.user_id;

stmt = (
    select(User)
    .where(
        User.is_active,
        User.name.ilike("a%"),
        User.id.in_([1, 2, 3]),
        User.nickname.is_not(None),
        (User.name == "Ada") | (User.name == "Grace"),
    )
    .order_by(User.created_at.desc(), User.id.asc())
    .limit(20).offset(40)
)

SELECT * FROM user_account
WHERE is_active = true
  AND name ILIKE 'a%'
  AND id IN (1, 2, 3)
  AND nickname IS NOT NULL
  AND (name = 'Ada' OR name = 'Grace')
ORDER BY created_at DESC, id ASC
LIMIT 20 OFFSET 40;

from sqlalchemy import and_, or_

stmt = select(User).order_by(User.created_at.desc(), User.id.desc()).limit(50)
if cursor:
    stmt = stmt.where(or_(
        User.created_at < cursor.created_at,
        and_(User.created_at == cursor.created_at, User.id < cursor.id),
    ))

SELECT * FROM user_account
WHERE created_at < :last_created_at
   OR (created_at = :last_created_at AND id < :last_id)
ORDER BY created_at DESC, id DESC
LIMIT 50;

from sqlalchemy import or_, select

stmt = select(User)
if term:
    stmt = stmt.where(or_(User.name.ilike(f"%{term}%"),
                           User.email.ilike(f"%{term}%")))
if ids is not None:
    stmt = stmt.where(User.id.in_(ids))  # [] yields no matches
stmt = stmt.where(User.nickname.is_(None), User.is_active.is_(True))

# Values are bound as parameters; escape %, _ yourself if term is literal.
users = session.scalars(stmt.order_by(User.id)).all()

SELECT * FROM user_account
WHERE (name ILIKE :pattern OR email ILIKE :pattern)
  AND id IN (:id_1, :id_2)  -- illustrative nonempty list
  AND nickname IS NULL AND is_active IS TRUE
ORDER BY id;
-- An empty IN list is compiled as an always-false condition.

from sqlalchemy import func, select

stmt = (select(User.id, User.name,
               func.count(Address.id).label("address_count"))
        .outerjoin(User.addresses)
        .group_by(User.id, User.name)
        .having(func.count(Address.id) >= 2)
        .order_by(func.count(Address.id).desc(), User.id)
        .limit(20))
for user_id, name, n in session.execute(stmt):
    print(user_id, name, n)

SELECT u.id, u.name, count(a.id) AS address_count
FROM user_account AS u
LEFT JOIN address AS a ON a.user_id = u.id
GROUP BY u.id, u.name
HAVING count(a.id) >= 2
ORDER BY count(a.id) DESC, u.id LIMIT 20;

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import relationship

class Address(Base):
    __tablename__ = "address"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255))
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    user: Mapped["User"] = relationship(back_populates="addresses")

class User(Base):
    __tablename__ = "user_account"
    # mapped columns omitted
    addresses: Mapped[list[Address]] = relationship(
        back_populates="user", cascade="all, delete-orphan"
    )

user.addresses.append(Address(email="[email protected]"))

CREATE TABLE address (
  id SERIAL PRIMARY KEY,
  email VARCHAR(255) NOT NULL,
  user_id INTEGER NOT NULL REFERENCES user_account(id)
);
INSERT INTO address (email, user_id)
VALUES ('[email protected]', :user_id);

# JOIN via a declared relationship
stmt = select(User).join(User.addresses).where(
    Address.email.endswith("@example.com")
)
users = session.scalars(stmt).unique().all()

# EXISTS expressions without spelling the join
work_users = select(User).where(
    User.addresses.any(Address.email.like("%@work.com"))
)
adas_addresses = select(Address).where(
    Address.user.has(User.name == "Ada")
)
# Users with no addresses; keep one row per User.
no_addresses = session.scalars(
    select(User).where(~User.addresses.any())
).all()
# A known parent instance is compared using its primary key.
owned = select(Address).where(Address.user == user)

SELECT u.* FROM user_account AS u
JOIN address AS a ON u.id = a.user_id
WHERE a.email LIKE '%@example.com';

SELECT u.* FROM user_account AS u WHERE EXISTS (
  SELECT 1 FROM address a
  WHERE a.user_id = u.id AND a.email LIKE '%@work.com'
);

SELECT a.* FROM address AS a WHERE EXISTS (
  SELECT 1 FROM user_account AS u
  WHERE u.id = a.user_id AND u.name = 'Ada'
);

SELECT u.* FROM user_account AS u WHERE NOT EXISTS (
  SELECT 1 FROM address AS a WHERE a.user_id = u.id
);
SELECT a.* FROM address AS a WHERE a.user_id = :user_id;

from sqlalchemy import Column, Table

user_role = Table(
    "user_role", Base.metadata,
    Column("user_id", ForeignKey("user_account.id"), primary_key=True),
    Column("role_id", ForeignKey("role.id"), primary_key=True),
)
class Role(Base):
    __tablename__ = "role"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40), unique=True)
    users: Mapped[list["User"]] = relationship(
        secondary=user_role, back_populates="roles"
    )
User.roles = relationship(Role, secondary=user_role, back_populates="users")

CREATE TABLE user_role (
  user_id INTEGER REFERENCES user_account(id),
  role_id INTEGER REFERENCES role(id),
  PRIMARY KEY (user_id, role_id)
);

class User(Base):
    __tablename__ = "user_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    profile: Mapped["Profile | None"] = relationship(back_populates="user")

class Profile(Base):
    __tablename__ = "profile"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(
        ForeignKey("user_account.id"), unique=True
    )
    bio: Mapped[str] = mapped_column(String(200))
    user: Mapped[User] = relationship(back_populates="profile")

user.profile = Profile(bio="Database learner")

CREATE TABLE profile (
  id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  user_id INTEGER NOT NULL UNIQUE REFERENCES user_account(id),
  bio VARCHAR(200) NOT NULL
);
INSERT INTO profile (user_id, bio) VALUES (:user_id, 'Database learner');

from sqlalchemy.ext.associationproxy import association_proxy

class Membership(Base):
    __tablename__ = "membership"
    user_id: Mapped[int] = mapped_column(
        ForeignKey("user_account.id"), primary_key=True)
    group_id: Mapped[int] = mapped_column(
        ForeignKey("group_account.id"), primary_key=True)
    role: Mapped[str] = mapped_column(String(30))
    user: Mapped["User"] = relationship(back_populates="memberships")
    group: Mapped["Group"] = relationship(back_populates="memberships")

class User(Base):
    __tablename__ = "user_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    memberships: Mapped[list[Membership]] = relationship(back_populates="user")
    groups = association_proxy("memberships", "group",
                               creator=lambda g: Membership(group=g, role="member"))

class Group(Base):
    __tablename__ = "group_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    memberships: Mapped[list[Membership]] = relationship(back_populates="group")

user.memberships.append(Membership(group=group, role="admin"))

CREATE TABLE membership (
  user_id INTEGER NOT NULL REFERENCES user_account(id),
  group_id INTEGER NOT NULL REFERENCES group_account(id),
  role VARCHAR(30) NOT NULL,
  PRIMARY KEY (user_id, group_id)
);
INSERT INTO membership (user_id, group_id, role)
VALUES (:user_id, :group_id, 'admin');

class Node(Base):
    __tablename__ = "node"
    id: Mapped[int] = mapped_column(primary_key=True)
    parent_id: Mapped[int | None] = mapped_column(ForeignKey("node.id"))
    parent: Mapped["Node | None"] = relationship(
        back_populates="children", remote_side=[id]
    )
    children: Mapped[list["Node"]] = relationship(back_populates="parent")

class Shipment(Base):
    __tablename__ = "shipment"
    id: Mapped[int] = mapped_column(primary_key=True)
    sender_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    recipient_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    sender: Mapped["User"] = relationship(foreign_keys=[sender_id])
    recipient: Mapped["User"] = relationship(foreign_keys=[recipient_id])

CREATE TABLE node (
  id INTEGER PRIMARY KEY,
  parent_id INTEGER NULL REFERENCES node(id)
);
SELECT child.* FROM node AS child WHERE child.parent_id = :parent_id;

CREATE TABLE shipment (
  id INTEGER PRIMARY KEY,
  sender_id INTEGER NOT NULL REFERENCES user_account(id),
  recipient_id INTEGER NOT NULL REFERENCES user_account(id)
);
SELECT s.*, sender.*, recipient.* FROM shipment AS s
JOIN user_account AS sender ON sender.id = s.sender_id
JOIN user_account AS recipient ON recipient.id = s.recipient_id;

class Project(Base):
    __tablename__ = "project"
    id: Mapped[int] = mapped_column(primary_key=True)
    tasks: Mapped[list["Task"]] = relationship(
        back_populates="project", cascade="all, delete-orphan",
        passive_deletes=True,
    )

class Task(Base):
    __tablename__ = "task"
    id: Mapped[int] = mapped_column(primary_key=True)
    project_id: Mapped[int] = mapped_column(
        ForeignKey("project.id", ondelete="CASCADE")
    )
    project: Mapped[Project] = relationship(back_populates="tasks")

project.tasks.remove(task)  # ORM emits DELETE for this orphan on flush
session.delete(project)     # database deletes unloaded tasks on DELETE

CREATE TABLE project (id INTEGER PRIMARY KEY);
CREATE TABLE task (
  id INTEGER PRIMARY KEY,
  project_id INTEGER NOT NULL REFERENCES project(id) ON DELETE CASCADE
);
DELETE FROM task WHERE id = :removed_task_id; -- orphan removal
DELETE FROM project WHERE id = :project_id;   -- DB cascades tasks

from sqlalchemy import ForeignKeyConstraint, and_

class User(Base):
    __tablename__ = "user_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    addresses: Mapped[list["Address"]] = relationship(back_populates="user")
    work_addresses: Mapped[list["Address"]] = relationship(
        primaryjoin=lambda: and_(User.id == Address.user_id,
                                 Address.kind == "work"),
        viewonly=True,
    )

class OrderLine(Base):
    __tablename__ = "order_line"
    order_id: Mapped[int] = mapped_column(primary_key=True)
    line_no: Mapped[int] = mapped_column(primary_key=True)
    tenant_id: Mapped[int] = mapped_column()
    product_id: Mapped[int] = mapped_column()
    __table_args__ = (ForeignKeyConstraint(
        ["tenant_id", "product_id"], ["product.tenant_id", "product.id"]
    ),)

SELECT a.* FROM address AS a
WHERE a.user_id = :user_id AND a.kind = 'work';

CREATE TABLE order_line (
  order_id INTEGER NOT NULL, line_no INTEGER NOT NULL,
  tenant_id INTEGER NOT NULL, product_id INTEGER NOT NULL,
  PRIMARY KEY (order_id, line_no),
  FOREIGN KEY (tenant_id, product_id)
    REFERENCES product(tenant_id, id)
);

from sqlalchemy.orm import joinedload, raiseload, selectinload

# Usually best for a collection: two predictable queries
users = session.scalars(
    select(User).options(selectinload(User.addresses))
).all()

# Good for a scalar many-to-one or one-to-one
addresses = session.scalars(
    select(Address).options(joinedload(Address.user))
).all()

# Development/async guard: unexpected lazy access raises
stmt = select(User).options(raiseload("*"))

-- selectinload(User.addresses)
SELECT u.* FROM user_account u;
SELECT a.* FROM address a WHERE a.user_id IN (1, 2, 3, ...);

-- joinedload(Address.user)
SELECT a.*, u.* FROM address a
LEFT OUTER JOIN user_account u ON u.id = a.user_id;

from sqlalchemy.orm import joinedload, selectinload

stmt = select(User).options(
    selectinload(User.roles).selectinload(Role.users)
)
users = session.scalars(stmt).all()  # user, then address, then message SQL

# Collection joinedload duplicates SQL rows; unique() collapses ORM parents.
stmt = select(User).options(joinedload(User.addresses))
users = session.scalars(stmt).unique().all()

# Filter on Address separately from the loader choice.
stmt = (select(User).join(User.addresses)
        .where(Address.email.like("%@work.com"))
        .options(selectinload(User.addresses)))

SELECT * FROM user_account;
SELECT r.* FROM role AS r JOIN user_role ur ON ur.role_id = r.id
WHERE ur.user_id IN (:id_1, :id_2, ...);
SELECT u.* FROM user_account AS u JOIN user_role ur ON ur.user_id = u.id
WHERE ur.role_id IN (:role_1, :role_2, ...);

SELECT u.*, a.* FROM user_account AS u
LEFT OUTER JOIN address AS a ON a.user_id = u.id;

SELECT u.* FROM user_account AS u
JOIN address AS a ON a.user_id = u.id WHERE a.email LIKE '%@work.com';
-- selectinload then separately fetches all addresses for matched users.

from sqlalchemy.orm import contains_eager, selectinload

stmt = (select(User).options(
            selectinload(User.addresses.and_(Address.email.like("%@work.com")))
        ).execution_options(populate_existing=True))
users = session.scalars(stmt).all()

stmt = (select(User).join(User.addresses)
        .where(Address.email.like("%@work.com"))
        .options(contains_eager(User.addresses))
        .execution_options(populate_existing=True))
users = session.scalars(stmt).unique().all()

SELECT u.* FROM user_account AS u;
SELECT a.* FROM address AS a
WHERE a.user_id IN (:id_1, :id_2, ...) AND a.email LIKE '%@work.com';

SELECT u.*, a.* FROM user_account AS u
JOIN address AS a ON a.user_id = u.id
WHERE a.email LIKE '%@work.com';

from sqlalchemy.orm import (WriteOnlyMapped, load_only, raiseload,
                            selectinload, with_loader_criteria)

stmt = select(User).options(
    load_only(User.id, User.name, raiseload=True),
    selectinload(User.addresses),
    with_loader_criteria(Address, Address.email.like("%@example.com")),
)
users = session.scalars(stmt).all()

# Alternative mapping for a very large child collection:
class AuditLog(Base):
    __tablename__ = "audit_log"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))

class User(Base):
    __tablename__ = "user_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    logs: WriteOnlyMapped[AuditLog] = relationship()

logs = session.scalars(user.logs.select().limit(50)).all()
user.logs.add(AuditLog())

SELECT id, name FROM user_account;
SELECT a.* FROM address AS a
WHERE a.user_id IN (:id_1, :id_2, ...) AND a.email LIKE '%@example.com';

SELECT * FROM audit_log WHERE user_id = :user_id LIMIT 50;
INSERT INTO audit_log (user_id) VALUES (:user_id);