Query Expressions¶

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

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],
    }

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)

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)

def column(self, name: str) -> ColumnExpression:
    """Return a target-table column qualified to this relation subquery."""
    return column(name, table=self.target_alias)

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

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

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

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

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

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}'")

def as_(self, alias: str) -> "ProjectionExpression":
    """Alias this expression when used as a projection."""
    return ProjectionExpression(self, alias)

def asc(
    self, *, nulls: Literal["first", "last"] | None = None
) -> "OrderExpression":
    """Order by this expression ascending."""
    return OrderExpression(self, "asc", nulls)

def desc(
    self, *, nulls: Literal["first", "last"] | None = None
) -> "OrderExpression":
    """Order by this expression descending."""
    return OrderExpression(self, "desc", nulls)

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)

def eq(self, value: Any) -> "QueryExpression":
    if value is None:
        return self.is_null()
    return predicate("eq", self, value)

def ne(self, value: Any) -> "QueryExpression":
    if value is None:
        return self.is_not_null()
    return predicate("ne", self, value)

def lt(self, value: Any) -> "QueryExpression":
    return predicate("lt", self, value)

def le(self, value: Any) -> "QueryExpression":
    return predicate("le", self, value)

def gt(self, value: Any) -> "QueryExpression":
    return predicate("gt", self, value)

def ge(self, value: Any) -> "QueryExpression":
    return predicate("ge", self, value)

def like(self, value: str) -> "QueryExpression":
    return predicate("like", self, value)

def ilike(self, value: str) -> "QueryExpression":
    return predicate("ilike", self, value)

def contains(self, value: str) -> "QueryExpression":
    return self.like(f"%{value}%")

def icontains(self, value: str) -> "QueryExpression":
    return self.ilike(f"%{value}%")

def startswith(self, value: str) -> "QueryExpression":
    return self.like(f"{value}%")

def istartswith(self, value: str) -> "QueryExpression":
    return self.ilike(f"{value}%")

def endswith(self, value: str) -> "QueryExpression":
    return self.like(f"%{value}")

def iendswith(self, value: str) -> "QueryExpression":
    return self.ilike(f"%{value}")

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)},
        ),
    )

def not_between(self, low: Any, high: Any) -> "QueryExpression":
    return not_(self.between(low, high))

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,
            },
        ),
    )

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,
            },
        ),
    )

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,
            },
        ),
    )

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)

def is_null(self) -> "QueryExpression":
    return QueryExpression(
        "leaf",
        legacy_filters=self._legacy_filter("is_null", True),
        expr=SqlExpression("unary", {"op": "is_null", "expr": self}),
    )

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}),
    )

def is_(self, value: Any) -> "QueryExpression":
    if value is not None:
        raise ValueError("only None is supported by is_()")
    return self.is_null()

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

def cast(self, type_name: str) -> "SqlExpression":
    """Cast this expression to a SQL type."""
    return SqlExpression("cast", {"expr": self, "type": type_name})

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)

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

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)

def with_cte(self, *ctes: "CommonTableExpression") -> "SelectExpressionQuery":
    """Return a copy with additional common table expressions."""
    return replace(self, ctes=self.ctes + tuple(ctes))

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