ExecutionPlan: run an effect or a stream with different resources per
step until it succeeds (Effect v4's unstable ExecutionPlan)::
from pyfect import execution_plan
plan = execution_plan.make(
execution_plan.Step(primary_layer, attempts=2),
execution_plan.Step(fallback_layer),
)
fx.run_sync(fetch.pipe(fx.with_execution_plan(plan)))A step provides a layer or a context and retries attempts times on its
schedule while while_(error) holds. The plan provides only what
every step provides. Inside an attempt, CurrentMetadata holds the
attempt number and step index; on_event observes AttemptStart /
AttemptSuccess / AttemptFailure for each attempt.
Classes#
Functions#
make#
make(first, *rest) -> ExecutionPlan[P, In, EE, RR]A plan of one or more steps (Effect's ExecutionPlan.make).
| Parameter | Type | Description |
|---|---|---|
first | Step[P, In, EE, RR] | |
rest | Step[P, In, EE, RR] = () |
Returns ExecutionPlan[P, In, EE, RR]
def make[P, In = object, EE = Never, RR = Never](
first: Step[P, In, EE, RR], /, *rest: Step[P, In, EE, RR]
) -> ExecutionPlan[P, In, EE, RR]:
"""A plan of one or more steps (Effect's ``ExecutionPlan.make``)."""
steps = (first, *rest)
for i, step in enumerate(steps):
if step.attempts is not None and step.attempts < 1:
raise ValueError(f"ExecutionPlan.make: step[{i}].attempts must be greater than 0")
return ExecutionPlan(steps)merge#
merge(first, *rest) -> ExecutionPlan[P, In, EE, RR]One plan running the steps of each plan in turn.
| Parameter | Type | Description |
|---|---|---|
first | ExecutionPlan[P, In, EE, RR] | |
rest | ExecutionPlan[P, In, EE, RR] = () |
Returns ExecutionPlan[P, In, EE, RR]
def merge[P, In = object, EE = Never, RR = Never](
first: ExecutionPlan[P, In, EE, RR], /, *rest: ExecutionPlan[P, In, EE, RR]
) -> ExecutionPlan[P, In, EE, RR]:
"""One plan running the steps of each plan in turn."""
return ExecutionPlan(tuple(step for plan in (first, *rest) for step in plan.steps))with_execution_plan#
with_execution_plan(plan, *, on_event=None) -> _WithExecutionPlan[Any, Any, Any, Any]Run an effect under plan: each step in turn, with the services
it provides, retried as the step says, until an attempt succeeds or
the plan runs out (then fail with the last error). Defects and
interruptions are not retried. Use with .pipe::
fetch.pipe(fx.with_execution_plan(plan))Each attempt sees its CurrentMetadata. on_event observes every
attempt's start and end, in order; its failures are ignored, and the
end is reported even when the attempt is interrupted.
| Parameter | Type | Description |
|---|---|---|
plan | ExecutionPlan[Any, Any, Any, Any] | |
on_event | OnEvent[Any, Any] | None = None |
Returns _WithExecutionPlan[Any, Any, Any, Any]
def with_execution_plan(
plan: ExecutionPlan[Any, Any, Any, Any], *, on_event: OnEvent[Any, Any] | None = None
) -> _WithExecutionPlan[Any, Any, Any, Any]:
"""Run an effect under ``plan``: each step in turn, with the services
it provides, retried as the step says, until an attempt succeeds or
the plan runs out (then fail with the last error). Defects and
interruptions are not retried. Use with ``.pipe``::
fetch.pipe(fx.with_execution_plan(plan))
Each attempt sees its ``CurrentMetadata``. ``on_event`` observes every
attempt's start and end, in order; its failures are ignored, and the
end is reported even when the attempt is interrupted."""
return _WithExecutionPlan(plan, on_event)