如何使用 SQLAlchemy 使用 SELECT COUNT(*) 计算行数？

Question

我想知道是否可以在 SQLAlchemy 中生成 SELECT COUNT(*) FROM TABLE 语句，而无需使用 execute() 明确要求它。 如果我使用：

session.query(table).count()

然后它会生成类似的东西：

SELECT count(*) AS count_1 FROM
    (SELECT table.col1 as col1, table.col2 as col2, ... from table)

在带有 InnoDB 的 MySQL 中速度明显较慢。我正在寻找一种不需要表具有已知主键的解决方案，正如Get the number of rows in table using SQLAlchemy 中所建议的那样。

Answer 1

以下是查找任何查询计数的方法。

aliased_query = alias(query)
db.session.query(func.count('*')).select_from(aliased_query).scalar()

如果您想探索更多选项或阅读详细信息，这里是link to the reference document。

Answer 2

我设法在两个图层上使用 SQLAlchemy 呈现以下 SELECT。

SELECT count(*) AS count_1
FROM "table"

从 SQL 表达式层使用

from sqlalchemy import select, func, Integer, Table, Column, MetaData

metadata = MetaData()

table = Table("table", metadata,
              Column('primary_key', Integer),
              Column('other_column', Integer)  # just to illustrate
             )   

print select([func.count()]).select_from(table)

ORM 层的使用

你只需继承 Query（你可能无论如何都有）并提供一个专门的 count() 方法，就像这个。

from sqlalchemy.sql.expression import func

class BaseQuery(Query):
    def count_star(self):
        count_query = (self.statement.with_only_columns([func.count()])
                       .order_by(None))
        return self.session.execute(count_query).scalar()

请注意order_by(None)会重置查询的顺序，与计数无关。

使用此方法，您可以在任何 ORM 查询上使用 count(*)，这将遵循已指定的所有 filter 和join 条件。

Answer 3

除了接受答案中的ORM 层的使用：count(*) 可以使用query.with_entities(func.count()) 为ORM 完成，如下所示：

session.query(MyModel).with_entities(func.count()).scalar()

它也可以用于更复杂的情况，当我们有连接和过滤器时 - 这里重要的是在连接之后放置 with_entities，否则 SQLAlchemy 可能会引发 Don't know how to join 错误。

例如：

• 我们有User模型（id，name）和Song模型（id，title，genre）
• 我们有用户歌曲数据 - UserSong 模型（user_id、song_id、is_liked），其中user_id + song_id 是主键）

我们希望获得一些用户喜欢的摇滚歌曲：

SELECT count(*) 
  FROM user_song
  JOIN song ON user_song.song_id = song.id 
 WHERE user_song.user_id = %(user_id)
   AND user_song.is_liked IS 1
   AND song.genre = 'rock'

可以通过以下方式生成此查询：

user_id = 1

query = session.query(UserSong)
query = query.join(Song, Song.id == UserSong.song_id)
query = query.filter(
    and_(
        UserSong.user_id == user_id, 
        UserSong.is_liked.is_(True),
        Song.genre == 'rock'
    )
)
# Note: important to place `with_entities` after the join
query = query.with_entities(func.count())
liked_count = query.scalar()

完整的例子是here。

Answer 4

如果您使用 SQL 表达式样式方法，如果您已经拥有表对象，则还有另一种方法来构造计数语句。

准备获取表格对象。也有不同的方法。

import sqlalchemy

database_engine = sqlalchemy.create_engine("connection string")

# Populate existing database via reflection into sqlalchemy objects
database_metadata = sqlalchemy.MetaData()
database_metadata.reflect(bind=database_engine)

table_object = database_metadata.tables.get("table_name") # This is just for illustration how to get the table_object                    

在table_object上发出计数查询

query = table_object.count()
# This will produce something like, where id is a primary key column in "table_name" automatically selected by sqlalchemy
# 'SELECT count(table_name.id) AS tbl_row_count FROM table_name'

count_result = database_engine.scalar(query)

Answer 5

我需要计算一个包含许多连接的非常复杂的查询。我使用连接作为过滤器，所以我只想知道实际对象的数量。 count() 不够，但我在这里的文档中找到了答案：

http://docs.sqlalchemy.org/en/latest/orm/tutorial.html

代码看起来像这样（计算用户对象）：

from sqlalchemy import func

session.query(func.count(User.id)).scalar() 

Answer 6

只查询一个已知列：

session.query(MyTable.col1).count()