I Deleted 18,347 Lines of Python Code Without Removing a Single Feature

Deleting code sounds easy until the code belongs to a running product.

A function may have no direct callers but still be loaded through a plugin registry. A serializer may look duplicated but quietly preserve an old field name used by one customer. A command may not appear in analytics because it runs from cron at 3:10 a.m. on the first Sunday of each month. Python makes this even more fun because imports, decorators, entry points, reflection, monkey patches, and strings can all become hidden edges in the dependency graph.

The project in this story was a multi-tenant reporting backend written in Python. It accepted events, stored normalized records, generated reports, exported CSV and JSON files, and delivered them through HTTP, email, and object storage. Nothing huge. Around 140 API endpoints, 46 background tasks, PostgreSQL, Redis, and a queue.

The repository contained 62,914 lines of Python excluding tests and migrations.

That number was not the actual problem. The problem was that a small change in report filtering could require edits in the API schema, a service class, a repository, a filter translator, a query builder, an export adapter, and several nearly identical tests. The system had layers, but the layers did not reduce complexity. They distributed it.

The first plan was a rewrite. Fortunately, that plan died before production did.

Instead, the question became much simpler:

How much code can disappear while externally observable behavior remains unchanged?

That wording changed the whole project.

The codebase was not large because the product was complex

The first useful discovery was slightly embarrassing. Most of the code did not represent product features. It represented different ways of reaching the same feature.

There were four report-building pipelines. One served the REST API, one generated scheduled reports, one handled CSV exports, and one existed for an old internal command-line tool. Each pipeline normalized filters, checked permissions, built nearly identical SQL, mapped database rows, and formatted the result.

The implementations had drifted just enough to look intentional.

One accepted an empty list as no filter. Another treated it as a query that matched nothing. One converted timestamps to UTC in the repository. Another did it in the serializer. The CSV path rounded decimal values during extraction, while the API rounded them during output formatting.

No single implementation was obviously wrong. That was the trap.

The codebase also contained 73 repository classes. Most wrapped fewer than three SQLAlchemy calls. Several existed only to rename methods:

fetch_report became get_reportstore_report became savefind_for_account became list_by_tenant

This did not isolate the database. Tests mocked repository methods so heavily that they could pass even when the real SQL was broken. The abstraction had removed useful details from tests while keeping all the complexity in production.

Have you ever opened a class and found that every method only calls another class with almost the same arguments? That was roughly one fifth of the backend.

Before deleting anything, we classified code into four rough groups:

• domain behavior that users could observe;

• integration code that talked to databases, queues, storage, or external APIs;

• coordination code that connected operations;

• structural code that existed mostly because an earlier design expected future complexity.

The last group was enormous.

Future flexibility had arrived years ago. The future itself had not.

Counting lines was the least important part

The target was not really 18,000 lines. That number appeared only after the work was finished.

At the start, line count was treated as a warning light, not a success metric. Removing ten clear lines can be better than removing a thousand lines and replacing them with a dependency nobody understands.

We tracked several other values:

• Python source lines excluding tests, generated files, and migrations;

• number of import edges between internal packages;

• number of modules loaded during each production workflow;

• test duration;

• mutation score for critical business rules;

• number of database queries per workflow;

• number of distinct implementations of the same domain operation;

• percentage of functions observed in production traces;

• number of mocks used in the test suite.

The mock count turned out to be surprisingly useful. At the beginning, the tests contained 2,214 calls to patch, monkeypatch, Mock, or AsyncMock. Many tests replaced five or six internal objects before reaching the function under test.

After the cleanup, that number dropped below 900.

This did not happen because mocking is bad. It happened because many internal boundaries disappeared. A pure function that receives values and returns values does not need a mocked repository factory, mocked settings object, mocked serializer registry, and mocked clock.

The import graph changed even more clearly. One package named reporting depended directly or indirectly on 41 internal packages. After the work, it depended on 17.

That was the first point where the project started feeling smaller, even before the line count dropped much.

Less code is nice. Fewer possible paths through the code are better.

First we had to define what a feature actually was

The phrase without losing a feature is dangerous because teams rarely have a complete list of features.

The product manager had a list of visible screens and API capabilities. The support team knew about undocumented customer workflows. Operations knew about maintenance commands. Finance knew about exports used during invoicing. The database knew about everything else.

So we built a behavioral inventory from several sources:

• OpenAPI routes;

• command-line entry points;

• queue task registrations;

• scheduler configuration;

• application logs;

