Bounded Contexts in Python — Deep Dive

Implement bounded context boundaries in Python with package isolation, anti-corruption layers, and inter-context communication.

Implementing bounded contexts in a Python monolith

This guide builds an e-commerce system with three bounded contexts — Catalog, Orders, and Fulfillment — inside a single Python codebase. We will enforce boundaries, implement communication patterns, and show how the same concept (Product) correctly differs across contexts.

Project structure

src/
  catalog/
    __init__.py
    domain/
      models.py       # Catalog's Product: title, description, images, price
      events.py        # ProductPublished, PriceChanged
      services.py      # CatalogService
    adapters/
      repository.py    # Product persistence
      api.py           # REST endpoints
  orders/
    __init__.py
    domain/
      models.py        # Orders' ProductSnapshot: id, name, price_at_order_time
      events.py        # OrderPlaced, OrderCancelled
      services.py      # OrderService
    adapters/
      repository.py
      api.py
    acl/
      catalog_translator.py  # Anti-corruption layer
  fulfillment/
    __init__.py
    domain/
      models.py        # Fulfillment's Item: sku, weight, dimensions, bin_location
      events.py        # ShipmentCreated, ShipmentDispatched
      services.py      # FulfillmentService
    adapters/
      repository.py
      api.py
    acl/
      orders_translator.py
  shared/
    events.py          # Event bus infrastructure
    types.py           # Shared value objects (Money, EntityId)

Different models for the same concept

Catalog’s Product

# catalog/domain/models.py
from dataclasses import dataclass, field
from decimal import Decimal
from uuid import UUID

@dataclass
class ProductImage:
    url: str
    alt_text: str
    sort_order: int

@dataclass
class Product:
    id: UUID
    title: str
    description: str
    base_price: Decimal
    currency: str
    images: list[ProductImage] = field(default_factory=list)
    tags: list[str] = field(default_factory=list)
    is_published: bool = False

    def publish(self) -> None:
        if not self.title or not self.description:
            raise ValueError("Cannot publish product without title and description")
        if self.base_price <= 0:
            raise ValueError("Price must be positive")
        self.is_published = True

    def change_price(self, new_price: Decimal) -> None:
        if new_price <= 0:
            raise ValueError("Price must be positive")
        self.base_price = new_price

Orders’ ProductSnapshot

# orders/domain/models.py
from dataclasses import dataclass
from decimal import Decimal
from uuid import UUID

@dataclass(frozen=True)
class ProductSnapshot:
    """Immutable snapshot of product info at time of ordering.
    Orders don't care about images, tags, or publication status."""
    product_id: UUID
    name: str
    price: Decimal
    currency: str

@dataclass
class OrderLine:
    product: ProductSnapshot
    quantity: int

    @property
    def total(self) -> Decimal:
        return self.product.price * self.quantity

@dataclass
class Order:
    id: UUID
    customer_id: UUID
    lines: list[OrderLine] = field(default_factory=list)
    status: str = "pending"

    @property
    def total(self) -> Decimal:
        return sum(line.total for line in self.lines)

    def add_line(self, product: ProductSnapshot, quantity: int) -> None:
        if quantity <= 0:
            raise ValueError("Quantity must be positive")
        self.lines.append(OrderLine(product=product, quantity=quantity))

Fulfillment’s Item

# fulfillment/domain/models.py
from dataclasses import dataclass
from uuid import UUID

@dataclass
class Dimensions:
    length_cm: float
    width_cm: float
    height_cm: float

    @property
    def volume_cm3(self) -> float:
        return self.length_cm * self.width_cm * self.height_cm

@dataclass
class Item:
    """Fulfillment doesn't know about prices, descriptions, or images.
    It knows about physical properties and warehouse locations."""
    sku: str
    product_id: UUID
    weight_kg: float
    dimensions: Dimensions
    bin_location: str
    is_fragile: bool = False

    @property
    def is_oversized(self) -> bool:
        return self.dimensions.volume_cm3 > 50000 or self.weight_kg > 30

