Prefect in practice

Define a two-task Prefect flow, run it locally, and observe task states — all in ordinary Python without any external infrastructure.

The previous lesson explained Prefect's flow/task model. Here you will write one. Because Prefect is not available in the in-browser runtime, the demo simulates its decorator behaviour using a lightweight wrapper — the structure and output mirror exactly what you would see running real Prefect locally.

Python — editable, runs in your browser

import time, random

# ── Prefect simulator (replicates local-run output) ──────────────────────────
_task_states = []

class TaskResult:
  def __init__(self, name, result):
      self.name = name
      self.result = result

def task(fn=None, retries=0, retry_delay_seconds=0):
  """Simulated @task decorator."""
  def decorator(f):
      def wrapper(*args, **kwargs):
          attempts = 0
          max_attempts = retries + 1
          while attempts < max_attempts:
              try:
                  attempts += 1
                  result = f(*args, **kwargs)
                  _task_states.append((f.__name__, "Completed"))
                  print(f"  [task:{f.__name__}] Completed")
                  return result
              except Exception as exc:
                  if attempts < max_attempts:
                      _task_states.append((f.__name__, f"Retrying ({attempts}/{retries})"))
                      print(f"  [task:{f.__name__}] Failed ({exc}), retrying in {retry_delay_seconds}s")
                      time.sleep(retry_delay_seconds * 0.01)  # speed up for demo
                  else:
                      _task_states.append((f.__name__, "Failed"))
                      print(f"  [task:{f.__name__}] Failed after {retries} retries: {exc}")
                      raise
      wrapper.__name__ = f.__name__
      return wrapper
  return decorator(fn) if fn is not None else decorator

def flow(fn=None, name=None):
  """Simulated @flow decorator."""
  def decorator(f):
      def wrapper(*args, **kwargs):
          flow_name = name or f.__name__
          print(f"Flow run '{flow_name}' starting")
          start = time.time()
          try:
              result = f(*args, **kwargs)
              elapsed = time.time() - start
              print(f"Flow run '{flow_name}' completed in {elapsed:.3f}s")
              return result
          except Exception as exc:
              elapsed = time.time() - start
              print(f"Flow run '{flow_name}' FAILED in {elapsed:.3f}s: {exc}")
              raise
      return wrapper
  return decorator(fn) if fn is not None else decorator

# ── Pipeline definition ──────────────────────────────────────────────────────
_fetch_attempts = 0

@task(retries=2, retry_delay_seconds=1)
def fetch_data(url: str) -> list:
  global _fetch_attempts
  _fetch_attempts += 1
  if _fetch_attempts <= 1:
      raise ConnectionError("connection reset")  # transient; will be retried
  print(f"    fetched 5 records from {url}")
  return [{"id": i, "value": i * 10} for i in range(1, 6)]

@task
def process_data(records: list) -> list:
  filtered = [r for r in records if r["value"] >= 20]
  print(f"    filtered to {len(filtered)} records (value >= 20)")
  return filtered

@flow(name="example-pipeline")
def pipeline(url: str = "https://api.example.com/data") -> list:
  raw     = fetch_data(url)
  cleaned = process_data(raw)
  return cleaned

# ── Run ──────────────────────────────────────────────────────────────────────
result = pipeline()
print()
print("Final result:", result)
print()
print("Task state log:")
for name, state in _task_states:
  print(f"  {name}: {state}")

Run the snippet and observe:

fetch_data fails on its first attempt and retries automatically.
process_data starts only after fetch_data succeeds — Prefect inferred the dependency from the fact that process_data receives fetch_data's return value.
The flow prints a completion time and the task state log shows the retry.

Real Prefect code

With Prefect installed (pip install prefect), the real version of this pipeline is nearly identical:

from prefect import flow, task
import requests

@task(retries=2, retry_delay_seconds=10, log_prints=True)
def fetch_data(url: str) -> list:
    response = requests.get(url, timeout=30)
    response.raise_for_status()
    records = response.json()
    print(f"Fetched {len(records)} records")
    return records

@task(log_prints=True)
def process_data(records: list) -> list:
    filtered = [r for r in records if r["value"] >= 20]
    print(f"Filtered to {len(filtered)} records")
    return filtered

@flow(name="example-pipeline", log_prints=True)
def pipeline(url: str = "https://api.example.com/data") -> list:
    raw = fetch_data(url)
    return process_data(raw)

if __name__ == "__main__":
    pipeline()

log_prints=True routes print() calls to the Prefect logger, making them visible in the Prefect UI. Run with python pipeline.py or schedule it via prefect deploy.

Prefect infers task dependencies from data flow. If you want two tasks to run in parallel, call them without passing the output of one to the other — Prefect will submit both immediately and wait for both to complete before proceeding.

Parametrised runs

Because the flow accepts url as a parameter, you can run it against different endpoints without touching the code:

# CLI (requires prefect deployment)
prefect flow-run create example-pipeline --param url=https://staging.api.example.com/data

Or from Python:

pipeline(url="https://staging.api.example.com/data")

This makes testing against staging trivial and production runs auditable — the parameters used for each run are stored in the Prefect run record.

Where to go next

Next: lab — DAG pipeline — express a four-step pipeline as both a Makefile and a Prefect flow, then compare the developer experience of each approach.

Finished reading? Mark it complete to track your progress.

Real Prefect code

Parametrised runs

Where to go next

On this page