• reverse-proxy access logs;

• audit events;

• production traces;

• support documentation;

• database tables and materialized views;

• external webhook configurations;

• object-storage file names;

• tests.

This produced 286 observable workflows.

A workflow was not just an endpoint. It was something like:

User requests a CSV report with two date filters,the account timezone is Europe/Riga,one column contains a nullable decimal,the result exceeds the synchronous export limit,a background task writes the file to object storage,and an email with a temporary link is sent.

That sounds annoyingly specific. It needed to be.

A generic test saying CSV export works would not catch a timezone conversion moving from the query layer to the formatting layer. It would not catch null values becoming zero. It would not catch a changed filename that broke a customer import script.

The inventory also revealed a few so-called features that nobody could trigger.

One endpoint had returned HTTP 410 for more than three years. Two command-line commands imported modules that no longer existed. A storage adapter for an abandoned provider could not be instantiated because its configuration model had been removed.

Were those still features?

We decided that code was removable when all four conditions were true:

• no supported workflow required it;

• no production observation showed it running;

• no documented compatibility contract included it;

• removal did not change recorded behavior of supported workflows.

That rule was conservative. It was supposed to be.

The first safety net was a behavioral snapshot, not more mocks

The existing tests were useful but not enough. They mostly checked internal calls. A service test might verify that repository.fetch was called with normalized filters, but it did not verify the actual SQL result or final API payload.

For the cleanup, we created a small characterization framework.

It executed real workflows against a temporary PostgreSQL database, captured externally visible output, normalized unstable values, and stored the result as a readable snapshot. The same scenario could then run against the old path and the new path.

Here is a simplified version of the harness.

