Python for Data Engineering

Functions

What

Reusable blocks of code that take input and (usually) return output.

Where

Anywhere you'd otherwise copy-paste the same logic more than once.

When

As soon as you do the same thing twice — wrap it in a function.

How

def name(params): ... return value.

def clean_price(raw_price: str) -> float:
    """Convert '$1,299.00' -> 1299.0"""
    return float(raw_price.replace('$', '').replace(',', ''))

def load_orders(*, source: str, batch_size: int = 500):
    # keyword-only args (after *) make call sites self-documenting
    ...

• Type hints (-> float, : str) aren't enforced at runtime but massively help readability and IDE support in pipeline code.
• *args, **kwargs, default values, and keyword-only args (*,) — know all four; ETL functions often take many optional parameters.
• Generators (yield) matter for DE specifically — process large files/row batches without loading everything into memory.

Worked example — the difference a generator makes when reading a large file:

# regular function: builds the entire list in memory before returning anything
def read_all_rows(path):
    rows = []
    with open(path) as f:
        for line in f:
            rows.append(line.strip().split(","))
    return rows   # a 5GB file means a 5GB+ list sitting in RAM

# generator: yields one row at a time, nothing is held in memory beyond the current row
def read_rows(path):
    with open(path) as f:
        for line in f:
            yield line.strip().split(",")

# both are consumed the same way, but only the generator scales to arbitrarily large files
for row in read_rows("huge_orders.csv"):
    process(row)

Calling read_rows(path) doesn't run the function's code at all — it just hands back a generator object. The code only actually runs, one yield at a time, when something loops over it (a for loop, next(), or a list comprehension). This "wait until it's actually needed" behavior is the same core idea behind Spark's lazy DataFrames (Week 9-10).

OOP basics

What

A way to bundle data and behavior together into a “class,” so related code stays organized.

Where

Anywhere you're modeling a real “thing” — a data source, a pipeline stage, a record.

When

When plain functions start needing to share the same state, like a connection or config.

How

class Name: with an __init__ method to set up its data, and other methods for behavior.

class DataSource:
    def __init__(self, name: str, base_url: str):
        self.name = name
        self.base_url = base_url

    def fetch(self):
        raise NotImplementedError

class OrdersAPI(DataSource):
    def fetch(self):
        # override the parent method
        return requests.get(f"{self.base_url}/orders").json()

• You don't need deep OOP theory — just classes, __init__, inheritance, and dataclasses.
• @dataclass is the pragmatic way to model rows/records without boilerplate.
• Pipelines are often organized as classes: one per data source, or one per pipeline stage (Extractor, Transformer, Loader).

from dataclasses import dataclass

@dataclass
class Order:
    order_id: int
    customer_id: int
    total_amount: float

Worked example — inheritance in a real pipeline: two data sources that share a retry/logging pattern but differ in how they actually fetch data.

class DataSource:
    def __init__(self, name: str, base_url: str):
        self.name = name
        self.base_url = base_url

    def fetch(self):
        raise NotImplementedError   # forces every subclass to implement this

    def run(self):
        # shared behavior every subclass gets for free
        print(f"[{self.name}] starting fetch")
        data = self.fetch()
        print(f"[{self.name}] got {len(data)} records")
        return data

class OrdersAPI(DataSource):
    def fetch(self):
        return requests.get(f"{self.base_url}/orders").json()

class ProductsCSV(DataSource):
    def fetch(self):
        import csv
        with open(self.base_url) as f:
            return list(csv.DictReader(f))

for source in [OrdersAPI("orders", "https://api.example.com"), ProductsCSV("products", "products.csv")]:
    source.run()

Notice run() is written once, on the parent class, and never rewritten — both subclasses share that exact same logging/orchestration behavior for free. Each subclass only has to write the one method (fetch) that's actually different for them. This is the shape most "Extractor" classes take in a real pipeline: a base class holds the shared behavior, and each subclass just fills in its own specific detail.

A quick note on @dataclass: it writes __init__ and a couple of other useful methods for you automatically, based on the fields you listed. That's why Order(1, 7, 59.99) == Order(1, 7, 59.99) comes out True, and printing an instance actually shows its field values — a plain class won't do either of those unless you write the code by hand.

Exception handling

What

