dtt -- data transformation tool
================================

Minimal Python data transformation framework. Explicit over implicit. No magic.
No config files. Every abstraction justifies its existence.

Dependencies: psycopg (for meta; bring your own driver for data)


CONCEPTS
--------

DAG
    An explicit registry of models and their dependencies. You create it,
    you pass it around, you run it. No global state.

        from dtt import DAG
        dag = DAG()

model
    A plain Python function that accepts (con, date) as its first two
    parameters. Register it on the DAG explicitly:

        def orders(con, date):
            con.execute("INSERT INTO orders ...")

        def order_summary(con, date):
            con.execute("INSERT INTO order_summary ...")

        dag.register(orders,        deps=[])
        dag.register(order_summary, deps=['orders'])

    A decorator form is also available when you prefer co-location:

        @dag.model(deps=['orders'])
        def order_summary(con, date):
            con.execute("INSERT INTO order_summary ...")

    Both enforce the (con, date) signature at registration time.

    Per-step retry policies override the run-level default when set:

        dag.register(orders,        deps=[],          max_retries=5)
        dag.register(sessions,      deps=['orders'],   max_retries=2)
        dag.register(daily_stats,   deps=['sessions'], max_retries=0)
        dag.register(order_summary, deps=['orders'])   # uses run-level default

run
    Executes all registered models in dependency order for one date.
    Validates the DAG before running anything. Logs to _runs and _logs tables.
    Retries with exponential backoff on failure.

        run_id = dag.run('2024-01-01', con, meta, max_retries=3)

    workers: run independent models in parallel (fan-out/fan-in):

        run_id = dag.run('2024-01-01', con, meta, workers=4)

    executor: replace the default fn(con, date) call with your own dispatch.
    Use this to run each step in a subprocess, a K8s Job, or any other
    environment. Must be callable(name: str, date) -> None.

        run_id = dag.run('2024-01-01', con, meta, executor=my_executor)

resume
    Re-run a previous run, skipping steps that already succeeded. Reads _runs
    for the given run_id to determine what completed. Logs append to the same
    run_id so the full history stays together.

        run_id = dag.resume(run_id, '2024-01-01', con, meta)

backfill
    Loops run() over a date range, inclusive on both ends. Validates the DAG
    once before starting. Accepts ISO date strings or datetime.date objects.

        run_ids = dag.backfill('2024-01-01', '2024-03-31', con, meta)
        run_ids = dag.backfill(date(2024, 1, 1), date(2024, 3, 31), con, meta)

    Returns one run_id per date. Stops on the first unrecoverable failure.


CONNECTIONS
-----------

con
    Your data warehouse connection. Opaque -- dtt passes it directly to each
    model. Bring your own driver (psycopg, clickhouse-driver, bigquery, etc).

meta
    A connection for observability. dtt creates two tables on first use:

        _runs  : run_id, step, date, status, started_at, finished_at, duration_ms, error
        _logs  : seq, run_id, ts, message

    dtt uses %s-style parameter placeholders (psycopg). For SQLite, wrap the
    connection in a thin adapter that substitutes ? for %s. See the example.

    PostgREST pointed at the meta Postgres gives you a zero-code query API:

        GET /runs?status=eq.error
        GET /runs?date=eq.2024-01-01&order=started_at.desc
        GET /logs?run_id=eq.abc-123&order=seq


COMPARISON
----------

                    dtt         dbt             Airflow
                    -------     -----------     -----------
    models          Python fn   SQL + Jinja     Python fn
    dependencies    explicit    {{ ref() }}     explicit
    config          none        profiles.yml    airflow.cfg
    scheduler       yours       yours           built-in
    UI              none        none            built-in
    observability   2 tables    run artifacts   full DB
    install size    ~200 lines  large           large
    Python needed   yes         no              yes

Use dtt when your team writes Python and you want to own the pipeline without
a framework's opinions. Use dbt when your authors are SQL-first and need
{{ ref() }} lineage. Use Airflow when you need a scheduler, a UI, and
multi-team visibility that a table and cron cannot provide.


SCHEDULING
----------

dtt has no built-in scheduler. The recommended production setup uses pg_cron
(a Postgres extension) to insert rows into a _jobs table on a schedule. A worker
process polls _jobs and calls dag.run(). Everything stays in Postgres -- no
external scheduler, no message broker.

    -- pg_cron schedule (runs inside Postgres)
    SELECT cron.schedule(
        'events-daily',
        '0 5 * * *',
        $$INSERT INTO _jobs (pipeline, date) VALUES ('events', CURRENT_DATE)$$
    );

    -- stale job recovery (run every few minutes via pg_cron)
    UPDATE _jobs SET status = 'pending'
    WHERE status = 'running'
    AND heartbeat_at < now() - interval '3 minutes';

The _jobs table:

    CREATE TABLE _jobs (
        id          SERIAL PRIMARY KEY,
        pipeline    TEXT NOT NULL,
        date        TEXT NOT NULL,
        status      TEXT NOT NULL DEFAULT 'pending',
        heartbeat_at TIMESTAMPTZ,
        created_at  TIMESTAMPTZ DEFAULT now()
    );

Manual triggers go through a small FastAPI app that inserts into _jobs directly.


KUBERNETES
----------

One Deployment, one pod, two containers (sidecar pattern):

    api        FastAPI. Accepts POST /pipelines/{name}/run (inserts into _jobs),
               GET /runs, GET /runs/{id}/logs, POST /runs/{id}/resume.

    worker     Polls _jobs with FOR UPDATE SKIP LOCKED. Calls dag.run(). Sends
               a heartbeat (UPDATE heartbeat_at) every 60 seconds so pg_cron can
               detect crashes and reset stale jobs to pending.

Both containers use the same image. The worker imports dtt and pipelines.py
directly -- no HTTP between them.

    # claim a job
    UPDATE _jobs SET status = 'running', heartbeat_at = now()
    WHERE id = (
        SELECT id FROM _jobs WHERE status = 'pending'
        ORDER BY created_at LIMIT 1
        FOR UPDATE SKIP LOCKED
    ) RETURNING *;

    # heartbeat (every 60s while dag.run() is executing)
    UPDATE _jobs SET heartbeat_at = now() WHERE id = %s;

    # complete
    UPDATE _jobs SET status = 'done' WHERE id = %s;

If the worker container crashes mid-run, the heartbeat stops. pg_cron detects
the stale heartbeat and resets the job to pending. The worker picks it up again
on restart. dtt's resume() can be used to skip already-successful steps.

_jobs vs _runs
    _jobs is the queue: one row per pending or in-flight pipeline run. It exists
    to coordinate dispatch -- who picks up what, and is the worker still alive.
    Rows can be deleted after completion.

    _runs is the history: one row per step per run, written by dtt, kept forever.
    Query it to see what succeeded, what failed, how long each step took. A single
    _jobs row triggers a run_id, which produces many _runs rows.


QUICKSTART
----------

    cd example
    python run.py 2024-01-01        # local SQLite
    python run.py 2024-01-01 2024-01-07  # backfill

    tilt up                         # k3d + Postgres + sidecar (see example/README)


DESIGN RULES
------------

- No YAML, no config files, no CLI parsing
- No file scanning or auto-discovery
- No global state: models register on an explicit DAG instance you create
- Failures caught early: signature validation at registration, DAG validation before any run
- Bring your own data connection; bring your own meta connection
- Free-threaded Python 3.14t safe


TESTS
-----

    uv run pytest tests/ -v