sqlmodel
sqlmodel copied to clipboard
tiangolo
How to write statements like "select count(*) from table_name where col_name>0"?
First Check
- [X] I added a very descriptive title to this issue.
- [X] I used the GitHub search to find a similar issue and didn't find it.
- [X] I searched the SQLModel documentation, with the integrated search.
- [X] I already searched in Google "How to X in SQLModel" and didn't find any information.
- [X] I already read and followed all the tutorial in the docs and didn't find an answer.
- [X] I already checked if it is not related to SQLModel but to Pydantic.
- [X] I already checked if it is not related to SQLModel but to SQLAlchemy.
Commit to Help
- [X] I commit to help with one of those options 👆
Example Code
How to write statements like "select count(*) from table_name where col_name>0"?
Description
How to write statements like "select count(*) from table_name where col_name>0"?
Operating System
Linux
Operating System Details
No response
SQLModel Version
0.0.8
Python Version
3.7.4
Additional Context
No response
Two options:
from typing import Optional
from sqlalchemy import func
from sqlmodel import Field, SQLModel, Session, create_engine
class Table(SQLModel, table=True):
id: Optional[int] = Field(primary_key=True)
column: int
engine = create_engine(
"sqlite:///",
echo=True,
)
with Session(engine) as session:
SQLModel.metadata.create_all(engine)
# Option #1:
print(session.query(Table).where(Table.column > 10).count())
"""
SELECT count(*) AS count_1
FROM (SELECT "table".id AS table_id, "table"."column" AS table_column
FROM "table"
WHERE "table"."column" > ?) AS anon_1
"""
# Option #2:
print(session.query(Table).with_entities(func.count()).where(Table.column > 10).scalar())
"""
SELECT count(*) AS count_1
FROM "table"
WHERE "table"."column" > ?
"""
@rxy1212 You can use count function via sqlalchemy.func object like in regular sql:
from typing import Optional
from sqlalchemy import func
from sqlmodel import Field, SQLModel, Session, create_engine, select
class Table(SQLModel, table=True):
id: Optional[int] = Field(primary_key=True)
column: int
engine = create_engine(
"sqlite:///",
echo=True,
)
SQLModel.metadata.create_all(engine)
stmt = select(func.count(Table.id)).where(Table.column > 10)
print(stmt)
"""
SELECT count("table".id) AS count_1
FROM "table"
WHERE "table"."column" > :column_1
"""
with Session(engine) as session:
for i in range(150):
session.add(Table(column=i))
session.flush()
print(session.scalar(stmt)) # 139
@meirdev @ThirVondukr just out of curiosity, how you guys found this solution? It's not in the official document
@Metal-joker what do you mean? I think all those are in sqlalchemy docs. SQLModel is a way to use SA with Pydantic models. https://docs.sqlalchemy.org/en/14/core/functions.html#sqlalchemy.sql.functions.count
This solution is not applicable anymore. Or I'm doing something wrong, but I don't have a where method, and I get a warning:
...sqlmodel/sql/expression.py:225: SAWarning: implicitly coercing SELECT object to scalar subquery; please use the .scalar_subquery() method to produce a scalar subquery.
return super().where(*whereclause) # type: ignore[arg-type]
I've found this gist, and adapted it for sqlmodel, here's what I've got:
from sqlmodel import Session
from sqlmodel.sql.expression import SelectOfScalar
def get_count(session: Session, q: SelectOfScalar) -> int:
count_q = q.with_only_columns(func.count()).order_by(None).select_from(*q.froms)
iterator = session.exec(count_q)
for count in iterator:
return count
return 0
This is a utility method, that can be used wherever we need it. For example:
import pydantic
class SearchSchema(pydantic.BaseModel):
opt_param: str | None = None
class AccountModel(SQLModel, table=True):
# Doesn't really matter TBH
opt_param: str
def get_accounts(session: Session, search_params: SearchSchema)
query = select(AccountModel)
if search_params.opt_param:
query = query.filter(AccountModel.opt_param == search.opt_param)
# You can add more filters here
total = get_count(db, query) # Here's your result
query = db.exec(query)
# Rest of the method here
Works fine for me, at least - for now.
a bit more update from makisukurisu's comment to not get deprecate warnings
from sqlmodel.sql.expression import SelectOfScalar
from sqlmodel import func
def get_count(session: Session, q: SelectOfScalar) -> int:
count_q = q.with_only_columns(func.count()).order_by(None).select_from(q.get_final_froms()[0])
iterator = session.exec(count_q)
for count in iterator:
return count
return 0
and use it like
# change to your code from below
with Session(engine) as sess:
query = select(YourTable).\
where(YourColumn == 1)
total = get_count(sess, query)
print(f'{total=}')
sess.close()
I have a simpler problem, just trying to do session.query(MyObject).count() and I get a warning:
<ipython-input-9-9445795a22fe>:1: DeprecationWarning:
🚨 You probably want to use `session.exec()` instead of `session.query()`.
`session.exec()` is SQLModel's own short version with increased type
annotations.
Or otherwise you might want to use `session.execute()` instead of
`session.query()`.
session.query(MyObject).count()
What is the expressive, short way to do something similar in SQLModel?
In the spirit of using SQLModel as a thin wrapper, I'd rather continue to use the SQLA syntax than build and maintain a (rather complicated) get_count helper as above. But maybe I'm missing something?
SQLModel has a func.count function that is working ok:
from sqlmodel import Session, col, func, select
from .database import engine
from .models import Page
with Session(engine) as session:
count = session.exec(select(func.count(col(Page.id))).where(Page.id > 100)).one()
