Python Basics: Ordering Dependent Tasks with graphlib

A report needs data before it can be rendered, and rendering must finish before publication. Python’s standard-library graphlib module can put tasks in an order that respects these dependencies.

Before you start

These three independent examples were tested on Linux with Bash, Python 3.12.14 and Python 3.13.5. No extra packages or environment variables are required. The commands use python3; replace it with python3.13 to use that interpreter.

Create a fresh working directory in one Bash session:

start_dir="$PWD"
work_dir="$(mktemp -d)" || exit 1
cd -- "$work_dir" || exit 1

Save each source block here using its stated filename. Avoid graphlib.py, which would shadow the standard-library module. These programs only print task names; they do not fetch data, create reports, or publish anything.

Find a valid task order

Save as demo_graphlib_order.py:

from graphlib import TopologicalSorter

dependencies = {
    "publish": ("render",),
    "render": ("load",),
}

order = list(TopologicalSorter(dependencies).static_order())
print("Order:", " -> ".join(order))
print("Empty graph:", list(TopologicalSorter({}).static_order()))

Run it:

python3 demo_graphlib_order.py

Output:

Order: load -> render -> publish
Empty graph: []

Each dictionary key is a task; its value lists the tasks that must finish first. Here, publish depends on render. A name used only as a dependency, such as load, is added automatically with no prerequisites. Check spelling carefully so a typo does not become an unintended task.

static_order() produces an iterator. Converting it to a list consumes that iterator. An empty graph simply has no tasks. See the TopologicalSorter documentation.

Process tasks as their dependencies finish

Save as demo_graphlib_ready.py:

from graphlib import TopologicalSorter

dependencies = {
    "fetch_users": (),
    "fetch_orders": (),
    "join": ("fetch_users", "fetch_orders"),
    "report": ("join",),
}
sorter = TopologicalSorter(dependencies)
sorter.prepare()

try:
    sorter.done("join")
except ValueError:
    print("Early completion: rejected")

while sorter.is_active():
    ready = sorted(sorter.get_ready())
    print("Ready:", ", ".join(ready))
    # Simulate successful completion of every task in this batch.
    sorter.done(*ready)

Run it:

python3 demo_graphlib_ready.py

Output:

Early completion: rejected
Ready: fetch_orders, fetch_users
Ready: join
Ready: report

prepare() freezes the graph and checks it. get_ready() returns tasks whose prerequisites have finished; done() records completion and can unlock later tasks. The intentional, caught error shows that you cannot mark a task done before it has been returned as ready. See the processing methods.

Both fetch tasks could run concurrently, but this loop only simulates completion. A real worker should call done() after successful work and handle failures separately. Forgetting completion can leave the sorter active without releasing more tasks.

Unrelated tasks can have several valid orders. Sorting each ready batch gives our string task names stable display order. Do not alphabetically sort the whole final order: that could put a dependent task before its prerequisite.

Detect a cycle and inspect what remains blocked

Save as demo_graphlib_cycle.py:

from graphlib import CycleError, TopologicalSorter

dependencies = {
    "draft": ("review",),
    "review": ("draft",),
    "publish": ("review",),
    "backup": (),
    "index": ("backup",),
}
sorter = TopologicalSorter(dependencies)

try:
    sorter.prepare()
except CycleError as error:
    print("Cycle members:", ", ".join(sorted(set(error.args[1]))))

completed = set()
while sorter.is_active():
    ready = sorted(sorter.get_ready())
    print("Can finish:", ", ".join(ready))
    completed.update(ready)
    sorter.done(*ready)

blocked = sorted(set(dependencies) - completed)
print("Blocked:", ", ".join(blocked))

Run it:

python3 demo_graphlib_cycle.py

Output:

Cycle members: draft, review
Can finish: backup
Can finish: index
Blocked: draft, publish, review

draft waits for review, which waits for draft. This intentional cycle raises CycleError, which we catch. Its second argument contains one detected cycle, including a repeated endpoint; the set removes that repetition for display. Multiple cycles may exist, so one reported cycle is not a complete diagnosis.

After this error, ready tasks can still be processed. Here, backup and index are independent of the cycle. publish is blocked by the cycle without being part of it. Our blocked calculation works because this example explicitly lists every task as a dictionary key. See cycle detection and partial progress.

Repair the dependency data and create a new sorter before trying again. The sorter does not decide which dependency to remove or whether partial work is appropriate for your application.

Clean up

In the same Bash session, remove only the files you created, then the empty directory:

rm -- demo_graphlib_order.py demo_graphlib_ready.py demo_graphlib_cycle.py
cd -- "$start_dir" || exit 1
rmdir -- "$work_dir"
unset start_dir work_dir

If rmdir reports remaining files, inspect them rather than deleting them blindly.

References

• Python 3.12 graphlib documentation
• Python 3.13 graphlib documentation
• CycleError details

Acknowledgements

This post was created with assistance from AI.