Machine definitions (parser, store, builder)¶

Use this pipeline when you want a canonical MachineDefinition (parse once, validate, store in DB) and build a StateMachine only when needed. For a single YAML file in a script, StateMachine.from_yaml() is usually enough.

See the Package structure guide for when to use each layer.

Types and parsing¶

MachineDefinition `dataclass` ¶

MachineDefinition(name: str, version: str, config: dict[str, Any], states: list[dict[str, Any]] = list(), transitions: list[dict[str, Any]] = list(), description: str | None = None, strict_mode: bool = True)

A fully parsed and validated FSM machine definition.

Analogous to PyCharter's ContractMetadata — the single canonical representation that flows between parser, store, and builder.

Attributes:

Name	Type	Description
`name`	`str`	Machine name (lowercase, underscores only).
`version`	`str`	Semantic version string.
`config`	`dict[str, Any]`	Full validated config dict (round-trip safe for persistence).
`states`	`list[dict[str, Any]]`	List of state definition dicts from config.
`transitions`	`list[dict[str, Any]]`	List of transition definition dicts from config.
`description`	`str \| None`	Human-readable description.
`strict_mode`	`bool`	Whether the machine enforces strict mode.

to_dict ¶

to_dict() -> dict[str, Any]

Return the full config dict (same as what gets persisted).

Source code in src/pystator/machine_parser/types.py

def to_dict(self) -> dict[str, Any]:
    """Return the full config dict (same as what gets persisted)."""
    return dict(self.config)

to_state_machine ¶

to_state_machine() -> Any

Build a StateMachine from this definition (no guards/actions bound).

Returns a pystator.machine.StateMachine. Import is deferred to avoid circular dependencies.

Source code in src/pystator/machine_parser/types.py

def to_state_machine(self) -> Any:
    """Build a StateMachine from this definition (no guards/actions bound).

    Returns a pystator.machine.StateMachine. Import is deferred to
    avoid circular dependencies.
    """
    from pystator.machine import StateMachine

    return StateMachine.from_dict(self.config)

parse_machine ¶

parse_machine(config: dict[str, Any], *, validate: bool = True) -> MachineDefinition

Parse an FSM config dict into a MachineDefinition.

Steps: 1. Resolve submachine references (inline expansion). 2. Optionally validate via ConfigValidator (Pydantic). 3. Extract canonical meta fields (name, version, description, strict_mode). 4. Return a frozen MachineDefinition.

Parameters:

Name	Type	Description	Default
`config`	`dict[str, Any]`	Raw FSM config dict (with meta, states, transitions).	required
`validate`	`bool`	If True, run Pydantic schema + semantic validation.	`True`

Returns:

Type	Description
`MachineDefinition`	A validated MachineDefinition.

Raises:

Type	Description
`ConfigurationError`	If validation fails.
`ValueError`	If required meta fields are missing or invalid.

Source code in src/pystator/machine_parser/parser.py

def parse_machine(
    config: dict[str, Any],
    *,
    validate: bool = True,
) -> MachineDefinition:
    """Parse an FSM config dict into a MachineDefinition.

    Steps:
    1. Resolve submachine references (inline expansion).
    2. Optionally validate via ConfigValidator (Pydantic).
    3. Extract canonical meta fields (name, version, description, strict_mode).
    4. Return a frozen MachineDefinition.

    Args:
        config: Raw FSM config dict (with meta, states, transitions).
        validate: If True, run Pydantic schema + semantic validation.

    Returns:
        A validated MachineDefinition.

    Raises:
        ConfigurationError: If validation fails.
        ValueError: If required meta fields are missing or invalid.
    """
    config = dict(config)  # shallow copy to avoid mutating caller's dict

    # Resolve submachines
    config = _resolve_submachines(config)

    # Validate
    if validate:
        _validate_config(config)

    # Extract meta
    meta = config.get("meta", {})
    name = meta.get("machine_name") or meta.get("name", "")
    if not name:
        raise ValueError("Machine config must have meta.machine_name (or meta.name)")

    # Normalize name
    from pystator.shared.name_validator import validate_name, validate_namespace

    name = validate_name(name, field_name="machine_name")

    namespace_raw = meta.get("namespace")
    if namespace_raw is not None and str(namespace_raw).strip():
        namespace = validate_namespace(str(namespace_raw), field_name="namespace")
        meta["namespace"] = namespace

    version = str(meta.get("version", "1.0.0"))
    description = meta.get("description")
    strict_raw = meta.get("strict_mode")
    if strict_raw is None:
        strict_mode = True
    elif isinstance(strict_raw, bool):
        strict_mode = strict_raw
    else:
        strict_mode = str(strict_raw).lower() in ("true", "1", "yes")

    states = config.get("states", [])
    transitions = config.get("transitions", [])

    return MachineDefinition(
        name=name,
        version=version,
        config=config,
        states=list(states),
        transitions=list(transitions),
        description=description,
        strict_mode=strict_mode,
    )

