"""Action availability for codegen-generated FastAPI projects.
An *action* is anything you can do to (or with) a resource: the
built-in CRUD ops (``get``, ``list``, ``create``, ``update``,
``delete``) plus any custom action endpoints declared in the
spec. Every action carries a single *guard* callable --
async def can_<name>(resource, session) -> bool
-- that decides two things at once: whether the current session
may execute the action, and whether the action should appear in
serialized responses so the frontend can show the corresponding
button. Object-scope guards see the resource instance; collection-
scope guards see ``None`` (there is no per-row resource yet).
Generated code emits one ``actions.py`` per app holding tuples of
:class:`ActionSpec` per resource (object and collection scopes
kept separate). The route-handler templates call
:func:`available_actions` against the right tuple to populate the
``actions`` field on response payloads, and call the matching
``can`` callable directly before executing each handler so the
visibility predicate and the authorization gate can never drift.
"""
from collections.abc import Awaitable, Callable, Iterable
from dataclasses import dataclass
from typing import Any, Literal, Self
from pydantic import BaseModel
Scope = Literal["object", "collection"]
"""Whether an action targets a single resource or a collection.
Object-scope actions take ``(resource, session)``; collection-
scope actions take ``(None, session)``. The frontend uses this
to decide where to render the button -- per row or once on the
list page.
"""
# Signature every guard conforms to: ``(resource, session) -> bool``.
# ``resource`` is the SQLAlchemy instance for object-scope actions
# or ``None`` for collection-scope actions; ``session`` is whatever
# the consumer's auth dep resolves -- the type is left open since
# consumers pick the session model.
CanCallable = Callable[[Any, Any], Awaitable[bool]]
[docs]
class ActionRef[NameT: str = str, ScopeT: str = Scope](BaseModel):
"""One action exposed in a serialized response.
Carries the bare minimum the frontend needs to render a
button: the action name (matches the operation name on the
backend) and its scope. Kept Pydantic so the OpenAPI schema
surfaces a stable shape; consumers downstream get a typed TS
interface for free.
Generic over the ``name`` and ``scope`` types so generated
code can subclass with a ``Literal`` union for ``name`` and a
single-member ``scope`` -- ``ActionRef[FooAction, Literal["object"]]``
-- and have the OpenAPI schema (and the TypeScript types
openapi-ts derives from it) carry the enum / const rather than
a bare ``str``. Narrowing through the type parameters keeps
the override compatible, so no field has to be redeclared.
Attributes:
name: Operation name (e.g. ``"publish"``, ``"update"``).
scope: ``"object"`` for per-row actions, ``"collection"``
for actions that target the resource as a whole.
formats: Available response formats (e.g. ``["json", "csv"]``).
``None`` when json-only (the default).
async_formats: Formats that run async by default.
``None`` when no async formats.
"""
name: NameT
scope: ScopeT
formats: list[str] | None = None
async_formats: list[str] | None = None
[docs]
@dataclass(frozen=True)
class ActionSpec:
"""Generator-emitted descriptor for one action.
Lives in the per-app ``actions.py`` registry. The route
handlers and serializers consume :class:`ActionSpec` tuples
via :func:`available_actions`; nothing outside generated code
should construct these by hand.
Attributes:
action: The typed :class:`Action` this spec describes -- the
single source of the ``<resource>:<operation>`` identity
(``AssetAction.view``). The bare operation name is exposed
via the :attr:`name` property so the response ``ActionRef``
and registry lookups keep their historical shape.
can: Async guard returning ``True`` when the action is
available. Bound to :func:`always_true` when the
spec did not declare a ``can`` dotted path.
is_object_action: ``True`` for object-scope actions,
``False`` for collection-scope actions. Drives the
:class:`ActionRef` ``scope`` field and disambiguates
which tuple a spec belongs in.
formats: Available response formats. ``None`` when json-only.
async_formats: Formats that run async. ``None`` when no async.
"""
action: Action
can: CanCallable
is_object_action: bool
formats: tuple[str, ...] | None = None
async_formats: tuple[str, ...] | None = None
@property
def name(self) -> str:
"""The bare operation name (e.g. ``"view"``) for this spec.
Derived from :attr:`action` so ``ActionRef.name`` and the
``find_can`` / ``available_actions`` lookups keep the historical
bare-name shape while the spec carries the full typed identity.
"""
return self.action.operation
@property
def scope(self) -> Scope:
"""Match :class:`ActionRef.scope` for this spec."""
return "object" if self.is_object_action else "collection"
[docs]
class Action(str):
"""A ``<resource>:<operation>`` permission code that carries its parts.
An :class:`Action` *is* the permission string (``"asset:view"``), so it
drops into anything that expects one -- the OPA gate, the audit row,
dict keys, equality -- unchanged, while exposing the two typed axes it
was built from: :attr:`resource` and :attr:`operation`.
Generated code builds one per action per resource
(``AssetAction.view = Action(ResourceType.ASSET, Operation.view)``);
spec-invariant fsh-lib sees only the ``str`` face plus the two
attributes, so it can recover the resource from an action alone
(``action.resource``) instead of taking it as a second argument.
"""
__slots__ = ("operation", "resource")
resource: str
operation: str
def __new__(cls, resource: str, operation: str) -> Self:
"""Build the ``resource:operation`` string carrying its parts."""
self = super().__new__(cls, f"{resource}:{operation}")
self.resource = resource
self.operation = operation
return self
def __reduce__(self) -> tuple[Any, ...]:
"""Round-trip through the parts (str pickles as its value)."""
# str.__new__ takes one arg; Action.__new__ takes two, so teach
# copy/pickle to rebuild from the resource + operation.
return (Action, (self.resource, self.operation))
def __repr__(self) -> str:
"""Return a repr showing the resource and operation."""
return (
f"Action(resource={self.resource!r}, operation={self.operation!r})"
)
[docs]
async def always_true(_resource: Any, _session: Any) -> bool:
"""Default guard used when an action declares no ``can`` path.
Returning ``True`` unconditionally matches the historical
behavior of generated handlers (auth handled at the route
level, no per-action gating); opting in to action gating is
additive.
"""
return True
[docs]
async def available_actions[T: BaseModel = ActionRef](
resource: Any,
session: Any,
specs: Iterable[ActionSpec],
ref_cls: type[T] = ActionRef, # type: ignore[assignment]
) -> list[T]:
"""Return the subset of *specs* whose guards pass for *session*.
The guard for each spec is awaited in declaration order; specs
whose guard returns ``False`` are dropped. Order is preserved
so the frontend can rely on a stable button layout driven by
the spec.
*ref_cls* lets generated code substitute a per-resource typed
``ActionRef`` subclass -- e.g. ``AssetObjectPermission`` whose
``name`` is a ``Literal["get", "update", "publish"]`` -- so
the OpenAPI schema (and the TypeScript types openapi-ts
derives from it) surface the enum rather than a bare ``str``.
The default keeps the historical untyped shape for callers
that don't go through codegen.
Args:
resource: The SQLAlchemy instance for object-scope dumps,
or ``None`` for collection-scope dumps.
session: Whatever the auth dep resolved -- passed through
to each guard untouched.
specs: Iterable of :class:`ActionSpec`; typically a tuple
literal from the generated per-app ``actions.py``.
ref_cls: Pydantic model used to construct each ref.
Defaults to :class:`ActionRef`.
Returns:
List of *ref_cls* instances, one per spec whose guard
returned ``True``, in spec order.
"""
refs: list[T] = []
for spec in specs:
if not await spec.can(resource, session):
continue
kwargs: dict[str, Any] = {"name": spec.name, "scope": spec.scope}
for field in ("formats", "async_formats"):
val = getattr(spec, field)
if val is not None:
kwargs[field] = list(val) if isinstance(val, tuple) else val
refs.append(ref_cls(**kwargs))
return refs
[docs]
def find_can(
specs: Iterable[ActionSpec],
name: str,
) -> CanCallable:
"""Return the guard for the action *name* in *specs*.
Generated handlers resolve their own row-level guard at
module-import time -- e.g. ``_CAN_LIST = find_can(...)`` --
so the per-request path is a single attribute lookup rather
than a tuple scan.
Args:
specs: An :class:`ActionSpec` tuple from the per-app
registry, typically the collection-scoped tuple for
row-level filters or the object-scoped tuple for
execution-time gates.
name: Operation name to look up (e.g. ``"list"``,
``"publish"``).
Returns:
The bound guard callable (a ``CanCallable``).
Raises:
KeyError: If *name* is not present in *specs* -- the
generated code should never reach this branch since
the registry is built from the same operation list.
"""
for spec in specs:
if spec.name == name:
return spec.can
msg = f"No action {name!r} in registry"
raise KeyError(msg)