Skip to content

SQLAlchemy & SQLModel Backend

Therismos provides first-class support for SQLAlchemy Core, ORM, and SQLModel.

Installation

pip install therismos[sqlalchemy]

SQLModel uses the same expression language, so no separate extra is needed.

Quick Start

import sqlalchemy as sa
from therismos import F
from therismos.sorting import SortSpec, SortCriterion, SortOrder
from therismos.grouping import GroupSpec, Aggregation, AggregationFunction
from therismos.expr.visitors.sqlalchemy import SQLAlchemyExprVisitor
from therismos.sorting.visitors.sqlalchemy import SQLAlchemySortSpecVisitor
from therismos.grouping.visitors.sqlalchemy import SQLAlchemyGroupSpecVisitor

filter_expr = (F("age", int) >= 21) & (F("status") == "active")
sort_spec = SortSpec([SortCriterion("age", SortOrder.DESCENDING)])
group_spec = GroupSpec(
    group_by=["status"],
    aggregations=[
        Aggregation("count", AggregationFunction.COUNT),
        Aggregation("avg_age", AggregationFunction.AVERAGE, "age"),
    ],
)

where = filter_expr.accept(SQLAlchemyExprVisitor())
sort = sort_spec.accept(SQLAlchemySortSpecVisitor())
grp = group_spec.accept(SQLAlchemyGroupSpecVisitor())

stmt = (
    sa.select(*grp.group_by, *grp.agg)
    .where(where)
    .group_by(*grp.group_by)
    .order_by(*sort.order_by)
)

Expression Filtering

SQLAlchemyExprVisitor returns a ColumnElement[bool] for .where() / .filter():

# SQLAlchemy Core
where = expr.accept(SQLAlchemyExprVisitor())
stmt = select(users).where(where)

# SQLAlchemy ORM
with Session(engine) as session:
    users = session.query(User).filter(where).all()

# SQLModel
with Session(engine) as session:
    users = session.scalars(select(User).where(where)).all()

Notes:

  • Between uses half-open [lower, upper) semantics: (col >= lower) AND (col < upper).
  • regex matching uses regexp_match() (PostgreSQL, SQLite 3.8.3+, MySQL 8.0+). Flags convert to inline modifiers: (?i) for re.IGNORECASE.

Sorting

SQLAlchemySortSpecVisitor returns a SQLAlchemySortSpec with order_by: tuple[UnaryExpression, ...]:

sort = spec.accept(SQLAlchemySortSpecVisitor())
stmt = select(users).order_by(*sort.order_by)
# ORDER BY created_at DESC, name ASC

Criteria with SortOrder.NONE are skipped.

Grouping

SQLAlchemyGroupSpecVisitor returns a SQLAlchemyGroupSpec with group_by and agg tuples:

grp = spec.accept(SQLAlchemyGroupSpecVisitor())
stmt = select(*grp.group_by, *grp.agg).group_by(*grp.group_by)
# SELECT category, region,
#        count(*) AS total,
#        min(price) AS min_price,
#        avg(price) AS avg_price,
#        percentile_cont(0.95) WITHIN GROUP (ORDER BY latency ASC) AS p95_latency
# GROUP BY category, region

Aggregation Function Mapping

Function SQL
COUNT count(*)
SUM sum(col)
MIN min(col)
MAX max(col)
AVERAGE avg(col)
STDDEV stddev(col)
MEDIAN percentile_cont(0.5) WITHIN GROUP (ORDER BY col ASC)
Q1 percentile_cont(0.25) WITHIN GROUP (ORDER BY col ASC)
Q3 percentile_cont(0.75) WITHIN GROUP (ORDER BY col ASC)
P01P99 percentile_cont(q) WITHIN GROUP (ORDER BY col ASC)

Note: Percentile functions use PostgreSQL's ordered-set aggregate syntax. They will fail at execution time on databases that do not support this syntax (e.g. SQLite, older MySQL).