SQLAlchemy 如何将未映射的 COUNT(*) 作为查询的结果？

Posted 2023-03-16

【中文标题】SQLAlchemy 如何将未映射的 COUNT(*) 作为查询的结果？

【英文标题】：SQLAlchemy how to get unmapped COUNT(*) as a result to a query?

【发布时间】：2022-01-09 18:25:02

【问题描述】：

我无法理解如何将未映射的列结果（例如 COUNT(*)）获取到映射查询。

我有两个这样的映射表：

class Actor(Base):
    __tablename__ = 'actor'
    id: int = Column(Integer)
    name: str = Column(String)

    movies = relationship("Movie", secondary=movie_actors_association_table, back_populates="actors")


class Movie(Base):
    __tablename__ = 'movie'
    id: int = Column(Integer)
    name: str = Column(String)
    year: int = Column(Integer)

    actors = relationship("Actor", secondary=movie_actors_association_table, back_populates="actors")
    

movie_actors_association_table 看起来像这样：

movie_actors_association_table = Table('movie_actors', Base.metadata,
                                       Column('movie_id', ForeignKey('movie.id'), primary_key=True),
                                       Column('actor_id', ForeignKey('actor.id'), primary_key=True)
                                       )

我的目标是按演员在特定年份（或年份范围）中的电影数量排序，但我也想得到结果中的计数。例如，如果 1999 年布拉德皮特拍了 5 部电影，安吉丽娜朱莉拍了 3 部电影，尼古拉斯凯奇拍了 15 部电影，那么我希望结果类似于 1999 年的(Nicolas Cage,15), (Brad Pitt,5), (Angelina Jolie,3)。

为此，我目前正在做的是：

current_filter = Movie.year == 1999
query_movies = Query(Movie).filter(current_filter)
query_movies_id = Query(Movie.id).filter(current_filter)


def query_to_txt_stmt(query):
    """converts Query object to SQL text statement"""
    stmt = query.statement
    stmt_complied = stmt.compile(dialect=sqlite.dialect(), compile_kwargs="literal_binds": True)
    print(stmt_complied)
    return stmt_complied

stmt = text(
    f"SELECT actor.*, COUNT(*) as my_count FROM "
    f"actor JOIN movie_actors on actor.id = movie_actors.actor_id "
    f"WHERE movie_actors.movie_id IN (query_to_txt_stmt(query_movies_id)) "
    f"GROUP BY actor.id ORDER BY my_count DESC"
    )


final_query = Query(Actor).from_statement(stmt)
final_query.session = db.session # the session created in an offscreen db class
results = final_query.all() #query executes here, but I get all Actor objects as a result. Without the count.

这种方法为我提供了正确的演员排序，但我没有在输出中得到实际的 COUNT。而且我不知道如何将其添加到结果中。这是我的问题的核心。如何在结果中添加未映射的列？例如，是否可以将 COUNT 结果映射到临时变量中的映射 Actor 类？

注意：我使用的是原始文本查询，因为我需要将 Query 对象作为参数传递，而不会将其附加到 session。但是当我尝试执行类似这样的操作时：（就像 @ 中建议的那样987654321@答案。）

query = Query(Actor, func.count(Actor.id).label("my_count")).join(movie_actors_association_table,movie_actors_association_table.c.actor_id == Actor.id).group_by(Actor.id).order_by(text("my_count DESC"))

然后像这样给它会话：

query.session = db.session
query.all()

它将无法运行查询，因为它生成的 SQL 不包含 my_count 的自定义标签。所以它说“没有这样的列'my_count'”

但是，如果我在会话中这样做：

db.session.query(Actor, func.count(Actor.id).label("my_count")).join(movie_actors_association_table,movie_actors_association_table.c.actor_id == Actor.id).group_by(Actor.id).order_by(text("my_count DESC"))

因此，如果我不使用原始文本，似乎我必须在编写查询时访问会话。我不希望那样。如果有人知道如何在此处不使用原始文本并且仍然保持独立于会话，我会很高兴听到它。

【参考方案1】：

你可以试试这个

db.session.query(Actor).join(movie_actors_association_table,movie_actors_association_table.c.actor_id == Actor.id)
.with_entities(Actor.id, func.count(movie_actors_association_table.c.actor_id).label("my_count"))
.group_by(Actor.id)
.order_by(desc(literal_column("my_count")))

【讨论】：

谢谢，这似乎有效。你知道如何过滤计数列吗？比如where my_count > 10？

另外，另一个问题是您的示例中的with_entitites 返回Actor.id 和my_count，因为我还需要实际的actor 对象。

如果你想过滤计数使用.having语句，或者你可以像任何列一样过滤它

有了实体，你可以返回你想要的任何东西，你可以检查另一个方法来返回整个对象

当我用 .with_entities(Actor, func.count(movie_actors_association_table.c.actor_id) 替换 .with_entities(Actor.id, func.count(movie_actors_association_table.c.actor_id) 时，我得到一个异常，说它找不到“my_count”列。然而，当它是 Actor.id

时它工作得很好