Functions, Modules & Configuration

In chapter 5 you wrote small functions such as parse_amount. A real job has many more steps. It reads a file, cleans each row, keeps the bad rows and writes the result.

If all of that sits in one long script, a small change is hard to make and hard to test. This chapter splits a job into clear parts.

A function is a named set of steps (chapter 1). Here is the plan for one orders job:

Each section builds one part. Every example uses the orders from chapter 4.

A good function does one job. You can say what goes in and what comes out.

normalize_order takes one raw row of text and returns one clean order. A bad row raises a ValueError (chapter 5):

from decimal import Decimal, InvalidOperation

STATUSES = {"paid", "refunded", "pending"}

def parse_amount(text):            # from chapter 5
    text = text.strip()
    if text == "":
        return None
    try:
        return Decimal(text)
    except InvalidOperation:
        raise ValueError(f"bad amount {text!r}")

def normalize_order(raw):
    if raw["status"] not in STATUSES:
        raise ValueError(f"bad status {raw['status']!r}")
    return {
        "order_id": int(raw["order_id"]),
        "customer_id": int(raw["customer_id"]),
        "status": raw["status"],
        "amount": parse_amount(raw["amount"]),
    }

print(normalize_order({"order_id": "105", "customer_id": "3",
                       "status": "pending", "amount": ""}))

The name says what the function does. normalize_order is clearer than a name like process.

Names inside a function

A function should get its values from its arguments, not from names outside it. This function tries to change an outside name:

total = 0

def add(amount):
    print(total)
    total = total + amount

add(5)

The line total = total + amount gives total a value inside the function. So Python treats total as a local name in the whole function. A local name belongs to one function only.

On the print line, the local total has no value yet. Pass the value in, and return the new value:

def add(total, amount):
    return total + amount

print(add(0, 5))

A pure function uses only its arguments and returns a result. It does not read files, read settings or write anything.

I/O means input and output: reading and writing files, databases and networks. An I/O wrapper is a small function that does the I/O and calls the pure functions.

transform_rows is pure. It cleans a list of rows and keeps the bad ones in a second list:

def transform_rows(raw_rows):
    accepted, rejected = [], []
    for line, raw in enumerate(raw_rows, start=2):
        try:
            accepted.append(normalize_order(raw))
        except ValueError as error:
            rejected.append({"line": line, "reason": str(error)})
    return accepted, rejected

accepted, rejected = transform_rows(raw_rows)
print(len(accepted))
print(rejected)

You can test transform_rows with a list in memory. No file is needed.

run_orders_job is the I/O wrapper. It reads the file, calls transform_rows and writes the clean rows:

import csv
import json

def run_orders_job(input_path, output_path):
    with open(input_path, encoding="utf-8", newline="") as file:
        raw_rows = list(csv.DictReader(file))
    accepted, rejected = transform_rows(raw_rows)
    with open(output_path, "w", encoding="utf-8") as file:
        for row in accepted:
            file.write(json.dumps(row, default=str) + "\n")
    return {"rows_read": len(raw_rows), "accepted": len(accepted),
            "rejected": len(rejected)}

print(run_orders_job("orders.csv", "orders_clean.ndjson"))

All six orders are clean, so nothing is rejected. When a job breaks, you now know where to look. Either the file, the cleaning or the writing failed.

A transform should return the good rows and the bad rows. If it drops the bad rows, nobody knows that data is missing.

Add a small report with the counts. The counts must add up:

accepted, rejected = transform_rows(raw_rows)
report = {"rows_read": len(raw_rows),
          "accepted": len(accepted),
          "rejected": len(rejected)}

print(report)
print(report["rows_read"] == report["accepted"] + report["rejected"])

Each rejected row keeps its line number and a reason. You can store it, count it and show it to the sender.

Return the same shape every time. Here it is always two lists. A function that returns a string for one input and a list for another makes every caller check the type first.

A signature is the first line of a function: its name, its arguments and its return type. Chapter 5 showed type hints, which are notes about types:

def normalize_order(raw: dict[str, str]) -> dict[str, object]:
    ...

An argument can have a default value. Python uses it when the caller gives no value.

Python works out a default value once, when it creates the function. It does not work it out again on each call. This surprises many people:

def add_tag(tag, tags=[]):
    tags.append(tag)
    return tags

print(add_tag("late"))
print(add_tag("big"))

Use None as the default, and make a new list inside the function:

def add_tag(tag, tags=None):
    if tags is None:
        tags = []
    tags.append(tag)
    return tags

print(add_tag("late"))
print(add_tag("big"))

