Debugging
Watch items flow and draw the process graph
.debug() can be inserted anywhere in the chain: every item passing through is printed with the process PID, then forwarded unchanged to the next step.
Signature
Pipeline.debug() -> Pipeline[InPacket]
Pipeline.wait_for_result(debug_graph: str | None = None)
Pipeline.print_graph(debug_graph: str)python
from olympipe import Pipeline
res = (
Pipeline(range(3))
.task(lambda x: x * 10)
.debug() # prints every item that passes here
.task(lambda x: x + 1)
.wait_for_result()
)
print(res)Output
DebugPipe[60979] 0
DebugPipe[60979] 10
DebugPipe[60979] 20
[1, 11, 21]How it works
Printing happens in a dedicated process, so lines may interleave with other print calls from your workers.
debug_graph / print_graph: the process graph
Pass debug_graph="name" to wait_for_result, wait_for_results or wait_for_completion: while waiting, Olympipe draws the process → queue graph and writes it to name.png, redrawn only when it changes. pipeline.print_graph("name") renders it once. You need the graphviz Python package (pip install graphviz) and Graphviz's dot executable; without the package, Olympipe prints a warning and carries on.
python
from olympipe import Pipeline
res = (
Pipeline(range(100))
.task(lambda x: x + 1, count=2)
.batch(10)
.wait_for_result(debug_graph="pipeline_graph") # writes pipeline_graph.png
)
print(len(res), "batches")Output
10 batches
# DOT source of the rendered graph (PIDs and queue ids vary):
digraph G {
61224 [label="(61224) TaskPipe[<lambda>]"]
"Q-sGArOhxB" -> 61224
61227 [label="(61227) TaskPipe[<lambda>]"]
"Q-sGArOhxB" -> 61227
61224 -> "Q-PGt4zD9u"
61227 -> "Q-PGt4zD9u"
61230 [label="(61230) BatchPipe[Batch:10]"]
"Q-PGt4zD9u" -> 61230
...
}