parse_machine_file ¶

parse_machine_file(path: str | Path, *, validate: bool = True, variables: dict[str, str] | None = None) -> MachineDefinition

Parse an FSM config from a YAML or JSON file.

Handles YAML/JSON detection, environment variable substitution, submachine resolution, and validation.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to .yaml, .yml, or .json file.	required
`validate`	`bool`	If True, run Pydantic schema + semantic validation.	`True`
`variables`	`dict[str, str] \| None`	Extra variables for ${VAR} substitution (in addition to env).	`None`

Returns:

Type	Description
`MachineDefinition`	A validated MachineDefinition.

Raises:

Type	Description
`FileNotFoundError`	If the file does not exist.
`ConfigurationError`	If parsing or validation fails.

Source code in src/pystator/machine_parser/parser.py

def parse_machine_file(
    path: str | Path,
    *,
    validate: bool = True,
    variables: dict[str, str] | None = None,
) -> MachineDefinition:
    """Parse an FSM config from a YAML or JSON file.

    Handles YAML/JSON detection, environment variable substitution,
    submachine resolution, and validation.

    Args:
        path: Path to .yaml, .yml, or .json file.
        validate: If True, run Pydantic schema + semantic validation.
        variables: Extra variables for ${VAR} substitution (in addition to env).

    Returns:
        A validated MachineDefinition.

    Raises:
        FileNotFoundError: If the file does not exist.
        ConfigurationError: If parsing or validation fails.
    """
    path = Path(path)
    if not path.exists():
        raise FileNotFoundError(f"Machine config file not found: {path}")

    try:
        content = path.read_text(encoding="utf-8")
    except OSError as e:
        raise ConfigurationError(
            f"Failed to read machine config: {e}", path=str(path)
        ) from e

    # Environment variable substitution
    content = _substitute_variables(content, str(path), variables or {})

    # Parse YAML or JSON
    config = _parse_file_content(content, path)

    # Resolve submachines relative to file directory
    config = _resolve_submachines(config, base_path=path.parent)

    # Validate
    if validate:
        _validate_config(config)

    # Extract meta and build definition
    meta = config.get("meta", {})
    name = meta.get("machine_name") or meta.get("name", "")
    if not name:
        raise ValueError(
            f"Machine config at {path} must have meta.machine_name (or meta.name)"
        )

    from pystator.shared.name_validator import validate_name, validate_namespace

    name = validate_name(name, field_name="machine_name")

    namespace_raw = meta.get("namespace")
    if namespace_raw is not None and str(namespace_raw).strip():
        namespace = validate_namespace(str(namespace_raw), field_name="namespace")
        meta["namespace"] = namespace

    version = str(meta.get("version", "1.0.0"))
    description = meta.get("description")
    strict_raw = meta.get("strict_mode")
    if strict_raw is None:
        strict_mode = True
    elif isinstance(strict_raw, bool):
        strict_mode = strict_raw
    else:
        strict_mode = str(strict_raw).lower() in ("true", "1", "yes")

    states = config.get("states", [])
    transitions = config.get("transitions", [])

    return MachineDefinition(
        name=name,
        version=version,
        config=config,
        states=list(states),
        transitions=list(transitions),
        description=description,
        strict_mode=strict_mode,
    )

Building runtime machines¶

build_machine ¶

build_machine(definition: MachineDefinition, *, guards: GuardRegistry | None = None, actions: ActionRegistry | None = None) -> StateMachine

Build a runtime StateMachine from a MachineDefinition.

Optionally binds guard and action registries so the machine is ready for event processing.

Parameters:

Name	Type	Description	Default
`definition`	`MachineDefinition`	A parsed MachineDefinition.	required
`guards`	`GuardRegistry \| None`	Optional guard registry to bind.	`None`
`actions`	`ActionRegistry \| None`	Optional action registry to bind.	`None`

