Separation of Concerns (I/O vs Logic)

By the end of this chapter, you should be able to:

Recognize the "god function" anti-pattern and explain why it is hard to test.
Split a pipeline into a thin I/O layer, pure logic functions, and a small orchestrator.
Use dependency injection to pass data into a function instead of letting the function fetch it.

In the early stages of learning data engineering, it's common to write scripts that do everything at once: read data, clean it, apply business rules, and write the result somewhere else. It works… until it doesn't.

As pipelines grow, this style quickly becomes:

Hard to test
Hard to reuse
Hard to debug
Hard to change without breaking something

This chapter covers separation of concerns, a design principle that makes pipelines easy to test, easy to change, and hard to break.

The Core Pattern

Most data pipelines at the high level follow this pattern:

Input -> Business Logic -> Output

<aside> 🎬 Animation: Separation of Concerns (I/O → Logic → Output)

</aside>

A common mistake is mixing these layers together!

The Problem: The "God Function"

Here's a very common anti-pattern (a pattern that seems helpful but actually causes problems):

import csv

def process_users():
    # Read
    with open("users.csv") as f:
        rows = list(csv.DictReader(f))

    # Clean
    rows = [r for r in rows if r["email"]]
    for r in rows:
        r["email"] = r["email"].lower()

    # Business rules
    rows = [r for r in rows if int(r["age"]) >= 18]

    # Write
    with open("clean_users.csv", "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=["name", "email", "age"])
        writer.writeheader()
        writer.writerows(rows)

This function:

Reads from disk
Applies business rules
Writes to disk

Why is this a problem?

You can't test the logic without touching the filesystem
You can't reuse the logic with data from an API or database
One small change risks breaking everything

This is where separation of concerns comes in.

<aside> 💡 Using AI to help: When you inherit a god function and want to refactor it, paste only the function body (⚠️ no surrounding repo code, secrets, or PII) into an LLM and ask it to split the function into an I/O layer and a pure-logic layer. Then check the suggestion against the rule below: I/O at the edges, no business decisions inside file handlers.

</aside>

The Humble Object Principle

A useful mental model is the Humble Object:

Keep the parts of your code that touch the outside world as thin as possible.

That means:

File reads
API calls
Database connections

All these steps should contain almost no logic.

Their only job is to:

Get data in
Pass it to the business logic
Send the result out

Business Logic as Pure Functions

Your business logic should live in pure functions whenever possible.

A pure function:

Takes inputs as arguments
Returns outputs
Does not read files, nor call APIs, nor write to databases

Example:

def clean_users(rows: list[dict]) -> list[dict]:
    cleaned = [r for r in rows if r["email"]]
    for r in cleaned:
        r["email"] = r["email"].lower()
    return cleaned

This makes it easy to understand, to reuse and trivial to test.

I/O Functions

I/O stands for Input / Output:

Usually, reading, writing files, calling APIs and querying databases are operations that can be slow, unreliable, and hard to test! For example: files may not exist, networks may fail, or credentials may be wrong. That's exactly why you want to isolate them.

An I/O function should answer only one question:

"How do I get data in or out?"

It should not decide:

What data is valid
What data should be kept
What business rules apply

Example of a good I/O function:

import csv

def read_users_csv(path: str) -> list[dict]:
    with open(path) as f:
        return list(csv.DictReader(f))

Simple and effective! The same problem appears any time I/O and business logic share a function: see the god function from the start of the chapter for the full version.

Dependency Injection

Another common mistake is letting functions fetch their own data:

def filter_adults():
    with open("users.csv") as f:
        rows = list(csv.DictReader(f))
    return [r for r in rows if int(r["age"]) >= 18]

This function decides where to get data and what to do with it. That's two responsibilities in one place. It's also impossible to test without a real users.csv file on disk.

The fix is dependency injection: instead of letting a function fetch its own data, you pass the data in as an argument.

def filter_adults(rows: list[dict]) -> list[dict]:
    return [r for r in rows if int(r["age"]) >= 18]

Now the function doesn't know or care where the data came from. It could be from a CSV, from an API, or from a test (the production branch below assumes a real users.csv and the filter_adults definition from above; the test branch runs standalone):

# In production
with open("users.csv") as f:
    rows = list(csv.DictReader(f))
adults = filter_adults(rows)

# In a test
test_rows = [{"name": "Alice", "age": "25"}, {"name": "Bob", "age": "16"}]
result = filter_adults(test_rows)
assert len(result) == 1

This is the same principle as the pure functions we discussed earlier, but applied as a deliberate design strategy: push I/O to the edges, inject data into the logic layer.

Putting It All Together

Here's the full refactored pipeline using everything from this chapter:

import csv

# io.py: thin I/O layer
def read_users(path: str) -> list[dict]:
    with open(path) as f:
        return list(csv.DictReader(f))

def save_users(rows: list[dict], path: str):
    with open(path, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=["name", "email", "age"])
        writer.writeheader()
        writer.writerows(rows)

# logic.py: pure business rules
def clean_users(rows: list[dict]) -> list[dict]:
    cleaned = [r for r in rows if r["email"]]
    for r in cleaned:
        r["email"] = r["email"].lower()
    return cleaned

def filter_adults(rows: list[dict]) -> list[dict]:
    return [r for r in rows if int(r["age"]) >= 18]

# main.py: orchestration
def run_pipeline(input_path, output_path):
    raw = read_users(input_path)
    cleaned = clean_users(raw)
    adults = filter_adults(cleaned)
    save_users(adults, output_path)

Notice how each layer has a single responsibility: I/O knows how to read/write, logic knows what to do with data, and the orchestrator decides when to call each step.

<aside> ⌨️ Hands on: Implement a clean_and_calculate(rows, vat_rate) pure function that cleans names and applies VAT, without mutating the original data. This is the logic layer in action!

</aside>

<aside> 🚀 Try it in the widget: https://lasse.be/simple-hyf-teach-widget/?week=2&chapter=separation_concerns&exercise=w2_separation_concerns__extract_logic&lang=python

</aside>

This pattern (I/O → Logic → Output) is so foundational to data engineering that it has a three-letter acronym you will hear constantly.

<aside> 🤓 Curious Geek: ETL vs ELT

</aside>

Time to apply this layering on a small dataset of your own.

<aside> 📝 Practice: Apply this pattern in Practice Exercise 3: Separate I/O from Logic.

</aside>

🧠 Knowledge Check

1. Why is the "god function" (doing Read → Transform → Write in one place) considered an anti-pattern?
1. In the "humble object" pattern, what logic should remain in the functions that touch the database or filesystem?
1. Rewrite this impure function signature def process_data(input_path, output_path) to be a Pure Function. What arguments should it take and what should it return?

Extra reading

Python Dependency Injection: A Practical Guide - Understanding DI patterns in Python
The Humble Object Pattern - Martin Fowler on keeping I/O boundaries thin
Separation of Concerns in Software Design - Visual guide to layered architecture

<aside> 💡 In the wild: The Meltano ELT framework is built entirely around separation of concerns. Each "tap" (extractor) and "target" (loader) is a standalone component that does one thing. The orchestrator just chains them together, exactly like the pattern you learned in this chapter.

</aside>

Next up: OOP vs Functional Programming, where you decide when to reach for a class versus a plain function, and meet the decorator syntax that makes Python's class tooling concise.