Chapter 46 — Design Patterns and Architecture: Organizing Growing Code

Series A1 — Professional Software Engineering

NoteIn this chapter, you will:
  • Meet the three most used design patterns: Strategy, Factory, Observer
  • Understand the SOLID principles through before-and-after examples
  • Separate “what is done” from “where the data lives” with ports and adapters
  • Organize an application into layers (interface, service, domain, adapter)
  • Check the architecture rules automatically, then reorganize the Series A1 running project

All scripts in this chapter are in the folder kode/ch46/. In the web version, you can also download everything at once (zip).

Our package from Chapter 45 looks tidy on the outside. But inside, imagine a restaurant kitchen where one person cooks, washes dishes, and counts the money. When the restaurant gets busy, chaos follows: the cook is busy counting money and the cashier is busy washing plates. The solution is division of labour. In software, division of labour that has been tested for years is called a design pattern (a recipe for a problem that keeps appearing) and architecture (the floor plan of the whole kitchen).

NoteREMEMBER

A design pattern is a tool, not decoration. Applying a pattern where it is not needed only adds complexity. Python also often simplifies patterns with language features (functions as objects, dicts, modules), so many “patterns” below turn out to be just a few lines.

Strategy: swappable formulas

You have many ways to compute a tree’s volume (cylinder, cone, with a form factor). Instead of branching ifs everywhere, wrap each formula as a function and hand it in to the user:

01_strategy.py
# Chapter 46 - Strategy pattern: formulas can be swapped without changing the user of the formula
import math


def cylinder_volume(diameter_cm, height_m):
    return math.pi * (diameter_cm / 200) ** 2 * height_m


def cone_volume(diameter_cm, height_m):
    return cylinder_volume(diameter_cm, height_m) / 3


def tree_shape_volume(diameter_cm, height_m):
    return cylinder_volume(diameter_cm, height_m) * 0.5     # example form factor


FORMULAS = {
    "cylinder": cylinder_volume,
    "cone": cone_volume,
    "tree": tree_shape_volume,
}


def compute_volume(diameter_cm, height_m, formula):
    """The caller does not care which formula; it just calls it."""
    return formula(diameter_cm, height_m)


for name, function in FORMULAS.items():
    print(f"{name:9} {compute_volume(30, 20, function):.3f} m3")
cylinder  1.414 m3
cone      0.471 m3
tree      0.707 m3

For a tree 30 cm in diameter and 20 m tall, the cylinder gives 1.414 m³, the cone a third of that (0.471), and the 0.5 form factor gives 0.707. The function compute_volume does not change at all if you add a fourth formula. That is the heart of Strategy: swappable algorithms.

Factory: one entrance for making objects

Field data arrives as several kinds of records (tree, fish, …). A Factory centralizes “this kind becomes which object”:

02_factory.py
# Chapter 46 - Factory pattern: one entrance that creates the right object for each record type
from dataclasses import dataclass


@dataclass
class TreeRecord:
    plot: str
    diameter_cm: float


@dataclass
class FishRecord:
    pond: str
    avg_weight_g: float


FACTORY = {"tree": TreeRecord, "fish": FishRecord}


def make_record(data):
    """Creates the right object from a dict that has a 'kind' key."""
    data = dict(data)
    kind = data.pop("kind")
    if kind not in FACTORY:
        raise ValueError(f"unknown record kind: {kind!r} (choices: {sorted(FACTORY)})")
    return FACTORY[kind](**data)


incoming = [
    {"kind": "tree", "plot": "A", "diameter_cm": 32.5},
    {"kind": "fish", "pond": "K1", "avg_weight_g": 540.5},
    {"kind": "livestock", "pen": "S1"},
]
for item in incoming:
    try:
        print(make_record(item))
    except ValueError as error:
        print("REJECTED:", error)
TreeRecord(plot='A', diameter_cm=32.5)
FishRecord(pond='K1', avg_weight_g=540.5)
REJECTED: unknown record kind: 'livestock' (choices: ['fish', 'tree'])

The kind livestock has not been registered, so it is rejected with a clear message naming the available choices. Adding a new kind only means adding one line to FACTORY.

Observer: giving news without knowing the receivers