Returns:

Type	Description
`StateMachine`	A fully configured StateMachine instance.

Source code in src/pystator/machine_builder/builder.py

def build_machine(
    definition: MachineDefinition,
    *,
    guards: GuardRegistry | None = None,
    actions: ActionRegistry | None = None,
) -> StateMachine:
    """Build a runtime StateMachine from a MachineDefinition.

    Optionally binds guard and action registries so the machine is
    ready for event processing.

    Args:
        definition: A parsed MachineDefinition.
        guards: Optional guard registry to bind.
        actions: Optional action registry to bind.

    Returns:
        A fully configured StateMachine instance.
    """
    from pystator.machine import StateMachine

    machine = StateMachine.from_dict(definition.config)
    if guards is not None:
        machine.bind_guards(guards)
    if actions is not None:
        machine.bind_actions(actions)
    return machine

build_machine_from_store ¶

build_machine_from_store(store: MachineStoreClient, name: str, version: str | None = None, *, guards: GuardRegistry | None = None, actions: ActionRegistry | None = None) -> StateMachine

Load a machine definition from a store and build a StateMachine.

Parameters:

Name	Type	Description	Default
`store`	`MachineStoreClient`	A connected MachineStoreClient.	required
`name`	`str`	Machine name to look up.	required
`version`	`str \| None`	Specific version, or None for latest.	`None`
`guards`	`GuardRegistry \| None`	Optional guard registry to bind.	`None`
`actions`	`ActionRegistry \| None`	Optional action registry to bind.	`None`

Returns:

Type	Description
`StateMachine`	A fully configured StateMachine instance.

Raises:

Type	Description
`MachineNotFoundError`	If the machine is not in the store.

Source code in src/pystator/machine_builder/builder.py

def build_machine_from_store(
    store: MachineStoreClient,
    name: str,
    version: str | None = None,
    *,
    guards: GuardRegistry | None = None,
    actions: ActionRegistry | None = None,
) -> StateMachine:
    """Load a machine definition from a store and build a StateMachine.

    Args:
        store: A connected MachineStoreClient.
        name: Machine name to look up.
        version: Specific version, or None for latest.
        guards: Optional guard registry to bind.
        actions: Optional action registry to bind.

    Returns:
        A fully configured StateMachine instance.

    Raises:
        MachineNotFoundError: If the machine is not in the store.
    """
    definition = store.get(name, version)
    if definition is None:
        raise MachineNotFoundError(name, version)
    return build_machine(definition, guards=guards, actions=actions)

MachineNotFoundError ¶

MachineNotFoundError(name: str, version: str | None = None)

Bases: Exception

Raised when a machine definition cannot be found in the store.

Source code in src/pystator/machine_builder/builder.py

def __init__(self, name: str, version: str | None = None) -> None:
    self.name = name
    self.version = version
    detail = f"Machine '{name}'"
    if version:
        detail += f" version '{version}'"
    detail += " not found in store"
    super().__init__(detail)

Definition stores¶

MachineStoreClient ¶

MachineStoreClient(connection_string: str | None = None)

Bases: ABC

Abstract base for machine definition storage.

All operations are keyed by (machine_name, version). Implementations extend this for specific backends.

Source code in src/pystator/machine_store/client.py

def __init__(self, connection_string: str | None = None) -> None:
    self.connection_string = connection_string
    self._connection: Any = None

connect `abstractmethod` ¶

connect() -> None

Establish backend connection.

Source code in src/pystator/machine_store/client.py

@abstractmethod
def connect(self) -> None:
    """Establish backend connection."""

disconnect ¶

disconnect() -> None

Close backend connection.

Source code in src/pystator/machine_store/client.py

def disconnect(self) -> None:
    """Close backend connection."""
    self._connection = None

save `abstractmethod` ¶

save(definition: MachineDefinition) -> str

Persist a machine definition (upsert by name + version).

Parameters:

Name	Type	Description	Default
`definition`	`MachineDefinition`	The parsed machine definition to store.	required

Returns:

Type	Description
`str`	The machine identifier (implementation-defined, e.g. UUID string).

Source code in src/pystator/machine_store/client.py

@abstractmethod
def save(self, definition: MachineDefinition) -> str:
    """Persist a machine definition (upsert by name + version).

    Args:
        definition: The parsed machine definition to store.

    Returns:
        The machine identifier (implementation-defined, e.g. UUID string).
    """

