lesson 198 Software Architecture · advanced

Dependency boundaries: interfaces, adapters and import rules

Business rules shouldn't know whether orders live in SQLite, a JSON file or a web API. The core states what it needs as an interface; adapters provide it; and an architecture test keeps imports pointing the right way.

You'll be able to

Describe what the core needs as a Protocol and pass an implementation in
Write adapters (in-memory, SQLite, files) that satisfy the same interface
Check dependency direction automatically by reading imports with ast

The idea

A dependency is something a module needs: another module, a database, a service. When the refund rules import sqlite3 and run SQL themselves, every test needs a database, and moving to another store means rewriting the rules. Instead, let the core say *what* it needs: an object with get(order_id) and save(order), described as a Protocol (see Data Models & Typing). Adapters implement that interface for each technology, and the program passes one in (the Advanced OOP & Design module's dependency injection, now at the scale of whole modules). Dependencies point inward: adapters import the core, never the other way round.

Run each example to see what it prints. Running examples doesn't count toward progress.

example-1.py · One service, two stores

import sqlite3
from typing import Protocol

class OrderStore(Protocol):
    def get(self, order_id):
        """Return the order dictionary, or None."""

    def save(self, order):
        """Store the order dictionary."""

class InMemoryOrders:
    def __init__(self):
        self.rows = {}

    def get(self, order_id):
        return self.rows.get(order_id)

    def save(self, order):
        self.rows[order["order_id"]] = dict(order)

class SqliteOrders:
    def __init__(self, connection):
        self.db = connection
        self.db.execute("CREATE TABLE IF NOT EXISTS orders (order_id TEXT PRIMARY KEY, status TEXT, total REAL)")

    def get(self, order_id):
        row = self.db.execute("SELECT order_id, status, total FROM orders WHERE order_id = ?", (order_id,)).fetchone()
        return None if row is None else {"order_id": row[0], "status": row[1], "total": row[2]}

    def save(self, order):
        self.db.execute("INSERT OR REPLACE INTO orders VALUES (?, ?, ?)", (order["order_id"], order["status"], order["total"]))

class RefundService:
    def __init__(self, store: OrderStore):
        self.store = store

    def refund(self, order_id):
        order = self.store.get(order_id)
        if order is None or order["status"] != "delivered":
            return False
        order["status"] = "refunded"
        self.store.save(order)
        return True

for store in [InMemoryOrders(), SqliteOrders(sqlite3.connect(":memory:"))]:
    store.save({"order_id": "O100282", "status": "delivered", "total": 35.84})
    service = RefundService(store)
    print(type(store).__name__, service.refund("O100282"), service.refund("O100282"), store.get("O100282")["status"])

Import rules decay unless something checks them. An architecture test reads each module's imports and fails when a forbidden one appears. The standard ast module parses source into a tree without running it (ast.parse(text); a real test would read each module's file): ast.walk visits every node, an ast.Import node lists names (each with .name), and an ast.ImportFrom node has the module it imports from.

example-2.py · An architecture test with ast

import ast

trees = {
    "domain.py": ast.parse("from dataclasses import dataclass\n\n@dataclass\nclass Order:\n    order_id: str\n    total: float\n"),
    "service.py": ast.parse("from domain import Order\nimport sqlite3\n"),
    "adapters.py": ast.parse("import sqlite3\nfrom domain import Order\n"),
}

def imported_modules(tree):
    names = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            names += [alias.name for alias in node.names]
        elif isinstance(node, ast.ImportFrom):
            names.append(node.module)
    return names

for module, tree in trees.items():
    print(module, imported_modules(tree))
print([m for m in ("domain.py", "service.py") if "sqlite3" in imported_modules(trees[m])])

When to draw a boundary: around anything slow, external or likely to change: databases, HTTP APIs, the clock, the file system, email. Mistakes to avoid: adapters leaking their types into the core (returning sqlite3.Row objects instead of plain data); the core importing a concrete adapter "just for the default"; and interfaces so wide (twenty methods) that every adapter is painful to write. Keep each interface to what the core actually calls.

Review

The core states what it needs (a Protocol); adapters implement it; the entry point wires them together.
Dependencies point inward: adapters import the core, never the reverse.
An architecture test with ast keeps import rules from eroding.
Next: changing the structure of existing code safely.

Practise this