Context, TransitionContext, and Sinks¶

Context dict¶

The context is a plain Python dict passed to every guard and action during a transition. It combines:

The entity's persistent context (stored between calls)
The event payload (data passed with the trigger)
Internal keys set by the engine and Orchestrator (prefixed with _)

Guards and actions receive it as their first argument:

def my_guard(ctx: dict) -> bool:
    return ctx.get("amount", 0) > 100

def my_action(ctx: dict) -> None:
    print(f"Processing entity {ctx.get('_entity_id')}")

`TransitionContext` — typed access to context¶

TransitionContext is a dict subclass that adds typed property accessors for the internal keys set by the engine. It is fully backward-compatible — every existing guard and action continues to work without changes.

Cast your context dict to TransitionContext inside a guard or action to get typed access:

from pystator.context import TransitionContext

def my_guard(ctx: dict) -> bool:
    tc = TransitionContext(ctx)
    event = tc.event            # Event | None
    entity = tc.entity_id       # str | None
    params = tc.action_params   # dict (action parameters)
    user = tc.user_data()       # dict — ctx without _ keys
    return user.get("amount", 0) > 100

Properties¶

Property	Type	Value
`tc.event`	`Event \\| None`	The `Event` object that triggered this transition
`tc.entity_id`	`str \\| None`	Entity ID set by the Orchestrator
`tc.action_params`	`dict`	Parameters injected for the current action (from YAML `params:`)
`tc.user_data()`	`dict`	All context keys that do not start with `_`

Import¶

from pystator.context import TransitionContext

`ContextValidatorFn` — validating context with the Orchestrator¶

When using the Orchestrator, you can pass a context_validator function that is called before every transition. Use it to enforce context shape, integrate with PyCharter for contract validation, or reject invalid context early.

from pystator import Orchestrator

def validate_order_context(entity_id: str, context: dict, trigger: str) -> None:
    """Raises ValueError if context is invalid."""
    if "order_id" not in context:
        raise ValueError(f"Missing order_id for entity {entity_id}")
    if trigger == "submit" and "amount" not in context:
        raise ValueError("submit requires 'amount' in context")

orchestrator = Orchestrator(
    machine=machine,
    store=store,
    context_validator=validate_order_context,
)

The validator signature is:

ContextValidatorFn = Callable[[str, dict[str, Any], str], None]
#                              entity_id  context   trigger

Raise any exception to reject the transition. The exception is caught by the Orchestrator and surfaced in the TransitionResult.

PyCharter integration¶

from pycharter import Validator
from pycharter.contract_parser import parse_contract_file

contract = parse_contract_file("contracts/order_context.yaml")
validator = Validator.from_contract(contract)

def pycharter_validator(entity_id: str, context: dict, trigger: str) -> None:
    result = validator.validate(context)
    if not result.is_valid:
        raise ValueError(f"Context validation failed: {result.errors}")

orchestrator = Orchestrator(
    machine=machine,
    store=store,
    context_validator=pycharter_validator,
)

Event sinks — `publish` and `emit_metric` actions¶

The built-in publish and emit_metric actions work through a sink abstraction. The default implementations log to Python's logging module; swap them out for production backends (Kafka, Redis Pub/Sub, Datadog, etc.) by implementing the protocols and calling the configure helpers.

Default behaviour¶

from pystator import builtin_registries

guards, actions = builtin_registries()
# publish     → LoggingEventSink (logs to pystator.events at INFO)
# emit_metric → LoggingMetricSink (logs to pystator.metrics at INFO)

Swapping in a production event sink¶

Implement EventSink:

from pystator.sinks import EventSink, configure_event_sink

class KafkaSink:
    def __init__(self, producer):
        self._producer = producer

    def send(self, topic: str, payload: dict, metadata: dict) -> None:
        self._producer.produce(
            topic,
            value=json.dumps({**payload, **metadata}).encode(),
        )

guards, actions = builtin_registries()
configure_event_sink(actions, KafkaSink(kafka_producer))

YAML:

transitions:
  - trigger: confirm
    source: pending
    dest: confirmed
    actions:
      - name: publish
        params:
          topic: "orders.confirmed"
          include: [order_id, customer_id]

Swapping in a production metric sink¶

Implement MetricSink:

from pystator.sinks import MetricSink, configure_metric_sink
from datadog import statsd

class DatadogSink:
    def emit(self, name: str, value: float, tags: dict) -> None:
        dd_tags = [f"{k}:{v}" for k, v in tags.items()]
        statsd.increment(name, tags=dd_tags)

guards, actions = builtin_registries()
configure_metric_sink(actions, DatadogSink())

YAML:

transitions:
  - trigger: confirm
    source: pending
    dest: confirmed
    actions:
      - name: emit_metric
        params:
          name: "orders.confirmed"
          type: counter
          tags:
            env: production

Sink protocols¶

class EventSink(Protocol):
    def send(self, topic: str, payload: dict, metadata: dict) -> None: ...

class MetricSink(Protocol):
    def emit(self, name: str, value: float, tags: dict[str, str]) -> None: ...

The metadata dict passed to EventSink.send always contains: - timestamp — UTC ISO 8601 string - state — the source or target state (context-dependent) - trigger — the trigger name