Queries with select(): filter, join, group_by

Description

select() is the universal query constructor in SQLAlchemy 2.0. It replaces 1.x's session.query(). Any SQL query you wrote in modules 2-3 can be expressed with select() — JOINs, subqueries, aggregations, window functions, all of it.

This capsule teaches you to build complex queries with the modern API. By the end, you will be able to translate any module 2 query into SQLAlchemy.

Basic syntax

from sqlalchemy import select
from app.models import User

stmt = select(User)
# SELECT users.id, users.email, ... FROM users

result = session.execute(stmt)
users = result.scalars().all()

select(User) returns User objects. If you want tuples (specific columns):

stmt = select(User.username, User.email)
# SELECT users.username, users.email FROM users

result = session.execute(stmt)
rows = result.all()
# [Row(username='maria', email='maria@blog.local'), ...]

WHERE: filtering

stmt = select(User).where(User.is_active == True)
# WHERE users.is_active = TRUE

# Multiple conditions (implicit AND)
stmt = select(User).where(
    User.is_active == True,
    User.email_verified == True,
)
# WHERE users.is_active = TRUE AND users.email_verified = TRUE

# Explicit AND/OR
from sqlalchemy import and_, or_

stmt = select(User).where(
    or_(
        User.is_admin == True,
        and_(User.is_active == True, User.email_verified == True),
    )
)

Operators

⚠️ Never use User.deleted_at == None — Python reads that as a special value. Always User.deleted_at.is_(None).

ORDER BY, LIMIT, OFFSET

stmt = (
    select(Post)
    .where(Post.published == True)
    .order_by(Post.published_at.desc())
    .limit(10)
    .offset(20)  # pagination
)

.desc() and .asc() for the direction (default: ASC).

Multiple columns:

.order_by(Post.published.desc(), Post.published_at.desc())

JOINs

INNER JOIN

stmt = (
    select(Post)
    .join(User, Post.author_id == User.id)
    .where(User.username == "maria")
)
# SELECT posts.* FROM posts INNER JOIN users ON posts.author_id = users.id WHERE users.username = 'maria'

If you have the relationship() configured, you can simplify:

stmt = (
    select(Post)
    .join(Post.author)  # uses the relationship
    .where(User.username == "maria")
)

LEFT JOIN (outer)

stmt = (
    select(User, Post)
    .outerjoin(Post, Post.author_id == User.id)
)
# LEFT OUTER JOIN

Or with the relationship:

stmt = select(User).outerjoin(User.posts)

Multiple joins

stmt = (
    select(Post.title, User.username, Category.name)
    .join(Post.author)
    .outerjoin(Post.category)
    .where(Post.published == True)
    .order_by(Post.published_at.desc())
)

SELECT with multiple entities / columns

# Tuples of objects
stmt = select(Post, User).join(Post.author)
result = session.execute(stmt).all()
# [(Post, User), (Post, User), ...]

# Mixing columns and entities
stmt = select(Post, User.username).join(Post.author)
# [(Post, "maria"), ...]

# Columns only
stmt = select(Post.title, Post.published_at)
# [Row(title=..., published_at=...), ...]

To extract:

result = session.execute(stmt)

# If it is ONE model: scalars()
posts = result.scalars().all()

# If it is MULTIPLE things: iterate the Rows
for post, username in session.execute(stmt):
    print(f"{username}: {post.title}")

Aggregations: COUNT, SUM, AVG, GROUP BY

from sqlalchemy import func

# Total published posts
total = session.execute(
    select(func.count(Post.id)).where(Post.published == True)
).scalar_one()

# Average title length
avg_len = session.execute(
    select(func.avg(func.length(Post.title)))
).scalar_one()

# Posts per author
stmt = (
    select(User.username, func.count(Post.id).label("num_posts"))
    .outerjoin(User.posts)
    .group_by(User.id, User.username)
    .order_by(func.count(Post.id).desc())
)

for username, num in session.execute(stmt):
    print(f"{username}: {num} posts")

func.X() reaches any SQL function: func.count(), func.avg(), func.now(), func.lower(), func.coalesce(), func.string_agg(), etc.

