Replaceable Django apps: package boundaries, an outbox and an upgrade test

Business software like ERPs, CRMs and customer portals is full of operations that belong together. Approving a purchase order reserves stock, creates a payable and queues a notification, and the user expects all of it to happen or none of it. In a single Django process with one Postgres database, that's one transaction.atomic() block. Split the same steps across services and you need sagas, retries and reconciliation jobs to get back the guarantee the database gave you for free.

That's why the usual problem with a growing Django codebase isn't the deployment shape but change coupling. Billing imports from accounts, accounts imports from notifications, a signal in one app quietly writes to another, and renaming a field means reading half the repo first. Microservices are often proposed as the cure, but they mostly turn an import problem into a network problem. What helps is boundaries inside the monolith that a machine checks, and a way to test that each module can be upgraded on its own.

This post builds that with a small, runnable example: a billing_app package with a narrow public API, an import contract, a transactional outbox for side effects, an upgrade test that migrates real seeded data from the previous release, and a gateway that lets you move billing into a separate service later without rewriting its callers. Everything was run with Django 6.1, psycopg 3.3, Postgres 18 and uv 0.13.

Start by making each high-churn domain a proper Python package with its own pyproject.toml, even if it lives in the same repo. For a single product, a uv workspace is enough: the host project depends on billing-app, and uv installs it from the local path in editable mode, so there's nothing to publish.

 1[project]
 2name = "host"
 3version = "0.0.0"
 4requires-python = ">=3.12"
 5dependencies = ["billing-app", "django>=5.2", "psycopg[binary]>=3.2", "httpx>=0.28"]
 6
 7[dependency-groups]
 8dev = ["pytest>=8", "pytest-django>=4.9", "import-linter>=2.1"]
 9
10[tool.uv]
11package = false
12
13[tool.uv.workspace]
14members = ["packages/*"]
15
16[tool.uv.sources]
17billing-app = { workspace = true }

Building versioned wheels and publishing them to a private index only pays off when the same app is installed in several projects, for example a billing or auth app reused across products or client projects. Then each project pins a version and upgrades when it's ready, and the upgrade test later in this post becomes the thing that makes that safe. For one product in one repo, the release overhead buys nothing that the workspace and the import contract don't already give you.

Inside the package, one module is the public API and everything else is private. Callers get plain data back, not model instances they could save or traverse into other apps:

 1@dataclass(frozen=True)
 2class InvoiceDTO:
 3    id: int
 4    account_id: int
 5    amount_cents: int
 6    status: str
 7
 8
 9def create_invoice(account_id: int, amount_cents: int) -> InvoiceDTO:
10    with transaction.atomic():
11        inv = Invoice.objects.create(account_id=account_id, amount_cents=amount_cents)
12        inv.number = f"INV-{inv.id:06d}"
13        inv.save(update_fields=["number"])
14        OutboxEvent.objects.create(
15            topic="invoice.created",
16            payload={"invoice_id": inv.id, "account_id": account_id, "amount_cents": amount_cents},
17        )
18    return InvoiceDTO(inv.id, inv.account_id, inv.amount_cents, inv.status)
19
20
21def get_invoice(invoice_id: int) -> InvoiceDTO:
22    inv = Invoice.objects.get(pk=invoice_id)
23    return InvoiceDTO(inv.id, inv.account_id, inv.amount_cents, inv.status)

Money is stored as integer cents, and the invoice stores account_id as a plain BigIntegerField rather than a ForeignKey to the accounts app. That keeps billing free of imports from accounts, which is what makes it reusable, but it's a real trade-off: the database no longer guarantees the account exists, and deleting an account doesn't cascade or block. If you take that route, add a periodic check that finds invoices pointing at missing accounts. Where the integrity matters more than the independence, a string reference like models.ForeignKey("accounts.Account", on_delete=models.PROTECT) keeps the constraint without a Python import, at the price of a migration dependency on the other app.

Boundaries that live in a wiki erode, so put them in import-linter, which reads contracts from pyproject.toml. This one forbids the package from knowing anything about the host project it's installed into:

 1[tool.importlinter]
 2root_packages = ["billing_app", "hostproj"]
 3
 4[[tool.importlinter.contracts]]
 5name = "billing_app does not know about the host"
 6type = "forbidden"
 7source_modules = ["billing_app"]
 8forbidden_modules = ["hostproj"]

Adding from hostproj import settings to the billing API makes lint-imports exit with status 1 and name the offending line:

 1billing_app is not allowed to import hostproj:
 2
 3-   billing_app.api -> hostproj.settings (l.29)

The same tool has layers and independence contracts for rules between domain packages, such as sales may call billing but billing may never import sales. Contracts like these are what stop the slow drift back to everything importing everything.

