"""Project-wide resource registry: value-provider engine.
Codegen emits one :class:`ResourceRegistry` per project, populated
declaratively with one :class:`ResourceEntry` per resource.
Subscripting it by slug -- ``registry[slug]`` -- yields a
per-resource handle; the generated ``_values`` route handler
delegates to it via ``registry[slug].values(...)``.
The filter catalog is no longer surfaced at runtime -- it lives
in the openapi spec (``x-fsh-list``) at build time, and the
codegen FE bakes it into per-resource hooks. ``ResourceRegistry``
keeps the value-provider plumbing (trigram autocomplete over enum
choices, ref labels, and free-text search columns).
The class is generic over the slug type (``Slug: str = str``) so a
codegen consumer can declare ``ResourceRegistry[ResourceType]`` and
get type-narrowed slug arguments on every method. The default of
``str`` keeps the class usable from hand-written code that doesn't
go through codegen.
Value endpoints are single-page -- autocomplete UX narrows by
typing more characters, not by paginating.
"""
from __future__ import annotations
from collections.abc import (
Awaitable,
Callable,
Mapping,
Sequence,
)
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Literal, Protocol
from fastapi import HTTPException
from pydantic import BaseModel
from sqlalchemy import (
Text,
cast,
column,
func,
literal,
select,
true,
union_all,
)
from sqlalchemy import (
inspect as sa_inspect,
)
from fsh_lib.actions import ActionRef, ActionSpec, available_actions
from fsh_lib.filter_values import FilterValuesRequest, resolved_limit
from fsh_lib.filters import FilterOperator, trigram_clause
from fsh_lib.values_table import values_table
if TYPE_CHECKING:
import enum as _enum_mod
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql import ColumnElement, Select
[docs]
class RowFilterFactory(Protocol):
"""Per-resource row-filter factory.
The registry calls it whenever it needs the row-visibility
clause for one resource's table: once per table a values
request will scan (the bound resource for plain search
columns, each ``Ref`` field's *target* resource for ref
suggestions), and once for the bound resource when a list
endpoint asks via :meth:`_BoundResource.view_filter`.
``action`` is the ``<slug>:view`` permission, ``resource_type``
the slug, and ``columns`` the ``{resource field -> SQLAlchemy
column}`` map the compiled residual may constrain. Return the
``WHERE`` clause to AND in, or ``None`` for no constraint.
"""
def __call__(
self,
*,
action: str,
resource_type: str,
columns: Mapping[str, Any],
) -> Awaitable[ColumnElement[bool] | None]:
"""Compile one resource's row filter."""
...
# Async callable that turns a single model row into a Pydantic
# default-rep instance: ``async fn(model_row, session) -> rep``.
# Stored on :class:`ResourceEntry` and awaited by
# :meth:`ResourceRegistry.hydrate_refs`. Uses ``Any`` rather than
# ``BaseModel`` for the return so user-supplied builders that return
# subclass instances type-check without variance gymnastics. Every
# serializer is async and takes ``session`` (the action-injection
# path awaits guards; plain reps ignore it), so the registry slot
# has a single shape.
DefaultRepSerializer = Callable[[Any, Any], Awaitable[Any]]
# =============================================================================
# Operator vocabulary.
#
# ``FilterOperator`` lives in :mod:`fsh_lib.filters` -- the same closed
# set that keys the execution-side dispatch table -- so a field can only
# advertise an operator the filter engine actually implements.
# =============================================================================
# Field specs — one frozen dataclass per ``FilterValueKind`` from
# :mod:`be.config.schema`. ``Ref`` covers both cross-resource FK and
# self-reference cases (codegen translates ``values: "self"`` to a
# ``Ref`` targeting the resource's own slug).
# =============================================================================
[docs]
@dataclass(frozen=True)
class Enum:
"""Enum-typed filter field.
Discovery emits ``{value, label}`` choices; the values endpoint
serves the same list ``q``-filterable through a Postgres
``VALUES`` clause.
"""
name: str
enum_class: type[_enum_mod.Enum]
operators: tuple[FilterOperator, ...] = (
FilterOperator.EQ,
FilterOperator.IN,
)
kind: Literal["enum"] = "enum"
[docs]
@dataclass(frozen=True)
class Ref:
"""Filter pointing at another resource (or this one).
The trigram subquery scores against the target's first
``search_columns`` entry on its :class:`ResourceEntry`;
targets without any search columns fall back to the
stringified pk.
"""
name: str
target: str
operators: tuple[FilterOperator, ...] = (
FilterOperator.EQ,
FilterOperator.IN,
)
kind: Literal["ref"] = "ref"
[docs]
@dataclass(frozen=True)
class LiteralField:
"""Numeric / date / datetime input rendered natively on the FE."""
name: str
type: str
operators: tuple[FilterOperator, ...] = (
FilterOperator.EQ,
FilterOperator.GT,
FilterOperator.GTE,
FilterOperator.LT,
FilterOperator.LTE,
)
kind: Literal["literal"] = "literal"
[docs]
@dataclass(frozen=True)
class Bool:
"""Boolean toggle."""
name: str
operators: tuple[FilterOperator, ...] = (FilterOperator.EQ,)
kind: Literal["bool"] = "bool"
FilterField = Enum | Ref | LiteralField | Bool
"""Sum of every supported filter-field shape."""
# =============================================================================
# Values response.
# =============================================================================
[docs]
class ValuesPage(BaseModel):
"""Response shape for ``POST /_values``.
Single-page only — autocomplete UX narrows by typing more
characters, not by paginating. ``results`` is
``[{"value": ..., "label": ...}]`` for enum / free-text /
single-field paths and the consumer's link-payload shape
(already ``model_dump``-ed) for resource search. Multi-column
union results add a ``"field"`` key indicating the source
column.
"""
results: list[dict[str, Any]]
# =============================================================================
# Resource entry.
# =============================================================================
[docs]
@dataclass(frozen=True)
class ResourceEntry:
"""One resource's registry-side declaration.
``search_columns`` are the model attributes used as the default
field list when the values endpoint is called with empty
``fields`` — they're trigram-matched the same way any other
field is, so the empty-fields path is just a multi-column
search over these defaults.
``default_rep_class`` and ``default_rep_serializer`` describe
the resource's cross-resource link shape — the Pydantic class
of its
:attr:`~be.config.schema.ResourceConfig.default_representation`
and the async ``(row, session) -> default_rep_class`` callable
that produces it. Both are ``None`` when the resource doesn't
declare a default representation; :meth:`ResourceRegistry.hydrate_refs`
then returns an empty list for that slug.
"""
model: type
pk: str
fields: tuple[FilterField, ...] = ()
search_columns: tuple[str, ...] = ()
opa_rep_class: type | None = None
"""The Pydantic class of the resource's OPA representation
(the projection point checks hand the policy as
``input.resource``). The row-filter compile derives its
unknowns from it dynamically (see
:meth:`_BoundResource._compile_slug_filter`): every rep field
that is a mapped column on ``model`` must be symbolic during
partial evaluation -- a field missing from the unknowns
partially evaluates as *absent*, so a rule constraining it
would silently widen or collapse the filter."""
default_rep_class: type | None = None
default_rep_serializer: DefaultRepSerializer | None = None
object_actions: tuple[ActionSpec, ...] = ()
"""Object-scope action specs. Drives the per-row
``registry[slug].actions(...)`` path."""
collection_actions: tuple[ActionSpec, ...] = ()
"""Collection-scope action specs. Drives the collection-scope
``registry[slug].actions(...)`` path."""
# =============================================================================
# Registry — public facade.
# =============================================================================
[docs]
class ResourceRegistry[Slug: str = str]:
"""Project-wide discovery + value-provider dispatcher.
Construct with a ``{slug: ResourceEntry}`` map at module load
time. Subscript by slug -- ``registry[slug]`` -- to get a
bound handle exposing ``actions`` / ``values`` for that
resource; :meth:`hydrate_refs` stays on the registry itself
since it dispatches on a runtime slug. Stateless after
construction -- safe to share across requests.
Generic over the slug type so a codegen consumer can declare
``ResourceRegistry[ResourceType]`` and get the project's
``ResourceType`` enum on the subscript. ``Slug`` defaults to
``str`` for non-codegen use.
"""
def __init__(self, entries: dict[Slug, ResourceEntry]) -> None:
"""Copy *entries* so the caller can mutate their original.
Keys are coerced to plain ``str`` internally so subsequent
lookups (driven by free-form slug arguments coming through
``Ref.target`` or values-endpoint payloads) line up with
whichever ``StrEnum`` member the caller passed in.
"""
self._entries: dict[str, ResourceEntry] = {
str(slug): entry for slug, entry in entries.items()
}
def __getitem__(self, slug: Slug) -> _BoundResource[Slug]:
"""Bind *slug* and return a per-resource handle.
``ResourceRegistry[ResourceType.WIDGET].values(...)`` reads
as "the widget resource's autocomplete values". The handle
is a cheap per-subscript view: the slug isn't validated
here -- each handle method 404s on an unregistered slug at
call time.
"""
return _BoundResource(self._entries, slug)
[docs]
def permission_codes(self) -> list[str]:
"""Every grantable ``<resource>:<action>`` permission code, sorted.
The closed catalogue folded from the registered resources: each
contributes ``<slug>:*`` plus one ``<slug>:<action>`` per object
and collection action, and ``"*"`` is the global wildcard. This
is the single source of truth for permission validation -- the
frontend's permission picker and (dumped to bundle data) the OPA
``valid_permissions`` set -- so there is no permissions *table*;
the registry *is* the catalogue.
"""
codes: set[str] = {"*"}
for slug, entry in self._entries.items():
codes.add(f"{slug}:*")
for spec in (*entry.object_actions, *entry.collection_actions):
codes.add(f"{slug}:{spec.name}")
return sorted(codes)
# -------- Cross-resource link hydration --------
[docs]
async def hydrate_refs(
self,
resource: Slug,
ids: Sequence[Any],
db: AsyncSession,
session: Any,
) -> list[dict[str, Any]]:
"""Fetch *ids* of *resource* and serialize them via its default rep.
Used by :func:`fsh_lib.saved_views.hydrate_view` (and
anything else dispatching by slug at runtime) to turn raw
ref ids into hydrated link payloads. Lenient on missing
slugs and dropped ids: an unknown *resource*, a resource
without a default representation, or an empty *ids* list
all return ``[]``; ids that don't resolve to a row are
silently skipped. Order of returned items mirrors *ids*.
"""
entry = self._entries.get(resource)
if entry is None or entry.default_rep_serializer is None or not ids:
return []
pk_col = getattr(entry.model, entry.pk)
stmt = select(entry.model).where(pk_col.in_(list(ids)))
rows = (await db.execute(stmt)).scalars().all()
# Stringify both sides so UUID columns and JSON-serialised
# ids (always strings) compare equal -- without this the
# ``.get`` lookup misses every row when the model's pk is
# a uuid.UUID.
by_id = {str(getattr(row, entry.pk)): row for row in rows}
items: list[dict[str, Any]] = []
for raw_id in ids:
row = by_id.get(str(raw_id))
if row is None:
continue
link = await entry.default_rep_serializer(row, session)
items.append(link.model_dump())
return items
class _BoundResource[Slug: str = str]:
"""A registry view with one resource's slug bound in.
Returned by ``ResourceRegistry[slug]``. The per-resource
generated ``_values`` route handler calls :meth:`values` on
it; :meth:`actions` covers the permissions endpoint. The
slug is named once at the subscript instead of threaded
through every call.
"""
def __init__(self, entries: dict[str, ResourceEntry], slug: Slug) -> None:
"""Wrap the registry's *entries* with *slug* bound in.
``entries`` is the registry's own dict, shared by reference
-- the handle is a cheap per-subscript view that never
outlives the request that created it.
"""
self._entries = entries
self._slug = slug
# -------- Action availability --------
async def actions[T: BaseModel = ActionRef](
self,
*,
session: Any,
obj: Any = None,
ref_cls: type[T] = ActionRef, # type: ignore[assignment]
) -> list[T]:
"""Return the visible actions for this resource in a scope.
``obj=None`` dispatches to ``collection_actions``; an
instance dispatches to ``object_actions``. Each spec's
``can`` guard is awaited; specs whose guard returns
``False`` are dropped, preserving registration order.
``ref_cls`` narrows the return type for callers that want a
typed per-resource ``ActionRef`` subclass; it overrides the
entry's stored ref class so the call site keeps a single
source of truth for the response shape.
"""
entry = self._require_entry()
specs = (
entry.object_actions
if obj is not None
else entry.collection_actions
)
return await available_actions(
resource=obj,
session=session,
specs=specs,
ref_cls=ref_cls,
)
# -------- Values --------
async def values(
self,
*,
fields: Sequence[str],
request: FilterValuesRequest,
db: AsyncSession,
session: Any = None, # noqa: ARG002 -- reserved for future hook use
row_filter: RowFilterFactory | None = None,
) -> ValuesPage:
"""Run a value-provider request for this resource.
Empty ``fields`` defaults to the resource's
``search_columns`` (see :class:`ResourceEntry`), so the
same multi-column pipeline serves both generic search and
per-filter narrowing. A request with ``search`` ranks by
trigram relevance; one without browses the field's distinct
values alphabetically, so an fk / enum picker opens
populated.
``row_filter`` (see :class:`RowFilterFactory`) is invoked
once per table the union will scan -- this resource for
plain search columns, each ref field's target resource for
ref suggestions -- and the returned clause is ANDed into
the matching suggestion queries. Enum suggestions come
from an in-memory members table (static schema, no rows)
and are never filtered.
"""
entry = self._require_entry()
if request.ids:
# Resolve mode keys the ids by field, so the map's keys are
# the field set (each field matched against only its own ids).
names = list(request.ids.keys())
elif fields:
names = list(fields)
else:
names = list(entry.search_columns)
return await self._run_multi_column_search(
entry, names, request, db, row_filter
)
# -------- Internal helpers --------
def _require_entry(self, slug: str | None = None) -> ResourceEntry:
"""Resolve *slug* (default: this handle's bound slug) or 404."""
resource = str(self._slug if slug is None else slug)
entry = self._entries.get(resource)
if entry is None:
raise HTTPException(
status_code=404,
detail=f"Unknown resource: {resource}",
)
return entry
# -------- Multi-column trigram union --------
async def _run_multi_column_search(
self,
entry: ResourceEntry,
names: list[str],
request: FilterValuesRequest,
db: AsyncSession,
row_filter: RowFilterFactory | None = None,
) -> ValuesPage:
"""UNION ``(field, value, label, score)`` per name, ranked together.
Each entry in ``names`` is either a registered filter
field (:class:`Enum` / :class:`Ref`) or a plain text
column on ``entry.model`` (the ``search_columns`` default
path). Bool / Literal fields
have no text to score against and 404 up-front, even with
no ``q``.
With a ``search`` query each field is trigram-ranked and the
union is ordered by relevance. Without one (the picker just
opened) the union *browses* instead: every field's distinct
values, ordered alphabetically, so an fk / enum filter opens
populated rather than blank. Both paths require ``pg_trgm``
only for the search branch. An empty ``names`` returns
nothing.
"""
for name in names:
spec = _find_field(entry, name)
if isinstance(spec, LiteralField):
raise HTTPException(
status_code=404,
detail=f"Field {name!r} has no value provider",
)
if spec is None and not hasattr(entry.model, name):
raise HTTPException(
status_code=404,
detail=f"Unknown filter field: {name}",
)
if not names:
return ValuesPage(results=[])
filters = await self._compile_row_filters(entry, names, row_filter)
search = request.search.strip() if request.search else ""
ids_map = request.ids or {}
if ids_map:
# Resolve mode: label lookup, each field against its own ids.
sub_queries = [
self._resolve_subquery(
entry, name, tuple(ids_map.get(name, ())), filters
)
for name in names
]
statement = union_all(*sub_queries).order_by(
column("value").asc(),
)
# One row per (field, id); never clamp below what was asked.
total_ids = sum(len(values) for values in ids_map.values())
limit = max(resolved_limit(request.limit), total_ids)
elif search:
sub_queries = [
self._trigram_subquery(entry, name, search, filters)
for name in names
]
statement = union_all(*sub_queries).order_by(
column("score").desc(),
column("value").asc(),
)
limit = resolved_limit(request.limit)
else:
sub_queries = [
self._browse_subquery(entry, name, filters) for name in names
]
statement = union_all(*sub_queries).order_by(
column("value").asc(),
)
limit = resolved_limit(request.limit)
statement = statement.limit(limit)
rows = (await db.execute(statement)).all()
return ValuesPage(
results=[
{
"field": row.field,
"value": row.value,
"label": row.label,
"score": float(row.score),
}
for row in rows
],
)
async def view_filter(
self,
*,
row_filter: RowFilterFactory,
) -> ColumnElement[bool]:
"""Compile this resource's ``view`` row filter for a list query.
Derives the action, the resource type, and the residual
column map from the registry entry -- the one place that
knows the model, pk, and owner column -- so list endpoints
and the values pipeline share a single wiring. Returns a
tautology when the factory imposes no constraint, so the
result can be ANDed in unconditionally.
"""
clause = await self._compile_slug_filter(str(self._slug), row_filter)
return true() if clause is None else clause
async def _compile_slug_filter(
self,
slug: str,
row_filter: RowFilterFactory,
) -> ColumnElement[bool] | None:
"""Compile one resource's row filter from its registry entry.
The residual column map is derived from the entry: ``id``
always binds the pk (generic RBAC may scope by instance
id), and every field of the OPA representation that is a
mapped column on the model rides along so custom rules
constraining it compile into the WHERE clause instead of
silently distorting the filter. Rep fields that aren't
mapped columns -- nested dumps (relationship traversals)
and the ``type`` discriminator (always sent as a known
value) -- have nothing a WHERE clause could bind and are
skipped.
"""
target = self._require_entry(slug)
columns: dict[str, Any] = {
"id": getattr(target.model, target.pk),
}
rep = target.opa_rep_class
if rep is not None:
mapped = sa_inspect(target.model).columns
for name in rep.model_fields:
if name in ("id", "type") or name not in mapped:
continue
columns[name] = getattr(target.model, name)
return await row_filter(
action=f"{slug}:view",
resource_type=slug,
columns=columns,
)
async def _compile_row_filters(
self,
entry: ResourceEntry,
names: list[str],
row_filter: RowFilterFactory | None,
) -> dict[str, ColumnElement[bool]]:
"""Compile one row filter per table the union will scan.
Plain search columns scan this resource's own rows; a
``Ref`` field scans its *target* resource's rows, so its
suggestions are constrained by the target's own ``view``
policy, not this resource's. Enum fields scan an
in-memory members table (static schema) and need no
filter. Each distinct slug is compiled once however many
fields share it.
"""
if row_filter is None:
return {}
slugs: list[str] = []
for name in names:
spec = _find_field(entry, name)
if isinstance(spec, (Enum, Bool)):
continue
slug = (
str(spec.target) if isinstance(spec, Ref) else str(self._slug)
)
if slug not in slugs:
slugs.append(slug)
clauses: dict[str, ColumnElement[bool]] = {}
for slug in slugs:
clause = await self._compile_slug_filter(slug, row_filter)
if clause is not None:
clauses[slug] = clause
return clauses
def _trigram_subquery(
self,
entry: ResourceEntry,
name: str,
query: str,
filters: Mapping[str, ColumnElement[bool]] | None = None,
) -> Select[Any]:
"""Build the trigram subquery for one field in the union.
``name`` is either a registered :class:`FilterField`
(dispatched by kind) or a plain model column from
``entry.search_columns`` (treated as a free-text trigram).
``filters`` maps slug -> compiled row filter (see
:meth:`_compile_row_filters`); the clause for whichever
table this subquery scans is ANDed in. The enum path has
no table to constrain.
"""
filters = filters or {}
spec = _find_field(entry, name)
if isinstance(spec, (Enum, Bool)):
return _members_select(spec.name, _member_rows(spec), query)
if isinstance(spec, Ref):
target = self._require_entry(spec.target)
target_pk = getattr(target.model, target.pk)
# Label by the target's first search_column when
# present; else its stringified pk. Must be text-
# shaped for ``similarity()`` to compose.
target_label = (
getattr(target.model, target.search_columns[0])
if target.search_columns
else cast(target_pk, Text)
)
stmt = select(
literal(spec.name).label("field"),
cast(target_pk, Text).label("value"),
target_label.label("label"),
func.similarity(target_label, query).label("score"),
).where(_text_match(target_label, query))
clause = filters.get(str(spec.target))
if clause is not None:
stmt = stmt.where(clause)
return stmt
# Plain search column — trigram against the model column.
column_attr = getattr(entry.model, name)
stmt = (
select(
literal(name).label("field"),
column_attr.label("value"),
column_attr.label("label"),
func.similarity(column_attr, query).label("score"),
)
.distinct()
.where(_text_match(column_attr, query))
)
clause = filters.get(str(self._slug))
if clause is not None:
stmt = stmt.where(clause)
return stmt
def _browse_subquery(
self,
entry: ResourceEntry,
name: str,
filters: Mapping[str, ColumnElement[bool]] | None = None,
) -> Select[Any]:
"""Build the no-``search`` "browse" subquery for one field.
The empty-query counterpart of :meth:`_trigram_subquery`:
same ``(field, value, label, score)`` shape and same row
filters, but with a constant ``score`` and no ``%`` match --
every value, so the picker opens populated. The outer union
orders by ``value`` and clamps the row count, so an enum's
full member set or an fk target's first page comes back.
"""
filters = filters or {}
spec = _find_field(entry, name)
if isinstance(spec, (Enum, Bool)):
return _members_select(spec.name, _member_rows(spec), None)
if isinstance(spec, Ref):
target = self._require_entry(spec.target)
target_pk = getattr(target.model, target.pk)
target_label = (
getattr(target.model, target.search_columns[0])
if target.search_columns
else cast(target_pk, Text)
)
stmt = select(
literal(spec.name).label("field"),
cast(target_pk, Text).label("value"),
target_label.label("label"),
literal(0.0).label("score"),
)
clause = filters.get(str(spec.target))
if clause is not None:
stmt = stmt.where(clause)
return stmt
column_attr = getattr(entry.model, name)
stmt = select(
literal(name).label("field"),
column_attr.label("value"),
column_attr.label("label"),
literal(0.0).label("score"),
).distinct()
clause = filters.get(str(self._slug))
if clause is not None:
stmt = stmt.where(clause)
return stmt
def _resolve_subquery(
self,
entry: ResourceEntry,
name: str,
ids: tuple[str, ...],
filters: Mapping[str, ColumnElement[bool]] | None = None,
) -> Select[Any]:
"""Build the label-resolve subquery for one field.
Like :meth:`_browse_subquery` but restricted to ``value IN
ids`` -- it returns ``(field, value, label, score)`` for
exactly the selected values so a default / restored filter's
chip can render its label. ``ids`` are checked against every
requested field; a field with no matching id contributes no
rows (an empty enum member set yields a no-row select rather
than invalid SQL). Same row filters as browse.
"""
filters = filters or {}
spec = _find_field(entry, name)
id_set = set(ids)
if isinstance(spec, (Enum, Bool)):
rows = [r for r in _member_rows(spec) if r.value in id_set]
if not rows:
return _empty_values_select(spec.name)
return _members_select(spec.name, rows, None)
if isinstance(spec, Ref):
target = self._require_entry(spec.target)
target_pk = getattr(target.model, target.pk)
target_label = (
getattr(target.model, target.search_columns[0])
if target.search_columns
else cast(target_pk, Text)
)
stmt = select(
literal(spec.name).label("field"),
cast(target_pk, Text).label("value"),
target_label.label("label"),
literal(0.0).label("score"),
).where(cast(target_pk, Text).in_(list(ids)))
clause = filters.get(str(spec.target))
if clause is not None:
stmt = stmt.where(clause)
return stmt
column_attr = getattr(entry.model, name)
stmt = (
select(
literal(name).label("field"),
column_attr.label("value"),
column_attr.label("label"),
literal(0.0).label("score"),
)
.distinct()
.where(column_attr.in_(list(ids)))
)
clause = filters.get(str(self._slug))
if clause is not None:
stmt = stmt.where(clause)
return stmt
# =============================================================================
# Lookup helpers.
# =============================================================================
def _empty_values_select(field_name: str) -> Select[Any]:
"""Build a no-row ``(field, value, label, score)`` select.
Keeps a UNION leg valid when a resolve request names a field none
of whose ids match (e.g. an enum with no requested member) --
``values_table`` rejects an empty row set, so synthesise the shape
with a ``WHERE false`` instead.
"""
return select(
literal(field_name).label("field"),
literal("").label("value"),
literal("").label("label"),
literal(0.0).label("score"),
).where(literal(False)) # noqa: FBT003 -- SQL false, not a bool flag
def _find_field(entry: ResourceEntry, name: str) -> FilterField | None:
return next(
(field for field in entry.fields if field.name == name),
None,
)
def _humanize_value(value: str) -> str:
"""``in_repair`` -> ``In repair`` -- a readable enum option label.
The ``_values`` endpoint is the option source for enum filters,
so the label it returns is what the picker shows; humanize the
raw member value rather than surfacing ``IN_REPAIR``.
"""
words = [w for w in value.replace("-", "_").split("_") if w]
if not words:
return value
first, *rest = words
return " ".join([first[:1].upper() + first[1:], *rest])
@dataclass(frozen=True)
class _ChoiceRow:
"""One ``(value, label)`` row for VALUES-clause enum tables."""
value: str
label: str
def _text_match(col: Any, query: str) -> ColumnElement[bool]:
"""One column's fuzzy match for the ``_values`` autocomplete.
Delegates to :func:`fsh_lib.filters.trigram_clause` so the filter
autocomplete and the list search box share one definition:
word_similarity (``col %> query``) for trigram-length queries, a
case-insensitive substring fallback for shorter ones (``"re"``
still finds ``"In repair"``).
"""
return trigram_clause(col, query)
def _member_rows(spec: Enum | Bool) -> list[_ChoiceRow]:
"""Build the ``(value, label)`` rows for an enum / bool table."""
if isinstance(spec, Bool):
return [
_ChoiceRow(value="true", label="Yes"),
_ChoiceRow(value="false", label="No"),
]
return [
_ChoiceRow(
value=str(member.value),
label=_humanize_value(str(member.value)),
)
for member in spec.enum_class
]
def _members_select(
field_name: str,
rows: list[_ChoiceRow],
query: str | None,
) -> Select[Any]:
"""Members-table subquery for an enum / bool field.
Trigram + substring ranked when *query* is given, else a flat
browse (constant score, every member). Both draw from a static
in-memory ``VALUES`` table, so neither is ever row-filtered.
"""
members = values_table(_ChoiceRow, rows, name=f"choice_{field_name}")
label_col = members.c.label
score = func.similarity(label_col, query) if query else literal(0.0)
stmt = select(
literal(field_name).label("field"),
members.c.value.label("value"),
label_col.label("label"),
score.label("score"),
)
if query:
stmt = stmt.where(_text_match(label_col, query))
return stmt