HAVING

stmt = (
    select(User.username, func.count(Post.id).label("n"))
    .join(User.posts)
    .group_by(User.id, User.username)
    .having(func.count(Post.id) >= 2)
)
# HAVING count(...) >= 2

FILTER (PostgreSQL-specific)

stmt = (
    select(
        User.username,
        func.count(Post.id).label("total"),
        func.count(Post.id).filter(Post.published == True).label("published"),
        func.count(Post.id).filter(Post.published == False).label("drafts"),
    )
    .outerjoin(User.posts)
    .group_by(User.id, User.username)
)

func.count(...).filter(condition) translates to count(...) FILTER (WHERE ...).

Subqueries

from sqlalchemy import select

# A scalar subquery
posts_count_per_user = (
    select(func.count(Post.id))
    .where(Post.author_id == User.id)
    .scalar_subquery()
)

stmt = select(User.username, posts_count_per_user.label("num_posts"))

IN with a subquery

admin_ids = select(User.id).where(User.is_admin == True)

stmt = select(Post).where(Post.author_id.in_(admin_ids))

EXISTS

from sqlalchemy import exists

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

CTEs (WITH ... AS)

# The CTE
post_counts = (
    select(Post.author_id, func.count().label("n"))
    .group_by(Post.author_id)
    .cte("post_counts")
)

# The main query using the CTE
stmt = (
    select(User.username, post_counts.c.n)
    .join(post_counts, User.id == post_counts.c.author_id)
)

post_counts.c.n reaches the CTE's n column (the c is like Table.c.column).

A recursive CTE

# A comment tree for a specific post
top_comments = (
    select(Comment)
    .where(Comment.post_id == post_id, Comment.parent_comment_id.is_(None))
    .cte("comment_tree", recursive=True)
)

# The "anchor" + the recursive part
top_comments = top_comments.union_all(
    select(Comment).join(top_comments, Comment.parent_comment_id == top_comments.c.id)
)

stmt = select(top_comments)

Recursive CTEs are rare but powerful. For the blog, a library like sqlalchemy-utils or direct SQL queries can read better.

Useful functions for the blog

Top N per category (a window function)

from sqlalchemy import func

# Posts: for each category, the 3 most recent
ranked = (
    select(
        Post,
        func.row_number().over(
            partition_by=Post.category_id,
            order_by=Post.published_at.desc(),
        ).label("rn"),
    )
    .where(Post.published == True)
    .subquery()
)

stmt = select(ranked).where(ranked.c.rn <= 3)

Case-insensitive search with an index

# Assumes idx_users_email_lower (module 3)
stmt = select(User).where(func.lower(User.email) == "maria@blog.local".lower())

Important: the expression must match the index. func.lower(User.email) == "maria@blog.local".lower() matches idx_users_email_lower.

Aggregating tags as a string (STRING_AGG)

from sqlalchemy import func

stmt = (
    select(
        Post.title,
        func.string_agg(Tag.name, ', ').label("tags"),
    )
    .join(Post.tags)
    .group_by(Post.id, Post.title)
)

Most commented posts

stmt = (
    select(
        Post.title,
        func.count(Comment.id).label("num_comments"),
    )
    .outerjoin(Post.comments)
    .where(Post.published == True)
    .group_by(Post.id, Post.title)
    .order_by(func.count(Comment.id).desc())
    .limit(10)
)

Every module 2 query, translated

Let's recall the 10 product queries from module 2, capsule 08. Here are some of them in SQLAlchemy:

Query 1: Main feed with author + category + tags

from sqlalchemy import func, select
from sqlalchemy.orm import selectinload  # capsule 07

stmt = (
    select(Post)
    .options(
        selectinload(Post.author),
        selectinload(Post.category),
        selectinload(Post.tags),
    )
    .where(Post.published == True)
    .order_by(Post.published_at.desc())
    .limit(10)
)

posts = session.execute(stmt).scalars().all()

for p in posts:
    print(f"{p.title} — {p.author.username} — {p.category.name if p.category else 'No cat'}")
    print(f"  Tags: {', '.join(t.name for t in p.tags)}")

