The Python API¶

This page lists every public name of the stateous package and of stateous.client. The text comes from the docstrings in the code.

Import each name from stateous, for example from stateous import Workflow. Import the client's errors from stateous.client.

Write a workflow¶

You write a workflow in flow.py, with Workflow, START and END. stateous compile reads it. See Write a workflow.

Workflow ¶

Workflow(
    name: str, *, context: type[BaseModel] | None = None
)

A process: its states, its events, and the transitions between them.

You write it in flow.py. stateous compile checks it and writes the graph file that a control plane serves. Each method refuses a mistake at once, with DefinitionError.

flow = Workflow("sdlc", context=Ticket)
coding = flow.state("coding", timeout="6h", retries=2)
testing = flow.state("testing", timeout="1h")
failed = flow.event("failed", payload=Tests)
flow.transition(testing, coding, on=failed, when="count('failed') < 3")

Make an empty workflow.

Parameters:

Name	Type	Description	Default
`name`	`str`	The workflow's name: `a-z`, `0-9`, `_` and `-`, with a letter first, at most 64 characters. It is also the name of the graph file.	required
`context`	`type[BaseModel] \| None`	The Pydantic model of each token's context. A token starts with it, and its routines read and write it as `ctx["key"]`. None: any JSON object.	`None`

Raises:

Type	Description
`DefinitionError`	The name is not valid, or `context` is not a Pydantic model class.

states `instance-attribute` ¶

states: dict[str, State] = {}

The declared states, by name, in declaration order.

events `instance-attribute` ¶

events: dict[str, Event] = {}

The declared events, by name. Stateous's own events are not among them.

transitions `instance-attribute` ¶

transitions: list[Transition] = []

The declared transitions, in declaration order.

state ¶

state(
    name: str,
    *,
    timeout: str | None = None,
    retries: int = 0,
    parallel: Sequence[Workflow] = (),
    join: str = "all",
    inputs: Sequence[str] = (),
    into: str | None = None,
) -> State

Add a state to the workflow.

When a token enters the state, the control plane makes one task. With parallel, it is a parallel state: it starts one child token for each workflow, and it has no task. See the docs, Concepts > Parallel and nested work.

Parameters:

Name	Type	Description	Default
`name`	`str`	The state's name: `a-z`, `0-9`, `_` and `-`, with a letter first. It is unique in the workflow.	required
`timeout`	`str \| None`	How long a token can stay in the state, for example `"30s"`, `"10m"`, `"6h"` or `"1d"`. Then the token gets the event `timeout`. None: no limit.	`None`
`retries`	`int`	How many more times a failed task runs. Then the token gets the event `error`. A parallel state has no task, so no retries.	`0`
`parallel`	`Sequence[Workflow]`	The child workflows of a parallel state. When the join is met, the token gets the event `joined`.	`()`
`join`	`str`	`"all"`: wait until each child ends. `"any"`: continue when the first child ends, and cancel the others.	`'all'`
`inputs`	`Sequence[str]`	The context keys that each child's context starts with. They must give each field that a child's context requires.	`()`
`into`	`str \| None`	For a parallel state: the context field that keeps the join's `Branches`, so that the routines of later states can read them.	`None`

Returns:

Type	Description
`State`	The state, to use in transitions.

Raises:

Type	Description
`DefinitionError`	The name is not valid, is used, or is Stateous's own. Or the timeout is not a duration, `retries` is not 0 or more, or `parallel`, `join`, `inputs` or `into` breaks a rule of a parallel state.

event ¶

event(
    name: str,
    *,
    payload: type[BaseModel] | None = None,
    into: str | None = None,
) -> Event

Add an event to the workflow. An event is an outcome: only an event moves a token.

Parameters:

Name	Type	Description	Default
`name`	`str`	The event's name, unique in the workflow. Do not declare Stateous's own events, `timeout`, `error` and `joined`: a transition uses them by name.	required
`payload`	`type[BaseModel] \| None`	The Pydantic model of the data that the event carries. None: no data. Guards read it; routines do not.	`None`
`into`	`str \| None`	The context field that keeps the payload when the event is accepted, so that routines can read it. Its type must take the payload model.	`None`

Returns:

Type	Description
`Event`	The event, to use in transitions.

Raises:

