### Async Initialization Example

Source: https://github.com/opsani/statesman/blob/main/README.md

Demonstrates the asynchronous construction and initialization of a Statesman state machine. This example shows how to create a state machine instance using `StateMachine.create()` and initialize it with states defined from an enum.

```python
import statesman


class States(statesman.StateEnum):
    starting = 'Starting'
    running = 'Running'
    stopping = 'Stopping'
    stopped = 'Stopped'


async def _example() -> None:
    state_machine = await statesman.StateMachine.create(
        states=statesman.State.from_enum(States)
    )
    print(f"Initialized state machine: {repr(state_machine)}")
```

--------------------------------

### Statesman State Machine Usage Example

Source: https://github.com/opsani/statesman/blob/main/README.md

Demonstrates how to instantiate and interact with a Statesman state machine. It shows starting the machine, triggering events, asserting state changes, and initializing the machine in a specific state.

```python
async def _examples():
    # Let's play.
    state_machine = ProcessLifecycle()
    await state_machine.start("ls -al")
    assert state_machine.command == "ls -al"
    assert state_machine.pid == 31337
    assert state_machine.state == ProcessLifecycle.States.starting

    await state_machine.run()
    assert state_machine.logs == ['Process pid 31337 is now running (command="ls -al")']

    await state_machine.stop()
    assert state_machine.logs == [
        'Process pid 31337 is now running (command="ls -al")',
        'Shutting down pid 31337 (command="ls -al")',
        'Terminated pid 31337 ("ls -al")',
    ]

    # Or start in a specific state
    state_machine = ProcessLifecycle(state=ProcessLifecycle.States.running)

    # Transition to a specific state
    await state_machine.enter_state(ProcessLifecycle.States.stopping)

    # Trigger an event
    await state_machine.trigger_event("stop", key="value")

```

--------------------------------

### State and Event Actions with Statesman

Source: https://github.com/opsani/statesman/blob/main/README.md

Illustrates attaching actions to states and events using decorators like `statesman.enter_state`, `statesman.after_event`, and the `guard` parameter in `statesman.event`. This example shows entry actions, guard conditions (both function references and lambdas), and after-event actions.

```python
import random
import statesman


class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = 'Starting...'
        running = 'Running...'
        stopping = 'Stopping...'
        stopped = statesman.InitialState('Terminated.')

    @statesman.event(States.stopped, States.starting)
    async def start(self) -> None:
        ... # Placeholder for event logic

    @statesman.enter_state(States.starting)
    async def _announce_start(self) -> None:
        print("enter:starting")

    def _can_run(self) -> bool:
        return False

    @statesman.event(States.starting, States.running, guard=_can_run)
    async def run(self) -> None:
        ... # Placeholder for event logic

    @statesman.event(States.running, States.stopping)
    async def stop(self) -> None:
        ... # Placeholder for event logic

    @statesman.after_event(stop)
    async def _announce_stop(self) -> None:
        print("after:stop")

    @statesman.event(
        States.stopping,
        States.stopped,
        guard=lambda: random.choice([True, False])
    )
    async def terminate(self) -> None:
        ... # Placeholder for event logic
```

--------------------------------

### Python State Machine with Process Lifecycle

Source: https://github.com/opsani/statesman/blob/main/README.md

Defines a state machine for managing a process lifecycle, including states like starting, running, stopping, and stopped. It demonstrates event handling for starting, running, stopping, and terminating the process, along with state entry and after-event actions.

```python
from typing import Optional, List
import statesman


class ProcessLifecycle(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = "Starting..."
        running = "Running..."
        stopping = "Stopping..."
        stopped = "Terminated."

    # Track state about the process we are running
    command: Optional[str] = None
    pid: Optional[int] = None
    logs: List[str] = []

    # initial state entry point
    @statesman.event(None, States.starting)
    async def start(self, command: str) -> None:
        """Start a process."""
        self.command = command
        self.pid = 31337
        self.logs.clear()  # Flush logs between runs

    @statesman.event(source=States.starting, target=States.running)
    async def run(self, transition: statesman.Transition) -> None:
        """Mark the process as running."""
        self.logs.append(f"Process pid {self.pid} is now running (command=\"{self.command}\")")

    @statesman.event(source=States.running, target=States.stopping)
    async def stop(self) -> None:
        """Stop a running process."""
        self.logs.append(f"Shutting down pid {self.pid} (command=\"{self.command}\")")

    @statesman.event(source=States.stopping, target=States.stopped)
    async def terminate(self) -> None:
        """Terminate a running process."""
        self.logs.append(f"Terminated pid {self.pid} (\"{self.command}\")")
        self.command = None
        self.pid = None

    @statesman.enter_state(States.stopping)
    async def _print_status(self) -> None:
        print("Entering stopped status!")

    @statesman.after_event("run")
    async def _after_run(self) -> None:
        print("running...")

    async def after_transition(self, transition: statesman.Transition) -> None:
        if transition.event and transition.event.name == "stop":
            await self.terminate()

```

--------------------------------

### Defining Initial States in Statesman

Source: https://github.com/opsani/statesman/blob/main/README.md

