Domain Module

The pycharter.domain module provides utilities for working with lifecycle bindings declared in domain-entity contracts. It is engine-agnostic: it reads the convention from the contract metadata and returns structured data that callers can pass to any FSM engine (e.g. PyStator).

See Domain Models and Lifecycle for the full guide.

Functions

get_lifecycle_binding

get_lifecycle_binding(
    metadata: dict[str, Any],
) -> dict[str, Any]

Read lifecycle binding from metadata (root lifecycle or governance_rules.lifecycle).

Returns the full lifecycle dict if it exists and contains at least state_machine_name; otherwise returns None. Keys may include state_machine_name, machine_version, state_field, entity_id_field (for precise integration with FSM engines like PyStator).

Parameters:

Name	Type	Description	Default
metadata	dict[str, Any]	Metadata dict (e.g. from a contract or metadata record), with optional "lifecycle" at root or under "governance_rules".	required

Returns:

Type	Description
dict[str, Any]	Lifecycle binding dict (e.g. state_machine_name, machine_version, state_field,
dict[str, Any]	entity_id_field), or None if not present or invalid.

Example

meta = {"governance_rules": {"lifecycle": {"state_machine_name": "order_lifecycle"}}} get_lifecycle_binding(meta)

Source code in src/pycharter/domain/__init__.py

def get_lifecycle_binding(metadata: dict[str, Any]) -> dict[str, Any]:
    """Read lifecycle binding from metadata (root lifecycle or governance_rules.lifecycle).

    Returns the full lifecycle dict if it exists and contains at least state_machine_name;
    otherwise returns None. Keys may include state_machine_name, machine_version,
    state_field, entity_id_field (for precise integration with FSM engines like PyStator).

    Args:
        metadata: Metadata dict (e.g. from a contract or metadata record),
            with optional "lifecycle" at root or under "governance_rules".

    Returns:
        Lifecycle binding dict (e.g. state_machine_name, machine_version, state_field,
        entity_id_field), or None if not present or invalid.

    Example:
        >>> meta = {"governance_rules": {"lifecycle": {"state_machine_name": "order_lifecycle"}}}
        >>> get_lifecycle_binding(meta)
        {'state_machine_name': 'order_lifecycle'}
    """
    return _get_lifecycle_dict(metadata)

get_domain_entity_info

get_domain_entity_info(
    contract_dict: dict[str, Any],
) -> dict[str, Any]

Extract domain entity info (schema + lifecycle) from a full contract dict.

Given a contract structure with optional "metadata" and "schema" (or raw schema at top level), returns a dict with state_machine_name, state_field, and optionally machine_version and entity_id_field for entities that have a lifecycle binding. Use this when calling an FSM engine (e.g. PyStator): entity_id = record[info["entity_id_field"]], machine_version for API pinning.

Parameters:

Name	Type	Description	Default
contract_dict	dict[str, Any]	Full contract dict (e.g. from store or build_contract), possibly with keys like metadata, schema, coercion_rules, validation_rules.	required

Returns:

Type	Description
dict[str, Any]	Dict with state_machine_name, state_field (default "status"), and when
dict[str, Any]	present in metadata: machine_version, entity_id_field. None if no binding.

Example

c = {"metadata": {"governance_rules": {"lifecycle": {"state_machine_name": "order_lifecycle"}}}} get_domain_entity_info©

Source code in src/pycharter/domain/__init__.py

def get_domain_entity_info(contract_dict: dict[str, Any]) -> dict[str, Any]:
    """Extract domain entity info (schema + lifecycle) from a full contract dict.

    Given a contract structure with optional "metadata" and "schema" (or raw schema
    at top level), returns a dict with state_machine_name, state_field, and
    optionally machine_version and entity_id_field for entities that have a
    lifecycle binding. Use this when calling an FSM engine (e.g. PyStator):
    entity_id = record[info["entity_id_field"]], machine_version for API pinning.

    Args:
        contract_dict: Full contract dict (e.g. from store or build_contract),
            possibly with keys like metadata, schema, coercion_rules, validation_rules.

    Returns:
        Dict with state_machine_name, state_field (default "status"), and when
        present in metadata: machine_version, entity_id_field. None if no binding.

    Example:
        >>> c = {"metadata": {"governance_rules": {"lifecycle": {"state_machine_name": "order_lifecycle"}}}}
        >>> get_domain_entity_info(c)
        {'state_machine_name': 'order_lifecycle', 'state_field': 'status'}
    """
    if not contract_dict or not isinstance(contract_dict, dict):
        return None
    metadata = contract_dict.get("metadata")
    if not metadata and "governance_rules" in contract_dict:
        metadata = contract_dict
    binding = get_lifecycle_binding(metadata) if metadata else None
    if not binding:
        return None
    out: dict[str, Any] = {
        "state_machine_name": binding["state_machine_name"],
        "state_field": binding.get("state_field") or DEFAULT_STATE_FIELD,
    }
    if binding.get("machine_version") is not None:
        out["machine_version"] = binding["machine_version"]
    if binding.get("entity_id_field"):
        out["entity_id_field"] = binding["entity_id_field"]
    return out

