Guards

Register and evaluate guards used in transitions.

GuardRegistry

GuardRegistry(*, strict: bool = True)

Registry and evaluator for guard functions (sync and async).

Handles both registration of guard callables and evaluation of guard specs on transitions (named, expression, check, composite).

Example

registry = GuardRegistry() registry.register("is_valid", lambda ctx: ctx.get("valid", False)) registry.evaluate("is_valid", {"valid": True}) True

Source code in src/pystator/guards.py

def __init__(self, *, strict: bool = True) -> None:
    self._guards: dict[str, AnyGuardFunc] = {}
    self._async_guards: set[str] = set()
    self._accepts_kwargs: set[str] = set()
    self._lock = threading.Lock()
    self.strict = strict

register

register(name: str, func: AnyGuardFunc) -> None

Register a guard function. Thread-safe.

Re-registration replaces the existing guard (allows overriding builtins). If the function accepts keyword arguments beyond ctx, parameters from the guard spec are passed as **kwargs instead of injected into the context dict.

Source code in src/pystator/guards.py

def register(self, name: str, func: AnyGuardFunc) -> None:
    """Register a guard function. Thread-safe.

    Re-registration replaces the existing guard (allows overriding builtins).
    If the function accepts keyword arguments beyond ``ctx``, parameters
    from the guard spec are passed as ``**kwargs`` instead of injected
    into the context dict.
    """
    if not name:
        raise ValueError("Guard name cannot be empty")
    with self._lock:
        self._guards[name] = func
        if asyncio.iscoroutinefunction(func) or (
            callable(func)
            and asyncio.iscoroutinefunction(getattr(func, "__call__", None))  # noqa: B004
        ):
            self._async_guards.add(name)
        else:
            self._async_guards.discard(name)
        # Introspect: does the function accept extra keyword args?
        try:
            sig = inspect.signature(func)
            has_extra = any(
                p.kind
                in (inspect.Parameter.KEYWORD_ONLY, inspect.Parameter.VAR_KEYWORD)
                for p in sig.parameters.values()
            )
            if has_extra:
                self._accepts_kwargs.add(name)
            else:
                self._accepts_kwargs.discard(name)
        except (ValueError, TypeError):
            self._accepts_kwargs.discard(name)

unregister

unregister(name: str) -> None

Unregister a guard function. Thread-safe.

Source code in src/pystator/guards.py

def unregister(self, name: str) -> None:
    """Unregister a guard function. Thread-safe."""
    with self._lock:
        if name not in self._guards:
            raise GuardNotFoundError(f"Guard '{name}' not found", guard_name=name)
        del self._guards[name]
        self._async_guards.discard(name)

get

get(name: str) -> AnyGuardFunc

Look up a registered guard callable by name.

Parameters:

Name	Type	Description	Default
name	str	The registered guard name.	required

Returns:

Type	Description
AnyGuardFunc	The guard callable.

Raises:

Type	Description
GuardNotFoundError	If no guard is registered under name.

Source code in src/pystator/guards.py

def get(self, name: str) -> AnyGuardFunc:
    """Look up a registered guard callable by name.

    Args:
        name: The registered guard name.

    Returns:
        The guard callable.

    Raises:
        GuardNotFoundError: If no guard is registered under *name*.
    """
    if name not in self._guards:
        raise GuardNotFoundError(f"Guard '{name}' not found", guard_name=name)
    return self._guards[name]

has

has(name: str) -> bool

Check whether a guard is registered.

Source code in src/pystator/guards.py

def has(self, name: str) -> bool:
    """Check whether a guard is registered."""
    return name in self._guards

is_async

is_async(name: str) -> bool

Return True if the named guard is an async function.

Source code in src/pystator/guards.py

def is_async(self, name: str) -> bool:
    """Return True if the named guard is an async function."""
    return name in self._async_guards

evaluate

evaluate(name: str, context: dict[str, Any]) -> bool

Evaluate a single guard synchronously.

Source code in src/pystator/guards.py

def evaluate(self, name: str, context: dict[str, Any]) -> bool:
    """Evaluate a single guard synchronously."""
    return self.get(name)(context)  # type: ignore[return-value]

evaluate_all

evaluate_all(guards: tuple[str, ...] | list[str], context: dict[str, Any], fail_fast: bool = True) -> GuardResult

