How to make reruns idempotent with targets¶

Declare where a task writes, and cereyan skips the task when that output already exists. Rerunning a day, a failed run, or a whole backfill then only does the missing work.

Declare a target¶

from datetime import date
from cereyan import flow, task, LocalTarget

writes = []

@task(output=lambda day: LocalTarget(f"data/{day}.csv"))
def export(day: date) -> None:
    writes.append(day)
    with LocalTarget(f"data/{day}.csv").open("w") as fh:
        fh.write("id,amount\n1,10\n")

@flow
def daily(day: date) -> None:
    export(day)

daily(date(2026, 3, 1))
daily(date(2026, 3, 1))
assert writes == [date(2026, 3, 1)]   # the second run skipped export

output= takes a Target, or a callable over the task's arguments returning one. Only the parameters named in the callable's signature are passed, so lambda day: ... works for a task with more arguments. When exists() is true the task run ends Skipped with the message output exists, records task_run.skipped, and the body never runs.

Write atomically¶

LocalTarget.open("w") writes to a temporary file next to the target and renames it into place on close. A crash mid-write leaves no partial file, so the next run does not mistake half an output for a finished one. Use temporary_path() when a library insists on writing a path itself:

from cereyan import LocalTarget

target = LocalTarget("data/report.txt")
with target.temporary_path() as tmp:
    with open(tmp, "w") as fh:
        fh.write("done")
assert target.exists()

Custom targets¶

Anything with an exists() method is a target: a row in a table, a key in an object store, a flag in an API.

from datetime import date
from cereyan import flow, task

class TableTarget:
    done: set = set()

    def __init__(self, name: str) -> None:
        self.name = name

    def exists(self) -> bool:
        return self.name in TableTarget.done

@task(output=lambda day: TableTarget(f"sales_{day}"))
def build(day: date) -> None:
    TableTarget.done.add(f"sales_{day}")

@flow
def nightly(day: date) -> None:
    build(day)

nightly(date(2026, 3, 2))
nightly(date(2026, 3, 2))

Remember what you started¶

A target says whether the work finished. When the work is an external job, the dangerous moment is earlier: the job was submitted, the task died, and the retry submits it again. Note the job id in the task's state store the moment you have it, and check for it first:

from cereyan import task, task_state

@task(retries=3, retry_delay=10)
def load_warehouse(day: str) -> str:
    job = task_state.get("job_id")
    if job is None:
        job = warehouse.submit(day)
        task_state.set("job_id", job)
    return warehouse.wait(job)

The note is scoped to this task in this run and is there on a retry, a crash rerun, or a retry from failure; see Task state.

Skip whole runs¶

Targets skip tasks, not runs. To avoid scheduling runs for work already done, give the flow bulk_complete= and let backfills skip those values before they are created.

Rerun on purpose¶

To recompute, delete the target (LocalTarget.remove()) and run again; there is no force flag, because the file is the truth.