如何在SQLModel（底层SQLAlchemy）中查询关联数量少于指定值的Paper对象

问题背景 #

你基于SQLModel（底层SQLAlchemy）定义了如下自关联模型：

from typing import Optional, List
from sqlmodel import SQLModel, Field, Relationship

class Paper(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    similar: List["PaperSimilarLink"] = Relationship(
        link_model=PaperSimilarLink,
        back_populates="similar",
        sa_relationship_kwargs=dict(
            primaryjoin="Paper.id==PaperSimilarLink.paper_id",
            secondaryjoin="Paper.id==PaperSimilarLink.similar_id",
        ),
    )

class PaperSimilarLink(SQLModel, table=True):
    paper_id: int = Field(default=None, foreign_key="paper.id", primary_key=True)
    similar_id: int = Field(default=None, foreign_key="paper.id", primary_key=True)
    paper: "Paper" = Relationship(
        back_populates="similar",
        sa_relationship_kwargs=dict(
            foreign_keys="[PaperSimilarLink.paper_id]",
        ),
    )
    similar: "Paper" = Relationship(
        back_populates="similar",
        sa_relationship_kwargs=dict(
            foreign_keys="[PaperSimilarLink.similar_id]",
        ),
    )
    similarity: Optional[float]

你尝试查询所有关联的PaperSimilarLink数量≤11的Paper对象，使用了如下代码：

papers = session.exec(select(Paper).where(Paper.similar.count() <= 11)).all()

但触发报错：

AttributeError: Neither 'InstrumentedAttribute' object nor 'Comparator' object associated with Paper.similar has an attribute 'count'

解决方案 #

Paper.similar是SQLAlchemy的InstrumentedAttribute，无法直接在查询条件中调用count()，需要通过聚合函数结合子查询或关联查询实现。

方法1：子查询+左外连接（推荐，包含无关联的Paper）

先通过子查询统计每个Paper的关联数量，再关联主表筛选：

from sqlalchemy import func, select

# 子查询：统计每个paper对应的similar link数量
count_subquery = (
    select(
        PaperSimilarLink.paper_id,
        func.count(PaperSimilarLink.similar_id).label("similar_count")
    )
    .group_by(PaperSimilarLink.paper_id)
    .subquery()
)

# 主查询：左外连接确保包含无关联的Paper，筛选数量≤11的记录
papers = session.exec(
    select(Paper)
    .outerjoin(count_subquery, Paper.id == count_subquery.c.paper_id)
    .where(
        (count_subquery.c.similar_count <= 11) | 
        (count_subquery.c.similar_count.is_(None))
    )
).all()

• 左外连接会保留没有任何关联PaperSimilarLink的Paper，此时similar_count为None，通过is_(None)将这类Paper也纳入结果。

方法2：GROUP BY + HAVING

如果不需要获取关联数量，仅做筛选，可直接用分组和聚合条件：

from sqlalchemy import func

# 仅筛选有关联且数量≤11的Paper
papers = session.exec(
    select(Paper)
    .join(Paper.similar)
    .group_by(Paper.id)
    .having(func.count(PaperSimilarLink.similar_id) <= 11)
).all()

如果需要包含无关联的Paper，改为左外连接：

papers = session.exec(
    select(Paper)
    .outerjoin(Paper.similar)
    .group_by(Paper.id)
    .having(func.count(PaperSimilarLink.similar_id) <= 11)
).all()

• 这里用func.count(PaperSimilarLink.similar_id)而非func.count(Paper.similar)，因为左外连接时无关联行的similar_id为NULL，count会忽略NULL值，得到正确的0；若用count(Paper.similar)会把NULL行统计为1，导致筛选错误。

内容的提问来源于stack exchange，提问作者phil0s0pher