from collections.abc import Iterable, Sequence from typing import Any, Protocol, Self from jinja2sql import Jinja2SQL, identifier from .utils import TransformTo, normalize_order_by class HasChoices(Protocol): @classmethod def choices(cls) -> list[Self]: ... @property def value(self) -> Self: ... def orderby_filter( self: Jinja2SQL, value: str | Sequence[str], transform_to: TransformTo = "snakecase", ci_fields: Sequence[str] | None = None, ) -> str: if ci_fields: return orderby_ci_filter( self, value, ci_fields=ci_fields, transform_to=transform_to ) if isinstance(value, str): value = [value] return identifier( self, ", ".join( f"{column} {direction} {nulls}".strip() for column, direction, nulls in ( normalize_order_by(v, transform_to=transform_to) for v in value ) ), ) def orderby_ci_filter( self: Jinja2SQL, value: str | Sequence[str], *, ci_fields: Sequence[str], transform_to: TransformTo = "snakecase", ) -> str: if isinstance(value, str): value = [value] ci = set(ci_fields) parts = [] for column, direction, nulls in ( normalize_order_by(v, transform_to=transform_to) for v in value ): col = f"{column} COLLATE 'und-ci-ai'" if column in ci else column part = f"{col} {direction} {nulls}".strip() parts.append(part) return identifier(self, ", ".join(parts)) def enum_values(self: Jinja2SQL, enum_cls: Iterable[Any]) -> str: return identifier( self, ", ".join( [f"('{index}', '{item.value}')" for index, item in enumerate(enum_cls)] ), ) def enum_choices(self: Jinja2SQL, enum_cls: HasChoices) -> str: return identifier( self, ", ".join( [ f"('{index}', '{item.value}')" for index, item in enumerate(enum_cls.choices()) ] ), ) def array_values(self: Jinja2SQL, enum_cls: Iterable[Any]) -> str: return identifier( self, ", ".join([f"'{item.value}'" for _, item in enumerate(enum_cls)]) ) def escape_like(value: str, escape_char: str = "*") -> str: return ( value.replace(escape_char, escape_char * 2) .replace("%", escape_char + "%") .replace("_", escape_char + "_") )