from __future__ import annotationsimport asyncioimport hashlibimport jsonimport refrom collections.abc import Awaitable, Callable, Mapping, Sequencefrom dataclasses import asdict, dataclassfrom datetime import UTC, datetimefrom decimal import Decimalfrom pathlib import Pathfrom typing import Any, Protocol, TypeAliasJSONScalar: TypeAlias = str | int | float | bool | NoneJSONValue: TypeAlias = JSONScalar | list["JSONValue"] | dict[str, "JSONValue"]class AsyncScenario(Protocol):    async def __call__(self, context: "ScenarioContext") -> "ObservedBehavior":        ...@dataclass(frozen=True, slots=True)class ScenarioContext:    tenant_id: str    user_id: str    now: datetime    database_url: str@dataclass(frozen=True, slots=True)class HttpObservation:    status_code: int    headers: Mapping[str, str]    body: JSONValue@dataclass(frozen=True, slots=True)class DatabaseObservation:    table: str    rows: Sequence[Mapping[str, JSONValue]]@dataclass(frozen=True, slots=True)class MessageObservation:    topic: str    payload: JSONValue@dataclass(frozen=True, slots=True)class FileObservation:    path: str    sha256: str    size: int@dataclass(frozen=True, slots=True)class ObservedBehavior:    http: Sequence[HttpObservation]    database: Sequence[DatabaseObservation]    messages: Sequence[MessageObservation]    files: Sequence[FileObservation]UNSTABLE_HEADERS = {    "date",    "server",    "x-request-id",    "x-runtime-ms",}UUID_PATTERN = re.compile(    r"\b[0-9a-f]{8}-[0-9a-f]{4}-"    r"[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-"    r"[0-9a-f]{12}\b",    flags=re.IGNORECASE,)def normalize_value(value: Any) -> JSONValue:    if value is None or isinstance(value, (str, int, float, bool)):        if isinstance(value, str):            return UUID_PATTERN.sub("<uuid>", value)        return value    if isinstance(value, Decimal):        return format(value, "f")    if isinstance(value, datetime):        if value.tzinfo is None:            raise ValueError("naive datetime found in observed behavior")        return value.astimezone(UTC).isoformat()    if isinstance(value, Mapping):        return {            str(key): normalize_value(item)            for key, item in sorted(value.items(), key=lambda pair: str(pair[0]))        }    if isinstance(value, Sequence) and not isinstance(value, (bytes, bytearray)):        return [normalize_value(item) for item in value]    raise TypeError(f"unsupported snapshot value: {type(value)!r}")def normalize_behavior(behavior: ObservedBehavior) -> dict[str, JSONValue]:    http = []    for item in behavior.http:        stable_headers = {            key.lower(): value            for key, value in item.headers.items()            if key.lower() not in UNSTABLE_HEADERS        }        http.append(            {                "status_code": item.status_code,                "headers": normalize_value(stable_headers),                "body": normalize_value(item.body),            }        )    database = [        {            "table": item.table,            "rows": normalize_value(                sorted(                    item.rows,                    key=lambda row: json.dumps(                        normalize_value(row),                        sort_keys=True,                    ),                )            ),        }        for item in behavior.database    ]    messages = [        {            "topic": item.topic,            "payload": normalize_value(item.payload),        }        for item in behavior.messages    ]    files = [        {            "path": item.path,            "sha256": item.sha256,            "size": item.size,        }        for item in behavior.files    ]    return {        "http": http,        "database": database,        "messages": messages,        "files": files,    }def snapshot_path(snapshot_directory: Path, scenario_name: str) -> Path:    safe_name = re.sub(r"[^a-zA-Z0-9_.-]+", "_", scenario_name)    return snapshot_directory / f"{safe_name}.json"async def assert_scenario_unchanged(    *,    scenario_name: str,    scenario: AsyncScenario,    context: ScenarioContext,    snapshot_directory: Path,    update_snapshots: bool = False,) -> None:    behavior = await scenario(context)    normalized = normalize_behavior(behavior)    serialized = json.dumps(        normalized,        indent=2,        sort_keys=True,        ensure_ascii=False,    ) + "\n"    path = snapshot_path(snapshot_directory, scenario_name)    if update_snapshots:        path.parent.mkdir(parents=True, exist_ok=True)        path.write_text(serialized, encoding="utf-8")        return    if not path.exists():        raise AssertionError(            f"snapshot does not exist for {scenario_name!r}; "            f"run with update_snapshots=True after reviewing the behavior"        )    expected = path.read_text(encoding="utf-8")    if serialized != expected:        expected_hash = hashlib.sha256(expected.encode()).hexdigest()[:12]        actual_hash = hashlib.sha256(serialized.encode()).hexdigest()[:12]        raise AssertionError(            f"behavior changed in {scenario_name!r}: "            f"expected={expected_hash}, actual={actual_hash}\n"            f"Inspect {path} and the generated observation before updating it."        )async def example_export_scenario(    context: ScenarioContext,) -> ObservedBehavior:    response = await call_application(        method="POST",        path="/v1/reports/export",        headers={            "x-tenant-id": context.tenant_id,            "x-user-id": context.user_id,        },        json_body={            "format": "csv",            "filters": {                "created_from": "2026-01-01T00:00:00+02:00",                "created_to": "2026-01-31T23:59:59+02:00",            },        },    )    await run_queued_tasks_until_idle()    rows = await read_database_rows(        table="report_exports",        tenant_id=context.tenant_id,    )    messages = await read_published_messages()    generated_files = await read_generated_files()    return ObservedBehavior(        http=[            HttpObservation(                status_code=response.status_code,                headers=response.headers,                body=response.json(),            )        ],        database=[            DatabaseObservation(                table="report_exports",                rows=rows,            )        ],        messages=[            MessageObservation(                topic=message.topic,                payload=message.payload,            )            for message in messages        ],        files=[            FileObservation(                path=file.path,                sha256=file.sha256,                size=file.size,            )            for file in generated_files        ],    )async def main() -> None:    await assert_scenario_unchanged(        scenario_name="async_csv_export_with_timezone",        scenario=example_export_scenario,        context=ScenarioContext(            tenant_id="tenant-demo",            user_id="user-demo",            now=datetime(2026, 2, 1, 12, 0, tzinfo=UTC),            database_url="postgresql://localhost/test",        ),        snapshot_directory=Path("tests/behavior_snapshots"),    )# These functions belong to the application-specific test environment.async def call_application(**kwargs: Any) -> Any:    raise NotImplementedErrorasync def run_queued_tasks_until_idle() -> None:    raise NotImplementedErrorasync def read_database_rows(**kwargs: Any) -> list[dict[str, JSONValue]]:    raise NotImplementedErrorasync def read_published_messages() -> list[Any]:    raise NotImplementedErrorasync def read_generated_files() -> list[Any]:    raise NotImplementedErrorif __name__ == "__main__":    asyncio.run(main())

These snapshots were not blindly approved golden files.

Every changed snapshot required review. If a timestamp moved because normalization was incomplete, we fixed the normalizer. If row ordering changed but the API never promised ordering, we decided whether the scenario should sort. If a JSON field disappeared, someone had to explain why.

The important part was that tests now protected visible behavior rather than class structure.