Three contexts, three Product-like models, each containing only what that context needs. No bloated universal model.

Anti-corruption layers

The Orders context needs product info from Catalog but should not depend on Catalog’s model. An anti-corruption layer (ACL) translates between them.

# orders/acl/catalog_translator.py
from typing import Protocol
from decimal import Decimal
from uuid import UUID
from orders.domain.models import ProductSnapshot

class CatalogClient(Protocol):
    def get_product(self, product_id: UUID) -> dict: ...

class CatalogTranslator:
    """Translates Catalog's product model into Orders' ProductSnapshot.
    Protects Orders from changes in Catalog's internal model."""

    def __init__(self, client: CatalogClient) -> None:
        self._client = client

    def get_product_snapshot(self, product_id: UUID) -> ProductSnapshot:
        raw = self._client.get_product(product_id)
        return ProductSnapshot(
            product_id=product_id,
            name=raw["title"],           # Catalog says "title", Orders says "name"
            price=Decimal(raw["base_price"]),
            currency=raw["currency"],
        )

Key ACL details:

  • Catalog calls it “title”; Orders calls it “name”. The ACL translates.
  • If Catalog adds or renames fields, only the ACL needs updating.
  • Orders never depends on Catalog’s model classes directly.

Inter-context communication with events

Domain events per context

# catalog/domain/events.py
@dataclass(frozen=True)
class ProductPublished:
    product_id: UUID
    title: str
    base_price: Decimal
    currency: str
    occurred_at: datetime

# orders/domain/events.py
@dataclass(frozen=True)
class OrderPlaced:
    order_id: UUID
    customer_id: UUID
    total: Decimal
    line_items: list[dict]
    occurred_at: datetime

# fulfillment/domain/events.py
@dataclass(frozen=True)
class ShipmentDispatched:
    shipment_id: UUID
    order_id: UUID
    tracking_number: str
    occurred_at: datetime

Event bus

# shared/events.py
from typing import Callable, Any
from collections import defaultdict

class InProcessEventBus:
    """Simple synchronous event bus for monolith contexts."""

    def __init__(self):
        self._handlers: dict[str, list[Callable]] = defaultdict(list)

    def subscribe(self, event_type: str, handler: Callable) -> None:
        self._handlers[event_type].append(handler)

    def publish(self, event: Any) -> None:
        event_type = type(event).__name__
        for handler in self._handlers.get(event_type, []):
            handler(event)

Wiring event handlers

# Fulfillment subscribes to Orders' events
event_bus.subscribe("OrderPlaced", fulfillment_service.handle_order_placed)

# Orders subscribes to Catalog's events (for price updates)
event_bus.subscribe("PriceChanged", order_service.handle_price_changed)

Each handler receives an event from another context and translates it through the ACL before acting on it.

Enforcing boundaries with import linting

Without enforcement, developers will take shortcuts across boundaries. Use import-linter to prevent this:

# pyproject.toml
[tool.importlinter]
root_package = "src"

[[tool.importlinter.contracts]]
name = "Context independence"
type = "independence"
modules = [
    "src.catalog.domain",
    "src.orders.domain",
    "src.fulfillment.domain",
]

[[tool.importlinter.contracts]]
name = "Orders does not import Catalog internals"
type = "forbidden"
source_modules = ["src.orders"]
forbidden_modules = ["src.catalog.domain", "src.catalog.adapters"]
# Orders can only use the ACL, not Catalog's internals

[[tool.importlinter.contracts]]
name = "Domain does not import adapters"
type = "layers"
layers = ["adapters", "domain"]
containers = ["src.catalog", "src.orders", "src.fulfillment"]

CI runs lint-imports and fails if any boundary is violated.

Testing across boundaries

Unit tests within a context

Test domain logic in isolation — no other contexts involved.

