Skip to content

arcanus.criteria

Pagination and filtering primitives.

arcanus.criteria

ScalarCriteria

Bases: ModelGeneric[P]

Source code in arcanus/criteria.py
class ScalarCriteria(ModelGeneric[P]):
    model_config = ConfigDict(
        populate_by_name=True,
        validate_by_alias=True,
        validate_by_name=True,
        arbitrary_types_allowed=True,
        extra="forbid",
    )

    criteria_type_mapping: ClassVar[
        dict[CriteriaValueType, type[BaseCriteria[Any]]]
    ] = {
        str: TextCriteria[str],
        UUID: NumericCriteria[UUID],
        int: NumericCriteria[int],
        float: NumericCriteria[float],
        bool: NumericCriteria[bool],
        datetime: NumericCriteria[datetime],
    }

    @classmethod
    def __get_generic_model_value_annotations__(
        cls, generic_model: TransmuterType
    ) -> dict[str, Any]:
        """Names and value annotations of criteria-eligible attributes.

        Regular scalar fields plus ``@provided`` computed fields (keyed by
        return type). Marking a computed field with :func:`arcanus.provided`
        declares it is backed by a provider-side expression (e.g. a SQLAlchemy
        ``hybrid_property``), making it filterable and orderable like a plain
        column; unmarked computed fields stay serialization-only.

        Fields marked ``exclude=True`` are hidden from serialization, so they
        are hidden here too: a value a client cannot read must not be
        filterable or orderable, or the criteria surface becomes an oracle for
        probing it.
        """
        annotations = {
            name: info.annotation
            for name, info in generic_model.__pydantic_fields__.items()
            if name not in generic_model.model_associations and not info.exclude
        }
        for name, computed in provided_computed_fields(generic_model).items():
            annotations.setdefault(name, computed.return_type)
        return annotations

    @classmethod
    def __get_generic_model_field_definitions__(
        cls, generic_model: TransmuterType
    ) -> dict[str, Any]:
        scalar_fields: dict[str, Any] = {}

        for name, annotation in cls.__get_generic_model_value_annotations__(
            generic_model
        ).items():
            # Normalize scalar field annotations before mapping them to criteria types.
            origin = get_origin(annotation)
            if origin is UnionType or str(origin) == "typing.Union":
                annotation = next(
                    (arg for arg in get_args(annotation) if arg is not type(None)),
                    object,
                )
                origin = get_origin(annotation)
            if origin is Literal:
                literal_values = get_args(annotation)
                if not literal_values:
                    continue
                criteria_type = cast(
                    type[BaseCriteria[Any]], cast(Any, ExactCriteria)[annotation]
                )
                scalar_fields[name] = (
                    criteria_type | None,
                    Field(default=None),
                )
                continue
            else:
                try:
                    if is_association(cast(type[Any], annotation)):
                        annotation = object
                except TypeError:
                    annotation = object
            if isinstance(annotation, type) and annotation in cls.criteria_type_mapping:
                criteria_value_type = cast(CriteriaValueType, annotation)
                criteria_type = cast(
                    type[BaseCriteria[Any]],
                    cls.criteria_type_mapping[criteria_value_type],
                )
                scalar_fields[name] = (
                    criteria_type | None,
                    Field(default=None),
                )

        return scalar_fields

    @classmethod
    def __get_generic_model_examples__(
        cls, generic_model: TransmuterType
    ) -> dict[str, CriteriaExample]:
        scalar_examples: dict[str, CriteriaExample] = {}

        def example_value(annotation: CriteriaValueType) -> CriteriaExampleValue:
            if annotation is str:
                return "string"
            if annotation is UUID:
                return "3fa85f64-5717-4562-b3fc-2c963f66afa6"
            if annotation is bool:
                return True
            if annotation is datetime:
                return "2026-05-25T16:59:31.252Z"
            if annotation is int:
                return 1
            if annotation is float:
                return 1.0
            return "string"

        def criteria_example(
            criteria_type: type[BaseCriteria[Any]], value: CriteriaExampleValue
        ) -> CriteriaExample:
            def attribute_example(attribute: str) -> Any:
                if attribute in {"in_", "not_in"}:
                    return [value]
                if attribute == "is_null":
                    return True
                return value

            return {
                (criteria_type.model_fields[attribute].alias or attribute): (
                    attribute_example(attribute)
                )
                for attribute, _ in criteria_type.criteria_operators
            }

        for name, annotation in cls.__get_generic_model_value_annotations__(
            generic_model
        ).items():
            origin = get_origin(annotation)
            if origin is UnionType or str(origin) == "typing.Union":
                annotation = next(
                    (arg for arg in get_args(annotation) if arg is not type(None)),
                    object,
                )
                origin = get_origin(annotation)
            if origin is Literal:
                literal_values = get_args(annotation)
                if not literal_values:
                    continue
                criteria_type = cast(
                    type[BaseCriteria[Any]], cast(Any, ExactCriteria)[annotation]
                )
                example = literal_values[0]
                scalar_examples[name] = criteria_example(
                    criteria_type,
                    example
                    if isinstance(example, (str, int, float, bool))
                    else "string",
                )
                continue
            else:
                try:
                    if is_association(cast(type[Any], annotation)):
                        annotation = object
                except TypeError:
                    annotation = object
            if isinstance(annotation, type) and annotation in cls.criteria_type_mapping:
                criteria_value_type = cast(CriteriaValueType, annotation)
                criteria_type = cast(
                    type[BaseCriteria[Any]],
                    cls.criteria_type_mapping[criteria_value_type],
                )
                scalar_examples[name] = criteria_example(
                    criteria_type, example_value(criteria_value_type)
                )
        return scalar_examples

    @cached_property
    def expressions(self) -> tuple[Expression[bool], ...]:
        expressions: list[Expression[bool]] = []
        generic_model = type(self).generic_model
        reserved = {"limit", "offset", "order_by", "cursor"}

        for name, value in self:
            if value is None or name in reserved:
                continue
            if isinstance(value, BaseCriteria):
                column = cast(
                    Column[CriteriaValue],
                    generic_model[name],
                )
                column_expressions = [
                    column.operate(operator, operator_value)
                    for attribute, operator in value.criteria_operators
                    if (operator_value := getattr(value, attribute)) is not None
                ]
                expressions.extend(column_expressions)

        return tuple(expressions)