Python's way of handling things going wrong, without crashing the whole program.

Where

Around any code that might fail — network calls, file reads, bad data.

When

Anywhere something outside your control could fail (an API, a file, user input).

How

Wrap risky code in try:, handle specific problems in except:, and clean up in finally:.

import time

def fetch_with_retry(url, max_retries=3):
    for attempt in range(1, max_retries + 1):
        try:
            response = requests.get(url, timeout=10)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            if attempt == max_retries:
                raise
            time.sleep(2 ** attempt)  # exponential backoff

• Catch specific exceptions, never a bare except: — swallowing errors silently is how broken pipelines go unnoticed.
• try / except / else / finally — finally is where you close connections/files regardless of success.
• Retry-with-backoff is a core DE pattern for flaky APIs/network calls.

Custom exceptions — plain Exception tells a caller nothing about what actually went wrong; a small hierarchy of specific exception types lets calling code react differently to different failure modes:

class PipelineError(Exception):
    """Base class for every error this pipeline can raise."""

class ValidationError(PipelineError):
    """Raised when incoming data fails a data-quality check."""

class SourceUnavailableError(PipelineError):
    """Raised when the upstream API/source can't be reached at all."""

def validate(df):
    if df["total_amount"].lt(0).any():
        raise ValidationError("Found negative order amounts")
    return df

try:
    clean = validate(raw_df)
except ValidationError as e:
    logger.error("Data quality check failed: %s", e)
    send_alert(str(e))          # bad data — alert a human, don't crash silently
except SourceUnavailableError:
    logger.warning("Source down, will retry next scheduled run")
    # transient — Airflow's retry logic (Week 7-8) will handle this automatically

This is the same thinking Airflow uses to decide whether to retry a failed task. A SourceUnavailableError is temporary and worth retrying. A ValidationError means the data itself is bad — retrying won't fix that, it'll just fail the exact same way again. That one should alert a human instead.

Files

What

Reading and writing data stored on disk.

Where

Anywhere your pipeline needs to read a source file or write output.

When

Whenever data lives in a file instead of a database or API.

How

with open(path) as f: — always use "with" so the file closes automatically.

with open("data/raw_orders.txt", "r") as f:
    for line in f:
        process(line.strip())

# always use context managers ("with") so files close automatically,
# even if an exception is raised mid-read

• Always use with open(...) — never manually call .close() and risk leaking file handles.
• For huge files, iterate line-by-line instead of .read()ing the whole file into memory.
• pathlib.Path is the modern way to handle file paths over raw string concatenation.

from pathlib import Path

raw_dir = Path("data") / "raw" / "orders"          # joins paths with /, works on any OS
raw_dir.mkdir(parents=True, exist_ok=True)          # create the folder tree if missing

for file in raw_dir.glob("*.csv"):                  # iterate matching files
    print(file.name, file.stat().st_size)            # filename and size in bytes

# reading a huge file in fixed-size chunks instead of line-by-line (e.g. for binary data)
with open("large_export.bin", "rb") as f:
    while chunk := f.read(1024 * 1024):   # 1 MB at a time; walrus operator assigns + checks in one line
        process_chunk(chunk)

Path lets you join paths with /, and gives you the filename, extension, and parent folder without any manual string slicing. It also works the same way on macOS, Linux, and Windows — code that hardcodes / or \ instead tends to break the moment it runs on a different operating system. The chunked-read pattern (while chunk := f.read(size)) is the general fix for any file too big to load all at once that doesn't have convenient line breaks.

JSON / CSV

What

Reading and writing two of the most common data formats: JSON (nested) and CSV (flat rows).

Where

Anywhere data is exchanged between systems — API responses, exported reports.

When

JSON for structured/nested data (like an API response), CSV for simple tabular data.

How

json.load()/json.dumps() for JSON, csv.DictReader/DictWriter for CSV.

import json, csv

with open("orders.json") as f:
    orders = json.load(f)

with open("orders.csv", newline="") as f:
    reader = csv.DictReader(f)
    rows = [row for row in reader]

with open("orders_clean.csv", "w", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=["order_id", "total"])
    writer.writeheader()
    writer.writerows(rows)