The create_invoice function above writes an OutboxEvent row in the same transaction as the invoice. That's the transactional outbox pattern, and it's one of the few patterns worth having from the start in a monolith. Emails, webhooks and messages to other systems are produced from committed rows, so a crash can't leave an invoice without its event or an event for an invoice that was rolled back. Sending the email directly inside the request has both problems, and calling transaction.on_commit fixes only the first one, because the process can still die between the commit and the send.

A relay drains the table. select_for_update(skip_locked=True) lets several workers run it at the same time without blocking each other or picking the same rows:

 1def publish_pending(send: Callable[[str, dict], None], batch_size: int = 100) -> int:
 2    # skip_locked lets several workers drain the table without blocking each other.
 3    # Delivery is at-least-once: if the commit fails after send(), the batch is sent again.
 4    with transaction.atomic():
 5        events = list(
 6            OutboxEvent.objects.select_for_update(skip_locked=True)
 7            .filter(published_at__isnull=True)
 8            .order_by("id")[:batch_size]
 9        )
10        for event in events:
11            send(event.topic, event.payload)
12        OutboxEvent.objects.filter(id__in=[e.id for e in events]).update(published_at=timezone.now())
13    return len(events)

Run it from a Celery beat task or a small management command loop. Delivery is at-least-once, so whatever consumes the events must tolerate duplicates, for example by keying on invoice_id. Published rows pile up, so delete or archive them on a schedule.

The contract tests for the package test exactly what the host relies on: the API round trip, that an event is published once, and that a failed write leaves no event behind.

 1def test_invoice_event_is_published_once():
 2    inv = api.create_invoice(account_id=7, amount_cents=500)
 3    sent = []
 4    assert publish_pending(lambda topic, payload: sent.append((topic, payload))) == 1
 5    assert publish_pending(lambda topic, payload: sent.append((topic, payload))) == 0
 6    assert sent == [("invoice.created", {"invoice_id": inv.id, "account_id": 7, "amount_cents": 500})]
 7
 8
 9def test_failed_write_leaves_no_event():
10    with pytest.raises(Exception):
11        api.create_invoice(account_id=7, amount_cents=None)
12    assert publish_pending(lambda *a: None) == 0

Each package owns its migrations, and no other app writes migrations for its models. That part is convention. The part that needs a test is upgrading: a migration that works on an empty test database can fail on a database with real rows, and the normal test suite only ever sees the empty case.

Here's the example. Version 1.1 of billing_app adds a unique invoice number. Declaring the field as CharField(max_length=20, unique=True, default="") and running makemigrations produces this:

 1operations = [
 2        migrations.AddField(
 3            model_name="invoice",
 4            name="number",
 5            field=models.CharField(default="", max_length=20, unique=True),
 6        ),
 7    ]

All four contract tests pass with it, because pytest-django creates a fresh database and migrates it before any invoice exists. Upgrading a database that already holds invoices from version 1.0 fails, because every existing row gets the same empty string:

 1psycopg.errors.UniqueViolation: could not create unique index "billing_invoice_number_key"
 2DETAIL:  Key (number)=() is duplicated.

The fix is to add the column as nullable, since NULLs don't collide in a unique index, and backfill it in batches:

 1def backfill_numbers(apps, schema_editor):
 2    Invoice = apps.get_model("billing", "Invoice")
 3    pending = Invoice.objects.filter(number__isnull=True).order_by("id")
 4    while batch := list(pending[:1000]):
 5        for inv in batch:
 6            inv.number = f"INV-{inv.id:06d}"
 7        Invoice.objects.bulk_update(batch, ["number"])
 8
 9
10class Migration(migrations.Migration):
11    dependencies = [("billing", "0001_initial")]
12
13    operations = [
14        # NULLs don't collide in a unique index, so existing rows are fine.
15        migrations.AddField(
16            model_name="invoice",
17            name="number",
18            field=models.CharField(max_length=20, unique=True, null=True),
19        ),
20        migrations.RunPython(backfill_numbers, migrations.RunPython.noop),
21    ]

On a big table, the backfill belongs in its own release rather than the same migration, because a Postgres migration runs in one transaction by default and holds its locks until the end. The point here is the test that catches the mistake. It builds the previous release's wheel from its Git tag and the current one from the working tree, installs the old one into a clean virtualenv, migrates and seeds data through the old public API, then installs the new wheel over it, migrates again, checks the seeded rows and runs the contract tests:

 1#!/usr/bin/env bash
 2# usage: upgrade-test.sh <previous-release-tag>
 3set -euo pipefail
 4from_tag=$1
 5work=$(mktemp -d)
 6
 7# Build the old wheel from its tag and the new wheel from the working tree.
 8git worktree add --quiet "$work/prev" "$from_tag"
 9(cd "$work/prev" && uv build --package billing-app --wheel --out-dir "$work/dist-prev")