with_identity_tiebreak

with_identity_tiebreak(entity: type[object], order_bys: tuple[T, ...]) -> tuple[T | Order[CriteriaValue], ...]

The ordering extended with the entity's identity keys, if absent.

Cursor pagination needs a total order: a nullable or non-unique trailing key cannot mint an unambiguous bookmark — a null yields no comparison at its level, and rows tied at a page boundary get skipped. Identity fields are unique and non-null, so every ordering ends with them, descending to match the newest-first default. Empty orderings, non-transmuter entities, transmuters without identity fields, and orderings that already name an identity pass through unchanged.

Source code in arcanus/criteria.py
def with_identity_tiebreak(
    entity: type[object],
    order_bys: tuple[T, ...],
) -> tuple[T | Order[CriteriaValue], ...]:
    """The ordering extended with the entity's identity keys, if absent.

    Cursor pagination needs a total order: a nullable or non-unique trailing
    key cannot mint an unambiguous bookmark — a null yields no comparison at
    its level, and rows tied at a page boundary get skipped. Identity fields
    are unique and non-null, so every ordering ends with them, descending to
    match the newest-first default. Empty orderings, non-transmuter entities,
    transmuters without identity fields, and orderings that already name an
    identity pass through unchanged.
    """
    if not order_bys or not isinstance(entity, TransmuterMetaclass):
        return order_bys
    identities = entity.model_identities
    ordered_names = {
        order.column.field_name for order in order_bys if isinstance(order, Order)
    }
    if not identities or ordered_names.intersection(identities):
        return order_bys
    return (*order_bys, *(entity[name].desc() for name in identities))

bookmark_after

bookmark_after(column: Column[CriteriaValue], value: CriteriaValue | None, descending: bool) -> Expression[bool] | None

Strictly-after comparison for one order key of a cursor bookmark.

Ordering is normalized to ASC NULLS LAST / DESC NULLS FIRST (the PostgreSQL defaults), so: - ascending, after a value: greater values, then the trailing null region; - ascending, after a null: nothing at this level (deeper keys tiebreak); - descending, after a value: smaller values only (nulls already passed); - descending, after a null: the entire non-null region.

Source code in arcanus/criteria.py
def bookmark_after(
    column: Column[CriteriaValue], value: CriteriaValue | None, descending: bool
) -> Expression[bool] | None:
    """Strictly-after comparison for one order key of a cursor bookmark.

    Ordering is normalized to ASC NULLS LAST / DESC NULLS FIRST (the
    PostgreSQL defaults), so:
    - ascending, after a value: greater values, then the trailing null region;
    - ascending, after a null: nothing at this level (deeper keys tiebreak);
    - descending, after a value: smaller values only (nulls already passed);
    - descending, after a null: the entire non-null region.
    """
    if value is None:
        if descending:
            return column.operate("is_null", False)
        return None
    if descending:
        return column < value
    return or_((column > value, column.operate("is_null", True)))