Using column_property for correlated subquery counts

Map a derived count with column_property(select(func.count(Comment.id)).where(Comment.post_id == Post.id).correlate_except(Comment).scalar_subquery()) — the number then arrives with each row, sortable and filterable, without loading the collection; and use query_expression() with with_expression() when only some queries need it. This guide belongs to hybrid properties, column properties and SQL expressions.

Quick Answer

A count of related rows is a number, and loading the rows to produce it is the most common avoidable cost in an ORM application.

Before — the collection loaded so Python can count it:

from sqlalchemy import select
from sqlalchemy.orm import selectinload

from blog.models import Post

posts = (await session.scalars(
    select(Post).order_by(Post.published_at.desc()).limit(20)
    .options(selectinload(Post.comments))      # every comment of every post
)).all()
counts = {post.id: len(post.comments) for post in posts}
# Two queries, and one of them returns every comment row on the page.
# Without the loader option, len(post.comments) raises MissingGreenlet.

After — the count computed in SQL, one value per row:

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


class Base(DeclarativeBase):
    pass


class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    comments: Mapped[list["Comment"]] = relationship(back_populates="post")


class Comment(Base):
    __tablename__ = "comments"
    id: Mapped[int] = mapped_column(primary_key=True)
    post_id: Mapped[int] = mapped_column(ForeignKey("posts.id", ondelete="CASCADE"), index=True)
    post: Mapped[Post] = relationship(back_populates="comments")


# Declared after both classes exist, because it references both.
Post.comment_count = column_property(
    select(func.count(Comment.id))
    .where(Comment.post_id == Post.id)
    .correlate_except(Comment)
    .scalar_subquery()
)

posts = (await session.scalars(
    select(Post).order_by(Post.published_at.desc()).limit(20)
)).all()
posts[0].comment_count        # an int, already loaded

The index on comments.post_id is not optional: without it the subquery scans the comments table once per returned row.

Execution Context & Async Workflow Integration

column_property() maps a SQL expression onto a class as though it were a column. When SQLAlchemy builds a SELECT for that class, the expression goes in the select list alongside the real columns, and the value comes back with the row. Nothing is lazy, nothing is cached, and nothing needs awaiting — which is exactly why it suits async code so well compared with counting a loaded collection.

Three parts of the construction matter.

.where(Comment.post_id == Post.id) is the correlation: it ties the subquery to whichever posts row the outer query is producing. .correlate_except(Comment) tells SQLAlchemy that only comments belongs in the subquery's FROM clause, and any other table it references — posts — comes from the enclosing statement. Without it, the subquery selects from both tables and counts every comment against every post, producing a number that is wrong in a way that looks plausible. .scalar_subquery() declares the result a single value, which is what allows it to appear in a select list at all.

Because the expression is in every SELECT for the class, the cost is paid by every query for the class — including ones that never read the attribute, and including the queries SQLAlchemy issues for relationship loading and refreshes. On PostgreSQL a correlated count over an indexed foreign key is an index-only scan and cheap, but "cheap per row" times "every query in the application" is a real number, and it is invisible in code that never mentions comment_count.

deferred=True moves the cost to first access, which sounds like the answer and is a trap under async: the deferred load is a lazy load, so reading post.comment_count raises MissingGreenlet unless the attribute was explicitly undeferred on the query or the object refreshed with await session.refresh(post, ["comment_count"]).

The async-friendly middle ground is query_expression(): the class declares a placeholder, and each query decides whether to fill it and with what expression:

from sqlalchemy import func, select
from sqlalchemy.orm import Mapped, mapped_column, query_expression, with_expression


class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    comment_count: Mapped[int] = query_expression()


count_comments = (
    select(func.count(Comment.id))
    .where(Comment.post_id == Post.id)
    .correlate_except(Comment)
    .scalar_subquery()
)

stmt = (
    select(Post)
    .order_by(Post.published_at.desc())
    .limit(20)
    .options(with_expression(Post.comment_count, count_comments))
)

Queries that do not apply the option leave the attribute None, so the cost appears exactly where the requirement is. That is the shape most production models settle on, and it is the same reasoning as choosing loader options per query rather than eager defaults on relationships.

Resolving Warnings, Errors & Common Mistakes

Sorting and filtering by the derived value is the reason many teams map it in the first place, and it works exactly like a column:

from sqlalchemy import select

from blog.models import Post

busiest = select(Post).order_by(Post.comment_count.desc()).limit(10)
discussed = select(Post).where(Post.comment_count > 50)

With query_expression() this needs care: the placeholder is not a SQL expression on its own, so ordering by it only works when the expression has been supplied for that query. Ordering by the subquery directly — order_by(count_comments.desc()) — is the version that always works and does not depend on the option being applied.

One more thing worth knowing: a filter on a correlated count cannot use an index on the child table to select parents efficiently. WHERE (SELECT count(*) ...) > 50 must evaluate the subquery for every candidate parent row. When that filter is common and the table is large, the honest answer is a stored counter column, which the last section covers.

Advanced: Grouped Joins, Filtered Counts and Several Aggregates

A correlated subquery is evaluated once per row the outer query returns. With LIMIT 20 that is twenty index lookups. Without a limit — an export, a report, a data feed — it is one lookup per row in the table, and a grouped join becomes the better plan:

