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

Query for just a single known column:

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

I managed to render the following SELECT with SQLAlchemy on both layers.

SELECT count(*) AS count_1
FROM "table"

Usage from the SQL Expression layer

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)

Usage from the ORM layer

You just subclass Query (you have probably anyway) and provide a specialized count() method, like this one.

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

Please note that order_by(None) resets the ordering of the query, which is irrelevant to the counting.

Using this method you can have a count(*) on any ORM Query, that will honor all the filter andjoin conditions already specified.

I needed to do a count of a very complex query with many joins. I was using the joins as filters, so I only wanted to know the count of the actual objects. count() was insufficient, but I found the answer in the docs here:

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

The code would look something like this (to count user objects):

from sqlalchemy import func


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

If you are using the SQL Expression Style approach there is another way to construct the count statement if you already have your table object.

Preparations to get the table object. There are also different ways.

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

Issuing the count query on the 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)

Addition to the Usage from the ORM layer in the accepted answer: count(*) can be done for ORM using the query.with_entities(func.count()), like this:

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

It can also be used in more complex cases, when we have joins and filters - the important thing here is to place with_entities after joins, otherwise SQLAlchemy could raise the Don't know how to join error.

For example:

• we have User model (id, name) and Song model (id, title, genre)
• we have user-song data - the UserSong model (user_id, song_id, is_liked) where user_id + song_id is a primary key)

We want to get a number of user's liked rock songs:

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'

This query can be generated in a following way:

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

Complete example is here.

Below is the way to find the count of any query.

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

Here is the link to the reference document if you want to explore more options or read details.

I'm not clear on what you mean by "without explicitly asking for it with execute()" So this might be exactly what you are not asking for. OTOH, this might help others.

You can just run the textual SQL:

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

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

but you can find the way if you checked debug watch

query = session.query(table.column).filter().with_entities(func.count(table.column.distinct())) count = query.scalar()

this worked for me.

Gives the query: SELECT count(DISTINCT table.column) AS count_1 FROM table where ...