Source code for fsh_lib.actions

"""Action availability for kiln-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

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]]




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.

    """

    name: NameT
    scope: ScopeT





@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:
        name: Operation name.
        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.

    """

    name: str
    can: CanCallable
    is_object_action: bool

    @property
    def scope(self) -> Scope:
        """Match :class:`ActionRef.scope` for this spec."""
        return "object" if self.is_object_action else "collection"





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





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.

    """
    return [
        ref_cls(name=spec.name, scope=spec.scope)
        for spec in specs
        if await spec.can(resource, session)
    ]





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)