SQLAlchemy Cheatsheet 04 — select(), where, joins, aggregates

from sqlalchemy import select

# All columns
stmt = select(User)

# Specific columns
stmt = select(User.id, User.email)

# Distinct
stmt = select(User.email).distinct()

result = await session.execute(stmt)

result.scalars().all()           # for select(User) → list[User]
result.scalars().first()
result.scalar_one()              # exactly 1; raises 0 or 2+
result.scalar_one_or_none()      # 0 or 1
result.all()                     # for multi-column → list[Row]
result.first()
result.one()
result.one_or_none()
result.mappings().all()          # list[dict]

stmt = (
    select(User)
    .where(User.email == "x")
    .where(User.active == True)               # AND
)

# OR / NOT
from sqlalchemy import or_, not_, and_

stmt = select(User).where(or_(User.role == "admin", User.id == 1))
stmt = select(User).where(not_(User.active))

# IN / NOT IN
stmt = select(User).where(User.id.in_([1, 2, 3]))
stmt = select(User).where(User.id.notin_([1, 2, 3]))

# LIKE / ILIKE
stmt = select(User).where(User.email.like("%@example.com"))
stmt = select(User).where(User.email.ilike("%@example.com"))   # PG / MySQL

# Null / not null
stmt = select(User).where(User.deleted_at.is_(None))
stmt = select(User).where(User.deleted_at.is_not(None))

# Between
stmt = select(Event).where(Event.ts.between(t1, t2))

# Contains / startswith / endswith
stmt = select(Post).where(Post.title.contains("foo"))
stmt = select(Post).where(Post.title.startswith("foo"))

stmt = (
    select(User)
    .order_by(User.created_at.desc(), User.id)
    .limit(20)
    .offset(40)
)

stmt = (
    select(User)
    .where(User.id > cursor)
    .order_by(User.id)
    .limit(20)
)

stmt = select(Post, User).join(User, Post.author_id == User.id)

# Via relationship (cleaner)
stmt = select(Post).join(Post.author)

# Outer join
stmt = select(Post).outerjoin(Comment).where(Comment.id == None)

# Aliasing
from sqlalchemy.orm import aliased
a = aliased(User); b = aliased(User)
stmt = select(a, b).join(b, a.referrer_id == b.id)

from sqlalchemy import func

# Count
total = await session.scalar(select(func.count(User.id)))

# Group by
stmt = (
    select(User.role, func.count(User.id).label("n"))
    .group_by(User.role)
)

# Having
stmt = (
    select(User.role, func.count(User.id).label("n"))
    .group_by(User.role)
    .having(func.count(User.id) > 10)
)

# Aggregates
func.sum(...)   func.avg(...)   func.min(...)   func.max(...)

sub = (
    select(Post.author_id, func.count(Post.id).label("posts"))
    .group_by(Post.author_id)
    .subquery()
)

stmt = (
    select(User, sub.c.posts)
    .join(sub, User.id == sub.c.author_id)
)

recent = (
    select(Post)
    .where(Post.created_at > datetime(2026, 1, 1))
    .cte("recent_posts")
)

stmt = select(User, recent).join(recent, User.id == recent.c.author_id)

roots = (
    select(Category)
    .where(Category.parent_id == None)
    .cte("cats", recursive=True)
)
ext = roots.union_all(
    select(Category).join(roots, Category.parent_id == roots.c.id)
)
stmt = select(ext)

from sqlalchemy import exists

stmt = select(User).where(
    select(Post).where(Post.author_id == User.id).exists()
)

stmt = select(
    Post.author_id,
    Post.title,
    func.row_number().over(
        partition_by=Post.author_id,
        order_by=Post.created_at.desc(),
    ).label("rn"),
)

from sqlalchemy import update, delete

await session.execute(
    update(User).where(User.email == "x").values(active=False)
)
await session.execute(
    delete(User).where(User.id == 5)
)
await session.commit()

stmt = (
    update(User)
    .where(User.id == 1)
    .values(name="x")
    .returning(User.id, User.name)
)
result = await session.execute(stmt)
print(result.all())

from sqlalchemy import insert

await session.execute(insert(User).values(email="[email protected]"))
await session.execute(insert(User), [{"email": ...}, ...])   # bulk

from sqlalchemy.dialects.postgresql import insert as pg_insert

stmt = pg_insert(User).values(email="[email protected]", name="Alice")
stmt = stmt.on_conflict_do_update(
    index_elements=["email"],
    set_={"name": stmt.excluded.name},
)

from sqlalchemy import text
stmt = text("SELECT * FROM users WHERE email = :email")
result = await session.execute(stmt, {"email": "[email protected]"})

print(stmt.compile(engine, compile_kwargs={"literal_binds": True}))