From Black Boxes to First-Class Tasks

58.1 Which Steps Deserve Deeper Integration? (20 min)

On Day 52 we wrapped every tool as a subprocess: simple and robust. Now, with the platform stable, we can ask for each step whether finer control is worth the coupling.

Cost of in-process integration: steps now share one Python process, so they also share DuckDB connections. Two DuckDB connections with different settings to the same file can't coexist in one process. That means in-process steps must run one after another and close their connections. Our flows are sequential at the writer level anyway (Day 57), and dlt and dbt close their connections when done. Keep it that way.

58.2 Refactoring Ingestion into a Library (25 min)

ingest.py (Day 30) mixes logic (run a source, collect errors) with a command-line interface (flags, exit codes). Prefect needs the logic as importable functions. The classic refactor: move logic into a module and keep the script as a thin wrapper.

Create macro_ingest/runner.py:

"""Ingestion logic as a library: used by ingest.py (CLI) and Prefect tasks."""
import time
from datetime import datetime, timezone

import dlt

from macro_ingest.boc import boc_source
from macro_ingest.fred import fred_source
from macro_ingest.resilience import ErrorLog
from macro_ingest.statcan import statcan_source

SOURCES = {"boc": boc_source, "fred": fred_source, "statcan": statcan_source}


def make_pipeline():
    return dlt.pipeline(pipeline_name="macro_ingest", destination="duckdb",
                        dataset_name="raw_dlt", export_schema_path="schemas/export")


def run_source(name: str, full_refresh: bool = False) -> dict:
    """Load one source into raw_dlt. Raises on failure; returns a summary on success."""
    started = datetime.now(timezone.utc)
    t0 = time.perf_counter()
    errors = ErrorLog()
    pipeline = make_pipeline()

    info = pipeline.run(SOURCES[name](error_log=errors),
                        refresh="drop_resources" if full_refresh else None)
    info.raise_on_failed_jobs()
    if errors.records:
        pipeline.run(errors.records, table_name="extraction_errors", write_disposition="append")

    summary = {"source": name, "status": "warning" if errors.records else "ok",
               "series_errors": len(errors), "duration_s": round(time.perf_counter() - t0, 1),
               "run_started_at": started, "message": None}
    pipeline.run([summary], table_name="ingest_runs", write_disposition="append")
    summary["error_examples"] = [f"{e['series_id']}: {e['message'][:80]}" for e in errors.records[:5]]
    return summary

Then rewrite ingest.py's main() to loop over sources and call run_source, keeping its flags and exit codes (0 / 2 / 1). The CLI behaves exactly as before. Verify:

python ingest.py --only boc; echo "exit: $?"
python -m pytest -q                      # Day 10 + Day 29 tests still pass

This "library + thin CLI" shape is worth applying to every script you write: it makes code testable, schedulable and reusable without duplication.

58.3 Per-Source dlt Tasks with Exception-Aware Retries (25 min)

Now each source becomes its own task, so FRED being down retries FRED only.

Which exceptions are transient?

In-process, failures arrive as exceptions, not exit codes. dlt wraps underlying errors (e.g. a PipelineStepFailed around a ConnectionError), so we inspect the whole exception chain:

Create orchestration/tasks/ingest.py:

"""Per-source dlt ingestion as Prefect tasks."""
from prefect import get_run_logger, task

from macro_ingest.runner import run_source
from orchestration.tasks.locks import duckdb_access

TRANSIENT = {"ConnectionError", "Timeout", "ReadTimeout", "ConnectTimeout",
             "ChunkedEncodingError", "CircuitOpen", "TransientStepError"}
NEVER_RETRY = {"DataValidationError"}                     # schema contract violations (Day 25)


def exception_chain(exc):
    """The exception and everything it was raised from (__cause__ / __context__)."""
    chain, e = [], exc
    while e is not None and e not in chain:
        chain.append(e)
        e = e.__cause__ or e.__context__
    return chain


def is_transient(exc) -> bool:
    names = {type(e).__name__ for e in exception_chain(exc)}
    if names & NEVER_RETRY:
        return False
    return bool(names & TRANSIENT)


def retry_if_transient_exception(task, task_run, state) -> bool:
    try:
        state.result()
    except Exception as e:
        return is_transient(e)
    return False


@task(retries=2, retry_delay_seconds=[300, 600], retry_jitter_factor=0.2,
      retry_condition_fn=retry_if_transient_exception, timeout_seconds=1800)
def ingest_source(name: str, full_refresh: bool = False) -> dict:
    logger = get_run_logger()
    with duckdb_access(write=True):
        summary = run_source(name, full_refresh)
    logger.info(f"{name}: {summary['status']} — {summary['series_errors']} series error(s), "
                f"{summary['duration_s']}s")
    for ex in summary.get("error_examples", []):
        logger.warning(f"{name} series error: {ex}")
    return summary

• exception_chain follows __cause__ (set by raise X from Y) and __context__ (set when an exception occurs while handling another). Wrapped network errors are still recognised.
• Contract violations are never retried, even if a network error appears elsewhere in the chain. A changed API shape won't fix itself in ten minutes.
• duckdb_access(write=True) is the lock helper you'll write on Day 59 (59.2). For today, create orchestration/tasks/locks.py with Day 57's simple version:

