"""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
import enum as _enum_mod
import inspect
from collections.abc import (
Callable,
Sequence,
)
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Literal
from fastapi import HTTPException
from pydantic import BaseModel
from sqlalchemy import (
String,
cast,
column,
func,
literal,
select,
union_all,
)
from fsh_lib.actions import ActionRef, ActionSpec, available_actions
from fsh_lib.filter_values import FilterValuesRequest, resolved_limit
from fsh_lib.values_table import values_table
if TYPE_CHECKING:
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql import Select
# Callable that turns a single model row into a Pydantic
# default-rep instance: ``fn(model_row, session) -> rep``. Stored
# on :class:`ResourceEntry` and consumed 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.
#
# Plain default-rep builders never need to issue I/O -- the row has
# already been fetched by ``hydrate_refs`` -- so the synchronous
# signature keeps the dispatch loop straightforward. Resources
# that opt into :attr:`ResourceConfig.include_actions_in_dump` get
# an async serializer (the action-injection path has to await
# guards); ``hydrate_refs`` detects the coroutine and awaits it,
# so the same registry slot handles both shapes.
DefaultRepSerializer = Callable[[Any, Any], Any]
# =============================================================================
# Operator vocabulary.
# =============================================================================
[docs]
class FilterOperator(_enum_mod.StrEnum):
"""Closed set of operators a filter field may declare."""
EQ = "eq"
NEQ = "neq"
GT = "gt"
GTE = "gte"
LT = "lt"
LTE = "lte"
CONTAINS = "contains"
STARTS_WITH = "starts_with"
IN = "in"
IS_NULL = "is_null"
# =============================================================================
# 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, ...] = ()
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)
# -------- 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
# The default-rep serializer is sync for plain reps and
# async when the resource opts into
# ``include_actions_in_dump`` (the action-injection path
# has to await guards). Branch on the return value
# rather than the static type so consumers can stay
# on either contract without registry plumbing.
link = entry.default_rep_serializer(row, session)
if inspect.isawaitable(link):
link = await link
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
) -> 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 trigram pipeline serves both generic
search and per-filter narrowing. The FE already has every
enum member from discovery, so a no-``search`` request for
an enum field returning nothing isn't a regression.
"""
entry = self._require_entry()
names = list(fields) if fields else list(entry.search_columns)
return await self._run_multi_column_search(entry, names, request, db)
# -------- 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,
) -> 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``.
Without ``q`` (or with an empty ``names``) the union
returns nothing — relevance scoring needs a query and
narrowing UX is "type to search". Requires ``pg_trgm``.
"""
for name in names:
spec = _find_field(entry, name)
if isinstance(spec, (Bool, 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 or not request.search:
return ValuesPage(results=[])
sub_queries = [
self._trigram_subquery(entry, name, request.search)
for name in names
]
statement = (
union_all(*sub_queries)
.order_by(column("score").desc(), column("value").asc())
.limit(resolved_limit(request.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
],
)
def _trigram_subquery(
self,
entry: ResourceEntry,
name: str,
query: str,
) -> 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).
"""
spec = _find_field(entry, name)
if isinstance(spec, Enum):
members_table = values_table(
_ChoiceRow,
[
_ChoiceRow(value=str(member.value), label=member.name)
for member in spec.enum_class
],
name=f"enum_{spec.name}",
)
label_col = members_table.c.label
return select(
literal(spec.name).label("field"),
members_table.c.value.label("value"),
label_col.label("label"),
func.similarity(label_col, query).label("score"),
).where(label_col.op("%")(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, String)
)
return select(
literal(spec.name).label("field"),
cast(target_pk, String).label("value"),
target_label.label("label"),
func.similarity(target_label, query).label("score"),
).where(target_label.op("%")(query))
# Plain search column — trigram against the model column.
column_attr = getattr(entry.model, name)
return (
select(
literal(name).label("field"),
column_attr.label("value"),
column_attr.label("label"),
func.similarity(column_attr, query).label("score"),
)
.distinct()
.where(column_attr.op("%")(query))
)
# =============================================================================
# Lookup helpers.
# =============================================================================
def _find_field(entry: ResourceEntry, name: str) -> FilterField | None:
return next(
(field for field in entry.fields if field.name == name),
None,
)
@dataclass(frozen=True)
class _ChoiceRow:
"""One ``(value, label)`` row for VALUES-clause enum tables."""
value: str
label: str