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

Question

我无法理解如何将未映射的列结果（例如 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"))

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

Answer 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")))