如何使用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中执行速度明显较慢。我正在寻找一种解决方案，不需要已知主键就能获取表中的行数，就像使用SQLAlchemy获取表中行数中所建议的那样。

- tiho

---

你的查询中的 table 是 Table、mapper 还是 Base 的实例？ - wberry

这是一个“Table”实例。 - tiho

可能是使用SQLAlchemy获取表中行数的重复问题。 - user

Answer 1

仅查询一个已知的列：

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

- Nathan Villaescusa

---

这将把计数构造成一个包装的选择语句。虽然MySQL的查询优化器应该会处理它。 - pi.

谢谢：虽然它并没有完全回答我的问题，但确实更快。我担心它会忽略 NULL 条目，但事实并非如此。然而，它仍比显式的 SELECT(*) 慢。 - tiho

@Nathan Villaescusa，您能否提供一个多列的示例呢？ 例如：session.query(MyTable.col1, MyTable.col2).count() SQL语句：SELECT col1, COUNT(col1) FROM "table" - Давид Шико

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条件。

- pi.

---

抱歉，我错了，它实际上并没有起作用。我的意思是，它确实做了你说的事情，但在我的实际使用情况中，它仍然没有产生SELECT(*)（这比你的建议快大约两倍）。 - tiho

请使用 sqlalchemy.sql.func.count() 代替 sqlalchemy.sql.text('count(*)')。 - ddaa

新的导入方式变成了类似于“from sqlalchemy.sql.functions import count”用于Python3。 - srknzl

对于简单的select ... from table查询，这可能会从查询中_删除_该表。 - Martijn Pieters

我目前在使用select([func.count()]).select_from(table)时收到了一个弃用警告，我改用func.count().select().select_from(table)来代替。 - undefined

Answer 3

我需要对一个非常复杂的带有许多连接的查询进行计数。我正在使用连接作为过滤器，因此我只想知道实际对象的数量。 count（）功能不足，但我在这里的文档中找到了答案：

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

代码大致如下（用于计算用户对象的数量）:

from sqlalchemy import func

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

- melchoir55

---

在您的示例中没有引用任何查询。 - Piotr Dobrogost

有人知道这个的速度与 session.query(User).count() 相比如何吗？这个问题也在这里被问到了：https://dev59.com/2GUq5IYBdhLWcg3wJNBA - Matt

Answer 4

除了接受的答案中ORM层的用法外，还可以使用query.with_entities(func.count())来进行ORM的计数(*), 例如： Usage from the ORM layer。

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

它也可以用于更复杂的情况，例如在有连接和筛选条件的情况下 - 在这里重要的是在连接后放置with_entities，否则SQLAlchemy可能会引发不知道如何连接错误。

例如:

• 我们有一个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()

完整的示例在这里。

- Borys Serebrov

---

这只适用于使用.join()的查询。否则，查询会被“优化”，WHERE子句会被简单地移除。例如session.query(Model).with_entities(func.count()).scalar() 生成SELECT count(*) AS count_1 ，我怀疑没有人搜索^ ^这个结果。 - Romain Vincent

@RomainVincent 对我来说，session.query(MyModel).filter(MyModel.rank < 100).with_entities(func.count()) 生成的是 SELECT count(*) AS count_1 FROM my_table WHERE my_table.rank < %(rank_1)s，因此您可能会遇到一些边缘情况或错误。 - Borys Serebrov

是的，我过于概括了我的回答。因此更好的说法应该是：“如果您不明确在query()参数中使用列，则此方法将无法正常工作。我必须承认我不知道SQLA会做什么或不会做什么，但从我的角度来看，这种方法似乎很危险，因为它可能导致意外的结果。” - Romain Vincent

Query.with_entities() 本质上与 Query.statement.with_only_columns()（在被接受的答案中）做的事情相同，并且存在相同的问题，即对于普通的 session.query(Model)，这会产生 SELECT COUNT(*)（没有 FROM）。在 https://gerrit.sqlalchemy.org/c/sqlalchemy/sqlalchemy/+/2973 实现之前，在 ORM 层面上没有很好的方法来解决这个问题。 - Martijn Pieters

我不知道这个功能从什么时候开始可用，但 query.with_only_columns(func.count(), maintain_column_froms=True) 似乎有效。 - Gabor Simon

Answer 5

这不是对十多年前发布的原始问题的答案，而是针对问题“如何制作SELECT COUNT(*)”的一般性答案。还要注意，这里的许多答案已经非常过时，只适用于旧版本的SQLAlchemy。

以下是SQLAlchemy 1.4.x和2.0.x的解决方案：

from sqlalchemy import func, select

class MyModel(Base):
    ...

statement = select(func.count()).select_from(MyModel)
count: int = session.execute(statement).scalar()

- Konstantin Smolyanin

Answer 6

如果您使用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)

- Ole

Answer 7

查询 = 会话.query(表.列).过滤().with_entities(func.count(表.列.distinct())) 计数 = 查询.scalar()

这对我很有效。

生成的查询语句为： SELECT count(DISTINCT 表.列) AS count_1 FROM table where ...

- gladiator

Answer 8

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

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

如果您想探索更多选项或阅读详细信息，请查看参考文档链接。

- Bilal Ahmed Yaseen

---

这仍然将别名查询封装到子选择中。没有名称的别名只是一个匿名子查询。你的答案生成的SQL与问题提出者想要避免的完全相同。 - Martijn Pieters

Answer 9

我不太清楚您所说的“without explicitly asking for it with execute()”是什么意思，因此可能恰恰不是您所要求的内容。但是这可能会对其他人有所帮助。

另一方面，您可以直接运行文本SQL：

your_query="""
SELECT count(*) from table
"""
the_count = session.execute(text(your_query)).scalar()

- Brian C.

---

是的，这正是我所不想做的事情 ;) - tiho

Answer 10

def test_query(val: str):    
   query = f"select count(*) from table where col1='{val}'"
   rtn = database_engine.query(query)
   cnt = rtn.one().count

但是如果你检查了调试监视器，你就可以找到方法。

- Jun Takeshita