Angee

Appearance

angee.workflows.steps ​

Step implementation seam for workflow definitions.

Step rows store a registry key in step_class. The configured class owns the behavior selected by that key and validates the row's config before the row is saved. Product addons register their own :class:StepImpl subclasses through ANGEE_WORKFLOW_STEP_CLASSES; row data never stores dotted import paths.

Implementations declare deterministic: deterministic implementations may be replayed for routing, while non-deterministic activity implementations are journaled by the runtime. Implementations may also declare decision_schema for typed resume payloads. Suspension persists no in-process state: resume_state on the future step-run journal is the only state surviving suspension, and an implementation must write any continuation facts there before returning a suspended result.

GATE_POLICIES ​

Seat aggregation policies supported by the built-in gate step.

TransientStepError ​

python
class TransientStepError(Exception)

Signal that a step implementation failed with a retryable condition.

StepRetryPolicy ​

python
@dataclass(frozen=True, slots=True)
class StepRetryPolicy()

Static queue retry policy declared by one step's JSON config.

DecisionSpec ​

python
@dataclass(frozen=True, slots=True)
class DecisionSpec()

Declaration for one awaited decision slot created while a step suspends.

StepResult ​

python
@dataclass(frozen=True, slots=True)
class StepResult()

Result returned by a workflow step implementation.

done(output, outcome) completes the step and routes by outcome. wait(until=...) records a durable timer wake. External events use :func:angee.workflows.engine.deliver, whose run-scoped generation prevents a delivery racing with this wait from being lost. suspend() pauses the step until an external resolution writes the next journal facts.

done ​

python
@classmethod
def done(cls, output: Any = None, outcome: str = "") -> Self

Return a completed step result.

wait ​

python
@classmethod
def wait(cls,
         *,
         until: datetime | None = None,
         resume_state: dict[str, Any] | None = None) -> Self

Return a durable wait result.

suspend ​

python
@classmethod
def suspend(
    cls,
    *,
    resume_state: dict[str, Any] | None = None,
    decisions: list[DecisionSpec] | tuple[DecisionSpec, ...] = ()
) -> Self

Return a suspended step result.

StepImpl ​

python
class StepImpl(ImplBase)

Base class for registry-selected workflow step implementations.

validate_config ​

python
@classmethod
def validate_config(cls, config: Any) -> None

Validate a step row's JSON config for this implementation.

run ​

python
def run(step_run: Any, *, now: datetime) -> StepResult

Execute one step-run journal row.

heartbeat ​

python
def heartbeat(step_run: Any, *, at: datetime | None = None) -> None

Refresh step_run's heartbeat while a long implementation is running.

retry_policy_from_config ​

python
def retry_policy_from_config(config: Any) -> StepRetryPolicy

Return the queue retry policy declared by config.

validate_retry_config ​

python
def validate_retry_config(config: Any) -> None

Validate the common per-step retry block.

HandlerStep ​

python
class HandlerStep(StepImpl)

Abstract activity step base registered as the built-in handler key.

WaitStep ​

python
class WaitStep(StepImpl)

Built-in timer wait step.

validate_config ​

python
@classmethod
def validate_config(cls, config: Any) -> None

Validate timer wait configuration.

run ​

python
def run(step_run: Any, *, now: datetime) -> StepResult

Return done once the timer or event condition has arrived.

GateStep ​

python
class GateStep(StepImpl)

Built-in gate step that suspends until Slice 4 decision rows exist.

validate_config ​

python
@classmethod
def validate_config(cls, config: Any) -> None

Validate declarative gate slot configuration.

run ​

python
def run(step_run: Any, *, now: datetime) -> StepResult

Suspend the step, keeping only durable resume state.

MapStep ​

python
class MapStep(StepImpl)

Built-in control step that maps one target step over a journaled item list.

validate_config ​

python
@classmethod
def validate_config(cls, config: Any) -> None

Validate the map declaration consumed by the engine.

engine_expanded_filter ​

python
@classmethod
def engine_expanded_filter(cls) -> dict[str, Any]

Return the ORM predicate for map parent rows the engine expands.

target_step ​

python
@classmethod
def target_step(cls, step_run: Any) -> Any

Return the configured target step for one map parent row.

items ​

python
@classmethod
def items(cls, step_run: Any) -> list[Any]

Return the item list resolved from this map step's config.

policy_passes ​

python
@classmethod
def policy_passes(cls, config: Any, output: Mapping[str, Any]) -> bool

Return whether aggregate map output satisfies config.

config_mapping ​

python
@classmethod
def config_mapping(cls, step_run: Any) -> Mapping[str, Any]

Return the map config as a mapping.

expression_value ​

python
@classmethod
def expression_value(cls, expression: Any, step_run: Any) -> Any

Resolve a map items expression against subject, run, or input.

lookup ​

python
@staticmethod
def lookup(value: Any, key: str) -> Any

Read one expression path segment from a mapping or object.

positive_int ​

python
def positive_int(value: Any, label: str) -> int

Return value as a positive integer or raise a config error.

non_negative_int ​

python
def non_negative_int(value: Any, label: str) -> int

Return value as a non-negative integer or raise a config error.

optional_number ​

python
def optional_number(value: Any, label: str) -> float | None

Return value as a number when present, or raise a config error.

optional_positive_int ​

python
def optional_positive_int(value: Any) -> int | None

Leniently return value as a positive integer when possible.

optional_non_negative_int ​

python
def optional_non_negative_int(value: Any) -> int | None

Leniently return value as a non-negative integer when possible.

Released under the AGPL-3.0 License.

Copyright © 2025–present Angee, Inc.