Demonstrates how to define initial states in a Statesman state machine using `statesman.InitialState` or by setting the initial state during initialization, entering states directly, or via events.

```python
import statesman


# Describe the initial state with `statesman.InitialState`
class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = 'Starting...'
        running = 'Running...'
        stopping = 'Stopping...'
        stopped = statesman.InitialState('Terminated.')

    @statesman.event(None, States.starting)
    def start(self) -> None:
        ...


async def _example() -> None:
    # Set at initialization time
    state_machine = StateMachine(state=StateMachine.States.stopping)

    # Enter a state directly
    state_machine = StateMachine()
    await state_machine.enter_state(StateMachine.States.running)

    # Via an event
    state_machine = StateMachine()
    await state_machine.start()
```

--------------------------------

### Sequencing State Machine Transitions in Python

Source: https://github.com/opsani/statesman/blob/main/README.md

Demonstrates how to use `statesman.SequencingMixin` to define and execute a linear sequence of state machine transitions. It includes defining states, events, and a sequence of asynchronous calls that are managed by the `sequence` and `next_transition` methods.

```python
from typing import List
import statesman


class StateMachine(statesman.SequencingMixin, statesman.StateMachine):
    class States(statesman.StateEnum):
        ready = statesman.InitialState("Ready")
        analyzing = "Analyzing"

        awaiting_description = "Awaiting Description"
        awaiting_measurement = "Awaiting Measurement"
        awaiting_adjustment = "Awaiting Adjustment"

        done = "Done"
        failed = "Failed"

    @statesman.event([States.ready, States.analyzing], States.awaiting_description)
    async def request_description(self) -> None:
        """Request a Description of application state from the servo."""
        ...

    @statesman.event([States.ready, States.analyzing], States.awaiting_measurement)
    async def request_measurement(self, metrics: List[str]) -> None:
        """Request a Measurement from the servo."""
        ...

    @statesman.event([States.ready, States.analyzing], States.awaiting_adjustment)
    async def recommend_adjustments(self, adjustments: List[str]) -> None:
        """Recommend Adjustments to the Servo."""
        ...


async def _example() -> None:
    state_machine = StateMachine()
    state_machine.sequence(
        state_machine.request_description(),
        state_machine.request_measurement(metrics=[...]),
        state_machine.recommend_adjustments([...]),
        state_machine.request_measurement(metrics=[...]),
        state_machine.recommend_adjustments([...]),
    )

    while True:
        transition = state_machine.next_transition()
        print(f"Executed transition: {repr(transition)}")
        if not transition:
            break

```

--------------------------------

### Entering States Directly

Source: https://github.com/opsani/statesman/blob/main/README.md

Demonstrates how to directly enter states in a StateMachine using the `enter_state` method. States can be referenced by name or by State object instance. This method triggers a transition between states.

```python
import statesman


class StateMachine(statesman.HistoryMixin, statesman.StateMachine):
    class States(statesman.StateEnum):
        first = "1"
        second = "2"
        third = "3"


async def _example() -> None:
    state_machine = StateMachine()
    await state_machine.enter_state(StateMachine.States.first)
    await state_machine.enter_state(StateMachine.States.second)
    await state_machine.enter_state(StateMachine.States.third)
```

--------------------------------

### Guard Configuration

Source: https://github.com/opsani/statesman/blob/main/README.md

Demonstrates how to configure guard behaviors within a StateMachine. Guards are executed sequentially and can be configured to silence, warn, or raise exceptions on failure using the `guard_with` setting.

```python
import statesman


class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = 'Starting...'
        running = 'Running...'
        stopping = 'Stopping...'
        stopped = statesman.InitialState('Terminated.')

    class Config:
        guard_with = statesman.Guard.exception
```

--------------------------------

### Defining and Triggering Events with Statesman

Source: https://github.com/opsani/statesman/blob/main/README.md

Demonstrates defining events using the `statesman.event` decorator with various source and target states, including sentinel values like `__any__` and `__active__`. It also shows how to trigger events programmatically.

```python
import statesman


class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        waiting = 'Waiting'
        running = 'Running'
        stopped = 'Stopped'
        aborted = 'Aborted'

    @statesman.event(None, States.waiting)
    async def start(self) -> None:
        ... # Placeholder for event logic

    @statesman.event(States.waiting, States.running)
    async def run(self) -> None:
        ... # Placeholder for event logic

    @statesman.event(States.running, States.stopped)
    async def stop(self) -> None:
        ... # Placeholder for event logic

    @statesman.event(States.__any__, States.aborted)
    async def abort(self) -> None:
        ... # Placeholder for event logic

    @statesman.event(
        States.__any__,
        States.__active__,
        type=statesman.Transition.Types.self
    )
    async def check(self) -> None:
        print("Exiting and reentering active state!")


async def _example() -> None:
    state_machine = await StateMachine.create()
    await state_machine.trigger_event("start")
    await state_machine.run()
    await state_machine.trigger_event(state_machine.stop)
```

--------------------------------

### Passing Data to Transitions

Source: https://github.com/opsani/statesman/blob/main/README.md

