without_dag¶
Bounded-concurrency execution of DAG-shaped async workflows for without, liftable into a Processor.
dataclass
¶
Node(
key: NodeKey,
dependencies: tuple[NodeKey, ...],
run: Callable[[tuple[object, ...]], Awaitable[object]],
)
One async step in a graph, named by key and wired by dependencies.
The narrow interface a graph-defining frontend lowers onto: a Node is a
value, not a place. run receives its dependencies' results as a tuple in
dependencies order and returns this node's single result. Results cross
this interface as object (the executor cannot know each step's type); a typed
frontend restores precision above it, exactly as without-web's Extractor
values are collected with object values and re-typed by into.
dataclass
¶
Plan(
by_key: Mapping[NodeKey, Node],
dependencies: Mapping[NodeKey, tuple[NodeKey, ...]],
consumers: Mapping[NodeKey, int],
)
A node set compiled into its input-independent scheduling structure.
Everything a run needs that does not depend on the input values: the nodes
by key, the dependency edges (as the graph graphlib wants), and the number
of consumers per key (so a result can be freed once its last dependent has
read it). Computed once and reused across runs as a value, so a fixed graph
driven per event never rebuilds any of it.
dataclass
¶
CompiledGraph(
nodes: tuple[Node, ...],
inputs: tuple[NodeKey, ...],
output: NodeKey,
limit: int | None,
_plan: Plan,
)
A frozen graph that is an async callable (*Ins) -> Out.
build returns one of these. It runs one bounded-concurrency execution per
call, seeding each entry the graph was opened over with the matching
positional argument and returning the value of its output. A single-input
graph is a plain Callable[[In], Awaitable[Out]], so it lifts into a
Processor with from_map. The scheduling structure is compiled once at
build into _plan and reused by every call and stream, so a graph driven
per event never recomputes it. nodes is kept so the structure is
recoverable (a future diagram is derived from this one declaration, not
maintained beside it).
Both the call and stream take a checkpoint of {node key: result}, the
shape stream itself emits. A node named in it is not run: its result is
taken as given and fed to its dependents, so a run picks up where a previous
one stopped. That is the whole of resumption, because a node's key is a name
the user chose (see node), so the pairs survive the process that produced
them: sink them to a Redis hash or a table row under the workflow's
idempotency key, and hand the hash back as the checkpoint after a crash.
stream(
*values: *Ins,
checkpoint: Mapping[NodeKey, object] = _NO_CHECKPOINT,
) -> AsyncGenerator[tuple[NodeKey, object]]
Run the whole graph, yielding each node's (key, result) as it completes.
The streaming counterpart to calling the graph: __call__ samples the
single output value (a behavior), stream reports every completion as
it happens (the events), letting the caller react as results land or read
several outputs. The positional arguments seed the graph's inputs,
checked against *Ins exactly as the call is; match a yielded key against
a Handle's key to pick out a node's result.
A node supplied by checkpoint is skipped, so a resumed run yields only
what this run actually computed. Collecting those pairs back into the
checkpoint is what advances it, and a checkpoint collected from a run is
closed under ancestry (a node completes only after its dependencies did),
so resuming from it re-runs exactly the nodes that had not finished.
dataclass
¶
A builder that records async steps and returns typed Handles.
of opens a graph over its entry types and hands back a tuple of one Handle
per type, node names a step and wires it to the handles it depends on, and build freezes the
result into a CompiledGraph. Because the graph carries its entry pack in
its type (Graph[*Ins]), build needs only the output handle: it recovers
the inputs the graph already knows, so there is no second place to keep in
sync. The builder itself is a frozen value; the only mutation is appending to
its interior list of recorded nodes. Each step's function receives its
dependencies' results as positional arguments in the order its handles were
passed.
staticmethod
¶
of(
a: type[A], b: type[B], c: type[C]
) -> tuple[
Graph[A, B, C],
tuple[Handle[A], Handle[B], Handle[C]],
]
of(
a: type[A], b: type[B], c: type[C], d: type[D]
) -> tuple[
Graph[A, B, C, D],
tuple[Handle[A], Handle[B], Handle[C], Handle[D]],
]
of(
a: type[A],
b: type[B],
c: type[C],
d: type[D],
e: type[E],
) -> tuple[
Graph[A, B, C, D, E],
tuple[
Handle[A],
Handle[B],
Handle[C],
Handle[D],
Handle[E],
],
]
of(
a: type[A],
b: type[B],
c: type[C],
d: type[D],
e: type[E],
f: type[F],
) -> tuple[
Graph[A, B, C, D, E, F],
tuple[
Handle[A],
Handle[B],
Handle[C],
Handle[D],
Handle[E],
Handle[F],
],
]
of(
a: type[A],
b: type[B],
c: type[C],
d: type[D],
e: type[E],
f: type[F],
g: type[G],
) -> tuple[
Graph[A, B, C, D, E, F, G],
tuple[
Handle[A],
Handle[B],
Handle[C],
Handle[D],
Handle[E],
Handle[F],
Handle[G],
],
]
of(
a: type[A],
b: type[B],
c: type[C],
d: type[D],
e: type[E],
f: type[F],
g: type[G],
h: type[H],
) -> tuple[
Graph[A, B, C, D, E, F, G, H],
tuple[
Handle[A],
Handle[B],
Handle[C],
Handle[D],
Handle[E],
Handle[F],
Handle[G],
Handle[H],
],
]
Open a graph over inputs, returning it and a tuple of one Handle per entry type.
An entry's key is its position (input:0, ...) rather than a name the
caller supplies, because an entry is fed positionally on every call and so
never has to be recovered the way a node's result does. A node may not
take one of these keys; node rejects the collision.
node(
key: NodeKey,
fn: Callable[[A, B, C], Awaitable[T]],
a: Handle[A],
b: Handle[B],
c: Handle[C],
) -> Handle[T]
node(
key: NodeKey,
fn: Callable[[A, B, C, D], Awaitable[T]],
a: Handle[A],
b: Handle[B],
c: Handle[C],
d: Handle[D],
) -> Handle[T]
node(
key: NodeKey,
fn: Callable[[A, B, C, D, E], Awaitable[T]],
a: Handle[A],
b: Handle[B],
c: Handle[C],
d: Handle[D],
e: Handle[E],
) -> Handle[T]
node(
key: NodeKey,
fn: Callable[[A, B, C, D, E, F], Awaitable[T]],
a: Handle[A],
b: Handle[B],
c: Handle[C],
d: Handle[D],
e: Handle[E],
f: Handle[F],
) -> Handle[T]
node(
key: NodeKey,
fn: Callable[
[A, B, C, D, E, F, G], Awaitable[T]
],
a: Handle[A],
b: Handle[B],
c: Handle[C],
d: Handle[D],
e: Handle[E],
f: Handle[F],
g: Handle[G],
) -> Handle[T]
node(
key: NodeKey,
fn: Callable[
[A, B, C, D, E, F, G, H], Awaitable[T]
],
a: Handle[A],
b: Handle[B],
c: Handle[C],
d: Handle[D],
e: Handle[E],
f: Handle[F],
g: Handle[G],
h: Handle[H],
) -> Handle[T]
Add a node named key computing fn from the handles it depends on,
returning its result handle.
fn is called with the dependencies' results as positional arguments in
the order their handles are passed. The overloads above tie each handle's
type to fn's matching parameter, so a mismatch is a static error.
The caller supplies key rather than the builder minting one: uniqueness
alone would be cheaper to generate, but a name chosen in the source is the
same on the other side of a crash, which is what lets a run's
(key, result) pairs be stored and handed back as a checkpoint. It must
be distinct from every key already in the graph, node or entry, since a
duplicate would otherwise quietly displace the node it collides with.
build(
*, output: Handle[Out], limit: int | None = None
) -> CompiledGraph[*Ins, Out]
Freeze the recorded steps into a callable graph over the graph's inputs.
The scheduling structure is compiled once here, so running the graph
repeats no graph analysis. limit caps how many nodes run concurrently;
it defaults to None, which leaves concurrency unbounded (every ready
node runs at once). Pass an integer to cap it when the steps contend for
a scarce resource.
dataclass
¶
Handle(key: NodeKey)
Bases: Generic[_T_co]
A typed reference to a node's future result.
The token the builder hands back from of/node and takes back as a
dependency. T is phantom: the handle carries only the node's key, but the
type flows through the wiring so a downstream step is checked against the
types of the handles it depends on. Because a caller can only pass handles
that already exist, a cycle is unrepresentable through this API.
async
¶
drive(
plan: Plan,
inputs: Mapping[NodeKey, object],
limit: int | None,
) -> AsyncGenerator[tuple[NodeKey, object]]
Run a compiled Plan, yielding each (key, result) the instant it completes.
The streaming core, and the events half of the model. Yields completions in
whatever order nodes finish; the only ordering guarantee is the causal one, a
node after the dependencies it consumed. limit caps how many nodes run
concurrently (None is unbounded).
inputs pre-supplies values by key: a key found there is marked done without
running and is never yielded, and its value is fed to its dependents as if it
had just been computed. A source key the graph is opened over and a node whose
result was captured by an earlier run are the same thing to the scheduler,
which is what makes a run resumable from a checkpoint of the (key, value)
pairs a previous one yielded. Only a node's own key is consulted, so a node
absent from inputs runs even when a dependent of it is already supplied; a
checkpoint captured from a run is closed under ancestry anyway, since a node
completes only after its dependencies did.
Scheduling drains completions off an asyncio.Queue that each spawned future
feeds via a done-callback attached once at spawn, rather than
asyncio.wait(..., FIRST_COMPLETED) re-registering a callback on every
in-flight future each step (O(W) callback churn per completion for a graph W
nodes wide). It reuses without.limit_concurrency's bounded-concurrency shape
but not its call, since the scheduler needs the completed task's NodeKey to
unlock successors, which that lazy source hides. Acyclicity is proven by TopologicalSorter.prepare,
which raises graphlib.CycleError. Each node runs once; a result is dropped
as soon as its last dependent has captured it. A node that raises fails the
whole run, cancelling in-flight siblings, which is also how closing the
iterator early tears the run down.
async
¶
evaluate(
plan: Plan,
target: NodeKey,
inputs: Mapping[NodeKey, object],
limit: int | None,
) -> object
Run every node in plan and return target's value: the behavior read.
A consumer of drive that runs the whole graph and keeps the one value the
caller wants, dropping the rest. target is a node whose completion supplies
the value, or a supplied input returned directly (an identity plan). There is
deliberately no early return on target: the graph is run to completion, so
the result reflects the whole graph and every node's effects have happened.
A target that is neither a defined node nor a supplied input raises
KeyError, matching drive, rather than silently reading back as None.