Skip to content

Tutorial: Order workflow

Build a small order lifecycle state machine with guards and actions, and run it from Python.

Overview

You will:

  1. Define an order FSM in YAML (states and transitions).
  2. Add guards so that “full fill” only fires when fill_qty >= order_qty.
  3. Register actions and run them after each transition.
  4. Simulate a short order lifecycle in a script.

Prerequisites

  • Python 3.11+
  • pip install pystator

Step 1: Create the FSM YAML

Create order_fsm.yaml:

meta:
  version: "1.0.0"
  machine_name: "order_management"
  strict_mode: true

states:
  - name: pending_new
    type: initial
    timeout:
      seconds: 5.0
      destination: timed_out_ack

  - name: open
    type: stable
    on_enter:
      - notify_ui_open
      - log_audit_trail

  - name: partially_filled
    type: stable

  - name: filled
    type: terminal

  - name: rejected
    type: terminal

  - name: canceled
    type: terminal

  - name: timed_out_ack
    type: terminal

transitions:
  - trigger: exchange_ack
    source: pending_new
    dest: open
    actions:
      - update_order_id

  - trigger: exchange_nack
    source: pending_new
    dest: rejected
    actions:
      - record_rejection_reason

  - trigger: execution_report
    source: [open, partially_filled]
    dest: filled
    guards:
      - is_full_fill
    actions:
      - update_positions
      - release_buying_power

  - trigger: execution_report
    source: open
    dest: partially_filled
    guards:
      - is_partial_fill
    actions:
      - update_positions
      - decrement_remaining_qty

  - trigger: cancel_request
    source: [open, partially_filled]
    dest: canceled
    guards:
      - is_cancellable
    actions:
      - send_cancel_to_exchange

Optional: Define state variables

You can declare which context keys your machine expects so that tooling (and optional validation) know the contract. Add a top-level state_variables list:

state_variables:
  - key: order_qty
    type: number
    description: "Total order quantity"
    default: 0
  - key: fill_qty
    type: number
    default: 0
  - key: in_flight
    type: boolean
    default: false
  - key: order_id
    type: string
    required: true
  • required: true — When you set meta.validate_context: true, the engine will reject transitions if this key is missing or null in the context.
  • default — Used by the UI “Fill from state variables” to pre-fill context JSON; optional for documentation.
  • type — When validate_context is true, the engine can check that values match (string, number, boolean, object).

Shorthand: state_variables: [order_id, order_qty, fill_qty] is equivalent to a list of { key: "..." } items. See the FSM config reference for full details.

Step 2: Load the machine and register guards

Guards decide whether a transition is allowed. We’ll implement is_full_fill, is_partial_fill, and is_cancellable:

from pathlib import Path
from pystator import StateMachine, GuardRegistry

config_path = Path("order_fsm.yaml")
machine = StateMachine.from_yaml(config_path)

guards = GuardRegistry()
guards.register(
    "is_full_fill",
    lambda ctx: ctx.get("fill_qty", 0) >= ctx.get("order_qty", 1),
)
guards.register(
    "is_partial_fill",
    lambda ctx: 0 < ctx.get("fill_qty", 0) < ctx.get("order_qty", 1),
)
guards.register(
    "is_cancellable",
    lambda ctx: not ctx.get("in_flight", False),
)

machine.bind_guards(guards)

Step 3: Register actions and create executor

Actions run after you persist the new state. Here we use print statements; in production you’d call APIs or DBs:

from pystator import ActionRegistry
from pystator.actions import ActionExecutor

actions = ActionRegistry()
actions.register("update_order_id", lambda ctx: print("  -> Updated order ID"))
actions.register("notify_ui_open", lambda ctx: print("  -> UI notified: order open"))
actions.register("log_audit_trail", lambda ctx: print("  -> Audit logged"))
actions.register("update_positions", lambda ctx: print("  -> Positions updated"))
actions.register("release_buying_power", lambda ctx: print("  -> Buying power released"))
actions.register("decrement_remaining_qty", lambda ctx: print("  -> Remaining qty decremented"))
actions.register(
    "record_rejection_reason",
    lambda ctx: print(f"  -> Rejection: {ctx.get('reason', 'unknown')}"),
)
actions.register("send_cancel_to_exchange", lambda ctx: print("  -> Cancel sent to exchange"))

executor = ActionExecutor(actions)

Step 4: Simulate the lifecycle

Keep “current state” in a variable (in a real app this would be in your DB), and call process then executor.execute:

order = {
    "id": "ORD-001",
    "state": "pending_new",
    "order_qty": 100,
}

def process_event(trigger: str, extra_context: dict | None = None):
    ctx = {**order, **(extra_context or {})}
    print(f"\n[{order['state']}] Event: {trigger}")
    result = machine.process(order["state"], trigger, ctx)
    if result.success:
        print(f"  -> {result.source_state} -> {result.target_state}")
        order["state"] = result.target_state or order["state"]
        if result.all_actions:
            executor.execute(result, ctx)
    else:
        print(f"  -> Failed: {result.error}")

# Run through a short lifecycle
process_event("exchange_ack")
process_event("execution_report", {"fill_qty": 50})
process_event("execution_report", {"fill_qty": 100})
process_event("cancel_request", {"in_flight": False})  # No-op: already filled

print(f"\nFinal state: {order['state']}")

Running this script should show: pending_new → open → partially_filled → filled, with actions printed after each transition. The last cancel_request does nothing because there is no transition from filled.

What you learned

  • States: initial, stable, terminal; optional on_enter, on_exit, timeout.
  • Transitions: trigger, source, dest, guards, actions.
  • Guards: Registered by name, bound to the machine; context drives the decision.
  • Actions: Registered by name; run only after you persist the new state (sandwich pattern).

Next steps