This gave us permission to make internal code ugly for a moment, move responsibilities, combine modules, and delete interfaces without rewriting hundreds of mocks after every step.

Static analysis found less dead code than expected

The first deletion pass used ordinary tools:

• coverage.py;

• Vulture;

• Ruff;

• import graph inspection;

• mypy;

• pytest;

• production traces.

The result looked impressive. Thousands of lines appeared unused.

Most of them were not immediately safe to delete.

Python allows code to be reached without a normal call expression. We had task functions registered by decorators, serializers loaded from dictionaries, exporters selected by configuration strings, Alembic hooks, Click commands, and package entry points.

A basic static analyzer sees this:

EXPORTERS = {    "csv": CsvExporter,    "json": JsonExporter,}

It may understand the class references. But if the registry itself is assembled through imports, plugins, or decorators, the graph becomes less obvious.

Runtime coverage had the opposite problem. Code not executed during an observation window is not necessarily dead. The monthly invoicing task did not run during our first two-week trace. A disaster-recovery command hopefully runs almost never.

So every deletion candidate received evidence from several directions.

We assigned a rough confidence score:

+3 no static references+3 not imported during any observed workflow+2 no production execution during 60 days+2 absent from documentation and scheduler configuration+1 no database artifacts associated with it-4 referenced by string-5 registered through a decorator or entry point-6 associated with maintenance, recovery, billing, or security

The score did not delete code automatically. It decided which code deserved human investigation first.

That distinction saved us from several stupid mistakes.

One module had no imports anywhere in the application. It looked completely dead. The container started it through python -m from a Kubernetes CronJob definition stored in another repository.

Another function had no callers because PostgreSQL invoked it through a notification listener that resolved handlers by name.

The boring lesson was that dead-code analysis is a repository problem only in small projects. In production, it is a system problem.

We built a runtime import and call inventory

To close the gap between static analysis and production reality, we added lightweight instrumentation.

Full function tracing would have been far too expensive, so we focused on module imports, registered entry points, task dispatch, HTTP handlers, command execution, and selected service boundaries.

Each observation included:

• deployment version;

• module or callable identifier;

• workflow category;

• tenant class rather than tenant identity;

• timestamp bucket;

• execution count;

• success or failure;

• caller category.

We did not send arguments or user data.

The inventory ran for sixty days. That covered daily, weekly, and monthly workflows, but we still manually reviewed low-frequency operations.

The following simplified tool combines AST analysis with runtime observations. It is not a perfect dead-code detector. It produces a review queue, which is much safer.

