Find the Seam Before You Cut: A Read/Write Census for Messy Modules

Most extractions fail for the same reason. The function you moved was writing module state all along. Tests still pass, coverage still looks fine, and production drifts three weeks later.

Length is not the signal. Shared mutable state is the signal. A 90-line function that only reads constants is safer to move than a 12-line function that appends to a module-level list.

So census the module before you read it. Then pin the blast radius. Then make the smallest change in one commit.

Everything below runs on the Python standard library. No plugins, no framework, no service accounts.

Step 1: Score every function by what it touches

Write the census before you form an opinion about the code. The script below parses one module and scores each function by four things: module-level writes, aliases, reads, and calls.

# seam_census.py -- rank functions by how hard they are to extract safely.
# Python 3.10+, standard library only.
from __future__ import annotations

import ast
import sys
from dataclasses import dataclass, field
from pathlib import Path

MUTATORS = {
    "append", "extend", "insert", "remove", "pop", "clear", "sort",
    "update", "setdefault", "add", "discard", "write", "close",
}


def target_names(node):
    if isinstance(node, ast.Name):
        return {node.id}
    if isinstance(node, (ast.Tuple, ast.List)):
        out = set()
        for elt in node.elts:
            out |= target_names(elt)
        return out
    return set()


def bound_names(stmts):
    """Names bound at module level, plus one level of nesting."""
    out = set()
    for node in stmts:
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
            out.add(node.name)
        elif isinstance(node, ast.Assign):
            for t in node.targets:
                out |= target_names(t)
        elif isinstance(node, (ast.AnnAssign, ast.AugAssign)) and node.target is not None:
            out |= target_names(node.target)
        elif isinstance(node, ast.Import):
            out |= {a.asname or a.name.split(".")[0] for a in node.names}
        elif isinstance(node, ast.ImportFrom):
            out |= {a.asname or a.name for a in node.names}
        elif isinstance(node, (ast.For, ast.AsyncFor)):
            out |= target_names(node.target) | bound_names(node.body) | bound_names(node.orelse)
        elif isinstance(node, (ast.With, ast.AsyncWith)):
            for item in node.items:
                if item.optional_vars is not None:
                    out |= target_names(item.optional_vars)
            out |= bound_names(node.body)
        elif isinstance(node, ast.Try):
            out |= bound_names(node.body) | bound_names(node.orelse) | bound_names(node.finalbody)
            for handler in node.handlers:
                if handler.name:
                    out.add(handler.name)
                out |= bound_names(handler.body)
        elif isinstance(node, (ast.If, ast.While)):
            out |= bound_names(node.body) | bound_names(node.orelse)
    return out


@dataclass
class Seam:
    qualname: str
    lineno: int
    writes: set[str] = field(default_factory=set)
    aliases: set[str] = field(default_factory=set)
    reads: set[str] = field(default_factory=set)
    calls: set[str] = field(default_factory=set)

    @property
    def score(self) -> int:
        return 3 * len(self.writes) + 2 * len(self.aliases) + len(self.reads) + len(self.calls)


def scan(fn, module_names):
    declared = {a.arg for a in fn.args.posonlyargs + fn.args.args + fn.args.kwonlyargs}
    if fn.args.vararg:
        declared.add(fn.args.vararg.arg)
    if fn.args.kwarg:
        declared.add(fn.args.kwarg.arg)
    declared |= bound_names(fn.body)

    seam = Seam(fn.name, fn.lineno)
    declared_global = set()

    for node in ast.walk(fn):
        if isinstance(node, ast.Global):
            declared_global |= set(node.names)
        elif isinstance(node, ast.Name) and node.id in module_names and node.id not in declared:
            if isinstance(node.ctx, ast.Store) and node.id in declared_global:
                seam.writes.add(node.id)
            elif isinstance(node.ctx, ast.Load):
                seam.reads.add(node.id)
        elif isinstance(node, ast.Call):
            func = node.func
            if isinstance(func, ast.Name) and func.id in module_names and func.id not in declared:
                seam.calls.add(func.id)
            elif isinstance(func, ast.Attribute) and func.attr in MUTATORS:
                base = func.value
                if isinstance(base, ast.Name) and base.id in module_names and base.id not in declared:
                    seam.aliases.add(f"{base.id}.{func.attr}()")
        elif isinstance(node, ast.Subscript) and isinstance(node.ctx, ast.Store):
            base = node.value
            if isinstance(base, ast.Name) and base.id in module_names and base.id not in declared:
                seam.aliases.add(f"{base.id}[...] =")

    seam.reads -= seam.writes
    return seam


def main(path: str) -> None:
    tree = ast.parse(Path(path).read_text(encoding="utf-8"))
    module_names = bound_names(tree.body)
    seams = [
        scan(node, module_names)
        for node in ast.walk(tree)
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
    ]
    seams.sort(key=lambda s: (-s.score, s.lineno))
    print(f"{'score':>5}  {'function':<26} writes alias reads calls")
    for seam in seams:
        print(
            f"{seam.score:>5}  {seam.qualname:<26} "
            f"{len(seam.writes):>6} {len(seam.aliases):>5} {len(seam.reads):>5} {len(seam.calls):>5}"
        )
        if seam.writes:
            print(f"       writes: {', '.join(sorted(seam.writes))}")