from contextlib import contextmanager

from prefect.concurrency.sync import concurrency


@contextmanager
def duckdb_access(write: bool):
    with concurrency("duckdb-writer", occupy=1) if write else _noop():
        yield


@contextmanager
def _noop():
    yield

The extract_load subflow

from prefect import flow


@flow(name="extract-load")
def extract_load_flow(sources: list[str] = ("boc", "fred", "statcan"),
                      full_refresh: bool = False) -> dict:
    return {f"ingest_{s}": ingest_source.with_options(name=f"ingest:{s}")(
                s, full_refresh, return_state=True)
            for s in sources}

Sources run sequentially: each needs the writer slot. The gain isn't speed but isolation: separate retries, timeouts, states and logs per provider.

Try it 58.3

Write three small unit tests for is_transient in tests/test_orchestration.py (run them with python -m pytest): a plain requests.ConnectionError → True; a RuntimeError raised from a ConnectionError → True; a RuntimeError raised from a class you name DataValidationError → False.

import requests

from orchestration.tasks.ingest import is_transient


class DataValidationError(Exception):
    pass


def test_plain_connection_error():
    assert is_transient(requests.ConnectionError("down"))


def test_wrapped_connection_error():
    try:
        try:
            raise requests.ConnectionError("down")
        except requests.ConnectionError as e:
            raise RuntimeError("pipeline step failed") from e
    except RuntimeError as wrapped:
        assert is_transient(wrapped)


def test_contract_violation_never_retried():
    try:
        try:
            raise DataValidationError("column frozen")
        except DataValidationError as e:
            raise RuntimeError("normalize failed") from e
    except RuntimeError as wrapped:
        assert not is_transient(wrapped)

58.4 dbt In-Process: dbtRunner with Model-Level Results (25 min)

dbt can be invoked from Python with dbtRunner. Unlike a subprocess, it returns structured results: every model, test, seed and snapshot with its status, timing and message. It also accepts callbacks that receive dbt's log events, which we forward to Prefect with the right level.

Create orchestration/tasks/dbt.py:

"""dbt as a first-class Prefect task (in-process dbtRunner)."""
import os
from contextlib import chdir

from dbt.cli.main import dbtRunner
from prefect import get_run_logger, task
from prefect.artifacts import create_table_artifact

from orchestration.settings import DBT_DIR
from orchestration.tasks.locks import duckdb_access


@task(timeout_seconds=1800)
def dbt_invoke(args: list[str], writes: bool = True) -> dict:
    """Run a dbt command in-process; forward logs; return node-level results."""
    logger = get_run_logger()

    def forward(event):
        level, msg = event.info.level, event.info.msg
        if level == "error":
            logger.error(msg)
        elif level == "warn":
            logger.warning(msg)
        elif level == "info":
            logger.info(msg)

    with duckdb_access(write=writes), chdir(DBT_DIR):
        res = dbtRunner(callbacks=[forward]).invoke(args)

    nodes = []
    for r in getattr(res.result, "results", None) or []:
        nodes.append({
            "node": r.node.name,
            "type": str(r.node.resource_type),
            "status": str(r.status),
            "seconds": round(r.execution_time or 0, 2),
            "message": (r.message or "")[:150],
        })
    problems = [n for n in nodes if n["status"] not in ("success", "pass", "RunStatus.Success",
                                                        "TestStatus.Pass")]
    if nodes:
        create_table_artifact(key="dbt-node-results", table=nodes,
                              description=f"dbt {' '.join(args)}")
    for p in problems:
        logger.warning(f"dbt {p['type']} {p['node']}: {p['status']} — {p['message']}")

    if not res.success:
        raise RuntimeError(f"dbt {' '.join(args)} failed: "
                           f"{len(problems)} problem node(s); {res.exception or ''}")
    return {"success": True, "nodes": len(nodes), "problems": len(problems)}

Notes:

• chdir(DBT_DIR) (Python 3.11+) runs dbt from the project folder, exactly as on the command line, so relative paths in profiles.yml (../macro.duckdb) and in the export post-hook (../data/marts) still resolve correctly. It restores the previous directory afterwards.
• The dbt-node-results artifact gives one row per model and test, so "which test failed last night?" is answered in the UI.
• Status strings differ slightly between dbt versions, which is why the problems filter accepts several spellings. Print nodes once to see yours.

Use it:

dbt_invoke.with_options(name="dbt:deps")(["deps", "--quiet"], writes=False)
dbt_invoke.with_options(name="dbt:build")(["build", "--target", "prod"])
dbt_invoke.with_options(name="dbt:docs")(["docs", "generate", "--target", "prod"], writes=False)

58.5 Targeted Rebuilds: Run dbt Only Downstream of What Changed (25 min)

Day 55's fingerprint was all-or-nothing: any change triggered a full dbt build. But if only the LFS table was republished, there's no need to rebuild CPI, GDP or open-economy models. dbt's graph selectors (Day 31) can express exactly "everything downstream of this source":