Evaluate multiple guards. Returns GuardResult.

Source code in src/pystator/guards.py

def evaluate_all(
    self,
    guards: tuple[str, ...] | list[str],
    context: dict[str, Any],
    fail_fast: bool = True,
) -> GuardResult:
    """Evaluate multiple guards. Returns GuardResult."""
    if not guards:
        return GuardResult.success()
    evaluated: list[tuple[str, bool]] = []
    for name in guards:
        result = self.evaluate(name, context)
        evaluated.append((name, result))
        if not result and fail_fast:
            return GuardResult.failure(name, evaluated=evaluated)
    for name, passed in evaluated:
        if not passed:
            return GuardResult.failure(name, evaluated=evaluated)
    return GuardResult.success(evaluated)

async_evaluate async

async_evaluate(name: str, context: dict[str, Any]) -> bool

Evaluate a single guard asynchronously.

Falls back to synchronous execution if the guard is not async.

Parameters:

Name	Type	Description	Default
name	str	Registered guard name.	required
context	dict[str, Any]	Context dict passed to the guard callable.	required

Returns:

Type	Description
bool	True if the guard passed, False otherwise.

Source code in src/pystator/guards.py

async def async_evaluate(self, name: str, context: dict[str, Any]) -> bool:
    """Evaluate a single guard asynchronously.

    Falls back to synchronous execution if the guard is not async.

    Args:
        name: Registered guard name.
        context: Context dict passed to the guard callable.

    Returns:
        True if the guard passed, False otherwise.
    """
    func = self.get(name)
    if self.is_async(name):
        return await func(context)  # type: ignore[misc]
    return func(context)  # type: ignore[return-value]

async_evaluate_all async

async_evaluate_all(guards: tuple[GuardSpec, ...] | tuple[str, ...] | list[str], context: dict[str, Any], fail_fast: bool = True) -> GuardResult

Evaluate multiple guards asynchronously.

Parameters:

Name	Type	Description	Default
guards	tuple[GuardSpec, ...] | tuple[str, ...] | list[str]	Guard specs or names to evaluate.	required
context	dict[str, Any]	Context dict passed to each guard.	required
fail_fast	bool	If True, stop on first failing guard.	True

Returns:

Type	Description
GuardResult	GuardResult summarising all evaluations.

Source code in src/pystator/guards.py

async def async_evaluate_all(
    self,
    guards: tuple[GuardSpec, ...] | tuple[str, ...] | list[str],
    context: dict[str, Any],
    fail_fast: bool = True,
) -> GuardResult:
    """Evaluate multiple guards asynchronously.

    Args:
        guards: Guard specs or names to evaluate.
        context: Context dict passed to each guard.
        fail_fast: If True, stop on first failing guard.

    Returns:
        GuardResult summarising all evaluations.
    """
    if not guards:
        return GuardResult.success()
    evaluated: list[tuple[str, bool]] = []
    for guard in guards:
        name = guard.name if isinstance(guard, GuardSpec) else guard
        if name is None:
            continue
        eval_ctx = context
        if isinstance(guard, GuardSpec) and guard.has_params:
            eval_ctx = {**context, GUARD_PARAMS_KEY: guard.params}
        result = await self.async_evaluate(name, eval_ctx)
        evaluated.append((name, result))
        if not result and fail_fast:
            return GuardResult.failure(name, evaluated=evaluated)
    for name, passed in evaluated:
        if not passed:
            return GuardResult.failure(name, evaluated=evaluated)
    return GuardResult.success(evaluated)

list_guards

list_guards() -> list[str]

Return all registered guard names.

Source code in src/pystator/guards.py

def list_guards(self) -> list[str]:
    """Return all registered guard names."""
    return list(self._guards.keys())

clear

clear() -> None

Remove all registered guards.

Source code in src/pystator/guards.py

def clear(self) -> None:
    """Remove all registered guards."""
    self._guards.clear()
    self._async_guards.clear()

can_transition

can_transition(transition: Transition, context: dict[str, Any]) -> GuardResult

Check if a transition is allowed based on its guards.

Parameters:

Name	Type	Description	Default
transition	Transition	The transition whose guards to evaluate.	required
context	dict[str, Any]	Context dict passed to each guard.	required

