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

Question

我想知道是否可以SELECT COUNT(*) FROM TABLE在SQLAlchemy中生成一个语句,而无需明确要求它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获取表中的行数所示.

Answer 1

我设法在两个层上使用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 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)重置查询的顺序,这与计数无关.

使用这种方法,你可以有count(*)任何的ORM查询,将兑现所有的filter和join已经规定的条件.

Answer 2

仅查询单个已知列:

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

Answer 3

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

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

代码看起来像这样(计算用户对象):

from sqlalchemy import func

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

Answer 4

除了接受的答案中ORM 层的Usage 之外：count(*) can be done for ORM using the query.with_entities(func.count())，如下所示：

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()

完整的例子在这里。

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()