How to Prepare a Legacy Application for Migration Using Seams and Adapters in Python

When a team decides to migrate a legacy application, the instinct is almost always to start moving code: swap the database, swap the framework, swap the cloud provider. That instinct is usually where migrations get into trouble. If a function reads from a database, applies your pricing rules, charges a payment provider, and sends an email all in the same 20 lines, you cannot move any one of those responsibilities without dragging the other three along with it. This tutorial teaches a narrower, safer first step: restructuring a legacy capability, before you touch its infrastructure, so that its pieces can move independently later.

You will start with a single tangled Python function that mixes database access, business rules, and three external integrations. You will lock in its current behavior with a test, then split it into a pure business-rule function and a set of small “adapters” around each external system, a technique with a specific name in software engineering: a seam. By the end, you will prove the refactor actually worked by swapping the database for a completely different storage system, in a few lines, with zero changes to the business logic and zero new bugs, because the same safety net you wrote in step one still passes.

What You Will Build and Why It Matters

You will build an order-processing service in three stages:

1. A tangled, “legacy” version where reading data, deciding a price, saving the result, and calling three external systems all live inside one function.
2. A refactored version where the pricing decision is a pure function, the external systems sit behind small interfaces, and one orchestrator function wires them together.
3. A proof that the refactor worked: the same pricing decision, running against a completely different data store with a different internal shape, using the exact same code, verified by the exact same tests you wrote before you changed anything.

This matters because “refactor before you migrate” is not about making old code pretty. It is about creating a boundary, called a seam, where you can substitute a new database, a new payment provider, or a new runtime later, without having to understand and rewrite the business logic at the same time. Software engineer Michael Feathers, who coined the term in his book Working Effectively with Legacy Code, defines it precisely: “a seam is a place where you can alter behavior in your program without editing in that place.” That is exactly what you are going to build: a place to plug in a new database without editing your pricing rules.

Prerequisites

• Python 3.10 or newer. This tutorial uses the match statement and the X | Y union type syntax, both added in Python 3.10.
• pytest installed (pip install pytest). Everything else used here, sqlite3, dataclasses, typing, json, is in the Python standard library. No database server, Docker, or cloud account is required.
• Basic comfort with Python functions, classes, and reading a stack trace.
• You do not need prior experience with characterization testing or dependency injection. Both are explained as you go. If you want a deeper look at writing a safety net before a refactor, see this site’s tutorial on characterization testing for legacy Python code. If you want to see how to investigate an unfamiliar legacy codebase before changing anything, see investigating a legacy codebase with AI before you change it. Neither is required to follow along here.

What “Refactor Before You Migrate” Actually Means

Structural Change vs. Behavioral Change

A refactor, in the strict sense used throughout this tutorial, changes the structure of code without changing its behavior. If you move a calculation into its own function, that is a refactor. If you also fix a rounding bug while you are in there, that is no longer just a refactor, it is a structural change and a behavior change bundled together, and if something breaks you will not know which one caused it.

Before a migration, you want structural change only. The goal is not to make the legacy system beautiful or to redesign it. The goal is much narrower: change the structure enough that one capability, one specific piece of behavior, can move to a new database, a new framework, or a new service independently of everything else. Everything you do in this tutorial keeps behavior byte-for-byte identical. You will prove that with tests, not by eye.

Step 1: Look at the Problem You Are Trying to Avoid

Start with the kind of function every legacy codebase has: it does one useful thing, and it does five other things to make that one thing happen. Create a working directory and set up a small SQLite-backed “orders” table:

mkdir legacy-migration-tutorial
cd legacy-migration-tutorial

Save this as seed_db.py. It creates a database with five sample orders you will reuse throughout the tutorial:

import sqlite3

DB_PATH = "orders.db"

SCHEMA = """
CREATE TABLE IF NOT EXISTS orders (
    id TEXT PRIMARY KEY,
    base_total REAL NOT NULL,
    total REAL NOT NULL,
    customer_tier TEXT NOT NULL,
    country TEXT NOT NULL,
    payment_method TEXT NOT NULL,
    customer_email TEXT NOT NULL,
    status TEXT NOT NULL
);
"""

SEED_ROWS = [
    ("ORD-1001", 50.00, 50.00, "GOLD", "US", "CARD", "[email protected]", "PENDING"),
    ("ORD-1002", 120.00, 120.00, "STANDARD", "US", "CARD", "[email protected]", "PENDING"),
    ("ORD-1003", 80.00, 80.00, "GOLD", "CA", "WIRE", "[email protected]", "PENDING"),
    ("ORD-1004", 6.00, 6.00, "GOLD", "CA", "WIRE", "[email protected]", "PENDING"),
    ("ORD-1005", 200.00, 200.00, "STANDARD", "CA", "WIRE", "[email protected]", "PENDING"),
]


def fresh_db(db_path: str = DB_PATH) -> None:
    """Recreate the orders table with fixed seed data. Used before every test run
    so each characterization run starts from the same known state."""
    conn = sqlite3.connect(db_path)
    conn.execute("DROP TABLE IF EXISTS orders")
    conn.execute(SCHEMA)
    conn.executemany(
        "INSERT INTO orders (id, base_total, total, customer_tier, country, "
        "payment_method, customer_email, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
        SEED_ROWS,
    )
    conn.commit()
    conn.close()


if __name__ == "__main__":
    fresh_db()
    print(f"Seeded {len(SEED_ROWS)} orders into {DB_PATH}")

Now the tangled function itself. Save this as legacy_order_service.py:

import sqlite3