from sqlalchemy import func, select

from blog.models import Comment, Post

comment_counts = (
    select(Comment.post_id, func.count().label("comment_count"))
    .group_by(Comment.post_id)
    .subquery()
)

stmt = (
    select(Post, func.coalesce(comment_counts.c.comment_count, 0).label("comment_count"))
    .outerjoin(comment_counts, comment_counts.c.post_id == Post.id)
    .order_by(Post.id)
)
rows = (await session.execute(stmt)).all()      # (Post, int) tuples

The outerjoin and coalesce together keep posts with no comments in the result with a count of zero, which an inner join would have dropped — a mistake that is easy to miss when test data happens to give every parent at least one child.

Filtered and multiple aggregates are where SQL earns its place. Several counts over the same child table should be one pass with FILTER, not several subqueries:

from sqlalchemy import func, select

from blog.models import Comment, Post

stats = (
    select(
        Comment.post_id,
        func.count().label("total"),
        func.count().filter(Comment.approved.is_(True)).label("approved"),
        func.count().filter(Comment.flagged.is_(True)).label("flagged"),
        func.max(Comment.created_at).label("last_comment_at"),
    )
    .group_by(Comment.post_id)
    .subquery()
)

func.count().filter(...) renders PostgreSQL's count(*) FILTER (WHERE ...), so three counts and a maximum are computed in a single scan of the relevant comments. Mapping those onto the class with column_property one subquery at a time would have meant four correlated subqueries per row.

The same FILTER trick works in a mapped column_property when only one filtered count is needed — "unapproved comments", say — and it is often the more useful number:

Post.pending_comment_count = column_property(
    select(func.count(Comment.id))
    .where(Comment.post_id == Post.id, Comment.approved.is_(False))
    .correlate_except(Comment)
    .scalar_subquery()
)

A partial index matching that predicate — Index("ix_comments_pending", Comment.post_id, postgresql_where=Comment.approved.is_(False)) — makes it as cheap as the unfiltered version. Partial and expression indexes are covered in creating partial and expression indexes from SQLAlchemy models.

When to Store the Count Instead

Every approach above computes the number on read. Past a certain read rate, the right answer is to compute it on write and store it — a denormalised counter column — and the decision hinges on one ratio: how often the count is read versus how often the underlying rows change.

A stored counter turns the read into a plain column access, indexable and free. It costs an extra write on every insert and delete of a child row, and it introduces the possibility of drift.

Maintaining it in the database is more reliable than maintaining it in the application, because it covers every writer:

from alembic import op


def upgrade() -> None:
    op.execute("ALTER TABLE posts ADD COLUMN comment_count integer NOT NULL DEFAULT 0")
    op.execute("""
        CREATE FUNCTION posts_comment_count() RETURNS trigger AS $$
        BEGIN
            IF TG_OP = 'INSERT' THEN
                UPDATE posts SET comment_count = comment_count + 1 WHERE id = NEW.post_id;
            ELSIF TG_OP = 'DELETE' THEN
                UPDATE posts SET comment_count = comment_count - 1 WHERE id = OLD.post_id;
            ELSIF NEW.post_id IS DISTINCT FROM OLD.post_id THEN
                UPDATE posts SET comment_count = comment_count - 1 WHERE id = OLD.post_id;
                UPDATE posts SET comment_count = comment_count + 1 WHERE id = NEW.post_id;
            END IF;
            RETURN NULL;
        END $$ LANGUAGE plpgsql;
    """)
    op.execute("""
        CREATE TRIGGER comments_maintain_count
        AFTER INSERT OR UPDATE OR DELETE ON comments
        FOR EACH ROW EXECUTE FUNCTION posts_comment_count();
    """)
    op.execute("""
        UPDATE posts p SET comment_count =
            (SELECT count(*) FROM comments c WHERE c.post_id = p.id)
    """)

In the model, map it as server-maintained so the ORM reads it but never writes it: mapped_column(server_default=text("0"), server_onupdate=FetchedValue()).

Two costs are worth stating plainly. Every comment insert now also updates the post row, which serialises concurrent inserts against the same post — fine for comments, potentially a hot-row problem for something like a view counter, where the usual answer is an append-only table aggregated periodically. And the backfill in that migration rewrites every row, so on a large table it belongs in batches, as in renaming a column without downtime.

Whichever mechanism maintains it, add a reconciliation check. A nightly job comparing the stored counter with a real count, and logging any row that disagrees, is a few lines and catches every class of drift — a trigger dropped by a migration, a bulk load that bypassed it, a restored backup:

from sqlalchemy import func, select

from blog.models import Comment, Post

drifted = (
    select(Post.id, Post.comment_count, func.count(Comment.id).label("actual"))
    .outerjoin(Comment, Comment.post_id == Post.id)
    .group_by(Post.id, Post.comment_count)
    .having(Post.comment_count != func.count(Comment.id))
)

Frequently Asked Questions

Related

• Hybrid Properties, Column Properties and SQL Expressions — The parent guide: derived values in Python and in SQL.
• Writing hybrid properties that work in Python and SQL — Derived values computed from columns on the same row.
• Using selectinload vs joinedload for N+1 prevention — When the rows, not just the count, are actually needed.
• Selecting the top N rows per group with row_number() — Per-group aggregates alongside per-group rows.