Source code for fsh_lib.actions

from collections.abc import Awaitable, Callable, Collection, Iterable
from dataclasses import dataclass
from typing import Any, Literal, Self

from pydantic import BaseModel

Scope = Literal["object", "collection"]

# Signature every guard conforms to: ``(resource, auth) -> bool``.
# ``resource`` is the SQLAlchemy instance for object-scope actions
# or ``None`` for collection-scope actions; ``auth`` is whatever
# the consumer's auth dep resolves -- the type is left open since
# consumers pick the auth-context model.
CanCallable = Callable[[Any, Any], Awaitable[bool]]




class ActionRef[NameT: str = str, ScopeT: str = Scope](BaseModel):
    name: NameT
    scope: ScopeT
    formats: list[str] | None = None
    async_formats: list[str] | None = None





@dataclass(frozen=True)
class ActionSpec:
    action: Action
    can: CanCallable
    is_object_action: bool
    formats: tuple[str, ...] | None = None
    async_formats: tuple[str, ...] | None = None

    @property
    def name(self) -> str:
        return self.action.operation

    @property
    def scope(self) -> Scope:
        return "object" if self.is_object_action else "collection"





class Action(str):
    __slots__ = ("operation", "resource")

    resource: str
    operation: str

    def __new__(cls, resource: str, operation: str) -> Self:
        self = super().__new__(cls, f"{resource}:{operation}")
        self.resource = resource
        self.operation = operation

        return self

    def __reduce__(self) -> tuple[Any, ...]:
        # 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 (
            f"Action(resource={self.resource!r}, operation={self.operation!r})"
        )



async def always_true(_resource: Any, _auth: Any) -> bool:  # noqa: ANN401
    return True


async def available_actions[T: BaseModel = ActionRef](
    resource: Any,  # noqa: ANN401
    auth: Any,  # noqa: ANN401
    action_specs: Iterable[ActionSpec],
    ref_cls: type[T] = ActionRef,  # type: ignore[assignment]
    allowed: Collection[str] | None = None,
) -> list[T]:
    refs: list[T] = []

    for spec in action_specs:
        if allowed is not None and spec.name not in allowed:
            continue

        if not await spec.can(resource, auth):
            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


def find_can(
    specs: Iterable[ActionSpec],
    name: str,
) -> CanCallable:
    for spec in specs:
        if spec.name == name:
            return spec.can

    msg = f"No action {name!r} in registry"
    raise KeyError(msg)