Illustrates how to pass arbitrary data to state machine transitions using positional and keyword arguments. Statesman uses introspection and type-hinting to deliver these arguments to callbacks and actions.

```python
import statesman


class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = 'Starting...'
        running = 'Running...'
        stopping = 'Stopping...'
        stopped = statesman.InitialState('Terminated.')

    @statesman.event(States.stopped, States.starting)
    async def start(self, process_name: str) -> None:
        ... # Placeholder for actual implementation

    @statesman.event(States.starting, States.running)
    async def run(self, uid: int, gid: int) -> None:
        ... # Placeholder for actual implementation

    @statesman.event(States.running, States.stopping)
    async def stop(self, *args, all: bool = False, **kwargs) -> None:
        ... # Placeholder for actual implementation


async def _example() -> None:
    state_machine = await StateMachine.create()
    await state_machine.start("servox")
    await state_machine.run(0, 31337)
    await state_machine.stop("one", "two", all=True, this="That")
```

--------------------------------

### State Machine Inheritable Callbacks

Source: https://github.com/opsani/statesman/blob/main/README.md

Shows how to implement inheritable callbacks for state machine transition lifecycle events. These callbacks, such as `guard_transition`, `before_transition`, `on_transition`, and `after_transition`, allow for custom logic at different stages of a transition.

```python
import statesman


class InheritableStateMachine(statesman.StateMachine):
    async def guard_transition(self, transition: statesman.Transition, *args, **kwargs) -> bool:
        return True

    async def before_transition(self, transition: statesman.Transition, *args, **kwargs) -> None:
        ... # Placeholder for actual implementation

    async def on_transition(self, transition: statesman.Transition, *args, **kwargs) -> None:
        ... # Placeholder for actual implementation

    async def after_transition(self, transition: statesman.Transition, *args, **kwargs) -> None:
        ... # Placeholder for actual implementation
```

--------------------------------

### Track State Machine History with HistoryMixin

Source: https://github.com/opsani/statesman/blob/main/README.md

Shows how to use `statesman.HistoryMixin` to automatically track the sequence of states entered by a state machine. The `history` attribute stores a list of states, allowing for post-execution analysis of the state transitions.

```python
import statesman


class StateMachine(statesman.HistoryMixin, statesman.StateMachine):
    class States(statesman.StateEnum):
        ready = statesman.InitialState("Ready")
        analyzing = "Analyzing"

        awaiting_description = "Awaiting Description"
        awaiting_measurement = "Awaiting Measurement"
        awaiting_adjustment = "Awaiting Adjustment"

        done = "Done"
        failed = "Failed"


async def _example() -> None:
    state_machine = StateMachine()
    await state_machine.enter_state(StateMachine.States.analyzing)
    await state_machine.enter_state(StateMachine.States.awaiting_measurement)
    await state_machine.enter_state(StateMachine.States.done)
    await state_machine.enter_state(StateMachine.States.failed)

    print(f"The history is: {state_machine.history}")
```

--------------------------------

### Transition Return Values and Types

Source: https://github.com/opsani/statesman/blob/main/README.md

Illustrates how to configure and retrieve different return types from state transitions. This includes boolean success indicators, arbitrary object outputs, tuples, lists of results, or the transition object itself. The return type can be specified globally or per-transition.

```python
import statesman


class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = 'Starting'
        running = 'Running'
        stopping = 'Stopping'
        stopped = 'Stopped'

    @statesman.event(None, States.starting, return_type=bool)
    async def start(self) -> int:
        return 31337


async def _example() -> None:
    state_machine = await StateMachine.create()
    bool_result = await state_machine.trigger_event("start")
    int_result = await state_machine.start(return_type=object)
    transition = await state_machine.enter_state(
        StateMachine.States.stopped,
        return_type=statesman.Transition
    )
    print(
        f"Return Values: state_machine={state_machine}
"
        f"bool_result={bool_result}
"
        f"int_result={int_result}
"
        f"transition={transition}"
    )
```

--------------------------------

### Restrict State Entry with statesman.Entry.forbid

Source: https://github.com/opsani/statesman/blob/main/README.md

Demonstrates how to configure the state machine to forbid the use of the `enter_state` method by setting `state_entry` to `statesman.Entry.forbid`. This prevents the state machine from entering states directly after initialization, enforcing transition-based state changes.

```python
import statesman


class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = 'Starting...'
        running = 'Running...'
        stopping = 'Stopping...'
        stopped = statesman.InitialState('Terminated.')

    class Config:
        state_entry = statesman.Entry.forbid
```

--------------------------------

### Introspecting State in Statesman

Source: https://github.com/opsani/statesman/blob/main/README.md

Shows how to introspect the current state of a Statesman state machine instance by comparing the `state` attribute with `statesman.State` objects or string values.

```python
import statesman


class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = 'Starting...'
        running = 'Running...'
        stopping = 'Stopping...'
        stopped = 'Terminated.'


async def _example() -> None:
    state_machine = StateMachine(state=StateMachine.States.stopping)
    state_machine.state == StateMachine.States.stopping  # => True
    state_machine.state == "stopping"  # => True
    state_machine.state == StateMachine.States.running  # => False
    state_machine.state == "stopped"  # => False
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.