Type	Description
`DefinitionError`	The name is not valid, is used, or is Stateous's own. Or the payload is not a Pydantic model class, or `into` is not a context field that takes the payload.

transition ¶

transition(
    source: State,
    target: State,
    *,
    on: Event | str | None = None,
    when: str | None = None,
    otherwise: bool = False,
) -> Transition

Add a transition: the event on moves a token from source to target. The events that a state can emit are the events of its transitions.

Parameters:

Name	Type	Description	Default
`source`	`State`	The state that the token is in, or `START`.	required
`target`	`State`	The state that the token moves to, or `END`.	required
`on`	`Event \| str \| None`	The event, or the name of one of Stateous's own events: `"timeout"`, `"error"` or `"joined"`. Leave it out from `START`.	`None`
`when`	`str \| None`	A guard that reads `ctx['key']`, `payload.field` and `count('event')`. Two guards on one state and event must not overlap: `stateous compile` checks this.	`None`
`otherwise`	`bool`	True for the fallback: the control plane takes it when no guard on the same state and event is true. It has no `when`.	`False`

Returns:

Type	Description
`Transition`	The transition.

Raises:

Type	Description
`DefinitionError`	A state is not of this workflow, or the event is missing or not allowed here. Or the guard does not parse, or it reads a field that the payload does not have. Or two transitions have no guard on the same state and event.

START `module-attribute` ¶

START = State('__start__', workflow='')

Where every token begins. Import it; do not declare it.

A transition from START takes no on=. The control plane leaves START at once, on __started__. When START has more than one transition, their guards on the starting context choose one.

END `module-attribute` ¶

END = State('__end__', workflow='')

Where a token completes. Import it; do not declare it.

A token that enters END is completed. END has no task, and no transition leaves it.

A parallel state with into= keeps the data of its join in the context, as Branches. See Parallel and nested work.

Branches ¶

Bases: BaseModel

The payload of joined: with join="all", one entry for each child; with join="any", the child that ended first.

Branch ¶

Bases: BaseModel

One ended child of a parallel state.

workflow `instance-attribute` ¶

workflow: str

The child's workflow.

token `instance-attribute` ¶

token: str

The child's id.

final `instance-attribute` ¶

final: str

The state that the child was in when it ended.

context `instance-attribute` ¶

context: dict[str, Any]

The child's context when it ended.

Write a routine¶

A routine is a function in the supervisor's folder. @on_state marks it, ctx is its Context, and it returns an Occurrence. The generated contract.py gives the StateRef of each state. See Write a routine.

on_state ¶

on_state(state: StateRef[Any]) -> Callable[[F], F]

Mark a function as the routine of a state. The supervisor runs it for each task of the state.

@on_state(states.testing)
def testing(ctx: TicketContext) -> events.passed | events.failed:
    return events.passed(Tests(passed=12, failed=0))

Parameters:

Name	Type	Description	Default
`state`	`StateRef[Any]`	The state, from `contract.states`.	required

Returns:

Type	Description
`Callable[[F], F]`	A decorator. It returns the function unchanged, with a mark.

Raises:

Type	Description
`TypeError`	`state` is not a `StateRef`.

Context ¶

Bases: Generic[C]

What a routine gets as ctx: the token's context C, and the task's facts.

The generated contract gives the type, for example TicketContext = Context[Ticket]. ctx["key"] reads a copy of a context value. ctx["key"] = value writes it. The control plane checks the writes against the context model, and saves them with the outcome.

A change in place, such as ctx["items"].append(2), changes nothing. Write the new value with ctx["items"] = ....

@on_state(states.coding)
async def coding(ctx: TicketContext) -> events.done | events.stuck:
    ctx.note(f"coding {ctx['repo']}, attempt {ctx.attempt}")
    ctx["branch"] = "fix-1"
    return events.done()

token `instance-attribute` ¶

token: str

The token's id.

task `instance-attribute` ¶

task: int

The task's id.

visit `instance-attribute` ¶

visit: int

1 the first time the token is in this state, 2 the next time, and so on.

attempt `instance-attribute` ¶

attempt: int

1, then 2 after a retry.

entered_by `instance-attribute` ¶

entered_by: EnteredBy | None

The name of the event that brought the token here.