Returns:

Type	Description
GuardResult	GuardResult indicating pass/fail with evaluation details.

Source code in src/pystator/guards.py

def can_transition(
    self,
    transition: Transition,
    context: dict[str, Any],
) -> GuardResult:
    """Check if a transition is allowed based on its guards.

    Args:
        transition: The transition whose guards to evaluate.
        context: Context dict passed to each guard.

    Returns:
        GuardResult indicating pass/fail with evaluation details.
    """
    if not transition.guards:
        return GuardResult.success()
    return self._evaluate_guards(transition.guards, context)

async_can_transition async

async_can_transition(transition: Transition, context: dict[str, Any]) -> GuardResult

Async version of :meth:can_transition.

Falls back to sync evaluation for non-async guards.

Source code in src/pystator/guards.py

async def async_can_transition(
    self,
    transition: Transition,
    context: dict[str, Any],
) -> GuardResult:
    """Async version of :meth:`can_transition`.

    Falls back to sync evaluation for non-async guards.
    """
    if not transition.guards:
        return GuardResult.success()
    # For async path, delegate to async_evaluate_all with GuardSpec objects
    return await self.async_evaluate_all(transition.guards, context, fail_fast=True)

evaluate_and_raise

evaluate_and_raise(transition: Transition, current_state: str, context: dict[str, Any]) -> None

Evaluate guards and raise GuardRejectedError if blocked.

Source code in src/pystator/guards.py

def evaluate_and_raise(
    self,
    transition: Transition,
    current_state: str,
    context: dict[str, Any],
) -> None:
    """Evaluate guards and raise GuardRejectedError if blocked."""
    result = self.can_transition(transition, context)
    if not result.passed:
        raise GuardRejectedError(
            message=result.message,
            current_state=current_state,
            trigger=transition.trigger,
            guard_name=result.guard_name or "unknown",
            guard_result=result.evaluated_guards,
        )

get_required_guards

get_required_guards(transitions: list[Transition] | tuple[Transition, ...]) -> set[str]

Return named guard names required by the given transitions.

Recurses into composite guards (allOf, anyOf, not). Excludes inline expression guards.

Source code in src/pystator/guards.py

def get_required_guards(
    self,
    transitions: list[Transition] | tuple[Transition, ...],
) -> set[str]:
    """Return named guard names required by the given transitions.

    Recurses into composite guards (allOf, anyOf, not).
    Excludes inline expression guards.
    """
    names: set[str] = set()
    for trans in transitions:
        for guard in trans.guards:
            _collect_guard_names(guard, names)
    return names

decorator

decorator(name: str | None = None) -> Callable[[AnyGuardFunc], AnyGuardFunc]

Decorator to register a guard function.

Example

@registry.decorator() ... def is_valid(ctx: dict) -> bool: ... return ctx.get("valid", False)

Source code in src/pystator/guards.py

def decorator(
    self, name: str | None = None
) -> Callable[[AnyGuardFunc], AnyGuardFunc]:
    """Decorator to register a guard function.

    Example:
        >>> @registry.decorator()
        ... def is_valid(ctx: dict) -> bool:
        ...     return ctx.get("valid", False)
    """

    def inner(func: AnyGuardFunc) -> AnyGuardFunc:
        self.register(name or func.__name__, func)
        return func

    return inner

GuardResult dataclass

GuardResult(passed: bool, guard_name: str | None = None, message: str = '', evaluated_guards: list[tuple[str, bool]] = list())

Result of guard evaluation.

Attributes:

Name	Type	Description
passed	bool	Whether all guards passed.
guard_name	str | None	Name of the last evaluated guard (if failed).
message	str	Explanation of the result.
evaluated_guards	list[tuple[str, bool]]	List of (guard_name, passed) tuples.

success classmethod

success(evaluated: list[tuple[str, bool]] | None = None) -> GuardResult

Create a passing guard result.

Parameters:

Name	Type	Description	Default
evaluated	list[tuple[str, bool]] | None	Optional list of (guard_name, passed) tuples.	None

Returns:

Type	Description
GuardResult	GuardResult with passed=True.

Source code in src/pystator/guards.py