def test_catalog_product_cannot_publish_without_title():
    product = Product(id=uuid4(), title="", description="desc",
                      base_price=Decimal("10"), currency="USD")
    with pytest.raises(ValueError, match="title and description"):
        product.publish()

def test_order_total_calculation():
    snap = ProductSnapshot(uuid4(), "Widget", Decimal("25"), "USD")
    order = Order(id=uuid4(), customer_id=uuid4())
    order.add_line(snap, 3)
    assert order.total == Decimal("75")

Integration tests across boundaries

Test that the ACL correctly translates and that events flow between contexts.

def test_acl_translates_catalog_product_to_order_snapshot():
    fake_client = FakeCatalogClient(products={
        product_id: {"title": "Gadget", "base_price": "49.99", "currency": "USD"}
    })
    translator = CatalogTranslator(fake_client)
    snapshot = translator.get_product_snapshot(product_id)
    assert snapshot.name == "Gadget"  # translated from "title"
    assert snapshot.price == Decimal("49.99")

def test_order_placed_triggers_fulfillment():
    bus = InProcessEventBus()
    fulfillment = FakeFulfillmentService()
    bus.subscribe("OrderPlaced", fulfillment.handle_order_placed)

    bus.publish(OrderPlaced(
        order_id=uuid4(), customer_id=uuid4(),
        total=Decimal("100"), line_items=[],
        occurred_at=datetime.now(timezone.utc),
    ))
    assert fulfillment.received_orders == 1

Evolving from monolith to services

When a bounded context needs independent scaling or deployment:

  1. Extract the package into its own repository
  2. Replace in-process event bus with a message broker (RabbitMQ, Kafka)
  3. Replace direct ACL calls with HTTP/gRPC clients
  4. Add a separate database (each service owns its data)

Because the boundaries are already clean, this extraction is mechanical rather than architectural. The domain models, events, and ACLs stay the same — only the communication infrastructure changes.

Common pitfalls

  1. Sharing database tables across contexts — This creates hidden coupling. Each context should own its data, even in a monolith (use separate schemas or table prefixes).
  2. Skipping the ACL — Directly importing another context’s models feels faster but creates tight coupling that makes future extraction painful.
  3. Too many contexts too early — Start with fewer, larger contexts. Split when you have evidence of diverging models, not when you speculate.
  4. Inconsistent event contracts — Use versioned event schemas and treat cross-context events as a public API.

Sizing a bounded context

A good bounded context is:

  • Owned by one team (3-8 people)
  • Independently deployable (even if currently in a monolith)
  • Focused on one subdomain with consistent vocabulary
  • Small enough to understand but large enough to be useful

If a context has more than 20 domain classes, consider splitting. If it has fewer than 3, it might be too granular.

The one thing to remember: Bounded contexts in Python are enforced through package structure, import linting, and anti-corruption layers — creating clear boundaries where each part of the system maintains its own model without contaminating others.

pythonarchitectureddd

See Also

  • Python Aggregate Pattern Why grouping related objects under a single gatekeeper prevents data chaos in your Python application.
  • Python Bulkhead Pattern Why smart Python apps put walls between their parts — like a ship that stays afloat even with a hole in the hull.
  • Python Circuit Breaker Pattern How a circuit breaker saves your app from crashing — explained with a home electrical fuse analogy.
  • Python Clean Architecture Why your Python app should look like an onion — and how that saves you from painful rewrites.
  • Python Connection Draining How to shut down a Python server without hanging up on people mid-conversation — like a store that locks the entrance but lets shoppers finish.

Related Topics

ab-testing How tech companies run thousands of experiments at once to improve their products — the scientific method applied to everything from button colors to recommendation algorithms. 3 levels → activation-functions Why neural networks need these tiny mathematical functions — and how ReLU's simplicity accidentally made deep learning possible. 3 levels → ai-agents-architecture How AI systems go from answering questions to actually doing things — the design patterns that turn language models into autonomous agents that browse, code, and plan. 3 levels →