memory `instance-attribute` ¶

memory: Memory

The routine's own memory, for this token and this state: a mapping of keys to JSON values. Each change is saved at once. del ctx.memory[key] deletes a key.

get ¶

get(key: str, default: Any = None) -> Any

Read a context value, with a default.

Parameters:

Name	Type	Description	Default
`key`	`str`	The context key.	required
`default`	`Any`	The value to return when the context has no `key`.	`None`

Returns:

Type	Description
`Any`	A copy of the value at `key`, or `default`.

note ¶

note(text: str) -> None

Add one line of progress to the token's history. The supervisor sends it at once.

Parameters:

Name	Type	Description	Default
`text`	`str`	The line. The spaces at its two ends are removed. Empty text does nothing.	required

Raises:

Type	Description
`ValueError`	The text has more than 500 characters.

schedule ¶

schedule(
    event: type[Occurrence[Any]] | Occurrence[Any],
    *,
    after: str,
) -> None

Set a timer. The clock fires event on this token after the time after, if the token is still in this state.

The timer is set when the control plane saves the report.

Parameters:

Name	Type	Description	Default
`event`	`type[Occurrence[Any]] \| Occurrence[Any]`	The event: a class of `events` without a payload, or an occurrence with its payload, for example `events.remind(note)`.	required
`after`	`str`	A duration from now, for example `"30s"`, `"10m"`, `"6h"` or `"1d"`.	required

Raises:

Type	Description
`ValueError`	`after` is not a duration, or the event needs a payload and `event` is a class.

writes ¶

writes() -> dict[str, Any]

The context keys that this routine wrote, with their values. The supervisor sends them with the report.

schedules ¶

schedules() -> list[Schedule]

The timers that this routine set. The supervisor sends them with the report.

StateRef `dataclass` ¶

StateRef(name: str, emits: tuple[type[O], ...] = ())

Bases: Generic[O]

A state as a supervisor sees it. The generated contract.py has one for each state, as states.<name>. Give it to @on_state.

You do not write a StateRef: stateous init supervisor writes them. The type argument is the union of the state's events. For example, StateRef("coding", emits=(done, stuck)) is a StateRef[done | stuck]. So outcome() returns a done | stuck.

name `instance-attribute` ¶

name: str

The state's name.

emits `class-attribute` `instance-attribute` ¶

emits: tuple[type[O], ...] = ()

The events that the state can emit: the events of its transitions.

may_emit ¶

may_emit(occurrence: Occurrence[Any]) -> bool

Tell if the state can emit an occurrence.

Parameters:

Name	Type	Description	Default
`occurrence`	`Occurrence[Any]`	An occurrence of an event.	required

Returns:

Type	Description
`bool`	True when the event of `occurrence` is one of `emits`.

outcome_schema ¶

outcome_schema() -> dict[str, Any]

The events that this state can emit, with their payloads, as one JSON Schema. Give it to an agent as the form of its final answer.

Returns:

Type	Description
`dict[str, Any]`	A JSON Schema with one branch for each event. An answer that agrees with it names one event and its payload.

Raises:

Type	Description
`OutcomeError`	The state can emit no event, so no answer agrees.

outcome ¶

outcome(data: Any) -> O

Make the occurrence that an agent's answer names.

Pydantic checks the payload as JSON, not in strict mode. So the string "3" agrees with an int field.

Parameters:

Name	Type	Description	Default
`data`	`Any`	The answer: `{"event": ..., "payload": ...}`.	required

Returns:

Type	Description
`O`	The occurrence of the event, with its payload.

Raises:

Type	Description
`OutcomeError`	`data` does not agree with `outcome_schema()`.

Occurrence ¶

Occurrence()

Occurrence(payload: P)

Occurrence(payload: P | None = None)

Bases: Generic[P]

One occurrence of an event, with its payload. A routine returns one.

The generated contract has one subclass for each event, in the class events. The type argument is the payload model, or None. For example, events.failed(Tests(passed=3, failed=1)) is an Occurrence[Tests], and events.done() is an Occurrence[None].

name `class-attribute` ¶

name: str

The event's name. Each subclass sets it.

payload `instance-attribute` ¶

payload: P = cast('P', payload)

The payload, or None for an event without a payload.

payload_model `classmethod` ¶

