Python Functions, Modules and Reusable Data Pipeline Code

Most pipeline scripts start as one long file that reads, transforms and writes in a single block. That works once and is painful afterwards: you cannot test the transformation without a database, and you cannot reuse it in another job. The fix is structure, not cleverness.

Separate reading, transforming and writing

Keep I/O at the edges and logic in the middle. A transformation function should take plain data in and return plain data out.

def normalise_order(raw: dict) -> dict:
    """Pure function: same input, same output, no I/O."""
    return {
        "order_id": int(raw["order_id"]),
        "customer": raw["customer"].strip().title(),
        "amount": round(float(raw["amount"]), 2),
    }

print(normalise_order({"order_id": "7", "customer": "  asha ", "amount": "19.999"}))

{'order_id': 7, 'customer': 'Asha', 'amount': 20.0}

Because normalise_order touches no files or databases, it is trivial to test and safe to reuse in a batch job, a streaming consumer or a notebook.

Pass configuration in, do not reach out for it

Functions that read global variables or environment variables deep inside are hard to test and surprising to call. Pass what they need as arguments, and read configuration once at the entry point.

from dataclasses import dataclass

@dataclass(frozen=True)
class LoadConfig:
    source_dir: str
    target_table: str
    batch_size: int = 1000

def plan_batches(row_count: int, config: LoadConfig) -> int:
    return (row_count + config.batch_size - 1) // config.batch_size

config = LoadConfig(source_dir="data/", target_table="orders")
print(plan_batches(2500, config))

3

A frozen dataclass documents every setting in one place and cannot be modified by accident halfway through a run.

Organise into modules

A small pipeline package might look like this:

orders_pipeline/
  __init__.py
  config.py        # LoadConfig, reading env vars or CLI args
  extract.py       # read files / APIs (I/O)
  transform.py     # pure functions such as normalise_order
  load.py          # write to the database (I/O)
  main.py          # wires them together
tests/
  test_transform.py

Each module has one reason to change. Tests import transform directly without touching files or credentials.

A thin entry point

def run(rows: list[dict]) -> list[dict]:
    return [normalise_order(r) for r in rows]

if __name__ == "__main__":
    print(run([{"order_id": "1", "customer": "ben", "amount": "5"}]))

The if __name__ == "__main__": guard means importing the module (from a test or an orchestrator) does not start the job. Only running it as a script does.

Pass dependencies in, too

The same idea applies to things a function talks to: a database connection, an HTTP fetch function, a clock. Pass them in, and production passes the real one while tests pass a fake:

from datetime import date

def build_report(fetch_orders, run_date: date) -> dict:
    """fetch_orders is any callable returning order dicts for a date."""
    orders = fetch_orders(run_date)
    return {"date": run_date.isoformat(), "orders": len(orders), "revenue": sum(o["amount"] for o in orders)}

fake_fetch = lambda d: [{"amount": 10.0}, {"amount": 5.5}]          # what a test would pass
print(build_report(fake_fetch, date(2026, 10, 5)))

{'date': '2026-10-05', 'orders': 2, 'revenue': 15.5}

Note that the run date is a parameter too. A function that calls date.today() inside gives a different answer on a rerun next week, which breaks retries and backfills.

Type hints

Type hints cost little and let editors and type checkers (mypy, pyright) catch mistakes such as passing a string where a number is expected, or forgetting that a value can be None, before the code runs. Python itself does not enforce them at run time.

from typing import TypedDict

class RawOrder(TypedDict):
    order_id: str
    customer: str
    amount: str

def parse(raw: RawOrder) -> dict[str, int | str | float]:
    return {"order_id": int(raw["order_id"]), "customer": raw["customer"].strip(), "amount": float(raw["amount"])}

def find_customer(orders: list[dict], name: str) -> dict | None:
    """Return the first order for a customer, or None. The `| None` forces callers to handle absence."""
    return next((o for o in orders if o["customer"] == name), None)

orders = [parse({"order_id": "7", "customer": " Asha ", "amount": "19.5"})]
print(find_customer(orders, "Asha"), find_customer(orders, "Ben"))
print(parse.__annotations__["return"])

{'order_id': 7, 'customer': 'Asha', 'amount': 19.5} None
dict[str, int | str | float]

Useful conventions for pipeline code:

• Built-in generics (list[str], dict[str, int]) and X | None work without imports on Python 3.10 and later.
• TypedDict describes dict-shaped records such as JSON payloads; a frozen dataclass suits records and configuration you create yourself.
• Annotate public functions at module boundaries first; that is where mistakes between team members happen.
• Run the type checker in CI next to the tests.

A one-line docstring that states what the function guarantees (“returns amounts in cents; never None”) is more useful than a comment that repeats the code.

Mutable default arguments

Default values are evaluated once, when def runs. A list or dict default is therefore shared by every call:

def add_row(row, batch=[]):          # bug: one list for every call
    batch.append(row)
    return batch

print(add_row(1), add_row(2))

def add_row_fixed(row, batch=None):
    batch = [] if batch is None else batch
    batch.append(row)
    return batch

print(add_row_fixed(1), add_row_fixed(2))

[1, 2] [1, 2]
[1] [2]

In a long-lived worker process this leaks rows from one run into the next. See the interview question on mutable defaults for more cases, including defaults that read the clock.

Common mistakes

1. Mixing database calls into transformation logic, so nothing can be tested offline.
2. Mutable default arguments: def f(rows=[]) shares one list across calls. Use None and create the list inside.
3. Reading environment variables inside deep helper functions.
4. Putting job execution at module import time without a __main__ guard.

Interview relevance

Interviewers often ask you to “clean up” a script or explain how you would test a pipeline. Talk about pure functions, I/O at the edges, configuration and dependencies passed in, type hints at module boundaries, and tests for the transformation layer (see testing with pytest).

Key takeaway

Keep transformations pure, push I/O to the edges, pass configuration in, and give the job a thin entry point. Testability and reuse follow.