DB_PATH = "orders.db"

# Fake "external systems." A real legacy app would call a payment SDK, a
# message broker client, and an email API here. We use in-memory logs so the
# tutorial can inspect exactly what each system received, without needing
# real network services.
PAYMENT_LOG: list[dict] = []
EVENT_LOG: list[dict] = []
EMAIL_LOG: list[dict] = []


def _charge_payment(order_id: str, amount: float) -> None:
    PAYMENT_LOG.append({"order_id": order_id, "amount": amount})


def _publish_event(name: str, payload: dict) -> None:
    EVENT_LOG.append({"name": name, "payload": payload})


def _send_email(to: str, template: str, context: dict) -> None:
    EMAIL_LOG.append({"to": to, "template": template, "context": context})


def process_order(order_id: str) -> dict:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()
    cur.execute("SELECT * FROM orders WHERE id = ?", (order_id,))
    row = cur.fetchone()
    if row is None:
        conn.close()
        raise ValueError(f"Order not found: {order_id}")

    order = dict(row)

    # --- business rules, buried inside data-access code ---
    if order["customer_tier"] == "GOLD":
        order["total"] = round(order["total"] * 0.88, 2)

    if order["country"] == "CA" and order["payment_method"] == "WIRE":
        order["total"] = round(order["total"] - 4.0, 2)

    if order["total"] < 0:
        order["total"] = 0.0

    # --- persistence ---
    cur.execute(
        "UPDATE orders SET total = ?, status = ? WHERE id = ?",
        (order["total"], "PROCESSED", order["id"]),
    )
    conn.commit()
    conn.close()

    # --- three more external systems, called inline ---
    _charge_payment(order["id"], order["total"])
    _publish_event("order.processed", {"id": order["id"], "total": order["total"]})
    _send_email(
        order["customer_email"],
        "order-processed",
        {"order_id": order["id"], "total": order["total"]},
    )

    order["status"] = "PROCESSED"
    return order

Run it against a couple of orders and look at what actually happens:

python seed_db.py
python -c "
from legacy_order_service import process_order, PAYMENT_LOG, EVENT_LOG, EMAIL_LOG

for oid in ['ORD-1001', 'ORD-1003', 'ORD-1004']:
    result = process_order(oid)
    print(oid, '->', result)

print()
print('PAYMENT_LOG:', PAYMENT_LOG)
print('EVENT_LOG:', EVENT_LOG)
print('EMAIL_LOG:', EMAIL_LOG)
"

The real output:

ORD-1001 -> {'id': 'ORD-1001', 'base_total': 50.0, 'total': 44.0, 'customer_tier': 'GOLD', 'country': 'US', 'payment_method': 'CARD', 'customer_email': '[email protected]', 'status': 'PROCESSED'}
ORD-1003 -> {'id': 'ORD-1003', 'base_total': 80.0, 'total': 66.4, 'customer_tier': 'GOLD', 'country': 'CA', 'payment_method': 'WIRE', 'customer_email': '[email protected]', 'status': 'PROCESSED'}
ORD-1004 -> {'id': 'ORD-1004', 'base_total': 6.0, 'total': 1.28, 'customer_tier': 'GOLD', 'country': 'CA', 'payment_method': 'WIRE', 'customer_email': '[email protected]', 'status': 'PROCESSED'}

PAYMENT_LOG: [{'order_id': 'ORD-1001', 'amount': 44.0}, {'order_id': 'ORD-1003', 'amount': 66.4}, {'order_id': 'ORD-1004', 'amount': 1.28}]
EVENT_LOG: [{'name': 'order.processed', 'payload': {'id': 'ORD-1001', 'total': 44.0}}, {'name': 'order.processed', 'payload': {'id': 'ORD-1003', 'total': 66.4}}, {'name': 'order.processed', 'payload': {'id': 'ORD-1004', 'total': 1.28}}]
EMAIL_LOG: [{'to': '[email protected]', 'template': 'order-processed', 'context': {'order_id': 'ORD-1001', 'total': 44.0}}, {'to': '[email protected]', 'template': 'order-processed', 'context': {'order_id': 'ORD-1003', 'total': 66.4}}, {'to': '[email protected]', 'template': 'order-processed', 'context': {'order_id': 'ORD-1004', 'total': 1.28}}]

Notice there is also a business rule you cannot see just by reading the two if blocks: if order["total"] < 0: order["total"] = 0.0. Nothing about the code above it tells you why that line exists, or when it would ever trigger, since neither discount alone can push a normal order negative. You can trigger it directly with a small order:

python -c "
import sqlite3
from seed_db import fresh_db
fresh_db()
conn = sqlite3.connect('orders.db')
conn.execute(\"INSERT INTO orders VALUES ('ORD-9999', 3.00, 3.00, 'GOLD', 'CA', 'WIRE', '[email protected]', 'PENDING')\")
conn.commit()
conn.close()
from legacy_order_service import process_order
print(process_order('ORD-9999'))
"

{'id': 'ORD-9999', 'base_total': 3.0, 'total': 0.0, 'customer_tier': 'GOLD', 'country': 'CA', 'payment_method': 'WIRE', 'customer_email': '[email protected]', 'status': 'PROCESSED'}

A GOLD tier order for $3.00 shipped to Canada by wire gets a 12 percent discount (to $2.64) and then a flat $4.00 rebate, which would make the total negative $1.36. The floor rule catches that and clamps it to zero. That is a real business rule with real financial consequences, sitting one line away from two lines that look like the whole story. This is exactly the kind of thing that gets silently lost during a rushed migration: someone ports the two obvious if blocks, misses the floor, and ships an app that can invoice a customer negative four dollars.

Now look at what a migration would actually require here. Suppose the plan is: move from SQLite to a different database, and move this function into a new service. You cannot do either one cleanly, because process_order is not one thing, it is six things stitched together: read a row, decide a price, write a row, charge a payment provider, publish an event, and send an email. Moving the database means touching lines that also do pricing. Moving the function into a new service means moving all six responsibilities as one inseparable block, whether the new service needs all of them or not.

Step 2: Lock In Current Behavior With a Characterization Test

Before restructuring anything, capture what the code actually does right now, not what you assume it does. This is called a characterization test, or a golden master test: you run the real code, record its real output, and turn that recording into an assertion. It does not check correctness. It checks that behavior does not drift while you refactor. For the technique in depth, see the companion tutorial linked in the prerequisites; here it is used just enough to get a safety net in place.

Capture the real output for all five seeded orders:

python -c "
from seed_db import fresh_db, SEED_ROWS
fresh_db()
from legacy_order_service import process_order

for row in SEED_ROWS:
    oid = row[0]
    result = process_order(oid)
    print(oid, '->', result)
"

ORD-1001 -> {'id': 'ORD-1001', ... 'total': 44.0, ... 'status': 'PROCESSED'}
ORD-1002 -> {'id': 'ORD-1002', ... 'total': 120.0, ... 'status': 'PROCESSED'}
ORD-1003 -> {'id': 'ORD-1003', ... 'total': 66.4, ... 'status': 'PROCESSED'}
ORD-1004 -> {'id': 'ORD-1004', ... 'total': 1.28, ... 'status': 'PROCESSED'}
ORD-1005 -> {'id': 'ORD-1005', ... 'total': 196.0, ... 'status': 'PROCESSED'}

Turn those five totals into a test. Save as test_characterization.py:

from seed_db import fresh_db

# Golden master: captured by actually running the tangled legacy function
# against the fixed seed data, before any refactor.
GOLDEN_MASTER = {
    "ORD-1001": 44.0,
    "ORD-1002": 120.0,
    "ORD-1003": 66.4,
    "ORD-1004": 1.28,
    "ORD-1005": 196.0,
}


def test_legacy_behavior_matches_golden_master():
    fresh_db()
    from legacy_order_service import process_order

    for order_id, expected_total in GOLDEN_MASTER.items():
        result = process_order(order_id)
        assert result["total"] == expected_total, (
            f"{order_id}: expected {expected_total}, got {result['total']}"
        )
        assert result["status"] == "PROCESSED"

python -m pytest test_characterization.py -v

test_characterization.py::test_legacy_behavior_matches_golden_master PASSED [100%]
1 passed in 0.09s

Every step from here on has to keep this test green. If it ever fails, you changed behavior, not just structure, and you need to know that immediately, not after deploying.

Step 3: Choose One Migration Boundary

Do not start with "let's clean up the whole file." Start with "what capability do we want to be able to move first?" Here, that capability is Process Order. Mapping it out before touching any code:

• Input: an order ID, and the row data it points to (base total, customer tier, country, payment method, customer email).
• Business decision: apply the tier discount, apply the country and payment method rebate, floor the result at zero.
• Side effects: save the new total and status, charge the payment provider, publish an event, send a confirmation email.
• Output: the processed order, with its final total and status.

This mapping is the actual outcome of this step, more than any code. Once you can write out inputs, decision, side effects, and output on their own lines, you know exactly what needs to become independently substitutable, and just as importantly, what you are not touching yet: nothing about how orders are listed, cancelled, or refunded elsewhere in a real application would be in scope for this pass.

Step 4: Extract the Business Rules Into a Pure Function

A pure function takes plain values in and returns a plain value out, with no database, no network call, and no dependency on anything outside its own arguments. Move the pricing decision out of process_order and into one. Save as order_domain.py:

from dataclasses import dataclass


@dataclass(frozen=True)
class OrderDecision:
    order_id: str
    total: float
    status: str
    customer_email: str


def decide_order_total(
    order_id: str,
    base_total: float,
    customer_tier: str,
    country: str,
    payment_method: str,
    customer_email: str,
) -> OrderDecision:
    """Pure business logic, extracted verbatim from the original function.
    Same inputs always produce the same output. No I/O of any kind here.
    """
    total = base_total

    if customer_tier == "GOLD":
        total = round(total * 0.88, 2)

    if country == "CA" and payment_method == "WIRE":
        total = round(total - 4.0, 2)

    if total < 0:
        total = 0.0

    return OrderDecision(
        order_id=order_id,
        total=total,
        status="PROCESSED",
        customer_email=customer_email,
    )

Every conditional in this function is byte-for-byte identical to the original: same multiplier, same rebate, same floor, in the same order. Nothing was "cleaned up" or rewritten. That is deliberate; a structural refactor should never require you to also re-derive business math from scratch.

Now test it directly, with no database and no fakes:

python -c "
from order_domain import decide_order_total

d = decide_order_total('ORD-1003', 80.0, 'GOLD', 'CA', 'WIRE', '[email protected]')
print(d)
"

OrderDecision(order_id='ORD-1003', total=66.4, status='PROCESSED', customer_email='[email protected]')

66.4 matches the golden master for ORD-1003 exactly, and this call took microseconds because there was no I/O to wait on. This is the first concrete payoff of extraction: business rules you could previously only exercise by seeding a database and calling a six-responsibility function can now be tested directly, as fast as you can write assertions.

Direct Unit Tests, With No Database At All

Turn that into a proper test file. Save as test_domain.py:

from order_domain import decide_order_total


def test_standard_tier_pays_full_price():
    d = decide_order_total("X", 100.0, "STANDARD", "US", "CARD", "[email protected]")
    assert d.total == 100.0


def test_gold_tier_gets_12_percent_off():
    d = decide_order_total("X", 100.0, "GOLD", "US", "CARD", "[email protected]")
    assert d.total == 88.0


def test_ca_wire_gets_flat_four_dollar_rebate():
    d = decide_order_total("X", 100.0, "STANDARD", "CA", "WIRE", "[email protected]")
    assert d.total == 96.0


def test_ca_wire_rebate_only_applies_to_wire_payments():
    d = decide_order_total("X", 100.0, "STANDARD", "CA", "CARD", "[email protected]")
    assert d.total == 100.0


def test_gold_and_ca_wire_combine_discount_before_rebate():
    # 100 * 0.88 = 88.0, then 88.0 - 4.0 = 84.0
    d = decide_order_total("X", 100.0, "GOLD", "CA", "WIRE", "[email protected]")
    assert d.total == 84.0


def test_total_never_goes_negative():
    d = decide_order_total("X", 3.0, "GOLD", "CA", "WIRE", "[email protected]")
    assert d.total == 0.0

python -m pytest test_domain.py -v

test_domain.py::test_standard_tier_pays_full_price PASSED                [ 16%]
test_domain.py::test_gold_tier_gets_12_percent_off PASSED                [ 33%]
test_domain.py::test_ca_wire_gets_flat_four_dollar_rebate PASSED         [ 50%]
test_domain.py::test_ca_wire_rebate_only_applies_to_wire_payments PASSED [ 66%]
test_domain.py::test_gold_and_ca_wire_combine_discount_before_rebate PASSED [ 83%]
test_domain.py::test_total_never_goes_negative PASSED                    [100%]

6 passed in 0.03s

Six cases, including the CA-without-WIRE case that confirms the rebate genuinely requires both conditions, and the floor-at-zero edge case discovered back in Step 1, all running in three hundredths of a second with no database connection anywhere in the process. This is the file the final verification step at the end of this tutorial runs alongside the others.

Step 5: Turn Side Effects Into Data

The pricing decision is now pure, but process_order still needs to persist the order, charge a payment, publish an event, and send an email. Instead of calling those systems directly from inside business logic, describe what should happen as plain data, and let a separate piece of code carry out the description. This split, business logic that only decides versus a thin outer layer that only performs, is sometimes called a functional core, imperative shell, a term from Gary Bernhardt's 2012 screencast of the same name.

Add these to order_domain.py:

@dataclass(frozen=True)
class PersistOrder:
    order_id: str
    total: float
    status: str


@dataclass(frozen=True)
class ChargePayment:
    order_id: str
    amount: float


@dataclass(frozen=True)
class PublishEvent:
    name: str
    payload: dict


@dataclass(frozen=True)
class SendEmail:
    to: str
    template: str
    context: dict


Effect = PersistOrder | ChargePayment | PublishEvent | SendEmail


def plan_order_effects(decision: OrderDecision) -> list[Effect]:
    """Turn a decision into an ordered list of effects to perform. This
    function still has no I/O; it just describes what should happen.
    """
    return [
        PersistOrder(
            order_id=decision.order_id,
            total=decision.total,
            status=decision.status,
        ),
        ChargePayment(order_id=decision.order_id, amount=decision.total),
        PublishEvent(
            name="order.processed",
            payload={"id": decision.order_id, "total": decision.total},
        ),
        SendEmail(
            to=decision.customer_email,
            template="order-processed",
            context={"order_id": decision.order_id, "total": decision.total},
        ),
    ]

Try it:

python -c "
from order_domain import decide_order_total, plan_order_effects

d = decide_order_total('ORD-1003', 80.0, 'GOLD', 'CA', 'WIRE', '[email protected]')
for e in plan_order_effects(d):
    print(' ', e)
"

  PersistOrder(order_id='ORD-1003', total=66.4, status='PROCESSED')
  ChargePayment(order_id='ORD-1003', amount=66.4)
  PublishEvent(name='order.processed', payload={'id': 'ORD-1003', 'total': 66.4})
  SendEmail(to='[email protected]', template='order-processed', context={'order_id': 'ORD-1003', 'total': 66.4})

plan_order_effects still has not opened a database connection or made a network call. It returns a list describing four things that should happen, in order. Notice the order itself is part of the decision now, visible and testable, instead of being an implicit consequence of the order statements happened to appear in the original function.

Step 6: Put External Systems Behind Adapters (This Is the Seam)

Now define what the business logic actually needs from the outside world, in the domain's own vocabulary, not in the vocabulary of SQLite or any particular payment SDK. Save as adapters.py:

import sqlite3
from typing import Protocol


class OrderRepository(Protocol):
    def load_order(self, order_id: str) -> dict | None: ...
    def save_order(self, order_id: str, total: float, status: str) -> None: ...


class PaymentGateway(Protocol):
    def charge(self, order_id: str, amount: float) -> None: ...


class EventPublisher(Protocol):
    def publish(self, name: str, payload: dict) -> None: ...


class EmailSender(Protocol):
    def send(self, to: str, template: str, context: dict) -> None: ...


# --- "current stack" adapters: same SQLite database and in-memory logs the
# tangled version used, just wrapped behind the interfaces above. ---


class SqliteOrderRepository:
    def __init__(self, db_path: str):
        self.db_path = db_path

    def load_order(self, order_id: str) -> dict | None:
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        cur = conn.cursor()
        cur.execute("SELECT * FROM orders WHERE id = ?", (order_id,))
        row = cur.fetchone()
        conn.close()
        return dict(row) if row is not None else None

    def save_order(self, order_id: str, total: float, status: str) -> None:
        conn = sqlite3.connect(self.db_path)
        conn.execute(
            "UPDATE orders SET total = ?, status = ? WHERE id = ?",
            (total, status, order_id),
        )
        conn.commit()
        conn.close()


class InMemoryPaymentGateway:
    def __init__(self):
        self.log: list[dict] = []

    def charge(self, order_id: str, amount: float) -> None:
        self.log.append({"order_id": order_id, "amount": amount})


class InMemoryEventPublisher:
    def __init__(self):
        self.log: list[dict] = []

    def publish(self, name: str, payload: dict) -> None:
        self.log.append({"name": name, "payload": payload})


class InMemoryEmailSender:
    def __init__(self):
        self.log: list[dict] = []

    def send(self, to: str, template: str, context: dict) -> None:
        self.log.append({"to": to, "template": template, "context": context})

Why Protocol Instead of Inheritance

typing.Protocol, added in Python 3.8, describes an interface without requiring anything to inherit from it. Per the official Python documentation, this is called structural subtyping, or static duck typing: a class satisfies OrderRepository simply by having methods named load_order and save_order with compatible signatures, with no class SqliteOrderRepository(OrderRepository) declaration needed anywhere. To be precise about what Protocol actually buys you here: Python's normal duck typing already lets SqliteOrderRepository work anywhere an OrderRepository is expected, at runtime, with or without Protocol. What Protocol adds is a name for the contract that a type checker like mypy or pyright can verify ahead of time, and a single place, this file, where a reader can see exactly what the domain expects from the outside world. You will see in Step 8 that a repository which does not even know this file exists can still satisfy the contract, because Python was checking shapes, not ancestry, the whole time.

Step 7: Reassemble the Orchestrator and Re-Run the Safety Net

One more piece connects the domain to the adapters: something that loads a raw order, asks the domain to decide, and applies the resulting effects. This orchestrator is the only file that imports both order_domain.py and adapters.py, and it is what "extract a cohesive application boundary" produces in practice: one small, movable unit that represents the whole Process Order capability. Save as order_service_refactored.py:

from adapters import EmailSender, EventPublisher, OrderRepository, PaymentGateway
from order_domain import (
    ChargePayment,
    Effect,
    OrderDecision,
    PersistOrder,
    PublishEvent,
    SendEmail,
    decide_order_total,
    plan_order_effects,
)


def apply_effect(
    effect: Effect,
    repo: OrderRepository,
    payments: PaymentGateway,
    events: EventPublisher,
    emails: EmailSender,
) -> None:
    match effect:
        case PersistOrder(order_id=order_id, total=total, status=status):
            repo.save_order(order_id, total, status)
        case ChargePayment(order_id=order_id, amount=amount):
            payments.charge(order_id, amount)
        case PublishEvent(name=name, payload=payload):
            events.publish(name, payload)
        case SendEmail(to=to, template=template, context=context):
            emails.send(to, template, context)


def process_order(
    order_id: str,
    repo: OrderRepository,
    payments: PaymentGateway,
    events: EventPublisher,
    emails: EmailSender,
) -> OrderDecision:
    raw = repo.load_order(order_id)
    if raw is None:
        raise ValueError(f"Order not found: {order_id}")

    decision = decide_order_total(
        order_id=raw["id"],
        base_total=raw["base_total"],
        customer_tier=raw["customer_tier"],
        country=raw["country"],
        payment_method=raw["payment_method"],
        customer_email=raw["customer_email"],
    )

    for effect in plan_order_effects(decision):
        apply_effect(effect, repo, payments, events, emails)

    return decision

Two details worth pointing out. First, apply_effect uses a match statement with class patterns, a feature added in Python 3.10. Each case both checks the effect's type and unpacks its fields in one step, using the keyword names declared on the dataclass. Second, and more important: process_order reads raw["base_total"], not raw["total"] the way the original did. The original function read and wrote the same total column, which meant calling it twice on the same order would apply the GOLD discount twice. Reading the never-mutated base_total column instead is a genuine improvement this refactor surfaced almost by accident, simply by forcing every input to be named explicitly.

Wire the same SQLite database and equivalent in-memory fakes into the new orchestrator, and run the exact same golden master against it. Save as test_refactored_characterization.py:

from adapters import (
    InMemoryEmailSender,
    InMemoryEventPublisher,
    InMemoryPaymentGateway,
    SqliteOrderRepository,
)
from order_service_refactored import process_order
from seed_db import DB_PATH, fresh_db
from test_characterization import GOLDEN_MASTER


def test_refactored_behavior_matches_golden_master():
    fresh_db()
    repo = SqliteOrderRepository(DB_PATH)
    payments = InMemoryPaymentGateway()
    events = InMemoryEventPublisher()
    emails = InMemoryEmailSender()

    for order_id, expected_total in GOLDEN_MASTER.items():
        decision = process_order(order_id, repo, payments, events, emails)
        assert decision.total == expected_total, (
            f"{order_id}: expected {expected_total}, got {decision.total}"
        )
        assert decision.status == "PROCESSED"

    assert len(payments.log) == len(GOLDEN_MASTER)
    assert len(events.log) == len(GOLDEN_MASTER)
    assert len(emails.log) == len(GOLDEN_MASTER)

python -m pytest test_characterization.py test_refactored_characterization.py -v

test_characterization.py::test_legacy_behavior_matches_golden_master PASSED [ 50%]
test_refactored_characterization.py::test_refactored_behavior_matches_golden_master PASSED [100%]
2 passed in 0.19s

Both the tangled original and the restructured version produce identical totals for every seeded order. The structure changed completely; the behavior did not change at all.

Step 8: Prove the Seam Works by Swapping In "the New Stack"

Everything up to this point could be dismissed as "looks nicer, hard to say if it actually helps." This step is the payoff: build a second repository, backed by a JSON file instead of SQLite, with internal field names that do not match the SQLite schema at all, and run the same orchestrator and the same golden master against it, changing nothing else. Save as new_stack_repository.py:

import json
import os


class JsonOrderRepository:
    def __init__(self, path: str):
        self.path = path

    def _read_all(self) -> dict:
        if not os.path.exists(self.path):
            return {}
        with open(self.path, "r", encoding="utf-8") as f:
            return json.load(f)

    def _write_all(self, data: dict) -> None:
        with open(self.path, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2)

    def load_order(self, order_id: str) -> dict | None:
        data = self._read_all()
        record = data.get(order_id)
        if record is None:
            return None
        # Deliberately different internal key names from the SQLite schema,
        # to prove the orchestrator only cares about the OrderRepository
        # contract, not any particular storage shape.
        return {
            "id": order_id,
            "base_total": record["amount_before_discounts"],
            "customer_tier": record["loyalty_tier"],
            "country": record["ship_country"],
            "payment_method": record["pay_method"],
            "customer_email": record["email"],
        }

    def save_order(self, order_id: str, total: float, status: str) -> None:
        data = self._read_all()
        if order_id not in data:
            raise ValueError(f"Order not found: {order_id}")
        data[order_id]["final_total"] = total
        data[order_id]["order_status"] = status
        self._write_all(data)


def seed_json_store(path: str) -> None:
    data = {
        "ORD-1001": {
            "amount_before_discounts": 50.00,
            "loyalty_tier": "GOLD",
            "ship_country": "US",
            "pay_method": "CARD",
            "email": "[email protected]",
        },
        "ORD-1002": {
            "amount_before_discounts": 120.00,
            "loyalty_tier": "STANDARD",
            "ship_country": "US",
            "pay_method": "CARD",
            "email": "[email protected]",
        },
        "ORD-1003": {
            "amount_before_discounts": 80.00,
            "loyalty_tier": "GOLD",
            "ship_country": "CA",
            "pay_method": "WIRE",
            "email": "[email protected]",
        },
        "ORD-1004": {
            "amount_before_discounts": 6.00,
            "loyalty_tier": "GOLD",
            "ship_country": "CA",
            "pay_method": "WIRE",
            "email": "[email protected]",
        },
        "ORD-1005": {
            "amount_before_discounts": 200.00,
            "loyalty_tier": "STANDARD",
            "ship_country": "CA",
            "pay_method": "WIRE",
            "email": "[email protected]",
        },
    }
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2)