from __future__ import annotationsimport argparseimport astimport jsonfrom collections import defaultdict, dequefrom dataclasses import asdict, dataclassfrom pathlib import Pathfrom typing import Iterable, Iterator@dataclass(frozen=True, slots=True)class Symbol:    module: str    qualname: str    path: str    line: int    kind: str    @property    def key(self) -> str:        return f"{self.module}:{self.qualname}"@dataclass(frozen=True, slots=True)class Reference:    source: str    target_name: str    kind: str    line: int@dataclass(frozen=True, slots=True)class Candidate:    symbol: Symbol    static_reference_count: int    runtime_execution_count: int    imported_at_runtime: bool    exported: bool    decorated: bool    suspicious_string_reference: bool    confidence: int    reasons: tuple[str, ...]class ModuleAnalyzer(ast.NodeVisitor):    def __init__(self, *, module: str, path: Path) -> None:        self.module = module        self.path = path        self.scope: deque[str] = deque()        self.symbols: list[Symbol] = []        self.references: list[Reference] = []        self.exports: set[str] = set()        self.decorated_symbols: set[str] = set()        self.string_literals: set[str] = set()    def current_qualname(self, name: str) -> str:        if not self.scope:            return name        return ".".join([*self.scope, name])    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:        qualname = self.current_qualname(node.name)        self.symbols.append(            Symbol(                module=self.module,                qualname=qualname,                path=str(self.path),                line=node.lineno,                kind="function",            )        )        if node.decorator_list:            self.decorated_symbols.add(qualname)        self.scope.append(node.name)        self.generic_visit(node)        self.scope.pop()    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:        qualname = self.current_qualname(node.name)        self.symbols.append(            Symbol(                module=self.module,                qualname=qualname,                path=str(self.path),                line=node.lineno,                kind="async_function",            )        )        if node.decorator_list:            self.decorated_symbols.add(qualname)        self.scope.append(node.name)        self.generic_visit(node)        self.scope.pop()    def visit_ClassDef(self, node: ast.ClassDef) -> None:        qualname = self.current_qualname(node.name)        self.symbols.append(            Symbol(                module=self.module,                qualname=qualname,                path=str(self.path),                line=node.lineno,                kind="class",            )        )        if node.decorator_list:            self.decorated_symbols.add(qualname)        self.scope.append(node.name)        self.generic_visit(node)        self.scope.pop()    def visit_Call(self, node: ast.Call) -> None:        target = dotted_name(node.func)        if target:            self.references.append(                Reference(                    source=self.module,                    target_name=target,                    kind="call",                    line=node.lineno,                )            )        self.generic_visit(node)    def visit_Name(self, node: ast.Name) -> None:        if isinstance(node.ctx, ast.Load):            self.references.append(                Reference(                    source=self.module,                    target_name=node.id,                    kind="name",                    line=node.lineno,                )            )    def visit_Attribute(self, node: ast.Attribute) -> None:        target = dotted_name(node)        if target:            self.references.append(                Reference(                    source=self.module,                    target_name=target,                    kind="attribute",                    line=node.lineno,                )            )        self.generic_visit(node)    def visit_Constant(self, node: ast.Constant) -> None:        if isinstance(node.value, str):            self.string_literals.add(node.value)    def visit_Assign(self, node: ast.Assign) -> None:        for target in node.targets:            if isinstance(target, ast.Name) and target.id == "__all__":                values = extract_string_sequence(node.value)                self.exports.update(values)        self.generic_visit(node)def dotted_name(node: ast.AST) -> str | None:    parts: list[str] = []    current = node    while isinstance(current, ast.Attribute):        parts.append(current.attr)        current = current.value    if isinstance(current, ast.Name):        parts.append(current.id)        return ".".join(reversed(parts))    return Nonedef extract_string_sequence(node: ast.AST) -> set[str]:    if not isinstance(node, (ast.List, ast.Tuple, ast.Set)):        return set()    result = set()    for item in node.elts:        if isinstance(item, ast.Constant) and isinstance(item.value, str):            result.add(item.value)    return resultdef module_name(root: Path, path: Path) -> str:    relative = path.relative_to(root).with_suffix("")    parts = list(relative.parts)    if parts[-1] == "__init__":        parts.pop()    return ".".join(parts)def iter_python_files(root: Path) -> Iterator[Path]:    for path in root.rglob("*.py"):        if any(            part in {                ".venv",                "venv",                "__pycache__",                "migrations",                "generated",            }            for part in path.parts        ):            continue        yield pathdef analyze_repository(root: Path) -> tuple[    list[Symbol],    list[Reference],    set[str],    set[str],    set[str],]:    symbols: list[Symbol] = []    references: list[Reference] = []    exports: set[str] = set()    decorated: set[str] = set()    strings: set[str] = set()    for path in iter_python_files(root):        module = module_name(root, path)        source = path.read_text(encoding="utf-8")        tree = ast.parse(source, filename=str(path))        analyzer = ModuleAnalyzer(module=module, path=path)        analyzer.visit(tree)        symbols.extend(analyzer.symbols)        references.extend(analyzer.references)        strings.update(analyzer.string_literals)        exports.update(            f"{module}:{name}"            for name in analyzer.exports        )        decorated.update(            f"{module}:{name}"            for name in analyzer.decorated_symbols        )    return symbols, references, exports, decorated, stringsdef load_runtime_inventory(path: Path) -> dict[str, int]:    if not path.exists():        return {}    raw = json.loads(path.read_text(encoding="utf-8"))    if not isinstance(raw, dict):        raise ValueError("runtime inventory must be a JSON object")    return {        str(key): int(value)        for key, value in raw.items()    }def load_imported_modules(path: Path) -> set[str]:    if not path.exists():        return set()    raw = json.loads(path.read_text(encoding="utf-8"))    if not isinstance(raw, list):        raise ValueError("import inventory must be a JSON list")    return {str(item) for item in raw}def reference_count_for_symbol(    symbol: Symbol,    references: Iterable[Reference],) -> int:    short_name = symbol.qualname.rsplit(".", maxsplit=1)[-1]    full_name = f"{symbol.module}.{symbol.qualname}"    count = 0    for reference in references:        if reference.target_name in {            short_name,            symbol.qualname,            full_name,        }:            count += 1    return countdef build_candidates(    *,    symbols: Iterable[Symbol],    references: Iterable[Reference],    exports: set[str],    decorated: set[str],    strings: set[str],    runtime: dict[str, int],    imported_modules: set[str],) -> list[Candidate]:    references = list(references)    result: list[Candidate] = []    for symbol in symbols:        static_count = reference_count_for_symbol(symbol, references)        runtime_count = runtime.get(symbol.key, 0)        imported = symbol.module in imported_modules        exported = symbol.key in exports        is_decorated = symbol.key in decorated        short_name = symbol.qualname.rsplit(".", maxsplit=1)[-1]        string_reference = any(            literal == short_name            or literal == symbol.qualname            or literal == symbol.key            for literal in strings        )        confidence = 0        reasons: list[str] = []        if static_count == 0:            confidence += 3            reasons.append("no static references")        if runtime_count == 0:            confidence += 3            reasons.append("not executed in runtime inventory")        if not imported:            confidence += 2            reasons.append("module not observed in runtime imports")        if exported:            confidence -= 3            reasons.append("explicitly exported")        if is_decorated:            confidence -= 5            reasons.append("decorated symbol may be registered dynamically")        if string_reference:            confidence -= 4            reasons.append("possible string-based reference")        if symbol.qualname.startswith("_"):            confidence += 1            reasons.append("private symbol")        result.append(            Candidate(                symbol=symbol,                static_reference_count=static_count,                runtime_execution_count=runtime_count,                imported_at_runtime=imported,                exported=exported,                decorated=is_decorated,                suspicious_string_reference=string_reference,                confidence=confidence,                reasons=tuple(reasons),            )        )    return sorted(        result,        key=lambda item: (            -item.confidence,            item.symbol.path,            item.symbol.line,        ),    )def main() -> None:    parser = argparse.ArgumentParser(        description="Build a review queue for potentially removable Python code."    )    parser.add_argument("root", type=Path)    parser.add_argument(        "--runtime",        type=Path,        default=Path("runtime-symbols.json"),    )    parser.add_argument(        "--imports",        type=Path,        default=Path("runtime-imports.json"),    )    parser.add_argument(        "--minimum-confidence",        type=int,        default=5,    )    args = parser.parse_args()    root = args.root.resolve()    symbols, references, exports, decorated, strings = analyze_repository(root)    runtime = load_runtime_inventory(args.runtime)    imported_modules = load_imported_modules(args.imports)    candidates = build_candidates(        symbols=symbols,        references=references,        exports=exports,        decorated=decorated,        strings=strings,        runtime=runtime,        imported_modules=imported_modules,    )    selected = [        candidate        for candidate in candidates        if candidate.confidence >= args.minimum_confidence    ]    print(        json.dumps(            [asdict(candidate) for candidate in selected],            indent=2,            sort_keys=True,        )    )if __name__ == "__main__":    main()