payload_model() -> type[BaseModel] | None

The payload model of the event.

Returns:

Type	Description
`type[BaseModel] \| None`	The model of the class's `Occurrence[Model]` base, or None when the event has no payload.

OutcomeError ¶

Bases: ValueError

An outcome does not fit its state. The state cannot emit the event, or the payload does not agree with the event's model. StateRef.outcome() raises it.

Talk to the control plane¶

A program outside uses Client to start, fire, read and steer tokens. Each method sends one request of the HTTP API. See Steer tokens.

Client ¶

Client(url: str, *, timeout: float = 30.0)

The control plane's HTTP API at one URL, for a program. Synchronous.

Each method sends one request, and returns the JSON answer. A failure raises a ClientError. Use the client in a with block, or call close() at the end.

from stateous import Client

with Client("http://127.0.0.1:8080") as client:
    client.start("TICKET-1", {"repo": "acme/api"})
    client.fire("TICKET-1", "done", key="gh-delivery-81231")

Make a client for one control plane.

Parameters:

Name	Type	Description	Default
`url`	`str`	The control plane's URL, for example `http://127.0.0.1:8080`.	required
`timeout`	`float`	How long to wait for each answer, in seconds.	`30.0`

Raises:

Type	Description
`ClientError`	`url` is not a valid URL.

start ¶

start(
    token: str, context: dict[str, Any] | None = None
) -> dict[str, Any]

Start a token. If the token exists, it is returned unchanged.

Parameters:

Name	Type	Description	Default
`token`	`str`	The token's id, for example `TICKET-1`.	required
`context`	`dict[str, Any] \| None`	The token's context. None: `{}`.	`None`

Returns:

Type	Description
`dict[str, Any]`	The token.

get ¶

get(token: str) -> dict[str, Any]

Read one token.

Parameters:

Name	Type	Description	Default
`token`	`str`	The token's id.	required

Returns:

Type	Description
`dict[str, Any]`	The token: its state, status, visit, context and the time it entered the state.

Raises:

Type	Description
`NotFound`	No token has this id.

tokens ¶

tokens(
    *,
    status: str | None = None,
    parent: str | None = None,
    limit: int | None = None,
) -> list[dict[str, Any]]

List the tokens, the most recently moved first.

Parameters:

Name	Type	Description	Default
`status`	`str \| None`	Only the tokens with this status, for example `"active"`.	`None`
`parent`	`str \| None`	Only the children of this token.	`None`
`limit`	`int \| None`	The most tokens to return: 1 to 1000. None: 100.	`None`

Returns:

Type	Description
`list[dict[str, Any]]`	The tokens.

history ¶

history(token: str) -> list[dict[str, Any]]

Read the history of a token, in order.

Parameters:

Name	Type	Description	Default
`token`	`str`	The token's id.	required

Returns:

Type	Description
`list[dict[str, Any]]`	The entries: each has `seq`, `at`, `kind` and `data`.

Raises:

Type	Description
`NotFound`	No token has this id.

fire ¶

fire(
    token: str,
    event: str,
    payload: dict[str, Any] | None = None,
    *,
    key: str | None = None,
) -> dict[str, Any]

Fire an event on a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	The token's id.	required
`event`	`str`	The event's name.	required
`payload`	`dict[str, Any] \| None`	The event's payload. None: no payload.	`None`
`key`	`str \| None`	An idempotency key, from 1 to 256 characters. The same key fires one time.	`None`

Returns:

Type	Description
`dict[str, Any]`	The token after the event. When `key` was used before, the answer is `{"duplicate": True, "token": ...}`, and nothing changes.

Raises:

Type	Description
`Refused`	The event cannot move the token, or the payload is not valid.

pause ¶

pause(token: str) -> dict[str, Any]

Pause an active token. Its events are held, its tasks wait, and its timers stand still.

Parameters:

Name	Type	Description	Default
`token`	`str`	The token's id. A child token cannot be paused: pause its top token.	required

Returns:

Type	Description
`dict[str, Any]`	The token.

resume ¶

resume(token: str) -> dict[str, Any]

Resume a paused token. The held events apply in the order that they arrived.

Parameters:

Name	Type	Description	Default
`token`	`str`	The token's id.	required

Returns:

Type	Description
`dict[str, Any]`	The token.

cancel ¶

cancel(token: str) -> dict[str, Any]

Cancel an active or a paused token for good.

Parameters:

Name	Type	Description	Default
`token`	`str`	The token's id.	required

Returns:

Type	Description
`dict[str, Any]`	The token.

report ¶

report(
    task: int,
    event: str,
    payload: dict[str, Any] | None = None,
    *,
    attempt: int | None = None,
) -> dict[str, Any]

Report the outcome of a task that waits. Its routine returned None, and the work went on outside.

Parameters:

Name	Type	Description	Default
`task`	`int`	The task's id: `ctx.task` in the routine.	required
`event`	`str`	The event's name.	required
`payload`	`dict[str, Any] \| None`	The event's payload. None: no payload.	`None`
`attempt`	`int \| None`	The attempt that the report is for. The control plane refuses the report (409) when the task is on a different attempt.	`None`

Returns:

Type	Description
`dict[str, Any]`	The token after the event.

workflow ¶

workflow() -> dict[str, Any]

Read the graph file that the control plane serves.

Returns:

Type	Description
`dict[str, Any]`	The graph file, as JSON.

close ¶

close() -> None

Close the connection. A with block calls it at the end.

ClientError ¶

Bases: Exception

The base of every error of the client. The control plane could not do the request, or the client could not send it. The message is one plain sentence.

NotFound ¶

NotFound(
    status: int,
    message: str,
    *,
    url: str,
    code: str | None = None,
)

Bases: Refused

The control plane answered 404: no token, task or path has this name.

Refused ¶

Refused(
    status: int,
    message: str,
    *,
    url: str,
    code: str | None = None,
)

Bases: ClientError

The control plane refused the request.

Use code in a program, for example attempt_over. The codes are in the docs, Reference > The HTTP API. message is for people, and it can change.

Attributes:

Name	Type	Description
`status`		The HTTP status, for example 409.
`message`		The reason that the control plane gave, as one sentence.
`code`		The stable code of the refusal. None when the answer has no code, for example an answer from a proxy.

The client makes it from the control plane's answer.

Unreachable ¶

Bases: ClientError

The control plane did not answer: the network failed, or no server listens at the URL.

The Python API¶

Write a workflow¶

Workflow ¶

states instance-attribute ¶

events instance-attribute ¶

transitions instance-attribute ¶

state ¶

event ¶

transition ¶

START module-attribute ¶

END module-attribute ¶

Branches ¶

Branch ¶

workflow instance-attribute ¶

token instance-attribute ¶

final instance-attribute ¶

context instance-attribute ¶

Write a routine¶

on_state ¶

Context ¶

token instance-attribute ¶

task instance-attribute ¶

visit instance-attribute ¶

attempt instance-attribute ¶

entered_by instance-attribute ¶

memory instance-attribute ¶

get ¶

note ¶

schedule ¶

writes ¶

schedules ¶

StateRef dataclass ¶

name instance-attribute ¶

emits class-attribute instance-attribute ¶

may_emit ¶

outcome_schema ¶

outcome ¶

Occurrence ¶

name class-attribute ¶

payload instance-attribute ¶

payload_model classmethod ¶

OutcomeError ¶

Talk to the control plane¶

Client ¶

start ¶

get ¶

tokens ¶

history ¶

fire ¶

pause ¶

resume ¶

cancel ¶

report ¶

workflow ¶

close ¶

ClientError ¶

NotFound ¶

Refused ¶

Unreachable ¶

states `instance-attribute` ¶

events `instance-attribute` ¶

transitions `instance-attribute` ¶

START `module-attribute` ¶

END `module-attribute` ¶

workflow `instance-attribute` ¶

token `instance-attribute` ¶

final `instance-attribute` ¶

context `instance-attribute` ¶

token `instance-attribute` ¶

task `instance-attribute` ¶

visit `instance-attribute` ¶

attempt `instance-attribute` ¶

entered_by `instance-attribute` ¶

memory `instance-attribute` ¶

StateRef `dataclass` ¶

name `instance-attribute` ¶

emits `class-attribute` `instance-attribute` ¶

name `class-attribute` ¶

payload `instance-attribute` ¶

payload_model `classmethod` ¶