When a pond’s pH is recorded, someone may need to know: a history log, a warning system, a dashboard. The recorder does not want to know them one by one. With Observer, receivers register themselves, and the recorder calls them all:

03_observer.py
# Chapter 46 - Observer pattern: the recorder notifies "listeners" without knowing them
class PhRecorder:
    def __init__(self):
        self._listeners = []

    def register(self, listener):
        self._listeners.append(listener)

    def record(self, pond, ph):
        for listener in self._listeners:
            listener(pond, ph)


history = []
warnings = []


def save_history(pond, ph):
    history.append((pond, ph))


def watch_ph(pond, ph):
    if not 6.5 <= ph <= 8.5:        # example limits, not farming advice
        warnings.append(f"pond {pond}: pH {ph} outside the example limits 6.5-8.5")


recorder = PhRecorder()
recorder.register(save_history)
recorder.register(watch_ph)

for pond, ph in [("K1", 7.2), ("K2", 6.1), ("K3", 8.9)]:
    recorder.record(pond, ph)

print("history:", history)
for message in warnings:
    print("WARNING:", message)
history: [('K1', 7.2), ('K2', 6.1), ('K3', 8.9)]
WARNING: pond K2: pH 6.1 outside the example limits 6.5-8.5
WARNING: pond K3: pH 8.9 outside the example limits 6.5-8.5

Pond K2 (pH 6.1) and pond K3 (pH 8.9) are outside the example limits of 6.5 to 8.5; these limits are for practice only, not farming advice. Note that PhRecorder does not know the function watch_ph exists; adding a new listener does not change the recorder.

SOLID: five principles for organizing classes and functions

SOLID is an acronym for five principles that make code easy to change:

Letter Principle In everyday words
S Single Responsibility one part, one job
O Open/Closed open to adding, closed to tearing apart
L Liskov Substitution a replacement must work wherever the original did
I Interface Segregation small, specific interfaces, not one giant interface
D Dependency Inversion depend on a contract, not on a concrete class

We look at the first two as before-and-after:

05_solid.py
# Chapter 46 - SOLID: single responsibility (S) and open for extension (O)

# ---- BEFORE: one function does everything, and must be changed for every new species ----
def do_everything_report(trees):
    total = 0
    for t in trees:
        if t["species"] == "Teak":
            total += t["volume"] * 4_000_000
        elif t["species"] == "Meranti":
            total += t["volume"] * 2_500_000
        # a new species? This function must be torn open again...
    return f"Total: Rp {total:,.0f}"          # calculating AND formatting are mixed


# ---- AFTER: each part has one job; a new species only adds data ----
PRICES = {"Teak": 4_000_000, "Meranti": 2_500_000}


def compute_total(trees, prices):
    return sum(t["volume"] * prices[t["species"]] for t in trees)


def format_rupiah(amount):
    return f"Total: Rp {amount:,.0f}"


trees = [
    {"species": "Teak", "volume": 1.2},
    {"species": "Meranti", "volume": 0.8},
]
print(do_everything_report(trees))
print(format_rupiah(compute_total(trees, PRICES)))

PRICES["Mahogany"] = 3_000_000                # extension: without touching any function
trees.append({"species": "Mahogany", "volume": 0.5})
print(format_rupiah(compute_total(trees, PRICES)))
print("old version for Mahogany (silently ignored):", do_everything_report(trees))
Total: Rp 6,800,000
Total: Rp 6,800,000
Total: Rp 8,300,000
old version for Mahogany (silently ignored): Total: Rp 6,800,000

The fourth line shows the danger of the old version: once a Mahogany tree exists, the old function silently ignores it (the total stays 6,800,000), while the new version counts 8,300,000. A bug like this is sneaky because there is no error. The new version is “open for extension”: a new species only adds data.

Ports and adapters, and dependency injection

Principle D gives birth to a very useful idea: the port (the contract: “I need something that can save and read”) and the adapter (a real way to fulfil it: memory, a JSON file, a database). In Python, a port is written with typing.Protocol:

04_port_adapter.py
# Chapter 46 - Port and Adapter: one contract, many ways to store
import json
from pathlib import Path
from typing import Protocol


class Storage(Protocol):
    def save(self, record: dict) -> None: ...
    def all(self) -> list[dict]: ...