• json.load/json.dump for files, json.loads/json.dumps for strings — mixing these up is a common bug.
• csv.DictReader/DictWriter give you column-name access instead of brittle index-based access.
• In practice you'll often reach for pandas (pd.read_csv/read_json) instead of raw csv/json once data needs transformation.

Worked example — real API responses are rarely flat. A nested JSON payload like this is extremely common:

order = {
    "order_id": 1042,
    "customer": {"id": 7, "name": "Asha Rao"},
    "items": [
        {"sku": "MOUSE-1", "qty": 2, "price": 25.0},
        {"sku": "DESK-1", "qty": 1, "price": 350.0}
    ]
}

# flatten one level of nesting manually
flat = {
    "order_id": order["order_id"],
    "customer_id": order["customer"]["id"],
    "customer_name": order["customer"]["name"],
    "item_count": len(order["items"]),
    "order_total": sum(i["qty"] * i["price"] for i in order["items"]),
}

# or let pandas do it for a whole batch of these at once
import pandas as pd
df = pd.json_normalize(orders_list, record_path="items", meta=["order_id", ["customer", "name"]])

json_normalize is built exactly for this. record_path tells it which nested list to turn into rows (here, items — one row per line item). meta tells it which parent-level fields to copy onto every one of those rows. This is probably the single most useful pandas function for turning messy, nested API responses into a flat table you can load straight into Postgres.

APIs — the requests library

What

A way for your code to ask another system for data over the internet.

Where

Any time your pipeline's data source is a web service instead of a file or database.

When

Whenever you need fresh data from an external system — a SaaS tool, a public API, another team's service.

How

requests.get(url, headers=..., params=..., timeout=...) then check the response and read .json().

import requests

response = requests.get(
    "https://api.example.com/orders",
    headers={"Authorization": f"Bearer {API_TOKEN}"},
    params={"page": 1, "page_size": 100},
    timeout=10,
)
response.raise_for_status()
data = response.json()

# pagination pattern
all_orders = []
page = 1
while True:
    resp = requests.get(url, params={"page": page}, timeout=10).json()
    if not resp["results"]:
        break
    all_orders.extend(resp["results"])
    page += 1

• Always set a timeout — an API call with no timeout can hang a pipeline forever.
• Always call raise_for_status() (or check status_code) — don't assume 200.
• Know the difference between query params, headers, and JSON body (json= vs data= in POST requests).
• This is exactly the pattern behind your Week 3 "REST API → Python → Transform → PostgreSQL" build.

Auth patterns you'll actually see — most APIs use one of three schemes, and mixing them up is a common source of 401 errors:

# 1. API key in a header (most common for simple third-party APIs)
requests.get(url, headers={"X-API-Key": API_KEY})

# 2. Bearer token (OAuth2 access token, JWT)
requests.get(url, headers={"Authorization": f"Bearer {access_token}"})

# 3. Basic auth (username/password, base64-encoded automatically by requests)
requests.get(url, auth=(username, password))

# a session reuses the same auth + connection across many calls — faster and less code
session = requests.Session()
session.headers.update({"Authorization": f"Bearer {access_token}"})
resp1 = session.get(f"{base_url}/orders")
resp2 = session.get(f"{base_url}/customers")  # auth header already attached

Rate limiting is the other API reality worth planning for up front: most APIs return 429 Too Many Requests once you exceed their limit, often with a Retry-After header telling you exactly how long to wait.

import time

def get_with_rate_limit_handling(url, **kwargs):
    resp = requests.get(url, timeout=10, **kwargs)
    if resp.status_code == 429:
        wait = int(resp.headers.get("Retry-After", 5))
        time.sleep(wait)
        return get_with_rate_limit_handling(url, **kwargs)   # retry once after waiting
    resp.raise_for_status()
    return resp

Logging

What

A structured way to record what your program is doing, instead of using print().

Where

Throughout any pipeline — start/end of steps, warnings, errors.

When

Always, in any code that isn't a one-off scratch script.

How

logger = logging.getLogger(name), then logger.info(...) / .warning(...) / .error(...).

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
logger = logging.getLogger("etl.orders")

logger.info("Starting extraction, batch_size=%d", 500)
logger.warning("Retrying request after timeout")
logger.error("Failed to load batch", exc_info=True)