if __name__ == "__main__":
    main(sys.argv[1])

Run it against one file.

python seam_census.py billing/legacy.py

The weights are deliberate, and you should tune them. A module-level write costs three points. A mutation through a shared object costs two. A read costs one, because constants are cheap to move.

Step 2: Read the table, not the source

The output below comes from a synthetic two-module fixture. It is not a production repo, and the numbers are not a benchmark. Use it to calibrate your reading of shape, not size.

score  function                   writes alias reads calls
    8  recompute_totals              1     2     1     0
       writes: LEDGER
    5  apply_fee                     0     1     2     1
    2  format_statement              0     0     2     0

Three things jump out immediately. recompute_totals is the only writer, and it also mutates a cache through a subscript assignment. apply_fee appends to LEDGER through an alias, which is a hidden write from the caller's perspective. format_statement is pure and cheap to move wherever you want.

The score is not a quality judgement. It is a risk estimate for one specific operation: moving code across a module boundary.

Step 3: Turn the table into a decision

Use this as a starting rule. Calibrate thresholds against your own repo, because your call graph will differ.

Score	Shape	Action this commit
0-2	Reads only, no alias	Extract now, one function, one commit
3-5	One alias, no writes	Extract, keep the alias call at the original call site
6+	Writes module state	Freeze state first, do not extract yet
any	Writes 2+ module names	Stop, the module has two responsibilities

The last row is the important one. If a function writes two module-level names, the module owns two separate concerns. Extracting the function moves the coupling instead of removing it.

Step 4: Pin the blast radius with one characterization test

Now write a test that asserts nothing about correctness. It asserts what the module mutates. That is the characterization part.

# tests/test_legacy_blast_radius.py
import copy
import importlib

import pytest

TARGET = "billing.legacy"


def snapshot(mod):
    snap = {}
    for name, value in vars(mod).items():
        if name.startswith("__") and name.endswith("__"):
            continue
        try:
            snap[name] = copy.deepcopy(value)
        except Exception:
            snap[name] = repr(value)  # last resort, unstable, allowlist it
    return snap


def changed(before, after):
    keys = set(before) | set(after)
    return sorted(k for k in keys if before.get(k) != after.get(k))


@pytest.mark.parametrize(
    "call, expected",
    [
        (lambda m: m.apply_fee("acct-1", 250), ["LEDGER"]),
        (lambda m: m.format_statement("acct-1"), []),
        (lambda m: m.recompute_totals(), ["LEDGER", "_cache", "RATES"]),
    ],
)
def test_blast_radius_is_frozen(call, expected):
    mod = importlib.import_module(TARGET)
    before = snapshot(mod)
    call(mod)
    assert changed(before, snapshot(mod)) == expected

Run it once against the untouched module. Whatever it reports becomes your baseline. If the baseline surprises you, that is a finding, and it is worth more than the refactor.

Step 5: Make the smallest change and re-run the census

Move exactly one function. Keep the alias call at the original call site if the census flagged one. Then re-run both artifacts.

The invariant is simple. The parametrized expected lists must not change. If the diff set grows by one key after the move, you introduced a second writer. Revert the move, not the test.

The second check is the census itself. The extracted function's reads should stay roughly the same. Its writes and aliases must be empty before extraction. If they are not, you have not found a seam yet.

One commit, one function, two artifacts. That is the whole loop.

Where a hosted workspace fits

The census needs two things: the repo on disk and a Python interpreter. Nothing else, which is why it works well as a first pass on unfamiliar code.

If the target repo's toolchain is not installable on your machine, MonkeyCode offers a free server option for running this kind of script in a hosted workspace, plus free model access for drafting the AST visitor for a language variant. Treat both as operator-supplied availability claims and check the current terms before you depend on them. Disclosure: This article was prepared as part of MonkeyCode's product outreach.

The census output stays the authority. Generated code that disagrees with the table is a bug in the generated code.

Limitations and who should skip this

Static AST analysis has blind spots, and they are large. getattr, exec, dynamic imports, decorators that rewrite globals, and monkeypatched tests are all invisible here. ast.walk also descends into nested scopes, so a closure that declares global gets counted as a write of the enclosing function.

The score over-counts read-only constants. A module that defines forty config strings will look riskier than it is. Raise the read weight to zero if your modules are config-heavy.

The repr fallback in the snapshot helper is genuinely unstable. Anything holding a file handle, socket, or lock should be allowlisted out of the diff rather than stringified.

Skip this approach if your repo already has mutation-tested coverage. Characterization tests are a substitute for missing observability, not a replacement for real assertions. Skip it for greenfield code too, since you can design the boundary instead of discovering it.

Do not skip it if you are three days into a module nobody on the team wrote. The table will tell you more in thirty seconds than an afternoon of scrolling.

Run the census on the one module you keep postponing. The read/write table is a better code-review artifact than a diff of a moved function.