class MemoryStorage:
    def __init__(self):
        self._data = []

    def save(self, record):
        self._data.append(record)

    def all(self):
        return list(self._data)


class JsonStorage:
    def __init__(self, path):
        self._path = Path(path)

    def save(self, record):
        data = self.all()
        data.append(record)
        self._path.write_text(json.dumps(data), encoding="utf-8")

    def all(self):
        if not self._path.exists():
            return []
        return json.loads(self._path.read_text(encoding="utf-8"))


def record_harvest(storage: Storage):
    """This function does not know where the data is stored."""
    storage.save({"block": "A", "yield_kg": 245.5})
    storage.save({"block": "B", "yield_kg": 310.0})
    return storage.all()


tmp = Path(__file__).parent / "_tmp"
tmp.mkdir(exist_ok=True)
file = tmp / "harvest.json"
file.unlink(missing_ok=True)

memory_result = record_harvest(MemoryStorage())
json_result = record_harvest(JsonStorage(file))
print("memory:", memory_result)
print("json  :", json_result)
print("both adapters give the same result:", memory_result == json_result)
memory: [{'block': 'A', 'yield_kg': 245.5}, {'block': 'B', 'yield_kg': 310.0}]
json  : [{'block': 'A', 'yield_kg': 245.5}, {'block': 'B', 'yield_kg': 310.0}]
both adapters give the same result: True

The function record_harvest works with any adapter. The biggest benefit appears in testing. When the inner parts are handed in from outside (dependency injection), a test can use a fake stand-in:

06_dependency_injection.py
# Chapter 46 - Dependency injection (the D in SOLID): inner parts are handed in, so testing is easy
class RealRepo:
    """In a real application this would read a database; here it only pretends."""

    def all_diameters(self):
        return [32.5, 45.2, 28.0]


class FakeRepo:
    """A stand-in for tests: we decide its content ourselves."""

    def all_diameters(self):
        return [10.0, 20.0]


class AverageService:
    def __init__(self, repo):
        self._repo = repo            # HANDED IN from outside, not created inside

    def average(self):
        numbers = self._repo.all_diameters()
        return sum(numbers) / len(numbers)


# Test: with the fake repo, the result is exactly predictable
test = AverageService(FakeRepo())
assert test.average() == 15.0
print("test with fake repo passed: average =", test.average())

# Real use: the same service, different storage
real = AverageService(RealRepo())
print("with real repo: average =", round(real.average(), 2))
test with fake repo passed: average = 15.0
with real repo: average = 35.23

The test with the fake repo can assert that the average is exactly 15.0 because we decide its content. With a real database, the test would have to prepare data first and become slow.

TipTIP

Rule of thumb: if a class creates its own database connection, file, or network link inside, it is hard to test. Change it so the connection is received through an argument (__init__).

Layered architecture and a short hexagonal note

Putting it all together, we organize the application into layers:

Layered architecture: dependency arrows only point inward, toward the domain.
  • Domain: the core rules (volume, timber value) and the ports. Knows nobody.
  • Service: the workflow (“compute the summary from all trees”). Knows only the domain.
  • Adapter: files, database, outside services; fulfils the ports owned by the domain.
  • Interface: API, CLI, dashboard; calls the service.

This style is often called hexagonal architecture (or ports and adapters): the application core sits in the middle, and the outside world attaches through adapters. The point is one rule: dependencies point inward. The domain must not import an adapter.

Running Project A1: reorganized into layers

We reorganize the timber value summary into the package field_app with four layers. Run the same service with two different adapters:

07_project_architecture.py
# Chapter 46 - Running project A1: a layered application; swap the adapter, same result
import sys
from pathlib import Path

folder = Path(__file__).parent
sys.path.insert(0, str(folder))

from field_app.adapter.repo_csv import CsvRepo
from field_app.adapter.repo_memory import MemoryRepo
from field_app.domain.model import Tree
from field_app.interface.report import print_report
from field_app.service.summary import SummaryService

sample = [
    Tree("Teak", "A", 32.5, 18.0), Tree("Meranti", "A", 45.2, 22.0),
    Tree("Mahogany", "B", 28.0, 15.0), Tree("Teak", "B", 51.5, 20.0),
    Tree("Sengon", "C", 24.0, 12.0), Tree("Teak", "A", 29.9, 16.0),
    Tree("Meranti", "C", 38.0, 19.0), Tree("Mahogany", "A", 30.0, 14.0),
]

