Copy-Paste Templates

<project>/
├── data/                # local only
├── pipelines/
│   ├── ingest_<source>.py
│   ├── transform_<entity>.py
│   └── build_<process>.py
├── sql/
│   ├── bronze/
│   ├── silver/
│   ├── gold/
│   └── audit/
├── tests/
├── scripts/
├── .env.example
├── pyproject.toml
└── README.md

"""<verb>_<layer>_<entity>: one-line description"""
from __future__ import annotations
import time, logging
from contextlib import contextmanager

log = logging.getLogger(__name__)

@contextmanager
def run_tracker(job: str, db):
    t0 = time.time()
    cur = db.cursor()
    cur.execute(
        "INSERT INTO audit.pipeline_runs(job, status) VALUES (%s,'running') RETURNING run_id",
        [job],
    )
    run_id = cur.fetchone()[0]
    db.commit()
    try:
        yield run_id
        cur.execute(
            "UPDATE audit.pipeline_runs SET status='success', duration_ms=%s WHERE run_id=%s",
            [int((time.time()-t0)*1000), run_id],
        )
    except Exception as e:
        cur.execute(
            "UPDATE audit.pipeline_runs SET status='failure', error=%s, duration_ms=%s WHERE run_id=%s",
            [str(e), int((time.time()-t0)*1000), run_id],
        )
        raise
    finally:
        db.commit()


def run(db) -> None:
    with run_tracker("transform_silver_claims", db) as run_id:
        cur = db.cursor()
        cur.execute("DELETE FROM silver.claims_clean WHERE service_date = CURRENT_DATE")
        cur.execute("""
            INSERT INTO silver.claims_clean (...)
            SELECT ... FROM bronze.raw_claims
            WHERE _loaded_at::date = CURRENT_DATE
        """)
        log.info("inserted %s rows (run_id=%s)", cur.rowcount, run_id)

-- silver.<entity>_clean
-- Source: bronze.raw_<source>_<entity>
-- Owner:  data-eng
-- Notes:  idempotent, dedupes on <business_key>

INSERT INTO silver.<entity>_clean
SELECT DISTINCT ON (<business_key>)
    <business_key>,
    -- typed, trimmed, conformed columns
    _source,
    _loaded_at,
    CURRENT_TIMESTAMP AS _updated_at
FROM bronze.raw_<source>_<entity>
WHERE _loaded_at >= :since
ORDER BY <business_key>, _loaded_at DESC
ON CONFLICT (<business_key>) DO UPDATE
SET ... ;

/project
  /extract
  /load
  /transform
  /validate
  /output

CREATE TABLE bronze.raw_<source> AS SELECT * FROM source;
CREATE TABLE silver.<entity>_clean AS SELECT * FROM bronze.raw_<source>;
CREATE TABLE gold.<metric> AS SELECT COUNT(*) FROM silver.<entity>_clean;

def run_pipeline():
    extract()
    load()
    transform()
    export()