10uv build --package billing-app --wheel --out-dir "$work/dist-new"
11git worktree remove --force "$work/prev"
12
13uv venv --quiet "$work/venv"
14export VIRTUAL_ENV="$work/venv"
15uv pip install --quiet "$work"/dist-prev/*.whl "django>=5.2" "psycopg[binary]" httpx pytest pytest-django
16
17cd host
18python() { "$VIRTUAL_ENV/bin/python" "$@"; }
19python manage.py migrate --no-input
20python manage.py shell -c "
21from billing_app.api import create_invoice
22for i in range(1, 501):
23    create_invoice(account_id=i % 17, amount_cents=i * 100)
24"
25
26uv pip install --quiet --reinstall --no-deps "$work"/dist-new/*.whl
27python manage.py migrate --no-input
28python manage.py shell -c "
29from billing_app.api import get_invoice
30from billing_app.models import Invoice
31assert Invoice.objects.count() == 500, Invoice.objects.count()
32assert get_invoice(Invoice.objects.order_by('id').first().id).amount_cents == 100
33print('seeded data survived the upgrade')
34"
35"$VIRTUAL_ENV/bin/pytest" -q tests/contracts

With the naive migration it stops at the UniqueViolation above. With the fixed one it prints seeded data survived the upgrade, all 500 invoices get numbers starting at INV-000001, and the contract tests pass. Seeding through the old version's public API rather than a SQL dump matters: it produces the data the old code would really have written, and it keeps working as the schema changes.

The script builds both wheels itself, which is the step that is easy to miss: a job that installs dist/*.whl without building it first is installing nothing, or a stale file left from somewhere else. In GitHub Actions, the job runs once per previous release you still support, against a Postgres service. This wiring is a sketch around the script above:

 1jobs:
 2  upgrade-test:
 3    runs-on: ubuntu-latest
 4    strategy:
 5      matrix:
 6        from_tag: ["billing-app-v1.0.0"]
 7    services:
 8      postgres:
 9        image: postgres:18
10        env:
11          POSTGRES_USER: app
12          POSTGRES_PASSWORD: pw
13          POSTGRES_DB: app
14        ports: ["5432:5432"]
15        options: >-
16          --health-cmd pg_isready --health-interval 5s --health-retries 10
17    steps:
18      - uses: actions/checkout@v7
19        with:
20          fetch-depth: 0
21      - uses: astral-sh/setup-uv@v10
22      - run: ./upgrade-test.sh ${{ matrix.from_tag }}
23        env:
24          DB_PORT: "5432"

fetch-depth: 0 is there because the script checks out the old tag with git worktree, and a shallow clone doesn't have it. Which versions go in the matrix is a policy decision. For an app reused across projects, list every version that some project still runs, because those are the upgrade paths someone will actually take. For a single product, the previous release is enough.

Sometimes a module does need to leave: it has a different scaling profile, a separate team with its own release cadence, or data that must be isolated for compliance. If callers already go through an interface, extraction is a configuration change rather than a rewrite. Define the interface as a Protocol, ship the in-process implementation with the package, and let settings choose:

 1class BillingGateway(Protocol):
 2    def create_invoice(self, account_id: int, amount_cents: int) -> int: ...
 3
 4
 5class LocalBillingGateway:
 6    def create_invoice(self, account_id: int, amount_cents: int) -> int:
 7        return api.create_invoice(account_id, amount_cents).id
 8
 9
10def get_gateway() -> BillingGateway:
11    path = getattr(settings, "BILLING_GATEWAY", "billing_app.gateway.LocalBillingGateway")
12    return import_string(path)()

The remote implementation lives in the host project, because where billing runs is a deployment concern, not a billing one:

 1class RemoteBillingGateway:
 2    def create_invoice(self, account_id: int, amount_cents: int) -> int:
 3        r = httpx.post(
 4            f"{os.environ['BILLING_URL']}/invoices",
 5            json={"account_id": account_id, "amount_cents": amount_cents},
 6            timeout=5.0,
 7        )
 8        r.raise_for_status()
 9        return r.json()["id"]

The example's contract tests include one that runs the same call through both gateways against a stub HTTP server and compares the results, and that test is the start of a safe extraction sequence. First run the remote service in shadow mode, calling it alongside the local path and logging differences without using its answers. Then switch one tenant or one workflow with a feature flag, keep the local path deployable so rolling back is a settings change, and widen only when the new service isn't producing more incidents than it removes. Note what the remote version loses: create_invoice was atomic with the rest of the caller's transaction, and over HTTP it isn't, so the caller now needs an idempotency key and a plan for the call that succeeded but timed out.

That loss is the reason to extract late and only with a concrete reason. A module that's merely large, or a team that's unhappy with the code layout, is better served by the package boundaries, import contracts and upgrade test above, which keep the single transaction and still let each part change on its own.