Skip to content

adag

Another DAG: a cached, content-addressed DAG engine for Python data pipelines.

A step's cache key is a hash of its fields, its inputs and its source code. Change one of them and that step and everything downstream recompute. Nothing else does.

Install

pip install adag

Requires Python 3.12 or newer.

Start here

  1. Why adag: decide if it fits. 2 minutes.
  2. Steps and pipelines: write two steps, run them, see the cache hit.
  3. Inputs: feed values and files into the graph.
  4. Caching: what is in a cache key, and what Config keeps out.
  5. Command line: run a target from a shell.

The first pipeline

import tempfile
from dataclasses import dataclass
from pathlib import Path

from pydantic import BaseModel

from adag import InputParam, Pipeline, Step


class Number(BaseModel):
    value: int


@dataclass(frozen=True)
class Square(Step[Number]):
    def run(self, x: int) -> Number:
        print("computing square")
        return Number(value=x * x)


@dataclass(frozen=True)
class AddOne(Step[Number]):
    def run(self, n: Number) -> Number:
        return Number(value=n.value + 1)


x = InputParam[int]("x")
total = AddOne()(n=Square()(x=x))
pipeline = Pipeline(total, cache_dir=Path(tempfile.mkdtemp()))

print(pipeline.execute(x=4))  # computes both steps
print(pipeline.execute(x=4))  # served from the cache

computing square
value=17
value=17

(output trimmed: log lines removed)

The second run reads both results from the cache and does not call Square.run again.

Find a feature

You want to Read
Keep a setting out of the cache key Caching
See why a step ran again Why a step recomputed
Cache a result that is not JSON Caching
Resume a long loop after a failure Batch steps
Run the same sub-graph for each key Fan-out
Await I/O, or run steps at the same time Async and parallel
Run branches in processes or on a scheduler Dispatch
Keep the cache in SQLite, SQL or Redis Cache stores
Save a pipeline and load it in another process Save and load
Draw the graph Visualization
Time a run or receive its events Observers
Upgrade from 0.1 or 0.2 Upgrade to 0.3.0
Look up a name API reference