dbt build --select source:statcan.lfs_full+

Per-source fingerprints

Extend orchestration/tasks/fingerprints.py so the fingerprint is a dictionary per dbt source:

SOURCE_QUERIES = {
    "source:boc.observations+":
        "select bit_xor(hash(series_id, date, value)) from raw_dlt.boc_observations",
    "source:fred.observations+":
        "select bit_xor(hash(series_id, date, value)) from raw_dlt.fred_observations",
    "source:statcan.observations+":
        "select bit_xor(hash(series_id, date, value)) from raw_dlt.statcan_observations",
    "source:statcan.cpi_full+":
        "select max(release_time)::varchar from raw_dlt.statcan_releases where product_id = '18100004'",
    "source:statcan.gdp_full+":
        "select max(release_time)::varchar from raw_dlt.statcan_releases where product_id = '36100104'",
    "source:statcan.lfs_full+":
        "select max(release_time)::varchar from raw_dlt.statcan_releases where product_id = '14100287'",
}


@task(cache_policy=None)
def source_fingerprints() -> dict:
    """{dbt selector: content/release fingerprint} plus the dbt code fingerprint."""
    fps = {}
    with duckdb_access(write=False):
        with duckdb.connect(ENV["MACRO_DUCKDB_PATH"], read_only=True) as con:
            for selector, sql in SOURCE_QUERIES.items():
                fps[selector] = str(con.sql(sql).fetchone()[0])
    fps["__code__"] = code_fingerprint()        # commit + uncommitted diff of macro_dbt (Day 55)
    return fps

(Move the Git part of Day 55's warehouse_fingerprint into a small code_fingerprint() helper.)

Deciding the selection

def plan_dbt_selection(current: dict, last: dict | None, force: bool) -> list[str] | None:
    """None = skip dbt; [] = full build; otherwise the list of selectors to build."""
    if force or not last or current.get("__code__") != last.get("__code__"):
        return []                                         # code changed (or first run): full build
    changed = [sel for sel, fp in current.items()
               if sel != "__code__" and fp != last.get(sel)]
    return changed or None

Why do code changes force a full build? A changed macro (Day 35) can affect any model. Rather than try to compute precisely which models a code change affects, play it safe. dbt's state:modified selector can do that precisely in CI setups, but it needs saved artifacts from the previous build. That's worth exploring later.

Try it 58.5

With dbt ls --select source:statcan.lfs_full+ --resource-type model, list what a new LFS release rebuilds. Does it include macro_monthly? Should it? Check your lineage graph (Day 40) and explain.

58.6 Mini-Project: extract_load and transform v2 (15 min)

Assemble the new transform subflow:

from prefect import flow, get_run_logger
from prefect.states import Completed
from prefect.variables import Variable

from orchestration.tasks.dbt import dbt_invoke
from orchestration.tasks.fingerprints import plan_dbt_selection, source_fingerprints


@flow(name="transform")
def transform_flow(target: str = "prod", force: bool = False) -> dict:
    logger = get_run_logger()
    var_name = f"dbt_last_fingerprints_{target}"
    current = source_fingerprints()
    selection = plan_dbt_selection(current, Variable.get(var_name, default=None), force)

    if selection is None:
        logger.info("No dbt inputs changed — skipping dbt")
        return {"dbt_build": Completed(name="Skipped", message="inputs unchanged")}

    select_args = ["--select", *selection] if selection else []
    logger.info(f"dbt build {'(full)' if not selection else ' '.join(selection)}")
    states = {
        "dbt_deps": dbt_invoke.with_options(name="dbt:deps")(["deps", "--quiet"], writes=False,
                                                              return_state=True)}
    states["dbt_build"] = dbt_invoke.with_options(name="dbt:build")(
        ["build", "--target", target, *select_args], return_state=True)
    if states["dbt_build"].is_completed():
        Variable.set(var_name, current, overwrite=True)          # markers after success
        states["dbt_docs"] = dbt_invoke.with_options(name="dbt:docs")(
            ["docs", "generate", "--target", target], writes=False, return_state=True)
    return states

Note the Variable name now includes the target, so dev and prod fingerprints never mix (Day 60 uses this).

Swap extract_load_flow and transform_flow into daily_refresh (keeping Day 53's outcome() policy logic and Day 57's history and hooks). Run it twice and compare the UI: per-source ingestion tasks, a dbt-node-results artifact, and on the second run a Skipped or targeted dbt build.

Day 58 checklist

• I can decide which steps to integrate in-process and which to keep as subprocesses, and I know the DuckDB-connection cost
• Ingestion logic is a library (runner.py); ingest.py is a thin CLI over it
• Each source is its own task, with retries only for transient exception chains
• dbt runs in-process with forwarded logs and a node-level results artifact
• dbt rebuilds only downstream of sources whose content or release changed; code changes force a full build

python -m pytest -q
git add . && git commit -m "Day 58: per-source dlt tasks, dbtRunner, targeted dbt selection" && git push