Daily ETL

A scheduled daily flow whose reruns and backfills are idempotent through targets.

Source: examples/daily_etl.py. Run it with python examples/daily_etl.py (no server needed).

A daily flow: extract a day's data, write it to a file, and never do the same day twice. The target on build makes reruns skip finished days, bulk_complete lets a backfill skip them before runs are even created, and the schedule fires the flow every morning once a server is running.

from datetime import date, timedelta

from cereyan import Abort, Cron, LocalTarget, exponential, flow, get_run_logger, task

FIRST_DAY = date(2024, 1, 1)


def output_for(day: date) -> LocalTarget:
    return LocalTarget(f"out/daily/{day}.csv")

Skipping finished days

bulk_complete receives every value a backfill is about to create and returns the ones already done; those runs are recorded as Skipped without dispatching.

def already_built(values: list[date]) -> set[date]:
    return {d for d in values if output_for(d).exists()}

The task

output= names the target; when it exists the task run ends Skipped and the body does not execute. Retries with an exponential delay cover transient failures, and retry_on says which those are: an OSError from the disk or the network is worth another attempt, any other exception fails the task run at once. A day before the source has data can never succeed, so Abort ends the task run without retrying and records abort in its state details.

@task(
    output=output_for,
    retries=2,
    retry_delay=exponential(base=0.5, maximum=30),
    retry_on=(OSError,),
)
def build(day: date) -> str:
    if day < FIRST_DAY:
        raise Abort(f"no source data before {FIRST_DAY}")
    target = output_for(day)
    get_run_logger().info("building %s", target.path)
    with target.open("w") as fh:
        fh.write("id,amount\n")
        for i in range(3):
            fh.write(f"{i},{(i + 1) * 10}\n")
    return target.path

The flow

Fires at 06:30 Istanbul time every day when served, never overlaps itself, and defaults to yesterday so a manual run does the most recent complete day. It is listed in the nightly group of its project.

@flow(
    group="nightly",
    schedule=Cron("30 6 * * *", timezone="Europe/Istanbul"),
    max_concurrent=1,
    on_overlap="skip",
    bulk_complete=already_built,
    run_name="daily-{day}",
)
def daily_etl(day: date = date.today() - timedelta(days=1)) -> str:
    build(day)
    return output_for(day).path

Run it twice

The second call for the same day skips build because its file exists; a skipped task returns None, so the flow returns the path itself.

if __name__ == "__main__":
    day = date(2026, 1, 15)
    first = daily_etl(day)
    second = daily_etl(day)
    assert first == second
    assert output_for(day).exists()
    print("built", first)