dagrun 0.1.0

Declarative DAG runner with pipelining and concurrency control.

dagrun

A small, dependency-free DAG runner for Python where you never declare an edge. Annotate your functions with the columns they read and produce, and dagrun will figure out the rest.

Example

from collections.abc import Iterable
from dagrun import Dag, DagRunner
from dagrun.model import PK, FK, Column, Entity


class Site(Entity):
    site_id = PK[str]
    name = Column[str | None]


class File(Entity):
    file_id = PK[str]
    site = FK[Site]
    name = Column[str | None]
    title = Column[str | None]


dag = Dag()


@dag.fn
def discover_sites(sp: Sharepoint) -> Iterable[Site[Site.name]]:
    for raw in sp.list_sites():
        yield Site(site_id=raw["id"], name=raw["name"])


@dag.fn
def discover_files(site: Site, sp: Sharepoint) -> Iterable[File[File.name]]:
    for raw in sp.list_files(site.site_id):
        yield File(file_id=raw["id"], site=site, name=raw["name"])


@dag.fn
def extract_title(file: File[File.name]) -> File[File.title]:
    return File(file_id=file.file_id, title=file.name.removesuffix(".txt"))


runner = DagRunner()
runner.provide(Sharepoint, Sharepoint)
runner.execute(dag)

Parameter annotations declare what a function reads (File[File.name]), return annotations declare what it produces, and the dependency graph is built from those. Reads and writes are keyed per column, so functions that enrich different columns of the same entity run independently.

What you get

• Compile-time checks. compile() fails on a missing dependency, a column produced twice, or a cycle, before anything runs.
• Incremental reruns. The store records when each (function, input) pair last ran. Pass max_age=timedelta(hours=6) to skip anything still fresh.
• Deferred nodes. A node can raise Defer() to skip the current input at runtime, writing and recording nothing so the next run reconsiders it.
• Pipelining. Tasks are scheduled and gated per input, so extract_title starts on the first file while discover_files is still finding the rest.
• Concurrency control. Each function declares a cost against named pools to bound resource usage and respect rate limits.
• Blob columns. A Blob column stashes bytes in a dedicated backend and keeps only a reference in the row.