philiprehberger-task-graph

Lightweight task dependency engine with topological execution.

Installation

pip install philiprehberger-task-graph

Usage

from philiprehberger_task_graph import TaskGraph

graph = TaskGraph()

@graph.task()
def fetch_data():
    return download()

@graph.task(depends=["fetch_data"])
def process_data():
    return transform()

@graph.task(depends=["process_data"])
def save_results():
    return store()

# Run tasks in dependency order
results = graph.run()

# Or run with parallelism
results = graph.run_parallel(max_workers=4)

Programmatic API

graph = TaskGraph()
graph.add_task("fetch", fetch_fn)
graph.add_task("process", process_fn, depends=["fetch"])
graph.add_task("save", save_fn, depends=["process"])

# Preview execution order
order = graph.dry_run()
# ["fetch", "process", "save"]

Timeout

Set a maximum execution time for a task. Raises TimeoutError if the task exceeds the limit.

@graph.task(timeout=30.0)
def slow_task():
    return long_running_operation()

# Or with add_task
graph.add_task("fetch", fetch_fn, timeout=10.0)

Retries

Automatically retry a task on failure. The task is retried up to N times before the exception propagates.

@graph.task(retries=3)
def flaky_task():
    return call_unreliable_api()

# Or with add_task
graph.add_task("fetch", fetch_fn, retries=2)

Timeout and Retries Combined

@graph.task(timeout=5.0, retries=2)
def resilient_task():
    return fetch_with_deadline()

Pass Dependency Results

Pass each dependency's return value as a keyword argument to the task function (named after the dependency).

graph = TaskGraph()
graph.add_task("source", lambda: 7)
graph.add_task("double", lambda source: source * 2, depends=["source"])

results = graph.run(pass_results=True)
# {"source": 7, "double": 14}

Execution Hooks

Register callbacks to observe task execution — useful for logging, metrics, or tracing without modifying the task functions.

graph = TaskGraph()

@graph.on_before_run
def log_start(name: str) -> None:
    print(f"starting {name}")

@graph.on_after_run
def log_done(name: str, result: object, duration: float) -> None:
    print(f"{name} done in {duration:.3f}s -> {result!r}")

@graph.on_error
def log_failure(name: str, exc: BaseException) -> None:
    print(f"{name} failed: {exc!r}")

The error hook fires only after all retries have been exhausted; the original exception still propagates after every hook has run.

Cycle Detection

from philiprehberger_task_graph import CycleError

# Raises CycleError if dependencies form a cycle
graph.run()

API

Development

pip install -e .
python -m pytest tests/ -v

Support

If you find this project useful:

⭐ Star the repo

🐛 Report issues

💡 Suggest features

❤️ Sponsor development

🌐 All Open Source Projects

💻 GitHub Profile

🔗 LinkedIn Profile

License

MIT