• Never use print() in pipeline code — logging gives you levels, timestamps, and the ability to route output (file, stdout, monitoring system).
• Levels: DEBUG < INFO < WARNING < ERROR < CRITICAL — set the threshold per environment.
• exc_info=True inside an except block logs the full traceback — essential for debugging failed pipeline runs.

Structured (JSON) logging — plain text logs are fine to read by eye, but once logs flow into CloudWatch (Week 5) or another log aggregator, structured JSON lines are far easier to search and alert on:

import logging, json

class JsonFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        })

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger = logging.getLogger("etl.orders")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

logger.info("Loaded batch")
# -> {"timestamp": "2026-08-18 10:03:12", "level": "INFO", "logger": "etl.orders", "message": "Loaded batch"}

Each log line is now its own small JSON object. A log tool can filter on level == "ERROR" or search logger == "etl.orders" directly, instead of trying to pattern-match free-form text. This is the format most real pipelines use once they're running somewhere other than your own laptop.

Virtual environments

What

An isolated space for a project's Python packages, separate from other projects.

Where

At the start of any new Python project.

When

Every time — avoid installing packages globally if you can.

How

python3 -m venv .venv, activate it, then pip install your packages.

python3 -m venv .venv
source .venv/bin/activate      # Windows: .venv\Scripts\activate
pip install pandas sqlalchemy psycopg2-binary requests
pip freeze > requirements.txt

# recreate elsewhere:
pip install -r requirements.txt

• Isolates project dependencies so different projects don't fight over package versions.
• requirements.txt (or pyproject.toml with a tool like Poetry/uv) makes the environment reproducible — critical once you're building Docker images in Week 6.

Secrets don't belong in code — pair a virtual environment with a .env file (never committed to Git) and python-dotenv to keep credentials out of your source entirely:

# .env  (add this filename to .gitignore!)
DB_PASSWORD=supersecret
API_TOKEN=abc123

# pipeline.py
from dotenv import load_dotenv
import os

load_dotenv()   # reads .env into the process environment
db_password = os.environ["DB_PASSWORD"]
api_token = os.environ["API_TOKEN"]

Real production systems work the same way — they just swap the .env file for something like AWS Secrets Manager, or environment variables set by Airflow/Docker. The pipeline code itself (os.environ[...]) never changes between your laptop and production — only where those values actually come from changes.

pandas Library

What

A Python library for working with tabular data (rows and columns) in memory.

Where

Anywhere you need to clean, reshape, or summarize data that fits on one machine.

When

For small-to-medium datasets, before or instead of using a database or Spark.

How

pd.read_csv()/read_json() to load, then filter/groupby/merge to transform.

import pandas as pd

df = pd.read_json("orders.json")
df["order_date"] = pd.to_datetime(df["order_date"])
df = df.dropna(subset=["customer_id"])
df["total_amount"] = df["total_amount"].astype(float)

summary = (
    df.groupby("customer_id")
      .agg(order_count=("order_id", "count"), revenue=("total_amount", "sum"))
      .reset_index()
)

df.to_csv("orders_clean.csv", index=False)

• Core operations for DE: read_csv/read_json/read_sql, dropna/fillna, astype, groupby, merge.
• merge() is pandas' JOIN — know how="inner"/"left"/"outer" just like SQL.
• Vectorize operations (whole-column operations) instead of looping row by row — loops in pandas are slow.