This tool generated many false positives. That was fine.

False positives cost review time. False confidence costs production incidents.

The output became a queue of questions:

• Why does this symbol exist?

• What loads it?

• Which workflow depends on it?

• Can we create a test that proves its purpose?

• Is it compatibility code or simply forgotten code?

• Would deleting it change data, timing, ordering, or error behavior?

Sometimes the answer was obvious. Sometimes a ten-line function required two days of archaeology.

The largest deletion came from merging four pipelines

The report pipelines accounted for a little over 7,000 deleted lines.

The solution was not a clever framework. We defined one domain request and one domain result.

All entry points translated their inputs into the same request:

HTTP requestscheduled taskCLI invocationCSV export

The shared pipeline then performed:

• permission evaluation;

• filter normalization;

• query planning;

• execution;

• result shaping.

Output-specific code only handled presentation.

The previous design had mixed presentation decisions into data retrieval. The CSV path requested strings from the database layer. The API path requested Python values. Scheduled reports had a third intermediate model.

The new path returned typed domain rows containing values such as datetime, Decimal, int, str, and None. JSON and CSV formatting happened afterward.

This removed duplicated query builders, serializers, validators, and tests.

It also revealed behavior differences we had accidentally shipped.

For example, a scheduled CSV report interpreted an end date as inclusive until midnight, while the API interpreted the same date as exclusive from the beginning of that day. Both implementations had tests. Both tests passed because they encoded different assumptions.

The characterization suite forced us to decide which behavior was contractual. We preserved both temporarily through explicit compatibility options, migrated saved schedules, and then removed the old branch after the migration was verified.

That was a recurring pattern:

• expose accidental variation;