check_state_alignment

check_state_alignment(
    contract_dict: dict[str, Any],
    fsm_state_names: list[str],
    state_field: str = "status",
) -> dict[str, Any]

Check if a contract's status enum aligns with FSM states.

Compares the enum values declared for state_field in the contract schema against the state names defined in an FSM. Returns a dict summarising alignment.

Parameters:

Name	Type	Description	Default
contract_dict	dict[str, Any]	Full contract dict (must contain a schema key with a JSON-Schema-style properties mapping).	required
fsm_state_names	list[str]	List of state names from the FSM definition.	required
state_field	str	Name of the property to inspect for enum values (default "status").	'status'

Returns:

Type	Description
dict[str, Any]	Dict with keys: - aligned (bool): True when both sets match exactly. - contract_states (set[str]): Enum values from the contract. - fsm_states (set[str]): State names from the FSM. - missing_from_contract (set[str]): States in FSM but not in the contract enum. - missing_from_fsm (set[str]): Enum values in the contract but not in the FSM.

Example

contract = {"schema": {"properties": {"status": {"enum": ["NEW", "ACTIVE"]}}}} check_state_alignment(contract, ["NEW", "ACTIVE", "CLOSED"]) {'aligned': False, 'contract_states': {'NEW', 'ACTIVE'}, 'fsm_states': {'NEW', 'ACTIVE', 'CLOSED'}, 'missing_from_contract': {'CLOSED'}, 'missing_from_fsm': set()}

Source code in src/pycharter/domain/__init__.py

def check_state_alignment(
    contract_dict: dict[str, Any],
    fsm_state_names: list[str],
    state_field: str = "status",
) -> dict[str, Any]:
    """Check if a contract's status enum aligns with FSM states.

    Compares the ``enum`` values declared for *state_field* in the contract
    schema against the state names defined in an FSM.  Returns a dict
    summarising alignment.

    Args:
        contract_dict: Full contract dict (must contain a ``schema`` key
            with a JSON-Schema-style ``properties`` mapping).
        fsm_state_names: List of state names from the FSM definition.
        state_field: Name of the property to inspect for enum values
            (default ``"status"``).

    Returns:
        Dict with keys:
            - ``aligned`` (bool): True when both sets match exactly.
            - ``contract_states`` (set[str]): Enum values from the contract.
            - ``fsm_states`` (set[str]): State names from the FSM.
            - ``missing_from_contract`` (set[str]): States in FSM but not
              in the contract enum.
            - ``missing_from_fsm`` (set[str]): Enum values in the contract
              but not in the FSM.

    Example:
        >>> contract = {"schema": {"properties": {"status": {"enum": ["NEW", "ACTIVE"]}}}}
        >>> check_state_alignment(contract, ["NEW", "ACTIVE", "CLOSED"])
        {'aligned': False, 'contract_states': {'NEW', 'ACTIVE'},
         'fsm_states': {'NEW', 'ACTIVE', 'CLOSED'},
         'missing_from_contract': {'CLOSED'}, 'missing_from_fsm': set()}
    """
    # Extract enum values from the contract schema
    contract_states: set[str] = set()
    schema = contract_dict.get("schema", {})
    if isinstance(schema, dict):
        props = schema.get("properties", {})
        field_def = props.get(state_field, {})
        enum_values = field_def.get("enum", [])
        contract_states = {str(v) for v in enum_values}

    fsm_states = set(fsm_state_names)

    return {
        "aligned": contract_states == fsm_states,
        "contract_states": contract_states,
        "fsm_states": fsm_states,
        "missing_from_contract": fsm_states - contract_states,
        "missing_from_fsm": contract_states - fsm_states,
    }

validate_lifecycle_binding

validate_lifecycle_binding(
    metadata: dict[str, Any],
) -> list[str]

Validate the structure of a lifecycle binding block.

Reads the lifecycle dict from metadata using :func:get_lifecycle_binding and checks that required fields are present and well-formed.

Parameters:

Name	Type	Description	Default
metadata	dict[str, Any]	Metadata dict (e.g. from a contract), with optional lifecycle at root or under governance_rules.	required

Returns:

Type	Description
list[str]	List of error message strings. An empty list means the binding
list[str]	is valid (or no lifecycle block is present at all).

Example