The same rule breaks a setting in a default. Here the default reads a setting from the computer:

import os

os.environ["BATCH_SIZE"] = "500"

def load(batch=int(os.environ["BATCH_SIZE"])):
    return batch

os.environ["BATCH_SIZE"] = "100"
print(load())

os.environ holds the environment variables. The section Environment-specific behavior explains them.

Configuration is the settings of one run: the batch date, the input path, the reject limit. A batch date is the day of data that the run handles.

If these values sit as fixed text in many functions, you cannot see what a run used. Put them in one object instead.

A dataclass is a short way to make a class that holds named values. With frozen=True, nobody can change the values after you create the object:

from dataclasses import dataclass, replace

@dataclass(frozen=True)
class JobConfig:
    batch_date: str
    input_path: str
    reject_limit: int = 10

config = JobConfig(batch_date="2026-06-19",
                   input_path="incoming/2026-06-19.csv")
print(config)

Pass config to the functions that need it. Then the settings of a run are in one place, and you can save them next to the result.

You cannot change a frozen object. To run again with one setting changed, replace() makes a new object:

rerun = replace(config, reject_limit=0)
print(rerun.reject_limit, config.reject_limit)

config.reject_limit = 5

Chapter 17, Functions, Decorators, Context Managers & Dataclasses, shows more about dataclasses.

The same job runs on your laptop and on a server. Each place is an environment. Each one can use a different folder or limit.

An environment variable is a named setting that the computer gives to a program. Python reads them from os.environ, a dict-like object.

Read them in one function, load_config, and turn them into a JobConfig. A missing required setting should stop the job at once:

def load_config(batch_date):
    root = os.environ["ORDERS_ROOT"]
    limit = int(os.environ.get("REJECT_LIMIT", "10"))
    return JobConfig(batch_date=batch_date,
                     input_path=f"{root}/{batch_date}.csv",
                     reject_limit=limit)

load_config("2026-06-19")

Now set the variable and call it again:

os.environ["ORDERS_ROOT"] = "incoming"
print(load_config("2026-06-19"))

On a real server, the operator sets ORDERS_ROOT before the job starts. The code above sets it only to show the example.

normalize_order never reads os.environ. It gets what it needs as arguments, so a test does not depend on the computer it runs on.

A module is one Python file. A package is a folder of modules. It usually has a file called __init__.py inside.

Split a job by what each file does, not by the order of the steps. Here is one layout for the orders job:

orders_job/
    __init__.py
    transform.py   # parse_amount, normalize_order, transform_rows
    config.py      # JobConfig, load_config
    io.py          # run_orders_job
    run.py         # main
tests/
    test_transform.py

Other files import what they need. An import makes the names of one module usable in another:

from orders_job.transform import normalize_order, transform_rows
from orders_job.config import load_config

When Python imports a module, it runs every top-level line of that file once. A top-level line is a line that is not inside a function.

Look at this small file, shop_job.py. Its first line prints, and its last lines start the job:

print("loading shop_job")

def main():
    print("job runs")

if __name__ == "__main__":
    main()

Now run the file, and then import it from another program:

The import ran the print line, but not the job. The if __name__ == "__main__": check stops the job when a test or another file imports the module.

So keep top-level lines for definitions. Start the real work inside main().

A dependency is anything a function uses that is not an argument: a file, the clock, the network or an environment variable.

The clock is a common one. A report that calls datetime.now() itself gives a different result on every run, so a test cannot check it.

Pass the clock in as an argument. A test can then pass a clock that always gives the same time:

from datetime import datetime, timezone

def build_report(accepted, rejected, now):
    return {"created_at": now().isoformat(),
            "accepted": accepted, "rejected": rejected}

def fixed_clock():
    return datetime(2026, 1, 22, 2, 0, tzinfo=timezone.utc)

print(build_report(2, 1, now=fixed_clock))

You do not need to do this for everything. Do it where the result must be exact: time, settings and files.

Chapter 13, Testing, Debugging, Logging & Operability, shows how to write the tests.

A job that is easy to change and test follows five rules:

1. Each function does one job, and gets its values as arguments.
2. Pure functions clean the data. A thin wrapper does the reading and writing.
3. A transform returns good rows, bad rows and counts that add up.
4. Settings live in one frozen config, read once from the environment.
5. Modules are named by their job, and main() starts the work.

Check yourself. Why does a default list in a signature cause trouble? (Answer: Python creates it once, and every call shares it.)

Each exercise opens in the Python workspace with its own data. Write the code, run it, and check the result against the tests.

Each one has a trap in the data or the settings, not in the loop.