Worker (Python API)

Long-running event consumer that claims rows from worker_events, runs the Orchestrator, and updates entity state. Operational guide: Worker.

Types and entrypoints

Worker

Worker(config: WorkerConfig | None = None, guards: GuardRegistry | None = None, actions: ActionRegistry | None = None, *, event_source: EventSource | None = None)

PyStator worker: a long-running event-processing service.

Usage::

from pystator.worker import Worker

worker = Worker()
await worker.start()
await worker.run()   # blocks until shutdown signal

With custom guards and actions::

from pystator import GuardRegistry, ActionRegistry
from pystator.worker import Worker

guards = GuardRegistry()
actions = ActionRegistry()

@guards.register("is_full_fill")
def is_full_fill(ctx):
    return ctx["fill_qty"] >= ctx["order_qty"]

worker = Worker(guards=guards, actions=actions)
await worker.start()
await worker.run()

Parameters:

Name	Type	Description	Default
config	WorkerConfig | None	Worker configuration. Defaults to :class:WorkerConfig which reads from pystator.cfg / environment.	None
guards	GuardRegistry | None	Guard registry bound to all loaded machines.	None
actions	ActionRegistry | None	Action registry bound to all loaded machines.	None
event_source	EventSource | None	Custom event source. When None, a :class:DatabaseEventSource is created from the config.	None

Source code in src/pystator/worker/runner.py

def __init__(
    self,
    config: WorkerConfig | None = None,
    guards: GuardRegistry | None = None,
    actions: ActionRegistry | None = None,
    *,
    event_source: EventSource | None = None,
) -> None:
    self._config = config or WorkerConfig()
    self._guards = guards or GuardRegistry()
    self._actions = actions or ActionRegistry()
    self._custom_event_source = event_source

    # Initialised in start()
    self._event_source: EventSource | None = None
    self._apply_data_event_source: EventSource | None = None
    self._registry: MachineRegistry | None = None
    self._processor: EventProcessor | None = None
    self._apply_data_processor: ApplyDataProcessor | None = None
    self._health: HealthChecker | None = None
    self._dlq_store: DeadLetterStore | None = None

    # Shutdown coordination
    self._shutdown_event = asyncio.Event()
    self._in_flight: int = 0
    self._in_flight_apply_data: int = 0
    self._started = False

health property

health: HealthChecker | None

Health checker (available after :meth:start).

dlq_store property

dlq_store: DeadLetterStore | None

Dead letter store (available after :meth:start when DLQ is enabled).

start async

start() -> None

Initialise connections, load machines, verify readiness.

Must be called before :meth:run.

Source code in src/pystator/worker/runner.py

async def start(self) -> None:
    """Initialise connections, load machines, verify readiness.

    Must be called before :meth:`run`.
    """
    config = self._config
    db_url = config.resolve_db_url()
    if db_url.startswith("sqlite:"):
        logger.info("Worker event queue database: %s", db_url)

    # Event source
    if self._custom_event_source is not None:
        self._event_source = self._custom_event_source
    else:
        self._event_source = DatabaseEventSource(db_url, config=config)

    # Apply-data event source (separate queue for isolation)
    self._apply_data_event_source = DatabaseApplyDataEventSource(
        db_url, config=config
    )

    # State store for the Orchestrators (sync protocol).
    # Use SQLAlchemy store if available, otherwise in-memory.
    state_store = self._create_state_store(db_url)

    # Machine registry
    self._registry = MachineRegistry(
        state_store=state_store,
        guards=self._guards,
        actions=self._actions,
        db_url=db_url,
        machine_source=config.machine_source,
        machine_dir=config.machine_dir,
    )
    self._registry.load()

    # Dead letter queue store (pystator.dlq backends)
    if config.dlq_enabled:
        if config.dlq_database_url:
            dlq_url = get_db_url(config.dlq_database_url, required=True)
            self._dlq_store = create_store(
                DLQConfig(
                    backend="database",
                    database_url=dlq_url,
                    table_name=config.dlq_table_name,
                )
            )
            logger.info(
                "DLQ store enabled (database table=%s)",
                config.dlq_table_name,
            )
        else:
            self._dlq_store = create_store(
                DLQConfig(backend="memory", max_memory_size=10_000)
            )
            logger.info("DLQ store enabled (in-memory)")

    # Processor
    self._processor = EventProcessor(
        registry=self._registry,
        event_source=self._event_source,
        dlq_store=self._dlq_store,
    )
    self._apply_data_processor = ApplyDataProcessor(
        registry=self._registry,
        event_source=self._apply_data_event_source,
        dlq_store=self._dlq_store,
    )

    # Health checker
    self._health = HealthChecker(
        event_source=self._event_source,
        registry=self._registry,
    )

    self._started = True
    logger.info(
        "Worker %s started (events=%d@%dms, apply_data=%d@%dms, machines=%d)",
        config.worker_id,
        config.concurrency,
        config.poll_interval_ms,
        config.apply_data_concurrency,
        config.apply_data_poll_interval_ms,
        self._registry.machine_count,
    )

