Context and Sinks API

TransitionContext

TransitionContext

Bases: dict

Context dict with typed access to internal metadata.

Internal keys (prefixed with _) are set by the engine and orchestrator. User data lives alongside them in the same dict.

Example::

def my_guard(ctx: dict) -> bool:
    tc = TransitionContext(ctx)
    event = tc.event          # typed access
    entity = tc.entity_id     # typed access
    return ctx.get("valid")   # still works as a normal dict

event property

event: Event | None

The Event object that triggered this transition.

entity_id property

entity_id: str | None

The entity ID (set by Orchestrator).

action_params property

action_params: dict[str, Any]

Parameters injected for the current action.

user_data

user_data() -> dict[str, Any]

Return a copy of the context without internal (_-prefixed) keys.

Source code in src/pystator/context.py

def user_data(self) -> dict[str, Any]:
    """Return a copy of the context without internal (``_``-prefixed) keys."""
    return {k: v for k, v in self.items() if not k.startswith("_")}

Sinks

EventSink

Bases: Protocol

Protocol for domain event sinks.

MetricSink

Bases: Protocol

Protocol for metric emission sinks.

LoggingEventSink

LoggingEventSink(logger_name: str = 'pystator.events')

Logs events at INFO. Zero-dependency default.

Source code in src/pystator/sinks.py

def __init__(self, logger_name: str = "pystator.events") -> None:
    self._logger = logging.getLogger(logger_name)

LoggingMetricSink

LoggingMetricSink(logger_name: str = 'pystator.metrics')

Logs metrics at INFO. Zero-dependency default.

Source code in src/pystator/sinks.py

def __init__(self, logger_name: str = "pystator.metrics") -> None:
    self._logger = logging.getLogger(logger_name)

configure_event_sink

configure_event_sink(registry: ActionRegistry, sink: EventSink | None = None) -> None

Register the publish action backed by the given sink.

If sink is None, uses :class:LoggingEventSink. Replaces the action if already registered.

Source code in src/pystator/sinks.py

def configure_event_sink(
    registry: ActionRegistry,
    sink: EventSink | None = None,
) -> None:
    """Register the ``publish`` action backed by the given sink.

    If *sink* is ``None``, uses :class:`LoggingEventSink`.
    Replaces the action if already registered.
    """
    if sink is None:
        sink = LoggingEventSink()
    if registry.has("publish"):
        registry.unregister("publish")
    registry.register("publish", create_publish_action(sink))

configure_metric_sink

configure_metric_sink(registry: ActionRegistry, sink: MetricSink | None = None) -> None

Register the emit_metric action backed by the given sink.

If sink is None, uses :class:LoggingMetricSink. Replaces the action if already registered.

Source code in src/pystator/sinks.py

def configure_metric_sink(
    registry: ActionRegistry,
    sink: MetricSink | None = None,
) -> None:
    """Register the ``emit_metric`` action backed by the given sink.

    If *sink* is ``None``, uses :class:`LoggingMetricSink`.
    Replaces the action if already registered.
    """
    if sink is None:
        sink = LoggingMetricSink()
    if registry.has("emit_metric"):
        registry.unregister("emit_metric")
    registry.register("emit_metric", create_metric_action(sink))

Import

from pystator.context import TransitionContext

from pystator.sinks import (
    EventSink,
    MetricSink,
    LoggingEventSink,
    LoggingMetricSink,
    configure_event_sink,
    configure_metric_sink,
)

Context key reference

Key	Set by	Description
_event	Engine	Event object for the current transition
_entity_id	Orchestrator	Entity ID being processed
_action_params	ActionExecutor	Parameters from YAML params: for the current action

All other keys are user-defined and come from the entity's persistent context or the event payload.

See also

• Context guide
• Orchestrator — context_validator parameter
• Hooks and Metrics