@classmethod
def success(cls, evaluated: list[tuple[str, bool]] | None = None) -> GuardResult:
    """Create a passing guard result.

    Args:
        evaluated: Optional list of (guard_name, passed) tuples.

    Returns:
        GuardResult with ``passed=True``.
    """
    return cls(passed=True, evaluated_guards=evaluated or [])

failure classmethod

failure(guard_name: str, message: str = '', evaluated: list[tuple[str, bool]] | None = None) -> GuardResult

Create a failing guard result with reason.

Parameters:

Name	Type	Description	Default
guard_name	str	Name of the guard that failed.	required
message	str	Human-readable failure reason.	''
evaluated	list[tuple[str, bool]] | None	Optional list of (guard_name, passed) tuples.	None

Returns:

Type	Description
GuardResult	GuardResult with passed=False.

Source code in src/pystator/guards.py

@classmethod
def failure(
    cls,
    guard_name: str,
    message: str = "",
    evaluated: list[tuple[str, bool]] | None = None,
) -> GuardResult:
    """Create a failing guard result with reason.

    Args:
        guard_name: Name of the guard that failed.
        message: Human-readable failure reason.
        evaluated: Optional list of (guard_name, passed) tuples.

    Returns:
        GuardResult with ``passed=False``.
    """
    return cls(
        passed=False,
        guard_name=guard_name,
        message=message or f"Guard '{guard_name}' rejected transition",
        evaluated_guards=evaluated or [],
    )

Transition-level evaluation (can_transition, async_can_transition, …) lives on GuardRegistry (formerly a separate GuardEvaluator).

Built-in guard helpers:

equals

equals(key: str, value: Any) -> Callable[[dict[str, Any]], bool]

Guard factory: context[key] == value.

Source code in src/pystator/guards.py

def equals(key: str, value: Any) -> Callable[[dict[str, Any]], bool]:
    """Guard factory: context[key] == value."""

    def guard(ctx: dict[str, Any]) -> bool:
        return ctx.get(key) == value

    return guard

greater_than

greater_than(key: str, value: Any) -> Callable[[dict[str, Any]], bool]

Guard factory: context[key] > value.

Source code in src/pystator/guards.py

def greater_than(key: str, value: Any) -> Callable[[dict[str, Any]], bool]:
    """Guard factory: context[key] > value."""

    def guard(ctx: dict[str, Any]) -> bool:
        v = ctx.get(key)
        if v is None:
            return False
        try:
            return v > value
        except TypeError:
            return False

    return guard

in_list

in_list(key: str, values: list[Any]) -> Callable[[dict[str, Any]], bool]

Guard factory: context[key] in values.

Source code in src/pystator/guards.py

def in_list(key: str, values: list[Any]) -> Callable[[dict[str, Any]], bool]:
    """Guard factory: context[key] in values."""

    def guard(ctx: dict[str, Any]) -> bool:
        return ctx.get(key) in values

    return guard

all_of

all_of(*guards: Callable[[dict[str, Any]], bool]) -> Callable[[dict[str, Any]], bool]

Compound guard: all must pass.

Source code in src/pystator/guards.py

def all_of(
    *guards: Callable[[dict[str, Any]], bool],
) -> Callable[[dict[str, Any]], bool]:
    """Compound guard: all must pass."""

    def guard(ctx: dict[str, Any]) -> bool:
        return all(g(ctx) for g in guards)

    return guard

any_of

any_of(*guards: Callable[[dict[str, Any]], bool]) -> Callable[[dict[str, Any]], bool]

Compound guard: at least one must pass.

Source code in src/pystator/guards.py

def any_of(
    *guards: Callable[[dict[str, Any]], bool],
) -> Callable[[dict[str, Any]], bool]:
    """Compound guard: at least one must pass."""

    def guard(ctx: dict[str, Any]) -> bool:
        return any(g(ctx) for g in guards)

    return guard

negate

negate(guard_fn: Callable[[dict[str, Any]], bool]) -> Callable[[dict[str, Any]], bool]

Guard that negates the result of another guard.

Source code in src/pystator/guards.py

def negate(
    guard_fn: Callable[[dict[str, Any]], bool],
) -> Callable[[dict[str, Any]], bool]:
    """Guard that negates the result of another guard."""

    def guard(ctx: dict[str, Any]) -> bool:
        return not guard_fn(ctx)

    return guard