"""Typed protocols for the store/searcher query contract.
These protocols define the backend-agnostic query interface shared by httk
data stores: the database layer in httk-store implements them over SQL, and
serving modules (such as *httk-serve*, whose in-memory store also conforms)
program against them. They mirror the query interface of the httk v1 database
layer (``httk.db`` ``FilteredCollection`` searchers), so lightweight fakes can
stand in for a real store in tests.
"""
from collections.abc import Iterable, Iterator
from dataclasses import dataclass
from typing import Any, Final, Literal, NamedTuple, Protocol, Self
__all__ = [
"ID_FIELD",
"ContinuationToken",
"CountUnavailableError",
"MultipleResultsError",
"NoResultError",
"PageOrder",
"PageableResultSetLike",
"PaginationCursorError",
"ResultPage",
"ResultRow",
"ResultRowLike",
"ResultSetLike",
"SearchExpression",
"SearchField",
"SearchResult",
"SearchVariable",
"Searcher",
"Store",
"UnsupportedQueryError",
]
[docs]
ID_FIELD: Final = "__id"
"""The backend field name used for the served entry identifier."""
[docs]
class UnsupportedQueryError(ValueError):
"""Report that a valid query operation is outside a store's supported profile."""
[docs]
class CountUnavailableError(RuntimeError):
"""Report that a store cannot provide an exact query count."""
[docs]
class NoResultError(LookupError):
"""Report that a result-set ``one()`` operation found no matching result."""
[docs]
class MultipleResultsError(LookupError):
"""Report that a result-set ``one()`` operation found multiple results."""
@dataclass(frozen=True, slots=True)
[docs]
class PageOrder:
"""Order a continuation page by one named scalar result projection.
``name`` identifies the name supplied to ``results()`` (or
:meth:`Searcher.output`), never a backend column object. The result-set
implementation validates that it is a root scalar projection before it
generates SQL.
:param name: The declared scalar output name used for ordering.
:param descending: Whether to order this field in descending order.
:param nulls: Whether null values sort first or last.
"""
[docs]
descending: bool = False
[docs]
nulls: Literal["first", "last"] = "last"
def __post_init__(self) -> None:
if not isinstance(self.name, str) or not self.name:
raise ValueError("PageOrder.name must be a non-empty string")
if not isinstance(self.descending, bool):
raise TypeError("PageOrder.descending must be bool")
if self.nulls not in {"first", "last"}:
raise ValueError("PageOrder.nulls must be 'first' or 'last'")
[docs]
class ContinuationToken(str):
"""Carry an opaque URL-safe continuation value.
It is a ``str`` subclass so normal JSON serializers preserve it as a
scalar value. Applications should pass a token returned by a page back
unchanged; data backends validate its version, structure, and result-plan
fingerprint before using any decoded value as a bound parameter.
:param value: The opaque continuation value.
"""
def __new__(cls, value: str) -> Self:
if not isinstance(value, str):
raise TypeError(f"ContinuationToken requires str, got {type(value).__name__}")
return str.__new__(cls, value)
@dataclass(frozen=True, slots=True)
[docs]
class ResultPage:
"""Represent an immutable continuation-page result.
``rows`` is always a tuple. Returned rows are ordinary persistent result
rows, not the expiring proxies produced by ``SqlResultSet.cursor()``.
``total`` is populated only when the caller explicitly asks for it.
:param rows: The persistent rows returned by the page.
:param next: The token for the next page, if one exists.
:param previous: The token for the previous page, if one exists.
:param total: The exact result count when requested, otherwise ``None``.
"""
[docs]
rows: tuple["ResultRowLike", ...]
[docs]
next: ContinuationToken | None
[docs]
previous: ContinuationToken | None
[docs]
total: int | None = None
def __post_init__(self) -> None:
if not isinstance(self.rows, tuple):
object.__setattr__(self, "rows", tuple(self.rows))
if self.total is not None and (
isinstance(self.total, bool) or not isinstance(self.total, int) or self.total < 0
):
raise ValueError("ResultPage.total must be a non-negative integer or None")
[docs]
class PageableResultSetLike(Protocol):
"""Expose optional continuation-page capability on a frozen result set.
This deliberately extends neither :class:`ResultSetLike` nor
:class:`Searcher`: stores that do not support seek pagination remain fully
conforming to the required portable contracts.
"""
[docs]
def page(
self,
*,
size: int,
order_by: Iterable[PageOrder],
cursor: ContinuationToken | None = None,
include_total: bool = False,
) -> ResultPage:
"""Return one ordered continuation page."""
...
[docs]
class ResultRow:
"""Represent one named result row by position, name, or attribute.
:param values: The row values in declaration order.
:param names: The corresponding output names.
:param resolver: An optional lazy value resolver.
:param guard: An optional callback that rejects access to expired values.
"""
__slots__ = ("_guard", "_names", "_resolver", "_values")
def __init__(
self,
values: tuple[Any, ...],
names: tuple[str, ...],
resolver: Any = None,
guard: Any = None,
) -> None:
self._values = values
self._names = names
self._resolver = resolver
self._guard = guard
@property
[docs]
def names(self) -> tuple[str, ...]:
"""Return the declared output names."""
self._check()
return self._names
@property
[docs]
def values(self) -> tuple[Any, ...]:
"""Return the row values in declaration order."""
self._check()
return tuple(self[index] for index in range(len(self._values)))
def _check(self) -> None:
if self._guard is not None:
self._guard()
def _value(self, index: int) -> Any:
if self._guard is not None:
self._guard()
value = self._resolver(index, self._values[index]) if self._resolver is not None else self._values[index]
activate = getattr(value, "_activate", None)
if activate is not None:
activate()
return value
def __getitem__(self, key: int | str) -> Any:
if isinstance(key, str):
try:
key = self._names.index(key)
except ValueError:
raise KeyError(key) from None
if not isinstance(key, int):
raise TypeError(f"result row indices must be integers or strings, got {type(key).__name__}")
return self._value(key)
def __getattr__(self, name: str) -> Any:
if name.startswith("_"):
raise AttributeError(name)
try:
return self[name]
except KeyError:
raise AttributeError(name) from None
def __iter__(self) -> Iterator[Any]:
return iter(self.values)
def __len__(self) -> int:
self._check()
return len(self._values)
def __eq__(self, other: object) -> bool:
if not isinstance(other, ResultRow):
return NotImplemented
return self.values == other.values
def __repr__(self) -> str:
return f"ResultRow({dict(zip(self.names, self.values, strict=True))!r})"
def __copy__(self) -> "ResultRow":
if self._guard is not None:
raise TypeError("expired cursor rows cannot be copied")
return type(self)(self._values, self._names, self._resolver, self._guard)
def __deepcopy__(self, memo: dict[int, Any]) -> "ResultRow":
if self._guard is not None:
raise TypeError("expired cursor rows cannot be copied")
return type(self)(self._values, self._names, self._resolver, self._guard)
def __reduce_ex__(self, protocol: Any) -> Any:
if self._guard is not None:
raise TypeError("expired cursor rows cannot be pickled")
return type(self), (self._values, self._names, self._resolver, self._guard)
[docs]
class ResultRowLike(Protocol):
"""Require named access to one result row."""
@property
[docs]
def names(self) -> tuple[str, ...]:
"""Return the row's declared output names."""
...
def __getitem__(self, key: int | str) -> Any:
"""Return a row value by index or output name."""
...
[docs]
class ResultSetLike(Protocol):
"""Require the common operations of a materialized result set."""
def __iter__(self) -> Iterator[ResultRowLike]:
"""Iterate over result rows."""
...
def __len__(self) -> int:
"""Return the exact number of rows."""
...
[docs]
def first(self) -> ResultRowLike | None:
"""Return the first row, or ``None`` when no row matches."""
...
[docs]
def one(self) -> ResultRowLike:
"""Return the only row, or raise when the count is not one."""
...
[docs]
def scalars(self, name: str | None = None) -> Iterator[Any]:
"""Iterate over one named scalar output."""
...
# column() is an optional backend capability; SQL stores expose it.
[docs]
class SearchExpression(Protocol):
"""Require composable backend search expressions."""
def __and__(self, other: "SearchExpression") -> "SearchExpression":
"""Conjoin this expression with ``other``."""
...
def __or__(self, other: "SearchExpression") -> "SearchExpression":
"""Disjoin this expression with ``other``."""
...
def __invert__(self) -> "SearchExpression":
"""Negate this expression."""
...
[docs]
class SearchField(Protocol):
"""Expose a queryable field of a search variable.
In addition to the methods below, fields support the rich comparison
operators (``==``, ``!=``, ``<``, ``<=``, ``>``, ``>=``), returning
:class:`SearchExpression`. The handlers invoke those via
``getattr(field, '__eq__')(value)`` since the comparison dunders cannot be
typed as expression-returning.
The three string-matching methods take **literal** text: no wildcard or
pattern syntax whatsoever crosses this contract, so ``%`` and ``_`` (and
any other metacharacter) match themselves. A backend is therefore free to
implement them with SQL ``LIKE`` over an escaped pattern, with a regular
expression, or with a full-text index — the choice is invisible here.
"""
[docs]
def has(self, value: Any) -> SearchExpression:
"""Match a list field containing ``value``."""
...
[docs]
def has_any(self, *values: Any) -> SearchExpression:
"""Match a list field containing any of ``values``."""
...
[docs]
def has_only(self, *values: Any) -> SearchExpression:
"""Match a list field containing no values outside ``values``."""
...
[docs]
def is_in(self, *values: Any) -> SearchExpression:
"""Match a root scalar field whose value is one of ``values``.
``None`` is an explicit member: it matches a null field value, and its
negation excludes nulls rather than inheriting SQL's three-valued
``NOT IN (..., NULL)`` behavior.
Backends define the corresponding semantics for child or set fields;
for example, a backend may use the existing ``has_only``-style
all-values reading for a child field.
"""
...
[docs]
def contains(self, text: str) -> SearchExpression:
"""Match values containing ``text`` as a literal substring."""
...
[docs]
def startswith(self, prefix: str) -> SearchExpression:
"""Match values beginning with the literal ``prefix``."""
...
[docs]
def endswith(self, suffix: str) -> SearchExpression:
"""Match values ending with the literal ``suffix``."""
...
def __getattr__(self, name: str) -> "SearchField":
"""The field ``name`` of the record this field refers to.
A field may hold a *reference* to another stored record, and attribute
access chains into that record's own fields — ``variable.ref.doi`` —
to any depth. Whether a given field is a reference is a property of the
stored class, not of this contract, so the chain is only resolved when
the query is built: a backend raises :class:`AttributeError` for a name
that is not a field of the referenced record, and for chaining off a
field that refers to nothing.
"""
...
[docs]
class SearchVariable(Protocol):
"""Bind a query variable to a target type whose attributes yield fields.
``always_true``/``always_false`` are reserved names: they are real methods
of the variable, never stored fields resolved through ``__getattr__``.
They exist so a translation layer can express a constant truth value
without inventing a probe field. A ``field == field`` probe is NULL-unsound,
since it yields NULL (not true) for a NULL field.
"""
[docs]
def always_true(self) -> SearchExpression:
"""An expression that matches every row."""
...
[docs]
def always_false(self) -> SearchExpression:
"""An expression that matches no row."""
...
def __getattr__(self, name: str) -> SearchField:
"""Return the named queryable field."""
...
[docs]
class SearchResult(NamedTuple):
"""Represent one match with declared output values and names.
``values`` holds one entry per :meth:`Searcher.output` call in declaration
order; it is a tuple, so ``values, names = result`` and ``result[0][0]``
both work.
"""
[docs]
values: tuple[Any, ...]
[docs]
class Searcher(Protocol):
"""Build one query and iterate its results.
Iteration yields one :class:`SearchResult` per match, so ``item[0][0]`` is
the first declared output of the match (typically the matched row object).
The expressions received by ``add`` are always ones produced by this same
backend's search variables, so implementations may type them as their own
expression class; a backend that needs a second (post-filter) evaluation
position decides that from the expression itself, not from the caller.
"""
[docs]
def variable(self, target: Any) -> Any:
"""Bind a query variable to ``target``."""
...
[docs]
def output(self, variable: Any, name: str) -> None:
"""Declare ``variable`` as a named result output."""
...
[docs]
def add(self, expression: Any) -> None:
"""Add a filter expression to the query."""
...
[docs]
def count(self) -> int:
"""Return the exact count of the current query."""
...
[docs]
def set_limit(self, limit: int) -> None:
"""Set the query limit."""
...
[docs]
def add_offset(self, offset: int) -> None:
"""Add an offset to the query."""
...
[docs]
def add_sort(self, field: Any, descending: bool) -> None:
"""Add a field sort to the query."""
...
def __iter__(self) -> Iterator[SearchResult]:
"""Iterate over the query's matches."""
...
[docs]
def results(self, **outputs: Any) -> ResultSetLike:
"""Return a result set for the requested named outputs."""
...
[docs]
class Store(Protocol):
"""Require a store that can create a query searcher."""
[docs]
def searcher(self) -> Searcher:
"""Create an empty searcher."""
...