from_memory = SummaryService(MemoryRepo(sample))
from_csv = SummaryService(CsvRepo(folder / "data" / "trees_sample.csv"))

print("== source: memory ==")
print_report(from_memory)
print("== source: CSV file ==")
print_report(from_csv)
print("same result:", from_memory.value_per_plot() == from_csv.value_per_plot())
== source: memory ==
  plot A: Rp 11,130,411
  plot B: Rp 9,717,723
  plot C: Rp 2,964,954
  count per species: {'Mahogany': 2, 'Meranti': 2, 'Sengon': 1, 'Teak': 3}
== source: CSV file ==
  plot A: Rp 11,130,411
  plot B: Rp 9,717,723
  plot C: Rp 2,964,954
  count per species: {'Mahogany': 2, 'Meranti': 2, 'Sengon': 1, 'Teak': 3}
same result: True

Two data sources (a list in memory and a CSV file) give identical results (same result: True), although the service and interface code were not changed at all. That is the real benefit of separating layers.

Architecture rules that live only in a document are broken quickly. So we check them with a program, reading all import lines with the ast module:

08_check_dependencies.py
# Chapter 46 - Checking the architecture rules automatically: the domain must not import outer layers
import ast
from pathlib import Path

PACKAGE = "field_app"
# layer -> layers it MAY import
ALLOWED = {
    "domain": set(),
    "service": {"domain"},
    "adapter": {"domain"},
    "interface": {"service", "domain"},
}


def internal_imports(source):
    """The layers a module imports (only from our own package)."""
    found = set()
    for node in ast.walk(ast.parse(source)):
        if isinstance(node, ast.ImportFrom) and node.module and node.module.startswith(PACKAGE + "."):
            found.add(node.module.split(".")[1])
    return found


def check(source, layer):
    return sorted(internal_imports(source) - ALLOWED[layer] - {layer})


root = Path(__file__).parent / PACKAGE
violations = 0
for layer in ALLOWED:
    for file in sorted((root / layer).glob("*.py")):
        if file.name == "__init__.py":
            continue
        wrong = check(file.read_text(encoding="utf-8"), layer)
        status = "OK" if not wrong else f"VIOLATION: imports {wrong}"
        violations += len(wrong)
        print(f"{layer + '/' + file.name:24} {status}")
print("number of violations:", violations)

# Test the checker itself with made-up bad code:
bad = f"from {PACKAGE}.adapter.repo_csv import CsvRepo"
print("bad code (domain imports adapter) detected:", check(bad, "domain"))
domain/model.py          OK
domain/port.py           OK
service/summary.py       OK
adapter/repo_csv.py      OK
adapter/repo_memory.py   OK
interface/report.py      OK
number of violations: 0
bad code (domain imports adapter) detected: ['adapter']

All modules comply (0 violations). The last line tests the checker itself with made-up bad code, and it is detected. A checker like this can be installed in CI (Chapter 48) so violations are caught automatically.

Exercises

  1. Add a fourth formula paraboloid_volume (half a cylinder, because the 0.5 form factor is the same) to FORMULAS and run again. Does compute_volume need to change?
  2. In 03_observer.py, add a third listener that counts how many recordings happened, without changing the class PhRecorder.
  3. Write a third adapter JsonRepo for field_app that reads the tree list from a JSON file. Make sure 08_check_dependencies.py still shows 0 violations.
  4. Challenge: change SummaryService to receive a price function (a strategy) through __init__, then test it with MemoryRepo and a fixed price of 1,000,000 per m³.

Chapter Summary

  • A design pattern = a recipe for a recurring problem; use it only when needed.
  • Strategy (swappable formulas), Factory (one entrance for making objects), Observer (news without knowing the receivers).
  • SOLID: one job per part, open for adding, equivalent replacements, small interfaces, depend on contracts.
  • Ports and adapters plus dependency injection make code easy to test and swap.
  • Layered architecture: interface, service, domain, adapter; dependencies only point inward. Check the rules with a program.

Next, in Chapter 47: modern Python (3.12 to 3.14), new features that make code shorter and safer.