Query 3: Top 5 authors by published posts

stmt = (
    select(
        User.username,
        User.full_name,
        func.count(Post.id).label("published_posts"),
        func.max(Post.published_at).label("last_post"),
    )
    .join(User.posts)
    .where(Post.published == True)
    .group_by(User.id, User.username, User.full_name)
    .order_by(func.count(Post.id).desc())
    .limit(5)
)

for username, full_name, count, last in session.execute(stmt):
    print(f"{username} ({full_name}): {count} posts, latest: {last}")

Query 4: Top 10 tags

stmt = (
    select(
        Tag.name,
        func.count(post_tags_table.c.post_id).label("usage_count"),
    )
    .join(post_tags_table, Tag.id == post_tags_table.c.tag_id)
    .group_by(Tag.id, Tag.name)
    .order_by(func.count().desc())
    .limit(10)
)

Query 8: Related posts (the same tag)

post_x = session.execute(select(Post).where(Post.slug == "postgresql-in-production")).scalar_one()
post_x_tag_ids = select(post_tags_table.c.tag_id).where(post_tags_table.c.post_id == post_x.id)

stmt = (
    select(Post, func.count(post_tags_table.c.tag_id).label("shared_tags"))
    .join(post_tags_table, Post.id == post_tags_table.c.post_id)
    .where(
        post_tags_table.c.tag_id.in_(post_x_tag_ids),
        Post.id != post_x.id,
        Post.published == True,
    )
    .group_by(Post.id)
    .order_by(func.count().desc())
    .limit(5)
)

We organize the complete blog in SQLAlchemy in capsule 08, with the Repository pattern.

Running raw SQL when you need it

Sometimes it is easier to write SQL directly. SQLAlchemy allows it:

from sqlalchemy import text

stmt = text("""
    SELECT username, count(*) AS n
    FROM users u
    JOIN posts p ON p.author_id = u.id
    WHERE p.published = true
    GROUP BY username
    ORDER BY n DESC
    LIMIT :limit
""").bindparams(limit=5)

result = session.execute(stmt)
for row in result:
    print(row.username, row.n)

Don't overuse text() — you lose the typing and the IDE stops helping you. Use it only when the ORM really falls short.

Composing reusable queries

SQLAlchemy queries are objects. You can pass them as arguments, modify them, reuse them:

def published_posts_query():
    """The base query for published posts."""
    return select(Post).where(Post.published == True)

def by_author(stmt, username):
    """Filters a query by author."""
    return stmt.join(Post.author).where(User.username == username)

def order_recent(stmt):
    """Orders by date descending."""
    return stmt.order_by(Post.published_at.desc())

# Composable usage
stmt = order_recent(by_author(published_posts_query(), "maria")).limit(10)
posts = session.execute(stmt).scalars().all()

This is NOT possible with text() — it is one of the ORM's advantages.

Common mistakes

MultipleResultsFound

session.execute(select(User)).scalar_one()
# ERROR if there is >1 user

The fix: use scalars().first() or scalars().all() if you expect multiple.

NoResultFound

session.execute(select(User).where(User.username == "doesnotexist")).scalar_one()
# ERROR if there is no match

The fix: scalar_one_or_none().

Forgetting .scalars()

result = session.execute(select(User)).all()
# Returns [Row(User=<User>), ...] — tuples with one element

result = session.execute(select(User)).scalars().all()
# Returns [<User>, <User>, ...] — a flat list

== with None

# ❌
.where(User.deleted_at == None)
# It works but it is a Python anti-pattern (linters complain)

# ✅
.where(User.deleted_at.is_(None))

Exercises

Exercise 1. Implement the plain-SQL "main feed" query as a select() with SQLAlchemy. Verify it returns the same results.

from sqlalchemy import select
from sqlalchemy.orm import selectinload
from app.database import db_session
from app.models import Post

with db_session() as session:
    posts = session.execute(
        select(Post)
        .options(selectinload(Post.author), selectinload(Post.category))
        .where(Post.published == True)
        .order_by(Post.published_at.desc())
        .limit(10)
    ).scalars().all()
    
    for p in posts:
        print(f"{p.title} — {p.author.username}")