get `abstractmethod` ¶

get(name: str, version: str | None = None) -> MachineDefinition | None

Retrieve a machine definition by name and optional version.

When version is None, returns the latest version.

Parameters:

Name	Type	Description	Default
`name`	`str`	Machine name.	required
`version`	`str \| None`	Specific version, or None for latest.	`None`

Returns:

Type	Description
`MachineDefinition \| None`	MachineDefinition if found, None otherwise.

Source code in src/pystator/machine_store/client.py

@abstractmethod
def get(self, name: str, version: str | None = None) -> MachineDefinition | None:
    """Retrieve a machine definition by name and optional version.

    When version is None, returns the latest version.

    Args:
        name: Machine name.
        version: Specific version, or None for latest.

    Returns:
        MachineDefinition if found, None otherwise.
    """

list `abstractmethod` ¶

list(name: str | None = None) -> list[MachineDefinition]

List machine definitions.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	If provided, filter to versions of this machine only.	`None`

Returns:

Type	Description
`list[MachineDefinition]`	List of MachineDefinition objects.

Source code in src/pystator/machine_store/client.py

@abstractmethod
def list(self, name: str | None = None) -> list[MachineDefinition]:
    """List machine definitions.

    Args:
        name: If provided, filter to versions of this machine only.

    Returns:
        List of MachineDefinition objects.
    """

delete `abstractmethod` ¶

delete(name: str, version: str) -> bool

Delete a specific machine version.

Parameters:

Name	Type	Description	Default
`name`	`str`	Machine name.	required
`version`	`str`	Version to delete.	required

Returns:

Type	Description
`bool`	True if deleted, False if not found.

Source code in src/pystator/machine_store/client.py

@abstractmethod
def delete(self, name: str, version: str) -> bool:
    """Delete a specific machine version.

    Args:
        name: Machine name.
        version: Version to delete.

    Returns:
        True if deleted, False if not found.
    """

InMemoryMachineStore ¶

InMemoryMachineStore()

Bases: MachineStoreClient

Dict-backed machine store for tests, notebooks, and embedded use.

Does not require a database. Machines are stored in a dict keyed by (name, version).

Source code in src/pystator/machine_store/in_memory.py

def __init__(self) -> None:
    super().__init__(connection_string=None)
    self._machines: dict[tuple[str, str], MachineDefinition] = {}

SQLAlchemyMachineStore ¶

SQLAlchemyMachineStore(connection_string: str)

Bases: MachineStoreClient

Machine store backed by SQLAlchemy (PostgreSQL or SQLite).

Uses the existing pystator.machines table (MachineModel).

Source code in src/pystator/machine_store/_sqlalchemy.py

def __init__(self, connection_string: str) -> None:
    super().__init__(connection_string=connection_string)
    self._engine: Any = None
    self._SessionFactory: Any = None

SQLAlchemyMachineStore requires SQLAlchemy (included with pystator[api] or pystator[worker]). It is also exported as from pystator.machine_store import SQLAlchemyMachineStore when dependencies are present.

Imports¶

from pystator.machine_parser import MachineDefinition, parse_machine, parse_machine_file
from pystator.machine_builder import build_machine, build_machine_from_store, MachineNotFoundError
from pystator.machine_store import MachineStoreClient, InMemoryMachineStore
# Optional, when SQLAlchemy is installed:
from pystator.machine_store import SQLAlchemyMachineStore

The same symbols are available from the top-level pystator package (from pystator import ...).

Machine definitions (parser, store, builder)¶

Types and parsing¶

MachineDefinition dataclass ¶

to_dict ¶

to_state_machine ¶

parse_machine ¶

parse_machine_file ¶

Building runtime machines¶

build_machine ¶

build_machine_from_store ¶

MachineNotFoundError ¶

Definition stores¶

MachineStoreClient ¶

connect abstractmethod ¶

disconnect ¶

save abstractmethod ¶

get abstractmethod ¶

list abstractmethod ¶

delete abstractmethod ¶

InMemoryMachineStore ¶

SQLAlchemyMachineStore ¶

Imports¶

MachineDefinition `dataclass` ¶

connect `abstractmethod` ¶

save `abstractmethod` ¶

get `abstractmethod` ¶

list `abstractmethod` ¶

delete `abstractmethod` ¶