• decide whether anyone depends on it;

• migrate data or callers;

• remove the branch;

• remove the compatibility option;

• remove the tests that existed only for the deleted path.

A lot of deletion work is delayed deletion. The safest pull request is often the one that makes future deletion boring.

Some tests had to be deleted before production code could go

This part felt wrong at first.

The project had around 11,000 lines of tests dedicated to internal classes that no longer represented meaningful boundaries. Those tests were not neutral. They actively prevented simplification.

A typical service test asserted that:

• a filter factory was called;

• the resulting filter object was passed to a query builder;

• the query builder result was passed to a repository;

• repository rows were passed to a mapper;

• mapped objects were passed to a presenter.

The test said nothing about whether the correct records were returned.

When we merged those components, hundreds of tests failed even though the behavioral snapshots stayed identical.

Keeping the tests would have required preserving pointless classes or rewriting the tests to mock the new pointless classes. Neither option improved confidence.

So we deleted them.

This was done carefully. Before removing an internal test, we identified which behavior it supposedly protected. If the behavior mattered, it moved into a domain test, integration test, property test, or characterization scenario. If nobody could name the behavior, the test disappeared with the structure.

The test suite became smaller and more useful.

Execution time dropped from 27 minutes to slightly under 11 minutes in CI. More importantly, failures became easier to understand. A failed test now tended to describe a broken rule rather than a missing call to a mock.

Coverage went down by a few percentage points during the process.

That did not worry us much.

One hundred percent coverage of wrapper methods had looked impressive but protected almost nothing. Lower coverage over fewer, more meaningful paths gave us more confidence.

Coverage tells you which code ran. It does not tell you whether the important idea was tested.

The home-grown framework was the hardest thing to remove

Every mature Python project seems to contain at least one private framework that nobody remembers approving.

Ours was a task framework built on top of the queue library.

It supported retries, delays, progress updates, task groups, task metadata, hooks, cancellation, and result storage. On paper, it looked important. In practice, the application used only a small part of it.

The framework contained 5,600 lines. Removing all of it was not possible because some behavior really was product-specific.

We traced every registered task and grouped the actual requirements:

• 31 tasks needed simple retry with backoff;

• 9 needed idempotency;

• 4 published progress;

• 2 supported cancellation;

• none used task groups correctly;

• one used result storage;

• hooks existed mostly for logging.

The replacement was about 1,400 lines of ordinary application code around the queue client.

The difficult part was not implementation. It was data compatibility.

Stored task records contained framework-specific state values. The UI expected those values. Monitoring queries grouped by them. Retry counters had slightly different meanings depending on which version created the task.

We created a migration layer that could read old records and write only the new format. Then we waited until no active old-format task remained. Only after that did the old model, adapters, serializers, and migration code disappear.

This deletion took six weeks and produced almost no exciting screenshots.

It was also the part most likely to have caused an incident if rushed.

Deleting code safely is often less about Python and more about the lifetime of data created by Python.

Code can vanish in one deployment. Stored assumptions cannot.

The last 2,000 lines were strangely difficult

Early deletion was satisfying. Entire packages disappeared. Import graphs became visibly smaller. Pull requests showed thousands of red lines.

The last part was slower.

Small helpers often contained old bug fixes whose context had disappeared. A date-normalization function looked redundant until a test revealed that it handled a daylight-saving transition for saved schedules. A custom JSON encoder seemed unnecessary until Decimal values began turning into floating-point numbers. A retry helper looked identical to the queue library implementation except for one rule preventing retries after a partial external side effect.

This stage required more reading than writing.

Git history helped, but not as much as expected. Commit messages often described what changed, not why the odd code was necessary. Issue trackers were better. Production incident notes were best.

We started adding comments only when a reasonable developer might simplify code and reintroduce an old failure.

A useful comment looked like this:

Do not replace this with datetime.replace(tzinfo=...).Saved local times may fall inside a daylight-saving transition.The conversion must use the tenant timezone rules for the target date.

A useless comment looked like this:

Convert time to timezone.

Some code survived because we could not prove it was removable. That was acceptable.

The goal was not the smallest possible repository. The goal was a repository where existing code had a known reason to exist.

Those are not the same thing.

What exactly disappeared

The final production-code reduction was 18,347 lines.

Roughly, it came from:

• 7,100 lines across duplicated report pipelines;

• 4,200 lines from the private task framework;

• 2,600 lines of repository wrappers and mapping classes;

• 1,900 lines of dead integrations and unreachable compatibility code;