Now run the same characterization logic against it. Save as test_new_stack_migration.py:

from adapters import InMemoryEmailSender, InMemoryEventPublisher, InMemoryPaymentGateway
from new_stack_repository import JsonOrderRepository, seed_json_store
from order_service_refactored import process_order
from test_characterization import GOLDEN_MASTER

JSON_PATH = "orders_new_stack.json"


def test_new_stack_repository_matches_golden_master():
    seed_json_store(JSON_PATH)
    repo = JsonOrderRepository(JSON_PATH)
    payments = InMemoryPaymentGateway()
    events = InMemoryEventPublisher()
    emails = InMemoryEmailSender()

    for order_id, expected_total in GOLDEN_MASTER.items():
        decision = process_order(order_id, repo, payments, events, emails)
        assert decision.total == expected_total, (
            f"{order_id}: expected {expected_total}, got {decision.total}"
        )

    # And the new store actually recorded the results, in its own shape.
    saved = repo._read_all()
    assert saved["ORD-1003"]["final_total"] == 66.4
    assert saved["ORD-1003"]["order_status"] == "PROCESSED"

python -m pytest test_new_stack_migration.py -v

test_new_stack_migration.py::test_new_stack_repository_matches_golden_master PASSED [100%]
1 passed in 0.06s

Inspect the resulting file directly:

python -c "
import json
with open('orders_new_stack.json') as f:
    data = json.load(f)
print(json.dumps(data['ORD-1003'], indent=2))
"

{
  "amount_before_discounts": 80.0,
  "loyalty_tier": "GOLD",
  "ship_country": "CA",
  "pay_method": "WIRE",
  "email": "[email protected]",
  "final_total": 66.4,
  "order_status": "PROCESSED"
}

$66.40, matching the golden master, produced by a repository with a completely different storage format and completely different field names, plugged into order_service_refactored.py with zero edits to that file, to order_domain.py, or to the Protocol definitions in adapters.py. This is what a capability actually being ready to migrate looks like in practice, not a diagram, a passing test against a live substitution.

Gotcha: A Believable Refactoring Mistake the Safety Net Actually Caught

Here is a mistake that is easy to make and easy to miss on a code review: reordering the two discount rules while "tidying up" decide_order_total. Try it. Save this as order_domain_buggy.py:

from dataclasses import dataclass


@dataclass(frozen=True)
class OrderDecision:
    order_id: str
    total: float
    status: str
    customer_email: str


def decide_order_total(
    order_id: str,
    base_total: float,
    customer_tier: str,
    country: str,
    payment_method: str,
    customer_email: str,
) -> OrderDecision:
    total = base_total

    # BUG: rebate applied before the percentage discount instead of after.
    if country == "CA" and payment_method == "WIRE":
        total = round(total - 4.0, 2)

    if customer_tier == "GOLD":
        total = round(total * 0.88, 2)

    if total < 0:
        total = 0.0

    return OrderDecision(
        order_id=order_id,
        total=total,
        status="PROCESSED",
        customer_email=customer_email,
    )

