from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Literal, Sequence
from sqlalchemy import inspect as sqlalchemy_inspect
from sqlalchemy.orm import aliased
from sqlalchemy.sql.selectable import Select
from edri.integrations.sqlalchemy.path import SQLAlchemyPathError, canonical_orm_path
__all__ = ["ResolvedSort", "SortBuilder", "apply_sorting"]
[docs]
@dataclass(frozen=True, slots=True)
class ResolvedSort:
"""A validated sort path; applying SQL joins is deliberately separate."""
root_model: type[Any]
path: tuple[str, ...]
order: Literal["asc", "desc"]
@property
def relationships(self) -> tuple[str, ...]:
return self.path[:-1]
@property
def column_name(self) -> str:
return self.path[-1]
[docs]
class SortBuilder:
"""Resolve explicit client sort fields to strict mapped ORM paths.
The builder performs no relationship guessing, raw SQL construction, or
implicit joins. A root field must be a scalar column. Every intermediate
segment in a dotted field must be a relationship and its final segment
must be a scalar column.
"""
[docs]
@staticmethod
def build(
table: Any | Sequence[Any],
names: list[str],
orders: list[str],
logger: Any | None = None,
) -> list[ResolvedSort]:
if not names:
return []
roots = tuple(table) if isinstance(table, (list, tuple)) else (table,)
result: list[ResolvedSort] = []
for index, raw_name in enumerate(names):
raw_order = orders[index] if index < len(orders) else None
order = SortBuilder._normalize_order(raw_order)
if order is None:
if logger:
logger.debug(
"Ignoring sort for %r due to invalid order %r",
raw_name,
raw_order,
)
continue
name = (raw_name or "").strip()
if not name:
continue
match: tuple[type[Any], tuple[str, ...]] | None = None
for root in roots:
try:
path = canonical_orm_path(root, name, require_scalar=True)
except (SQLAlchemyPathError, TypeError):
continue
match = (root, path)
break
if match is None:
if logger:
logger.warning("Ignoring unknown or non-scalar sort field %r", name)
continue
root, path = match
result.append(ResolvedSort(root, path, order))
return result
@staticmethod
def _normalize_order(order: str | None) -> Literal["asc", "desc"] | None:
if not order:
return None
normalized = order.strip().lower()
return normalized if normalized in ("asc", "desc") else None
[docs]
def apply_sorting(statement: Select[Any], sorts: Sequence[ResolvedSort]) -> Select[Any]:
"""Explicitly add the JOINs and ORDER BY clauses required by sort plans."""
joined_entities: dict[tuple[type[Any], tuple[str, ...]], Any] = {}
for resolved in sorts:
current_entity: Any = resolved.root_model
for depth, relationship_name in enumerate(resolved.relationships, start=1):
relationship_path = resolved.relationships[:depth]
key = (resolved.root_model, relationship_path)
target = joined_entities.get(key)
if target is None:
mapper = sqlalchemy_inspect(current_entity).mapper
relationship = mapper.relationships[relationship_name]
root_name = sqlalchemy_inspect(resolved.root_model).local_table.name
alias_name = "sort_" + "_".join((root_name, *relationship_path))
target = aliased(relationship.mapper.class_, name=alias_name)
relationship_attr = getattr(current_entity, relationship_name).of_type(target)
statement = statement.outerjoin(target, relationship_attr)
joined_entities[key] = target
current_entity = target
column = getattr(current_entity, resolved.column_name)
statement = statement.order_by(getattr(column, resolved.order)())
return statement