run async

run() -> None

Block and process events until a shutdown signal is received.

Spawns config.concurrency poller tasks. Each poller claims events, processes them, then sleeps for poll_interval_ms before the next claim.

Handles SIGTERM and SIGINT for graceful shutdown.

Source code in src/pystator/worker/runner.py

async def run(self) -> None:
    """Block and process events until a shutdown signal is received.

    Spawns ``config.concurrency`` poller tasks.  Each poller claims
    events, processes them, then sleeps for ``poll_interval_ms``
    before the next claim.

    Handles ``SIGTERM`` and ``SIGINT`` for graceful shutdown.
    """
    if not self._started:
        raise RuntimeError("Worker.start() must be called before run()")

    # Install signal handlers
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, self._request_shutdown, sig)

    config = self._config
    tasks: list[asyncio.Task[None]] = []

    for idx in range(config.concurrency):
        task = asyncio.create_task(
            self._poll_loop(idx),
            name=f"worker-poller-{idx}",
        )
        tasks.append(task)

    apply_tasks: list[asyncio.Task[None]] = []
    for idx in range(config.apply_data_concurrency):
        task = asyncio.create_task(
            self._poll_apply_data_loop(idx),
            name=f"worker-apply-data-poller-{idx}",
        )
        apply_tasks.append(task)

    logger.info(
        "Worker %s running (events=%d poller(s), apply_data=%d poller(s))",
        config.worker_id,
        config.concurrency,
        config.apply_data_concurrency,
    )

    # Wait for shutdown signal
    await self._shutdown_event.wait()

    logger.info("Shutdown requested, draining in-flight events...")

    # Cancel pollers (they will stop after their current iteration)
    for task in tasks:
        task.cancel()
    for task in apply_tasks:
        task.cancel()

    # Wait for in-flight events to complete (with timeout)
    await self._drain(config.drain_timeout_s)

    # Cleanup
    await self._cleanup()

    logger.info("Worker %s stopped", config.worker_id)

shutdown async

shutdown() -> None

Programmatically trigger a graceful shutdown.

Source code in src/pystator/worker/runner.py

async def shutdown(self) -> None:
    """Programmatically trigger a graceful shutdown."""
    self._shutdown_event.set()

WorkerConfig dataclass

WorkerConfig(db_url: str | None = None, poll_interval_ms: int = 500, concurrency: int = 5, apply_data_poll_interval_ms: int = 250, apply_data_concurrency: int = 5, drain_timeout_s: int = 30, max_attempts: int = 5, apply_data_max_attempts: int = 5, machine_source: str = 'db', machine_dir: str | None = None, worker_id: str = _default_worker_id(), claim_lease_timeout_s: int = 300, retry_backoff_base_ms: int = 1000, retry_backoff_max_ms: int = 300000, dlq_enabled: bool = True, dlq_database_url: str | None = None, dlq_table_name: str = 'dead_letter_queue')

submit_event async

submit_event(machine_name: str, entity_id: str, trigger: str, *, context: dict[str, Any] | None = None, fires_at: datetime | None = None, idempotency_key: str | None = None, max_attempts: int = 5, db_url: str | None = None) -> str

Convenience function: submit one event to the worker queue.

Writes directly to the worker_events table. Can be called from any process that has database access.

Parameters:

Name	Type	Description	Default
machine_name	str	Target FSM machine name.	required
entity_id	str	Entity identifier (e.g. "order-123").	required
trigger	str	Event trigger name (e.g. "fill").	required
context	dict[str, Any] | None	Optional dict passed to guards and actions.	None
fires_at	datetime | None	When the event becomes eligible. None = now.	None
idempotency_key	str | None	Optional dedup key.	None
max_attempts	int	Maximum processing attempts.	5
db_url	str | None	Database URL. Falls back to the standard pystator config resolution chain.	None

Returns:

Type	Description
str	The event_id of the enqueued event.

Source code in src/pystator/worker/__init__.py