Worked example — merge (pandas' JOIN):

orders = pd.DataFrame({
    "order_id": [1, 2, 3],
    "customer_id": [1, 1, 2],
    "total_amount": [59.99, 12.50, 350.00],
})
customers = pd.DataFrame({
    "customer_id": [1, 2, 3],
    "customer_name": ["Asha", "Ravi", "Meera"],
})

merged = orders.merge(customers, on="customer_id", how="left")
#    order_id  customer_id  total_amount customer_name
# 0         1            1         59.99          Asha
# 1         2            1         12.50          Asha
# 2         3            2        350.00          Ravi

# customer 3 (Meera) never appears — how="left" keeps every row from `orders`, the left frame,
# just like a SQL LEFT JOIN keeps every row from the left table

Vectorization, concretely — this is the single biggest pandas performance mistake beginners make:

# slow: Python-level loop, re-evaluates row by row (~1000x slower on large data)
for i in range(len(df)):
    df.loc[i, "total_with_tax"] = df.loc[i, "total_amount"] * 1.08

# fast: vectorized — pandas applies the operation to the whole column at once in C
df["total_with_tax"] = df["total_amount"] * 1.08

# when logic is too complex for a single expression, np.where / np.select beat .apply()
import numpy as np
df["tier"] = np.select(
    [df["total_amount"] >= 500, df["total_amount"] >= 100],
    ["high", "medium"],
    default="low",
)

.apply() with a Python function is still going row-by-row behind the scenes, even though it looks tidy. Only reach for it when there's genuinely no vectorized way to do it — np.where/np.select already cover most conditional-column situations, and run much faster.

SQLAlchemy

What

A Python library that connects your code to a SQL database.

Where

Anywhere Python needs to read from or write to Postgres (or another database).

When

Whenever you're moving data between Python and a database.

How

create_engine(connection_string), then df.to_sql() or engine.connect() to run SQL.

from sqlalchemy import create_engine, text

engine = create_engine("postgresql+psycopg2://user:pass@localhost:5432/ecommerce")

# write a dataframe straight to Postgres
df.to_sql("orders_clean", engine, if_exists="append", index=False)

# raw SQL via SQLAlchemy Core
with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM orders WHERE total_amount > :amt"), {"amt": 100})
    for row in result:
        print(row)

• SQLAlchemy is the standard Python-to-database bridge — create_engine manages the connection pool.
• df.to_sql(...) is the fastest path from a pandas transform to a Postgres table.
• You don't need the full ORM (declarative models) for DE work initially — Core (engine + text()) covers most pipeline needs.

Transactions with SQLAlchemy — engine.connect() alone doesn't commit automatically; wrap writes in engine.begin() so a failure partway through rolls back everything, the same atomicity guarantee from Week 1-2:

with engine.begin() as conn:   # begin() auto-commits on success, auto-rolls-back on exception
    conn.execute(text("DELETE FROM staging_orders WHERE batch_id = :bid"), {"bid": batch_id})
    conn.execute(
        text("INSERT INTO staging_orders (order_id, total_amount, batch_id) VALUES (:oid, :amt, :bid)"),
        [{"oid": 1, "amt": 59.99, "bid": batch_id}, {"oid": 2, "amt": 12.50, "bid": batch_id}],
    )
# if the INSERT raises an exception, the DELETE above is rolled back too — nothing is left half-done

This "delete, then insert, all in one transaction" pattern is how you make a load safe to re-run for a given batch_id (Week 4's idempotency idea) — it always clears out its own previous attempt first, before writing anything new.

psycopg2

What

The lower-level Python driver specifically for PostgreSQL.

Where

Direct database access when you need more control than SQLAlchemy gives you.

When

For fast bulk loads, or when you need raw cursor-level control.

How

psycopg2.connect(...), open a cursor, execute SQL, then commit().

import psycopg2
from psycopg2.extras import execute_values

conn = psycopg2.connect(host="localhost", dbname="ecommerce", user="postgres", password="postgres")
cur = conn.cursor()

cur.execute("SELECT customer_id, total_amount FROM orders LIMIT 5;")
rows = cur.fetchall()

execute_values(
    cur,
    "INSERT INTO orders_clean (order_id, total_amount) VALUES %s",
    [(1, 99.5), (2, 150.0)],
)
conn.commit()
cur.close()
conn.close()

• The lower-level Postgres driver — SQLAlchemy actually uses psycopg2 under the hood for Postgres connections.
• execute_values batches inserts — far faster than looping single-row INSERTs.
• Remember: conn.commit() is required or your writes won't persist; always close cursor/connection (or use with blocks).

Bulk-loading large files fast — for genuinely large loads, Postgres's COPY command (exposed via copy_expert) is an order of magnitude faster than any row-by-row INSERT, because it streams data directly instead of executing one SQL statement per row:

with open("orders_clean.csv") as f, conn.cursor() as cur:
    cur.copy_expert(
        "COPY orders_clean (order_id, customer_id, total_amount) FROM STDIN WITH CSV HEADER",
        f,
    )
conn.commit()

And for reading results too large to fit in memory, a server-side (named) cursor streams rows from Postgres in batches instead of pulling the entire result set into your Python process at once:

with conn.cursor(name="server_side_cursor") as cur:   # naming the cursor makes it server-side
    cur.itersize = 2000                                # fetch 2000 rows at a time under the hood
    cur.execute("SELECT * FROM orders")
    for row in cur:            # iterates lazily, never loads all rows into memory at once
        process(row)

In real code, prefer with open(...) as f, conn.cursor() as cur:-style blocks (or a try/finally) over the plain open/close calls shown at the top of this section — those were kept simple on purpose, just to show the core API clearly.

Practice project

Build the pipeline: REST API → Python → Transform → PostgreSQL

What

A small pipeline that pulls data from an API and loads it into Postgres.

Where

Your own machine, using everything from this page together.

When

After you're comfortable with each piece individually — this ties them together.

How

extract() with requests, transform() with pandas, load() with SQLAlchemy.

import requests, logging
import pandas as pd
from sqlalchemy import create_engine

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("pipeline")

def extract(url: str) -> pd.DataFrame:
    resp = requests.get(url, timeout=10)
    resp.raise_for_status()
    return pd.DataFrame(resp.json())

def transform(df: pd.DataFrame) -> pd.DataFrame:
    df["order_date"] = pd.to_datetime(df["order_date"])
    df = df.dropna(subset=["customer_id"])
    df["total_amount"] = df["total_amount"].astype(float)
    return df

def load(df: pd.DataFrame, engine, table: str):
    df.to_sql(table, engine, if_exists="append", index=False)
    logger.info("Loaded %d rows into %s", len(df), table)

if __name__ == "__main__":
    engine = create_engine("postgresql+psycopg2://postgres:postgres@localhost:5432/ecommerce")
    raw = extract("https://api.example.com/orders")
    clean = transform(raw)
    load(clean, engine, "orders_clean")

Extending it with retry, validation, and idempotency — combining every concept from this page into one small, realistic pipeline:

import time
import requests, logging
import pandas as pd
from sqlalchemy import create_engine, text

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
logger = logging.getLogger("pipeline")

class ValidationError(Exception):
    pass

def extract(url: str, max_retries: int = 3) -> pd.DataFrame:
    for attempt in range(1, max_retries + 1):
        try:
            resp = requests.get(url, timeout=10)
            resp.raise_for_status()
            return pd.DataFrame(resp.json())
        except requests.exceptions.RequestException as e:
            logger.warning("Attempt %d/%d failed: %s", attempt, max_retries, e)
            if attempt == max_retries:
                raise
            time.sleep(2 ** attempt)

def transform(df: pd.DataFrame) -> pd.DataFrame:
    df["order_date"] = pd.to_datetime(df["order_date"])
    df = df.dropna(subset=["customer_id"])
    df["total_amount"] = df["total_amount"].astype(float)
    return df

def validate(df: pd.DataFrame) -> pd.DataFrame:
    if df["total_amount"].lt(0).any():
        raise ValidationError("Found negative order amounts")
    if df["order_id"].duplicated().any():
        raise ValidationError("Found duplicate order_id values")
    return df

def load(df: pd.DataFrame, engine, table: str, batch_id: str):
    with engine.begin() as conn:                                       # idempotent: same batch_id always replaces cleanly
        conn.execute(text(f"DELETE FROM {table} WHERE batch_id = :bid"), {"bid": batch_id})
        df["batch_id"] = batch_id
        df.to_sql(table, conn, if_exists="append", index=False)
    logger.info("Loaded %d rows into %s (batch=%s)", len(df), table, batch_id)

if __name__ == "__main__":
    engine = create_engine("postgresql+psycopg2://postgres:postgres@localhost:5432/ecommerce")
    batch_id = time.strftime("%Y-%m-%d")
    try:
        raw = extract("https://api.example.com/orders")
        clean = validate(transform(raw))
        load(clean, engine, "orders_clean", batch_id)
    except ValidationError as e:
        logger.error("Data quality check failed, not loading: %s", e)
        raise

Everything from this page shows up here, doing real work: extract retries with backoff (Exceptions), validate raises a custom exception on bad data (Exceptions/OOP), load wraps a delete-and-insert in one transaction keyed by batch_id so re-running the same day never double-loads it (SQLAlchemy transactions, Week 4 idempotency), and every step logs instead of printing (Logging).