Skip to content

06 – Async Execution

AsyncNode subclasses run I/O-bound steps (HTTP, DB, file reads) via Engine.run_async. Sync and async nodes can be mixed in the same graph — the engine dispatches each node to the right path automatically.

🎯 Goal

Fetch a document from a remote service, then run a synchronous post-processor.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import asyncio
from dagpipe import AsyncNode, Node, Schema, State, Payload, Graph, Engine

class FetchDoc(AsyncNode):
    id = "web.fetch"

    async def resolve_async(self, state: State):
        body = await fetch_body(state.get("url"))     # your async I/O
        yield self.fork(state, payload_update={"body": body, "fetched": True})

class Summarize(Node):
    id = "text.summarize"

    def resolve(self, state: State):
        words = len(state.get("body", "").split())
        yield self.fork(state, payload_update={"word_count": words})

class DocState(State):
    schema = Schema({
        "url": str,
        "body": str | None,
        "fetched": bool | None,
        "word_count": int | None,
    })

🏃 Run it

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
async def main():
    graph = Graph()
    graph.add_edge(FetchDoc(), Summarize())

    engine = Engine(graph)
    results = await engine.run_async(DocState(payload=Payload({"url": "https://example.com"})))

    assert results[0].get("fetched") is True
    assert results[0].get("word_count") > 0

asyncio.run(main())

🔍 How dispatch works

Engine.run_async checks each node at execution time:

1
2
node is AsyncNode ? → await node.run_async(state)   # resolve_async()
                    → node.run(state)               # resolve()
  • AsyncNode.resolve() is a no-op (returns ()), so a sync engine (run) treats async nodes as if they pruned the branch.
  • AsyncNode.resolve_async() may be a plain async def returning an iterable, or an async generator — both are supported.
  • run_async validates that yielded objects are State (raises TypeError otherwise), exactly like the sync path.

🔍 Async step-wise variant

run_steps_async pairs with async nodes for progress + resume:

1
2
async for step in engine.run_steps_async(root):
    print(step.index, step.node_id, len(step.states), step.completed)

💡 Tips

  • Async nodes and sync nodes interleave freely — no need to split phases.
  • For true concurrency across branches, combine async nodes with a library like asyncio.gather inside a single wrapper node.
  • Don't call run_async on an engine whose nodes are all sync — you pay event loop overhead for nothing; run is fine there.
  • Cancellation patterns (timeouts, retries) belong inside resolve_async, not the engine — keep the engine generic.