85

我想知道是否可以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 获取表中的行数中所建议的那样。

4

7 回答 7

93

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

使用此方法,您可以count(*)在任何 ORM 查询上使用,这将遵守所有已指定的filterjoin条件。

于 2012-10-17T20:03:28.237 回答
90

仅查询一个已知列:

session.query(MyTable.col1).count()
于 2012-10-17T19:56:13.183 回答
28

我需要对具有许多连接的非常复杂的查询进行计数。我使用连接作为过滤器,所以我只想知道实际对象的数量。count() 是不够的,但我在这里的文档中找到了答案:

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

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

from sqlalchemy import func

session.query(func.count(User.id)).scalar() 
于 2015-06-04T01:48:42.263 回答
14

在接受的答案中添加来自 ORM 层的用法: count(*) 可以使用 为 ORM 完成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()

完整的例子在这里

于 2019-09-14T10:19:00.760 回答
1

如果您正在使用 SQL 表达式样式方法,那么如果您已经拥有表对象,则还有另一种方法来构造 count 语句。

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

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)
于 2019-07-03T08:31:21.963 回答
0

我不清楚您所说的“没有使用execute()明确要求”是什么意思所以这可能正是您不要求的。OTOH,这可能对其他人有所帮助。

您可以只运行文本 SQL:

your_query="""
SELECT count(*) from table
"""
the_count = session.execute(text(your_query)).scalar()
于 2022-02-09T21:10:43.357 回答
-1

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

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

如果您想探索更多选项或阅读详细信息,这里是参考文档的链接。

于 2021-03-08T07:57:24.693 回答