validate_lifecycle_binding({"lifecycle": {"state_machine_name": ""}}) ['state_machine_name must be a non-empty string']

Source code in src/pycharter/domain/__init__.py

def validate_lifecycle_binding(metadata: dict[str, Any]) -> list[str]:
    """Validate the structure of a lifecycle binding block.

    Reads the lifecycle dict from *metadata* using
    :func:`get_lifecycle_binding` and checks that required fields are
    present and well-formed.

    Args:
        metadata: Metadata dict (e.g. from a contract), with optional
            ``lifecycle`` at root or under ``governance_rules``.

    Returns:
        List of error message strings.  An empty list means the binding
        is valid (or no lifecycle block is present at all).

    Example:
        >>> validate_lifecycle_binding({"lifecycle": {"state_machine_name": ""}})
        ['state_machine_name must be a non-empty string']
    """
    binding = get_lifecycle_binding(metadata)
    if binding is None:
        # get_lifecycle_binding returns None for missing OR empty state_machine_name.
        # Check if a lifecycle block actually exists — if so, validate it.
        raw = metadata.get("lifecycle") or (
            metadata.get("governance_rules", {}).get("lifecycle")
        )
        if not raw or not isinstance(raw, dict):
            return []  # No lifecycle block at all = nothing to validate
        binding = raw

    errors: list[str] = []

    # state_machine_name is required and must be a non-empty string
    smn = binding.get("state_machine_name")
    if not isinstance(smn, str) or not smn.strip():
        errors.append("state_machine_name must be a non-empty string")

    # machine_version, if present, must be semver
    if "machine_version" in binding:
        v = binding["machine_version"]
        if not isinstance(v, str) or not re.match(r"^\d+\.\d+\.\d+$", v):
            errors.append("machine_version must be semver (e.g. '1.0.0')")

    # state_field, if present, must be a non-empty string
    if "state_field" in binding:
        sf = binding["state_field"]
        if not isinstance(sf, str) or not sf.strip():
            errors.append("state_field must be a non-empty string")

    # entity_id_field, if present, must be a non-empty string
    if "entity_id_field" in binding:
        eid = binding["entity_id_field"]
        if not isinstance(eid, str) or not eid.strip():
            errors.append("entity_id_field must be a non-empty string")

    return errors

Constants

Name	Default	Description
DEFAULT_STATE_FIELD	"status"	Field name used as the entity state when none is specified in the lifecycle block

Import

from pycharter.domain import (
    get_lifecycle_binding,
    get_domain_entity_info,
    check_state_alignment,
    validate_lifecycle_binding,
    DEFAULT_STATE_FIELD,
)

Lifecycle contract schema

# In contract metadata:
metadata:
  lifecycle:
    state_machine_name: order_lifecycle   # required
    machine_version: "1.0.0"              # optional — pin FSM version
    state_field: status                   # optional, default "status"
    entity_id_field: order_id             # optional — field carrying entity ID

# Alternatively under governance_rules:
metadata:
  governance_rules:
    lifecycle:
      state_machine_name: order_lifecycle

Usage examples

Reading a lifecycle binding

from pycharter.domain import get_lifecycle_binding

meta = {
    "governance_rules": {
        "lifecycle": {
            "state_machine_name": "order_lifecycle",
            "machine_version": "2.0.0",
            "state_field": "order_status",
            "entity_id_field": "order_id",
        }
    }
}

binding = get_lifecycle_binding(meta)
# {'state_machine_name': 'order_lifecycle', 'machine_version': '2.0.0', ...}

Getting domain entity info from a full contract

from pycharter import build_contract
from pycharter.domain import get_domain_entity_info

contract = build_contract("orders", "1.0.0", store)
info = get_domain_entity_info(contract)

if info:
    machine = info["state_machine_name"]
    state_field = info["state_field"]
    entity_id_field = info.get("entity_id_field", "id")
    print(f"FSM: {machine}, state field: {state_field}")

Checking alignment between contract and FSM states

from pycharter.domain import check_state_alignment

result = check_state_alignment(
    contract_dict=contract,
    fsm_state_names=["NEW", "PROCESSING", "SHIPPED", "DELIVERED", "CANCELLED"],
    state_field="order_status",
)

if not result["aligned"]:
    if result["missing_from_contract"]:
        print(f"Add to contract enum: {result['missing_from_contract']}")
    if result["missing_from_fsm"]:
        print(f"Add to FSM: {result['missing_from_fsm']}")

Validating a lifecycle binding block

from pycharter.domain import validate_lifecycle_binding

errors = validate_lifecycle_binding(meta)
if errors:
    for err in errors:
        print(f"  ✗ {err}")

See also

• Domain Models and Lifecycle guide
• Contract Parser