async def submit_event(
    machine_name: str,
    entity_id: str,
    trigger: str,
    *,
    context: dict[str, Any] | None = None,
    fires_at: datetime | None = None,
    idempotency_key: str | None = None,
    max_attempts: int = 5,
    db_url: str | None = None,
) -> str:
    """Convenience function: submit one event to the worker queue.

    Writes directly to the ``worker_events`` table.  Can be called from
    any process that has database access.

    Args:
        machine_name: Target FSM machine name.
        entity_id: Entity identifier (e.g. ``"order-123"``).
        trigger: Event trigger name (e.g. ``"fill"``).
        context: Optional dict passed to guards and actions.
        fires_at: When the event becomes eligible.  ``None`` = now.
        idempotency_key: Optional dedup key.
        max_attempts: Maximum processing attempts.
        db_url: Database URL.  Falls back to the standard pystator
            config resolution chain.

    Returns:
        The ``event_id`` of the enqueued event.
    """
    from pystator.worker.event_sources.database import DatabaseEventSource

    if db_url is not None:
        resolved_url = db_url
    else:
        from pystator.db.config import get_db_url

        resolved_url = get_db_url(None, required=False)

    source = DatabaseEventSource(resolved_url)
    try:
        event = WorkerEvent(
            machine_name=machine_name,
            entity_id=entity_id,
            trigger=trigger,
            context=context,
            fires_at=fires_at,
            idempotency_key=idempotency_key,
            max_attempts=max_attempts,
        )
        return await source.submit(event)
    finally:
        await source.close()

submit_event_sync

submit_event_sync(machine_name: str, entity_id: str, trigger: str, *, context: dict[str, Any] | None = None, fires_at: datetime | None = None, idempotency_key: str | None = None, max_attempts: int = 5, db_url: str | None = None) -> str

Synchronous wrapper around :func:submit_event.

Suitable for calling from non-async code (scripts, Django views, Flask routes, etc.).

Source code in src/pystator/worker/__init__.py

def submit_event_sync(
    machine_name: str,
    entity_id: str,
    trigger: str,
    *,
    context: dict[str, Any] | None = None,
    fires_at: datetime | None = None,
    idempotency_key: str | None = None,
    max_attempts: int = 5,
    db_url: str | None = None,
) -> str:
    """Synchronous wrapper around :func:`submit_event`.

    Suitable for calling from non-async code (scripts, Django views,
    Flask routes, etc.).
    """
    return asyncio.run(
        submit_event(
            machine_name,
            entity_id,
            trigger,
            context=context,
            fires_at=fires_at,
            idempotency_key=idempotency_key,
            max_attempts=max_attempts,
            db_url=db_url,
        )
    )

Related models

WorkerEvent dataclass

WorkerEvent(machine_name: str, entity_id: str, trigger: str, context: dict[str, Any] | None = None, fires_at: datetime | None = None, idempotency_key: str | None = None, max_attempts: int = 5)

Immutable event envelope submitted by callers.

Parameters:

Name	Type	Description	Default
machine_name	str	Name of the FSM machine to route to.	required
entity_id	str	Identifier for the entity (e.g. "order-123").	required
trigger	str	The event trigger name (e.g. "fill").	required
context	dict[str, Any] | None	Optional dict passed to guards and actions.	None
fires_at	datetime | None	When the event becomes eligible for processing. None means immediately.	None
idempotency_key	str | None	Optional client-supplied dedup key. If a pending/claimed event with the same key already exists, the submission is silently skipped.	None
max_attempts	int	Maximum number of processing attempts before the event moves to dead-letter.	5

ClaimedEvent dataclass

ClaimedEvent(event_id: str, machine_name: str, entity_id: str, trigger: str, context: dict[str, Any], attempt: int, max_attempts: int, fires_at: datetime = (lambda: now(UTC))(), claimed_at: datetime = (lambda: now(UTC))())

An event that has been atomically claimed by a worker.

The :class:EventProcessor receives this after a successful claim from the event source. event_id is used to complete or fail the event after processing.

EventStatus

Bases: str, Enum

Lifecycle status of a worker event.

PENDING class-attribute instance-attribute

PENDING = 'pending'

Event is queued and waiting to be claimed.

CLAIMED class-attribute instance-attribute

CLAIMED = 'claimed'

Event has been claimed by a worker and is being processed.

COMPLETED class-attribute instance-attribute

COMPLETED = 'completed'

Event was processed successfully.

FAILED class-attribute instance-attribute

FAILED = 'failed'

Processing failed but may be retried.

DEAD_LETTER class-attribute instance-attribute

DEAD_LETTER = 'dead_letter'

Processing failed and max attempts have been exhausted.

Imports

from pystator.worker import Worker, WorkerConfig, submit_event, submit_event_sync

Install: pip install pystator[worker] (database URL required at runtime).