Flows and Tasks

51.1 Orchestration Concepts (20 min)

Workflows as Python, not as diagrams

Some orchestrators (Airflow is the best known) make you declare a static graph of tasks up front. Prefect builds the graph dynamically from your Python code: if you call task B with the output of task A, B depends on A. Loops, if statements and function calls work as usual.

@flow
def refresh():
    data = extract()          # task A
    clean = transform(data)   # task B: depends on A through its input
    if clean.shape[0] > 0:    # ordinary Python control flow
        publish(clean)        # task C

For economists who already write Python, that's the big attraction: orchestration looks like the code you already write.

What Prefect records for you

Every time a flow runs, Prefect records who ran what, with which parameters, when, for how long, in what order, what was logged, and how it ended. That's the audit trail a data team needs when a dashboard number is questioned three weeks later.

51.2 Installing Prefect and Starting the UI (23 min)

Prefect goes into the main environment, next to dlt and dbt, because it will call their Python code directly. GX stays isolated in .venv-gx and will be called as a separate process (Day 50's pattern).

cd /workspaces/macro-data-course
source .venv/bin/activate
pip install -U prefect
pip freeze > requirements.txt
prefect version

Version note. This course uses Prefect 3.x. Prefect 1 and 2 had different APIs (e.g. Prefect 1's Flow context manager), and many online examples still use them. Check the version any example targets before copying it.

Start a local Prefect server

The Prefect server stores run history and serves the UI. Open a second terminal (it will keep running):

source .venv/bin/activate
prefect server start --host 0.0.0.0

Codespaces will offer to open port 4200. In your first terminal, tell Prefect where the API is:

prefect config set PREFECT_API_URL="http://127.0.0.1:4200/api"

If the UI loads in your browser but shows "can't connect to the API", tell the UI to use the Codespaces forwarded address (Codespaces provides these variables automatically), then restart the server:

prefect config set PREFECT_UI_API_URL="https://${CODESPACE_NAME}-4200.${GITHUB_CODESPACES_PORT_FORWARDING_DOMAIN}/api"

Without a server: if PREFECT_API_URL isn't set, Prefect starts a temporary server for each run. The flow works, but history disappears afterwards. Keep the server terminal running during these three days.

Project skeleton

mkdir -p orchestration/tasks orchestration/flows
touch orchestration/__init__.py orchestration/tasks/__init__.py orchestration/flows/__init__.py

We'll run flows as modules from the repository root, e.g. python -m orchestration.flows.hello, so that imports like from orchestration.settings import ... work.

51.3 Your First Flow (20 min)

Create orchestration/flows/hello.py:

from prefect import flow


@flow(log_prints=True)
def hello_macro(target: float = 2.0):
    """A first flow: compare a real-rate calculation with the inflation target."""
    overnight, cpi_yoy = 2.25, 3.0            # values from the course notebook
    real_rate = overnight - cpi_yoy
    print(f"Real policy rate: {real_rate:+.2f}% (inflation target {target}%)")
    return real_rate


if __name__ == "__main__":
    hello_macro()

python -m orchestration.flows.hello

Look at the terminal output: besides your print, Prefect logged that it created a flow run (with a random, memorable name like "cerulean-otter"), and that the run finished in state Completed.

Open the UI (port 4200) → Runs. Click your run: you'll see its parameters (target: 2.0), logs, duration and final state.

What the decorator added:

log_prints=True makes every print() inside the flow a log line, which is handy for learning. For production code we'll use a proper logger (51.6).

Make it fail

Add raise ValueError("CPI data missing") before the return, run again, and look at the run in the UI: state Failed, with the error message and traceback attached. Remove the line again.

51.4 Tasks: The Observable Steps Inside a Flow (25 min)

A task is a function decorated with @task, called from inside a flow. Each call becomes a task run with its own state, logs and timing.

Create orchestration/flows/cpi_steps.py:

import time

from prefect import flow, task


@task
def fetch_cpi_levels() -> list[float]:
    """Illustrative CPI levels (Day 1). Tomorrow we fetch real data."""
    time.sleep(1)                               # pretend this is a slow API call
    return [161.3, 161.8, 162.4, 162.6, 163.0, 163.4,
            163.9, 164.0, 163.7, 164.1, 164.3, 164.2, 166.1]


@task
def year_over_year(levels: list[float]) -> float:
    return 100 * (levels[-1] / levels[-13] - 1)


@task
def classify(yoy: float, target: float = 2.0) -> str:
    if yoy > target + 1:
        return "above the 1–3% control range"
    if yoy < target - 1:
        return "below the 1–3% control range"
    return "within the 1–3% control range"


@flow(log_prints=True)
def cpi_steps(target: float = 2.0):
    levels = fetch_cpi_levels()
    yoy = year_over_year(levels)
    verdict = classify(yoy, target)
    print(f"CPI inflation {yoy:.1f}% — {verdict}")


if __name__ == "__main__":
    cpi_steps()

python -m orchestration.flows.cpi_steps

In the UI, open the flow run and look at the timeline (or "graph") view: three task runs in sequence, each with its own duration. fetch_cpi_levels takes about a second; the others are instant.

How dependencies arise

You never declared "year_over_year runs after fetch_cpi_levels". Prefect saw that year_over_year received the output of fetch_cpi_levels and recorded the dependency. That's the dynamic graph from 51.1.

What makes a good task?

Rule of thumb: one task per external system interaction or meaningful stage: one API, one database write, one tool invocation. Pure calculations inside a stage can stay ordinary functions.

Try it 51.4

Add a fourth task, real_rate(overnight: float, yoy: float) -> float, and make the flow print the real policy rate, using 2.25 as the overnight rate. In the UI, which tasks does real_rate depend on?

@task
def real_rate(overnight: float, yoy: float) -> float:
    return overnight - yoy

# inside the flow, after computing yoy:
    r = real_rate(2.25, yoy)
    print(f"Real policy rate: {r:+.2f}%")

51.5 Parameters and Type Hints (18 min)

A flow's function arguments are its parameters. Prefect records them for every run and validates them against the type hints, converting where it safely can.

from datetime import date

from prefect import flow


@flow(log_prints=True)
def parameter_demo(series_ids: list[str], start_date: date = date(2006, 1, 1),
                   lookback_days: int = 14):
    print(f"series={series_ids} start={start_date!r} lookback={lookback_days}")


if __name__ == "__main__":
    parameter_demo(["V39079", "FXCADUSD"], start_date="2024-01-01", lookback_days="30")

Run it and look at the output: start_date arrived as the text "2024-01-01" but was converted to a date, and "30" became the integer 30. Invalid values fail before any work starts:

parameter_demo(["V39079"], start_date="not a date")    # the run fails immediately, with a clear error

Why this matters:

• Every run documents its inputs. "Which start date did the backfill use?" is answered in the UI.
• Bad inputs fail fast, before touching APIs or the warehouse.
• Later, deployments (Days 54–57) let you run flows with different parameters from the UI or a schedule. Type hints then generate the input form automatically.

Parameters should be small, serialisable values (text, numbers, dates, lists, dictionaries). Don't pass DataFrames or database connections as flow parameters; pass where to find the data instead.

51.6 Logging (10 min)

print with log_prints=True is fine for learning. In real flows, use Prefect's run logger, which attaches each message to the right flow or task run, with a level:

from prefect import flow, get_run_logger, task


@task
def check_freshness(age_days: int, tolerance: int) -> bool:
    logger = get_run_logger()
    if age_days > tolerance:
        logger.warning(f"Series is {age_days} days old (tolerance {tolerance})")
        return False
    logger.info(f"Series fresh ({age_days} days old)")
    return True


@flow
def logging_demo():
    logger = get_run_logger()
    logger.info("Checking freshness of two series")
    check_freshness(3, 10)
    check_freshness(45, 10)


if __name__ == "__main__":
    logging_demo()

In the UI, filter the run's logs by level to see only warnings. That's the same level scheme as Day 9's logging module: DEBUG, INFO, WARNING, ERROR, CRITICAL.

51.7 Mini-Project: A CPI Briefing Flow (20 min)

Create orchestration/flows/cpi_briefing.py: a flow that fetches real data and produces a short inflation briefing.

Requirements:

1. Parameter latest_n: int = 25: how many months to fetch.
2. Task fetch_cpi(latest_n): POST to StatCan's getDataFromVectorsAndLatestNPeriods for vector 41690973 (CPI All-items, Day 10) and return a list of (date, value) pairs.
3. Task compute_measures(points): return a dictionary with the latest month, YoY inflation, the YoY one year earlier, and the change in YoY (acceleration, Day 19).
4. Task briefing(measures, target=2.0): return a two-sentence text, logged at INFO.
5. Use get_run_logger() (no print).
6. Run it twice, once with latest_n=13 and once with the default, and compare the runs' parameters in the UI.

"""A small real-data flow: CPI inflation briefing from the StatCan WDS API."""
from datetime import date

import requests
from prefect import flow, get_run_logger, task

WDS = "https://www150.statcan.gc.ca/t1/wds/rest/getDataFromVectorsAndLatestNPeriods"


@task
def fetch_cpi(latest_n: int) -> list[tuple[date, float]]:
    r = requests.post(WDS, json=[{"vectorId": 41690973, "latestN": latest_n}], timeout=30)
    r.raise_for_status()
    item = r.json()[0]
    if item.get("status") != "SUCCESS":
        raise RuntimeError(f"StatCan returned status {item.get('status')}")
    points = [(date.fromisoformat(dp["refPer"][:10]), float(dp["value"]))
              for dp in item["object"]["vectorDataPoint"] if dp.get("value") is not None]
    get_run_logger().info(f"Fetched {len(points)} CPI observations")
    return sorted(points)


@task
def compute_measures(points: list[tuple[date, float]]) -> dict:
    values = [v for _, v in points]
    if len(values) < 25:
        yoy_prev = None
    else:
        yoy_prev = 100 * (values[-13] / values[-25] - 1)
    yoy = 100 * (values[-1] / values[-13] - 1)
    return {"latest": points[-1][0].isoformat(), "yoy": yoy, "yoy_prev": yoy_prev,
            "accel": None if yoy_prev is None else yoy - yoy_prev}


@task
def briefing(m: dict, target: float = 2.0) -> str:
    position = ("above" if m["yoy"] > target + 1 else
                "below" if m["yoy"] < target - 1 else "within")
    text = (f"CPI inflation was {m['yoy']:.1f}% in {m['latest'][:7]}, "
            f"{position} the Bank of Canada's 1–3% control range.")
    if m["accel"] is not None:
        direction = "higher" if m["accel"] > 0 else "lower"
        text += f" That is {abs(m['accel']):.1f}pp {direction} than a year earlier."
    get_run_logger().info(text)
    return text


@flow(name="cpi-briefing")
def cpi_briefing(latest_n: int = 25, target: float = 2.0) -> str:
    points = fetch_cpi(latest_n)
    measures = compute_measures(points)
    return briefing(measures, target)


if __name__ == "__main__":
    cpi_briefing()

Day 51 checklist

• I can explain flows, tasks, runs, states, parameters and subflows
• Prefect is installed, the local server runs, and I can browse runs in the UI from Codespaces
• I can turn functions into flows and tasks, and read the dynamic dependency graph
• I choose sensible task boundaries (one external interaction or meaningful stage per task)
• Flow parameters are typed, validated and recorded
• I log with get_run_logger() at appropriate levels

git add orchestration requirements.txt
git commit -m "Day 51: Prefect flows, tasks, parameters, logging"
git push