Exercise 2. Count how many posts and comments there are, grouped by author.

from sqlalchemy import func, select
from app.database import db_session
from app.models import User, Post, Comment

with db_session() as session:
    stmt = (
        select(
            User.username,
            func.count(Post.id.distinct()).label("posts"),
            func.count(Comment.id.distinct()).label("comments"),
        )
        .outerjoin(Post, Post.author_id == User.id)
        .outerjoin(Comment, Comment.author_id == User.id)
        .group_by(User.id, User.username)
    )
    
    for u, p, c in session.execute(stmt):
        print(f"{u}: {p} posts, {c} comments")

Exercise 3. Find users who have commented but never published, using EXISTS and NOT EXISTS.

from sqlalchemy import select, exists
from app.database import db_session
from app.models import User, Post, Comment

with db_session() as session:
    stmt = (
        select(User.username)
        .where(
            exists().where(Comment.author_id == User.id),
            ~exists().where(Post.author_id == User.id, Post.published == True),
        )
    )
    
    for (username,) in session.execute(stmt):
        print(username)

Exercise 4. Implement "tags related to postgresql" (co-occurrence, query 9 from module 2).

from sqlalchemy import func, select
from sqlalchemy.orm import aliased
from app.database import db_session
from app.models import Tag, post_tags_table

with db_session() as session:
    # The base tag
    tag_pg = session.execute(select(Tag).where(Tag.slug == "postgresql")).scalar_one()
    
    # Aliases to join post_tags with itself
    pt1 = aliased(post_tags_table)
    pt2 = aliased(post_tags_table)
    
    stmt = (
        select(
            Tag.name.label("related"),
            func.count().label("co_occurrence"),
        )
        .select_from(pt1)
        .join(pt2, pt1.c.post_id == pt2.c.post_id)  # the same post
        .join(Tag, Tag.id == pt2.c.tag_id)
        .where(pt1.c.tag_id == tag_pg.id, pt2.c.tag_id != tag_pg.id)
        .group_by(Tag.id, Tag.name)
        .order_by(func.count().desc())
    )
    
    for name, n in session.execute(stmt):
        print(f"{name}: {n}")

Exercise 5. Build a search_posts(session, query: str, tag: str | None = None) function that returns posts whose title contains query (case-insensitive), optionally filtered by tag.

from sqlalchemy import func, select
from sqlalchemy.orm import Session, selectinload
from app.models import Post, Tag

def search_posts(session: Session, query: str, tag: str | None = None) -> list[Post]:
    stmt = (
        select(Post)
        .options(selectinload(Post.author), selectinload(Post.tags))
        .where(
            Post.published == True,
            Post.title.ilike(f"%{query}%"),
        )
    )
    
    if tag:
        stmt = stmt.join(Post.tags).where(Tag.slug == tag)
    
    return list(session.execute(stmt).scalars().unique().all())

# Usage
from app.database import db_session
with db_session() as session:
    results = search_posts(session, "postgres", tag="tutorial")
    for p in results:
        print(p.title)

Summary

• select(Model) or select(Model.col1, Model.col2) to build queries
• .where() with ==, !=, <, .in_(), .like(), .is_(None), etc.
• .join(Model) or .join(Model.relationship) (preferred if the relationship exists)
• .outerjoin() for a LEFT JOIN
• .order_by(), .limit(), .offset() for sorting and pagination
• func.count(), func.sum(), func.avg(), func.X() for any SQL function
• .group_by() + .having() for groupings
• func.X(...).filter(condition) for FILTER WHERE
• .cte() for Common Table Expressions
• Results: .scalars().all() for lists of models, .execute().all() for tuples
• Subqueries: .scalar_subquery() for scalars, a plain select(...) inside .in_()
• Composable: queries are Python objects, chainable and reusable

In the next capsule, the most important performance topic: eager vs lazy loading and the N+1 problem.

Additional Resources

1. SELECT Statement — SQLAlchemy 2.0
2. ORM Query Guide
3. func — SQL Function Construction
4. Window Functions

Next: Capsule 07 — Eager vs Lazy Loading.