Query expressions¶
Expression helpers build serializable SQL AST fragments consumed by the Rust compiler.
Use these helpers when dictionary filters are not enough for the query you need.
Core classes¶
dataclass
¶
QueryExpression(connector, legacy_filters=None, children=(), expr=None)
Boolean query expression used by filters and typed SQL compilation.
to_where()
Lower simple AND expressions into the current Rust filter contract.
Source code in ormdantic/expressions.py
def to_where(self) -> dict[str, Any]:
"""Lower simple AND expressions into the current Rust filter contract."""
if self.connector == "leaf":
return self._legacy_filter_payload()
if self.connector == "and":
merged: dict[str, Any] = {}
for child in self.children:
merged.update(child.to_where())
return merged
raise ValueError("OR expressions require native disjunction support")
to_filter_tree()
Return the recursive payload consumed by the Rust filter normalizer.
Source code in ormdantic/expressions.py
def to_filter_tree(self) -> dict[str, Any]:
"""Return the recursive payload consumed by the Rust filter normalizer."""
if self.connector == "leaf":
return {"connector": "leaf", "filters": self._legacy_filter_payload()}
return {
"connector": self.connector,
"children": [child.to_filter_tree() for child in self.children],
}
supports_legacy_filters()
Return whether this expression can lower into legacy filter specs.
Source code in ormdantic/expressions.py
def supports_legacy_filters(self) -> bool:
"""Return whether this expression can lower into legacy filter specs."""
if self.connector == "leaf":
return self.legacy_filters is not None
return all(child.supports_legacy_filters() for child in self.children)
to_expression_payload(ctx=None)
Source code in ormdantic/expressions.py
def to_expression_payload(
self, ctx: SerializationContext | None = None
) -> dict[str, Any]:
if self.expr is None:
raise ValueError("query expression does not have a typed SQL expression")
return self.expr.to_expression_payload(ctx)
dataclass
¶
RelationExpression(source_table, source_pk, relationship, target_table, target_pk, target_alias, correlation_source_column, correlation_target_column, kind, outer_alias=None)
Registered relationship helper for typed predicates and aggregate ordering.
column(name)
Return a target-table column qualified to this relation subquery.
Source code in ormdantic/expressions.py
def column(self, name: str) -> ColumnExpression:
"""Return a target-table column qualified to this relation subquery."""
return column(name, table=self.target_alias)
any(where=None)
Match rows where a collection relationship has at least one row.
Source code in ormdantic/expressions.py
def any(self, where: QueryExpression | None = None) -> QueryExpression:
"""Match rows where a collection relationship has at least one row."""
self._require_kind("collection", "any")
return exists(self._select(literal(1), where=where))
none(where=None)
Match rows where no related row satisfies the optional predicate.
Source code in ormdantic/expressions.py
def none(self, where: QueryExpression | None = None) -> QueryExpression:
"""Match rows where no related row satisfies the optional predicate."""
return not_exists(self._select(literal(1), where=where))
every(where)
Match collection rows where every related row satisfies a predicate.
Source code in ormdantic/expressions.py
def every(self, where: QueryExpression) -> QueryExpression:
"""Match collection rows where every related row satisfies a predicate."""
self._require_kind("collection", "every")
return not_exists(self._select(literal(1), where=not_(where)))
has(where=None)
Match rows where a scalar relationship target satisfies a predicate.
Source code in ormdantic/expressions.py
def has(self, where: QueryExpression | None = None) -> QueryExpression:
"""Match rows where a scalar relationship target satisfies a predicate."""
self._require_kind("scalar", "has")
return exists(self._select(literal(1), where=where))
count(where=None)
Return a scalar subquery counting rows for this relationship.
Source code in ormdantic/expressions.py
def count(self, where: QueryExpression | None = None) -> SqlExpression:
"""Return a scalar subquery counting rows for this relationship."""
return subquery(self._select(func("COUNT", raw_sql_safe("*")), where=where))
ColumnExpression(name, table=None)
Bases: SqlExpression
Column helper for building query expressions.
Source code in ormdantic/expressions.py
def __init__(self, name: str, table: str | None = None) -> None:
object.__setattr__(self, "name", name)
object.__setattr__(self, "table", table)
object.__setattr__(self, "kind", "column")
data: dict[str, Any] = {"name": name}
if table is not None:
data["table"] = table
object.__setattr__(self, "data", data)
to_expression_payload(ctx=None)
Source code in ormdantic/expressions.py
def to_expression_payload(
self, ctx: SerializationContext | None = None
) -> dict[str, Any]:
ctx = ctx or SerializationContext()
kind = self.kind
if kind in {"column", "param", "literal", "raw_safe"}:
return {"kind": kind, **self.data}
if kind == "binary":
return {
"kind": "binary",
"op": self.data["op"],
"left": expression_payload(self.data["left"], ctx),
"right": expression_payload(self.data["right"], ctx),
}
if kind == "unary":
return {
"kind": "unary",
"op": self.data["op"],
"expr": expression_payload(self.data["expr"], ctx),
}
if kind == "function":
return {
"kind": "function",
"name": self.data["name"],
"args": [
expression_payload(arg, ctx) for arg in self.data.get("args", ())
],
}
if kind == "between":
return {
"kind": "between",
"expr": expression_payload(self.data["expr"], ctx),
"low": expression_payload(self.data["low"], ctx),
"high": expression_payload(self.data["high"], ctx),
}
if kind == "in_list":
return {
"kind": "in_list",
"expr": expression_payload(self.data["expr"], ctx),
"values": [
expression_payload(value, ctx)
for value in self.data.get("values", ())
],
"negated": self.data.get("negated", False),
}
if kind == "case":
return {
"kind": "case",
"whens": [
{
"when": expression_payload(condition, ctx),
"then": expression_payload(value, ctx),
}
for condition, value in self.data.get("whens", ())
],
"else": (
expression_payload(self.data["else"], ctx)
if "else" in self.data
else None
),
}
if kind == "cast":
return {
"kind": "cast",
"expr": expression_payload(self.data["expr"], ctx),
"type": self.data["type"],
}
if kind == "tuple":
return {
"kind": "tuple",
"values": [
expression_payload(value, ctx)
for value in self.data.get("values", ())
],
}
if kind == "subquery":
return {
"kind": "subquery",
"query": self.data["query"].to_query_payload(ctx),
}
if kind == "exists":
return {
"kind": "exists",
"query": self.data["query"].to_query_payload(ctx),
"negated": self.data.get("negated", False),
}
if kind == "in_subquery":
return {
"kind": "in_subquery",
"expr": expression_payload(self.data["expr"], ctx),
"query": self.data["query"].to_query_payload(ctx),
"negated": self.data.get("negated", False),
}
if kind == "window":
return {
"kind": "window",
"expr": expression_payload(self.data["expr"], ctx),
"partition_by": [
expression_payload(expr, ctx)
for expr in self.data.get("partition_by", ())
],
"order_by": [
window_order_payload(order, ctx)
for order in self.data.get("order_by", ())
],
}
raise ValueError(f"unsupported expression kind '{kind}'")
as_(alias)
Alias this expression when used as a projection.
Source code in ormdantic/expressions.py
def as_(self, alias: str) -> "ProjectionExpression":
"""Alias this expression when used as a projection."""
return ProjectionExpression(self, alias)
asc(*, nulls=None)
Order by this expression ascending.
Source code in ormdantic/expressions.py
def asc(
self, *, nulls: Literal["first", "last"] | None = None
) -> "OrderExpression":
"""Order by this expression ascending."""
return OrderExpression(self, "asc", nulls)
desc(*, nulls=None)
Order by this expression descending.
Source code in ormdantic/expressions.py
def desc(
self, *, nulls: Literal["first", "last"] | None = None
) -> "OrderExpression":
"""Order by this expression descending."""
return OrderExpression(self, "desc", nulls)
over(*, partition_by=(), order_by=())
Use this expression as a SQL window expression.
Source code in ormdantic/expressions.py
def over(
self,
*,
partition_by: Sequence[SerializableExpression] = (),
order_by: Sequence["OrderExpression | SerializableExpression"] = (),
) -> "SqlExpression":
"""Use this expression as a SQL window expression."""
return over(self, partition_by=partition_by, order_by=order_by)
eq(value)
Source code in ormdantic/expressions.py
def eq(self, value: Any) -> "QueryExpression":
if value is None:
return self.is_null()
return predicate("eq", self, value)
ne(value)
Source code in ormdantic/expressions.py
def ne(self, value: Any) -> "QueryExpression":
if value is None:
return self.is_not_null()
return predicate("ne", self, value)
lt(value)
Source code in ormdantic/expressions.py
def lt(self, value: Any) -> "QueryExpression":
return predicate("lt", self, value)
le(value)
Source code in ormdantic/expressions.py
def le(self, value: Any) -> "QueryExpression":
return predicate("le", self, value)
gt(value)
Source code in ormdantic/expressions.py
def gt(self, value: Any) -> "QueryExpression":
return predicate("gt", self, value)
ge(value)
Source code in ormdantic/expressions.py
def ge(self, value: Any) -> "QueryExpression":
return predicate("ge", self, value)
like(value)
Source code in ormdantic/expressions.py
def like(self, value: str) -> "QueryExpression":
return predicate("like", self, value)
ilike(value)
Source code in ormdantic/expressions.py
def ilike(self, value: str) -> "QueryExpression":
return predicate("ilike", self, value)
contains(value)
Source code in ormdantic/expressions.py
def contains(self, value: str) -> "QueryExpression":
return self.like(f"%{value}%")
icontains(value)
Source code in ormdantic/expressions.py
def icontains(self, value: str) -> "QueryExpression":
return self.ilike(f"%{value}%")
startswith(value)
Source code in ormdantic/expressions.py
def startswith(self, value: str) -> "QueryExpression":
return self.like(f"{value}%")
istartswith(value)
Source code in ormdantic/expressions.py
def istartswith(self, value: str) -> "QueryExpression":
return self.ilike(f"{value}%")
endswith(value)
Source code in ormdantic/expressions.py
def endswith(self, value: str) -> "QueryExpression":
return self.like(f"%{value}")
iendswith(value)
Source code in ormdantic/expressions.py
def iendswith(self, value: str) -> "QueryExpression":
return self.ilike(f"%{value}")
between(low, high)
Source code in ormdantic/expressions.py
def between(self, low: Any, high: Any) -> "QueryExpression":
return QueryExpression(
"leaf",
legacy_filters=self._legacy_range("ge", low)
| self._legacy_range("le", high),
expr=SqlExpression(
"between",
{"expr": self, "low": bind_value(low), "high": bind_value(high)},
),
)
not_between(low, high)
Source code in ormdantic/expressions.py
def not_between(self, low: Any, high: Any) -> "QueryExpression":
return not_(self.between(low, high))
in_(values)
Source code in ormdantic/expressions.py
def in_(self, values: Sequence[Any]) -> "QueryExpression":
return QueryExpression(
"leaf",
legacy_filters=self._legacy_filter("in", list(values)),
expr=SqlExpression(
"in_list",
{
"expr": self,
"values": [bind_value(value) for value in values],
"negated": False,
},
),
)
in_query(query, *, negated=False)
Compare this expression against the result of a typed subquery.
Source code in ormdantic/expressions.py
def in_query(
self, query: "SelectExpressionQuery", *, negated: bool = False
) -> "QueryExpression":
"""Compare this expression against the result of a typed subquery."""
return QueryExpression(
"leaf",
expr=SqlExpression(
"in_subquery",
{
"expr": self,
"query": query,
"negated": negated,
},
),
)
not_in(values)
Source code in ormdantic/expressions.py
def not_in(self, values: Sequence[Any]) -> "QueryExpression":
return QueryExpression(
"leaf",
legacy_filters=self._legacy_filter("not_in", list(values)),
expr=SqlExpression(
"in_list",
{
"expr": self,
"values": [bind_value(value) for value in values],
"negated": True,
},
),
)
not_in_query(query)
Compare this expression against rows absent from a typed subquery.
Source code in ormdantic/expressions.py
def not_in_query(self, query: "SelectExpressionQuery") -> "QueryExpression":
"""Compare this expression against rows absent from a typed subquery."""
return self.in_query(query, negated=True)
is_null()
Source code in ormdantic/expressions.py
def is_null(self) -> "QueryExpression":
return QueryExpression(
"leaf",
legacy_filters=self._legacy_filter("is_null", True),
expr=SqlExpression("unary", {"op": "is_null", "expr": self}),
)
is_not_null()
Source code in ormdantic/expressions.py
def is_not_null(self) -> "QueryExpression":
return QueryExpression(
"leaf",
legacy_filters=self._legacy_filter("is_not_null", True),
expr=SqlExpression("unary", {"op": "is_not_null", "expr": self}),
)
is_(value)
Source code in ormdantic/expressions.py
def is_(self, value: Any) -> "QueryExpression":
if value is not None:
raise ValueError("only None is supported by is_()")
return self.is_null()
is_not(value)
Source code in ormdantic/expressions.py
def is_not(self, value: Any) -> "QueryExpression":
if value is not None:
raise ValueError("only None is supported by is_not()")
return self.is_not_null()
cast(type_name)
Cast this expression to a SQL type.
Source code in ormdantic/expressions.py
def cast(self, type_name: str) -> "SqlExpression":
"""Cast this expression to a SQL type."""
return SqlExpression("cast", {"expr": self, "type": type_name})
set(value)
Assign this column to a typed expression or bound value.
Source code in ormdantic/expressions.py
def set(self, value: Any) -> "AssignmentExpression":
"""Assign this column to a typed expression or bound value."""
if self.kind != "column":
raise ValueError("only column expressions can be assigned")
return AssignmentExpression(self.data["name"], value)
dataclass
¶
SelectExpressionQuery(table, projections, table_alias=None, ctes=(), where=None, group_by=(), having=None, order_by=(), limit=None, offset=None, distinct=False)
Serializable SELECT statement for the Rust SQL AST compiler.
to_query_payload(ctx=None)
Source code in ormdantic/expressions.py
def to_query_payload(
self, ctx: SerializationContext | None = None
) -> dict[str, Any]:
ctx = ctx or SerializationContext()
payload: dict[str, Any] = {
"table": self.table,
"values": ctx.values,
}
if self.table_alias is not None:
payload["table_alias"] = self.table_alias
if self.ctes:
payload["ctes"] = [cte.to_cte_payload(ctx) for cte in self.ctes]
payload["projections"] = [
projection.to_projection_payload(ctx) for projection in self.projections
]
if self.where is not None:
payload["where"] = self.where.to_expression_payload(ctx)
if self.group_by:
payload["group_by"] = [
expression_payload(expr, ctx) for expr in self.group_by
]
if self.having is not None:
payload["having"] = self.having.to_expression_payload(ctx)
if self.order_by:
payload["order_by"] = [
order.to_order_payload(ctx) for order in self.order_by
]
if self.limit is not None:
payload["limit"] = self.limit
if self.offset is not None:
payload["offset"] = self.offset
if self.distinct:
payload["distinct"] = True
payload["values"] = ctx.values
return payload
to_expression_payload(ctx=None)
Serialize this SELECT as a scalar subquery expression.
Source code in ormdantic/expressions.py
def to_expression_payload(
self, ctx: SerializationContext | None = None
) -> dict[str, Any]:
"""Serialize this SELECT as a scalar subquery expression."""
return subquery(self).to_expression_payload(ctx)
with_cte(*ctes)
Return a copy with additional common table expressions.
Source code in ormdantic/expressions.py
def with_cte(self, *ctes: "CommonTableExpression") -> "SelectExpressionQuery":
"""Return a copy with additional common table expressions."""
return replace(self, ctes=self.ctes + tuple(ctes))
dataclass
¶
UpdateExpressionQuery(table, assignments, where=None)
Serializable UPDATE statement for the Rust SQL AST compiler.
to_query_payload()
Source code in ormdantic/expressions.py
def to_query_payload(self) -> dict[str, Any]:
ctx = SerializationContext()
payload: dict[str, Any] = {
"table": self.table,
"assignments": [
assignment.to_assignment_payload(ctx) for assignment in self.assignments
],
"values": ctx.values,
}
if self.where is not None:
payload["where"] = self.where.to_expression_payload(ctx)
payload["values"] = ctx.values
return payload
Helper functions¶
column(name, *, table=None)
Create a query expression column reference.
Source code in ormdantic/expressions.py
def column(name: str, *, table: str | None = None) -> ColumnExpression:
"""Create a query expression column reference."""
return ColumnExpression(name, table)
projection(expr, alias=None)
Create a selectable projection.
Source code in ormdantic/expressions.py
def projection(
expr: SerializableExpression, alias: str | None = None
) -> ProjectionExpression:
"""Create a selectable projection."""
return ProjectionExpression(expr, alias)
assignment(column_name, expr)
Create a typed UPDATE assignment.
Source code in ormdantic/expressions.py
def assignment(
column_name: str, expr: SerializableExpression | Any
) -> AssignmentExpression:
"""Create a typed UPDATE assignment."""
return AssignmentExpression(column_name, expr)
select_query(table, *projections, table_alias=None, with_=(), where=None, group_by=(), having=None, order_by=(), limit=None, offset=None, distinct=False)
Build a serializable SELECT expression query.
Source code in ormdantic/expressions.py
def select_query(
table: str,
*projections: ProjectionExpression | SerializableExpression,
table_alias: str | None = None,
with_: Sequence[CommonTableExpression] = (),
where: QueryExpression | None = None,
group_by: Sequence[SerializableExpression] = (),
having: QueryExpression | None = None,
order_by: Sequence[OrderExpression] = (),
limit: int | None = None,
offset: int | None = None,
distinct: bool = False,
) -> SelectExpressionQuery:
"""Build a serializable SELECT expression query."""
normalized = tuple(
item if isinstance(item, ProjectionExpression) else ProjectionExpression(item)
for item in projections
)
return SelectExpressionQuery(
table=table,
projections=normalized,
table_alias=table_alias,
ctes=tuple(with_),
where=where,
group_by=tuple(group_by),
having=having,
order_by=tuple(order_by),
limit=limit,
offset=offset,
distinct=distinct,
)
update_query(table, *assignments, where=None)
Build a serializable UPDATE expression query.
Source code in ormdantic/expressions.py
def update_query(
table: str,
*assignments: AssignmentExpression,
where: QueryExpression | None = None,
) -> UpdateExpressionQuery:
"""Build a serializable UPDATE expression query."""
return UpdateExpressionQuery(table=table, assignments=assignments, where=where)
literal(value)
Create an inline SQL literal.
Source code in ormdantic/expressions.py
def literal(value: None | bool | int | str) -> SqlExpression:
"""Create an inline SQL literal."""
return SqlExpression("literal", {"value": value})
raw_sql_safe(sql)
Opt into a raw SQL fragment that is trusted by the caller.
Source code in ormdantic/expressions.py
def raw_sql_safe(sql: str) -> SqlExpression:
"""Opt into a raw SQL fragment that is trusted by the caller."""
return SqlExpression("raw_safe", {"sql": sql})
case(*whens, else_=None)
Create a SQL CASE expression.
Source code in ormdantic/expressions.py
def case(
*whens: tuple[QueryExpression, SerializableExpression | Any],
else_: SerializableExpression | Any | None = None,
) -> SqlExpression:
"""Create a SQL CASE expression."""
data: dict[str, Any] = {"whens": whens}
if else_ is not None:
data["else"] = else_
return SqlExpression("case", data)
cast(expr, type_name)
Cast an expression to a SQL type.
Source code in ormdantic/expressions.py
def cast(expr: SerializableExpression | Any, type_name: str) -> SqlExpression:
"""Cast an expression to a SQL type."""
return SqlExpression("cast", {"expr": expr, "type": type_name})
tuple_(*values)
Create a SQL tuple expression.
Source code in ormdantic/expressions.py
def tuple_(*values: SerializableExpression | Any) -> SqlExpression:
"""Create a SQL tuple expression."""
return SqlExpression("tuple", {"values": values})
cte(name, query, *, columns=(), recursive=False)
Create a common table expression for a typed SELECT query.
Source code in ormdantic/expressions.py
def cte(
name: str,
query: SelectExpressionQuery,
*,
columns: Sequence[str] = (),
recursive: bool = False,
) -> CommonTableExpression:
"""Create a common table expression for a typed SELECT query."""
return CommonTableExpression(
name=name,
query=query,
columns=tuple(columns),
recursive=recursive,
)
count(expr=None)
Source code in ormdantic/expressions.py
def count(expr: SerializableExpression | None = None) -> SqlExpression:
return func("COUNT", raw_sql_safe("*") if expr is None else expr)
sum(expr)
Source code in ormdantic/expressions.py
def sum(expr: SerializableExpression) -> SqlExpression: # noqa: A001
return func("SUM", expr)
avg(expr)
Source code in ormdantic/expressions.py
def avg(expr: SerializableExpression) -> SqlExpression:
return func("AVG", expr)
min(expr)
Source code in ormdantic/expressions.py
def min(expr: SerializableExpression) -> SqlExpression: # noqa: A001
return func("MIN", expr)
max(expr)
Source code in ormdantic/expressions.py
def max(expr: SerializableExpression) -> SqlExpression: # noqa: A001
return func("MAX", expr)
exists(query, *, negated=False)
Build an EXISTS (SELECT ...) predicate.
Source code in ormdantic/expressions.py
def exists(query: "SelectExpressionQuery", *, negated: bool = False) -> QueryExpression:
"""Build an `EXISTS (SELECT ...)` predicate."""
return QueryExpression(
"leaf",
expr=SqlExpression("exists", {"query": query, "negated": negated}),
)
not_exists(query)
Build a NOT EXISTS (SELECT ...) predicate.
Source code in ormdantic/expressions.py
def not_exists(query: "SelectExpressionQuery") -> QueryExpression:
"""Build a `NOT EXISTS (SELECT ...)` predicate."""
return exists(query, negated=True)
not_(expr)
Negate a boolean expression.
Source code in ormdantic/expressions.py
def not_(expr: QueryExpression) -> QueryExpression:
"""Negate a boolean expression."""
return QueryExpression(
"leaf", expr=SqlExpression("unary", {"op": "not", "expr": expr})
)
group(expr)
Explicit grouping helper for readability.
Source code in ormdantic/expressions.py
def group(expr: QueryExpression) -> QueryExpression:
"""Explicit grouping helper for readability."""
return QueryExpression("leaf", expr=expr.expr)
over(expr, *, partition_by=(), order_by=())
Create a SQL window expression.
Source code in ormdantic/expressions.py
def over(
expr: SerializableExpression,
*,
partition_by: Sequence[SerializableExpression] = (),
order_by: Sequence["OrderExpression | SerializableExpression"] = (),
) -> SqlExpression:
"""Create a SQL window expression."""
return SqlExpression(
"window",
{
"expr": expr,
"partition_by": tuple(partition_by),
"order_by": tuple(order_by),
},
)
subquery(query)
Use a typed SELECT as a scalar subquery expression.
Source code in ormdantic/expressions.py
def subquery(query: "SelectExpressionQuery") -> SqlExpression:
"""Use a typed SELECT as a scalar subquery expression."""
return SqlExpression("subquery", {"query": query})