• 1,300 lines of duplicated validation;

• 700 lines of unused feature-flag branches;

• 547 lines of miscellaneous helpers and abandoned experiments.

The test suite also lost around 8,000 lines, although new behavioral, property, and integration tests were added during the same period.

The final production codebase contained about 44,500 lines.

Nothing became magically perfect.

There were still old modules. Some names were awkward. A few abstractions remained because replacing them offered no practical value. The system still had bugs because software apparently insists on being software.

But several things changed measurably:

• CI time fell from 27 minutes to about 11;

• cold worker startup became around 30 percent faster;

• internal import edges dropped by more than half;

• the number of test mocks fell by roughly 60 percent;

• report-related incidents became easier to reproduce;

• new export formats required one formatter instead of another pipeline;

• developers touched fewer files per feature.

The last metric mattered most.

Before the cleanup, a normal reporting change modified around nine files. Afterward, similar changes usually touched three or four.

That is what less code should buy: fewer places where one idea can hide.

What did not work

Several approaches wasted time.

Starting with line count was one of them. It encouraged deleting short helpers while ignoring large duplicated workflows. The useful unit was not a line. It was a reason for change.

Trying to infer dead code from coverage alone was another. Rare workflows and external entry points made that unsafe.

Creating a giant abstraction before merging the report pipelines also failed. The first attempt produced a generic processing engine with configurable stages. It reduced duplication but added a new language inside Python. After two weeks, it was deleted.

The final shared pipeline was intentionally boring.

We also tried replacing many small classes with large service objects. This reduced file count but created objects with too many responsibilities. Fewer files are not automatically simpler.

One proposed pull request removed almost 3,000 lines at once. Reviewers could not meaningfully verify it. We split it into nine changes with preserved behavior after each one.

Large deletions look heroic in a diff. Small deletions are easier to trust.

The worst mistake was assuming that unused means unwanted.

Some compatibility code was unused because migrations had succeeded. That was removable. Other compatibility code was unused because the relevant customer workflow had not occurred during the observation period. That was not.

The difference required context no tool could provide automatically.

Rules that survived the project

Several habits remained after the cleanup.

New abstractions now need at least two real consumers. A hypothetical second use is not enough.

Internal interfaces are not created merely because an external interface exists. A Protocol wrapping one concrete class is allowed, but it needs a testing, ownership, or substitution reason.

Tests should prefer behavior over orchestration. Mocking is still used at expensive or dangerous boundaries, not between every internal function.

Compatibility paths need an expiration condition. A comment saying remove later is not a plan. A database query, metric, date, or migration state is.

Feature flags need owners and removal dates. A disabled branch without either becomes archaeological material.

Production observations should be designed before risky simplification. Logs added after a failure are useful, but slightly late.

Most importantly, deletion must have its own acceptance criteria.

Adding a feature has an obvious reason. Removing code often gets approved because the diff looks cleaner. That is not enough.

For every major deletion, we now ask:

• Which behavior remains?

• How is that behavior verified?

• Which dependency or path disappears?

• Which data format remains readable?

• What would make rollback necessary?

• How will we know that an unknown caller still exists?

These questions slow the first pull request and speed up everything after it.

The uncomfortable conclusion

The project did not contain 18,000 lines of obviously bad code.

Most of it had been reasonable when written.

A second report pipeline appeared because changing the first one looked risky. A repository wrapper appeared to improve testability. A task hook solved one urgent logging problem. A compatibility serializer protected an old customer. A feature flag made a release safer.

Complexity arrived through locally sensible decisions.

That is why periodic cleanup matters. Nobody sits down on Monday and decides to build an unmaintainable backend. The system grows by keeping yesterday’s solution after tomorrow’s requirements arrive.

The biggest change was not the smaller repository.

It was learning to treat code as inventory.

Every function has a cost even when nobody edits it. It appears in searches, type checking, coverage reports, dependency graphs, stack traces, onboarding, reviews, security scans, and the mental model of anyone trying to understand the system.

Unused code is not free storage. It is active uncertainty.

Would I set another line-removal target? Probably not.

The number made a good title, but it would be a bad engineering objective. The useful objective is reducing the number of concepts needed to explain the product.

Still, deleting 18,347 lines felt pretty good.

The nicest pull request comment came near the end. A reviewer opened a diff, looked through it, and wrote:

This removes three classes. What replaces them?

The answer was one sentence:

Nothing. We do not need them anymore.

That was the whole point.