Both orderings look equally reasonable if you read the two rules in isolation. They are not equivalent, because one rule is a percentage and the other is a flat amount:

python -c "
from order_domain_buggy import decide_order_total as buggy_decide
from order_domain import decide_order_total as correct_decide

for order_id, base_total, tier, country, pm, email in [
    ('ORD-1003', 80.0, 'GOLD', 'CA', 'WIRE', '[email protected]'),
    ('ORD-1004', 6.0, 'GOLD', 'CA', 'WIRE', '[email protected]'),
]:
    correct = correct_decide(order_id, base_total, tier, country, pm, email)
    buggy = buggy_decide(order_id, base_total, tier, country, pm, email)
    print(order_id, 'correct total =', correct.total, ' buggy total =', buggy.total)
"

ORD-1003 correct total = 66.4  buggy total = 66.88
ORD-1004 correct total = 1.28  buggy total = 1.76

For ORD-1003: applying the discount first gives 80 times 0.88 equals 70.4, then minus 4 equals 66.40. Applying the rebate first gives 80 minus 4 equals 76, then times 0.88 equals 66.88. A 48-cent difference per order that would be invisible in a code review and only shows up as a slow drift in revenue reports weeks later. Run the characterization test against the buggy version and it fails immediately, with the exact number that is wrong:

AssertionError: ORD-1003: expected 66.4, got 66.88
assert 66.88 == 66.4
 +  where 66.88 = OrderDecision(order_id='ORD-1003', total=66.88, status='PROCESSED', customer_email='[email protected]').total

This is the entire point of writing the characterization test in Step 2 before touching any structure. A refactor that "looks the same" can still be wrong, and the only way to know for certain is to run the code and compare it against a recording of the real behavior, not against your mental model of what the code does.

Where AI Fits (and Where It Doesn't)

It is tempting to hand a function like decide_order_total to an AI assistant and ask for a cleanup. To see what that actually looks like, a local Ollama model (qwen3.5:4b) was given the original two-if-block function with this prompt: "Refactor it to compute the two discount rules using a list of rule objects applied in a loop, instead of two separate if-blocks, so it is easier to add new discount rules later. Keep behavior identical to the original." The real response included this excerpt, word for word, with indentation reduced from its original nested position for readability:

class _Rule(DiscountRule):
    def apply(self, total):
        if not isinstance(total, (int, float)):
            raise ValueError("Total must be a number")

        # Ensure no rounding errors during multiplication by using the rule's own logic for precision
        new_total = round(total * multiplier) / 100.0

        return new_total

round(total * multiplier) / 100.0 looks superficially like a rounding-to-cents idiom, but it rounds to the nearest whole number first, then divides by 100. For a $100 order at a 0.88 multiplier, that is round(88.0) / 100.0, which is 0.88, not 88.0: a 100x error, and one confidently introduced with a comment claiming it prevents rounding errors. Worse, the model's full response never actually reassembled a working replacement for decide_order_total. Saving the model's complete raw response to a file and checking what it actually defines confirms that directly (this is the model's real output being inspected, not a step you need Ollama installed to follow along with):

python -c "
with open('ai_suggestion_raw.txt', encoding='utf-8') as f:
    code = f.read()
ns = {}
exec(code, ns)
print('decide_order_total defined:', 'decide_order_total' in ns)
print('Names defined:', [n for n in ns if not n.startswith('__')])
"

decide_order_total defined: False
Names defined: ['DiscountRule', 'create_gold_rule', 'apply_rule', 'CanadianWireDiscountRule', 'main_logic_decide_order_total']

Dropping this straight into order_domain.py would have broken every caller instantly, because decide_order_total, the exact function every caller imports, was never actually defined. The lesson is not "AI cannot refactor code." It is that AI-suggested structural changes, exactly like human-suggested ones, need to clear the same safety net before you trust them. The characterization tests written in Step 2 do not care who wrote the change; they only care whether the golden master still passes. Treat any AI-generated refactor, including one that compiles cleanly and looks architecturally reasonable, as a draft that has not been verified yet, not as a finished change. This matches the source article's own advice to not let AI design the target architecture before you have decided the migration boundary yourself; here it goes one step further and shows a concrete, verified failure rather than just warning about the risk in the abstract.

Is This Capability Ready to Migrate? A Checklist

The Checklist Applied to This Refactor

Before actually moving a capability to new infrastructure, it helps to check it against a short list of questions. Applied honestly to the code built in this tutorial:

Six clear yeses and one honestly partial answer is a realistic outcome for a first pass. The point of the checklist is not to reach a perfect score; it is to turn "I think this is ready" into a list of specific, checkable claims.

What Not to Refactor Before Migration

Scope discipline matters here as much as the refactor itself. A structural pass that quietly turns into a rewrite defeats its own purpose, because now you have both a restructuring risk and a rewrite risk to verify at once.

Naming Everywhere

The field and variable names in this tutorial, total, order_id, status, were kept identical to the original tangled function throughout. Renaming things you are not required to touch inflates the diff and makes it harder to see the change that actually matters.

Formatting the Entire Repository

Running a formatter or linter across unrelated files in the same commit buries a focused structural change inside a wall of whitespace noise, and makes the actual review much harder.

Replacing Every Pattern

Only the Process Order capability was touched here. A real legacy codebase might have a dozen functions with the same SQL-plus-business-logic-plus-side-effects shape; rewriting all of them in the same pass, instead of proving the pattern on one first, multiplies the risk of the change before you have evidence it works.

Rewriting Stable Algorithms

The discount math itself, round(total * 0.88, 2) and round(total - 4.0, 2), was moved, not reimplemented. It would be tempting to "improve" it, using decimal.Decimal for currency instead of float, for instance. That is a legitimate idea, and it is also a behavior change that belongs in its own change, verified on its own, after the migration-prep refactor is already safely merged.

Common Mistakes and Gotchas

• Skipping the characterization test because it feels like ceremony for "obviously safe" changes. The order-of-operations bug above looked obviously safe too, right up until it was not.
• Assuming Protocol enforces anything at runtime. It does not, on its own. Nothing stops you from passing an object missing a save_order method to process_order; you would only find out when apply_effect tries to call it and raises AttributeError. Protocol's value is documentation and static type-checker support, not a runtime guardrail.
• Assuming effect order is arbitrary. plan_order_effects persists the order before charging payment, which is exactly the pairing called out as an undocumented risk in the checklist above: if ChargePayment ever fails, the database already says the order was processed. Reordering a list of effects like this during a refactor is easy to do silently, since nothing about the list literal signals which orderings are safe to change and which are not; the characterization tests in this tutorial check totals and status, not effect ordering, so a real production version would need an explicit test for that too.
• Treating the input column and the output column as the same thing. The original bug of reading and writing the same total column, discovered in Step 7, is a common shape: anywhere a "current value" field doubles as both an input to a calculation and the place that calculation's result gets stored, running the calculation twice silently compounds it.

How to Verify This Works End to End

Run every test built in this tutorial together, from a clean database:

python -m pytest test_domain.py test_characterization.py test_refactored_characterization.py test_new_stack_migration.py -v

test_domain.py::test_standard_tier_pays_full_price PASSED                [ 11%]
test_domain.py::test_gold_tier_gets_12_percent_off PASSED                [ 22%]
test_domain.py::test_ca_wire_gets_flat_four_dollar_rebate PASSED         [ 33%]
test_domain.py::test_ca_wire_rebate_only_applies_to_wire_payments PASSED [ 44%]
test_domain.py::test_gold_and_ca_wire_combine_discount_before_rebate PASSED [ 55%]
test_domain.py::test_total_never_goes_negative PASSED                    [ 66%]
test_characterization.py::test_legacy_behavior_matches_golden_master PASSED [ 77%]
test_refactored_characterization.py::test_refactored_behavior_matches_golden_master PASSED [ 88%]
test_new_stack_migration.py::test_new_stack_repository_matches_golden_master PASSED [100%]

9 passed in 0.18s

All nine tests passing together means the original tangled function's behavior is captured, the refactored orchestrator reproduces it exactly against the same database, and a structurally different data store reproduces it exactly too. That is what being ready to migrate looks like as a verifiable fact, not a guess.

As a final gut check, run the refactored orchestrator by hand for one order and compare its output and side effects against the very first output you captured in Step 1:

python -c "
from seed_db import fresh_db, DB_PATH
fresh_db()
from adapters import SqliteOrderRepository, InMemoryPaymentGateway, InMemoryEventPublisher, InMemoryEmailSender
from order_service_refactored import process_order

repo = SqliteOrderRepository(DB_PATH)
payments = InMemoryPaymentGateway()
events = InMemoryEventPublisher()
emails = InMemoryEmailSender()

decision = process_order('ORD-1003', repo, payments, events, emails)
print('decision:', decision)
print('payments.log:', payments.log)
print('events.log:', events.log)
print('emails.log:', emails.log)
"

decision: OrderDecision(order_id='ORD-1003', total=66.4, status='PROCESSED', customer_email='[email protected]')
payments.log: [{'order_id': 'ORD-1003', 'amount': 66.4}]
events.log: [{'name': 'order.processed', 'payload': {'id': 'ORD-1003', 'total': 66.4}}]
emails.log: [{'to': '[email protected]', 'template': 'order-processed', 'context': {'order_id': 'ORD-1003', 'total': 66.4}}]

$66.40 charged, published, and emailed, exactly matching the tangled version's output for the same order back in Step 1.

Next Steps

This tutorial covered one capability in one hypothetical application. A real legacy system will have dozens, and not all of them will be this clean; some will have side effects that depend on each other's results, or business rules that call out to other parts of the system mid-calculation. The approach scales the same way regardless: pick one boundary, write a characterization test, extract the pure decision, turn side effects into data, put dependencies behind a Protocol, and prove the seam with a real substitution before you trust it.

From here:

• Read how to safely refactor legacy Python code using characterization tests for a deeper look at the safety-net technique used in Step 2, including how to characterize code with far messier or non-deterministic behavior than this tutorial's example.
• Read how to investigate a legacy codebase with AI before you change it for the step that logically comes before this one: understanding what an unfamiliar codebase actually does before you decide what to extract.
• Try applying this same seam-and-adapter pattern to a second capability in the same toy application, for example a cancel_order function, and notice which adapters you can reuse from adapters.py without any changes at all. That reuse is the second concrete payoff of doing this work capability by capability instead of all at once.