Modular Monolith in Practice: Enforced Module Boundaries (Python)

In the previous post, we built the Orders feature using five architectures — from Simple MVC to Microservices. Each one was tested and compared side by side.

But there's one architecture we haven't covered: Modular Monolith.

Python doesn't have an internal keyword like C# or enforced module access like Java. Module boundaries in Python — by default — depend entirely on convention and team discipline. This can become a problem as a codebase grows.

This post shows how __init__.py and import-linter can serve as an effective substitute.

---

The Problem It Solves

In Feature-Based Structure, nothing prevents this:

# features/orders/service.py
from features.notifications._internal._service import NotificationsService
# direct import into another module's internal class — Python doesn't complain

One developer in a hurry, one import that was "only temporary", and hidden coupling starts to accumulate. Python has no built-in mechanism to prevent it.

Modular Monolith in Python uses two layers of protection:

__init__.py — explicitly defines what can be accessed from outside the module

import-linter — validates in CI that no import violates the defined rules

---

The Scenario: Orders Feature

Same as the previous series:

POST /orders — create a new order

GET /orders/{id} — fetch order details

Validation: product stock must be available

Notification: send email after successful order

---

Folder Structure

/src
  /modules
    /orders
      __init__.py               ← public API of the orders module
      /contracts
        dto.py                  ← CreateOrderDto, OrderDto
        events.py               ← OrderCreatedEvent
        ports.py                ← INotificationPort (Protocol)
      /_internal
        _service.py
        _repository.py
        _entity.py
        _test_service.py
    /notifications
      __init__.py               ← public API of the notifications module
      /contracts
        ports.py                ← INotificationService (Protocol)
      /_internal
        _service.py
        _adapter.py             ← implements INotificationPort
        _test_service.py
    /products
      __init__.py
      /contracts
        dto.py
        ports.py
      /_internal
        _service.py
  /shared
    /events
      event_bus.py
    /kernel
      base_entity.py
  /presentation
    main.py                     ← FastAPI app, wires all modules

The conventions:

The _ prefix on folder and file names signals private/internal

Other modules may only import from modules/*/(__init__.py)

Enforced by import-linter in CI

---

Public API via __init__.py

The __init__.py file is the module's official contract — what is exported here is what other modules are allowed to use.

# modules/orders/__init__.py
from modules.orders.contracts.dto import CreateOrderDto, OrderDto
from modules.orders.contracts.events import OrderCreatedEvent
from modules.orders.contracts.ports import INotificationPort
from modules.orders._internal._service import OrderService

__all__ = [
    "CreateOrderDto",
    "OrderDto",
    "OrderCreatedEvent",
    "INotificationPort",
    "OrderService",  # exposed for DI wiring in main.py
]

# CORRECT — from the public API
from modules.orders import CreateOrderDto, OrderService

# WRONG — directly to internal (import-linter will reject this in CI)
from modules.orders._internal._repository import OrderRepository

---

Python Protocol as Interface

Python has no interface keyword. Instead, we use Protocol from the typing module — this enables structural typing without explicit inheritance.

# modules/orders/contracts/ports.py
from typing import Protocol

class INotificationPort(Protocol):
    async def send_order_confirmation(self, order_id: str) -> None:
        ...

Any class that has a send_order_confirmation(order_id: str) method is automatically considered an implementation of INotificationPort — no implements declaration or inheritance required. This is type-safe duck typing.

---

Orders Module Implementation

# modules/orders/_internal/_service.py
from modules.orders.contracts.dto import CreateOrderDto, OrderDto
from modules.orders.contracts.ports import INotificationPort

class OrderService:
    def __init__(self, repo, notifier: INotificationPort):
        self._repo = repo
        self._notifier = notifier

    async def create_order(self, dto: CreateOrderDto) -> OrderDto:
        product = await self._repo.find_product_by_id(dto.product_id)
        if product is None or product["stock"] < dto.quantity:
            raise ValueError("Insufficient stock")

        order = await self._repo.save({
            "product_id": dto.product_id,
            "quantity": dto.quantity,
        })
        await self._notifier.send_order_confirmation(order["id"])

        return OrderDto(id=order["id"], product_id=order["product_id"])

OrderService doesn't know about NotificationService from the notifications module. It only knows INotificationPort — a Protocol defined in the orders module's own contracts.

---

Inter-Module Communication

Pattern 1 — Dependency Injection via Protocol

The notifications module implements INotificationPort from the orders module. Wiring is done in main.py.

# modules/notifications/_internal/_adapter.py
from modules.orders import INotificationPort
from ._service import NotificationsService

class NotificationAdapter:
    def __init__(self, service: NotificationsService):
        self._service = service

    async def send_order_confirmation(self, order_id: str) -> None:
        await self._service.send({
            "type": "order_confirmed",
            "order_id": order_id,
        })

# presentation/main.py — composition root
from modules.orders import OrderService
from modules.notifications._internal._adapter import NotificationAdapter
from modules.notifications._internal._service import NotificationsService

notification_service = NotificationsService()
notification_adapter = NotificationAdapter(notification_service)
order_service = OrderService(repo=order_repository, notifier=notification_adapter)

Pattern 2 — In-Process Event Bus

# shared/events/event_bus.py
from collections import defaultdict
from typing import Callable, Type, TypeVar

T = TypeVar("T")

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

    def subscribe(self, event_type: Type[T], handler: Callable[[T], None]):
        self._handlers[event_type].append(handler)

    async def publish(self, event):
        for handler in self._handlers.get(type(event), []):
            await handler(event)

# modules/orders/_internal/_service.py
from modules.orders.contracts.events import OrderCreatedEvent

class OrderService:
    def __init__(self, repo, event_bus: EventBus):
        self._repo = repo
        self._event_bus = event_bus

    async def create_order(self, dto: CreateOrderDto) -> OrderDto:
        order = await self._repo.save(dto)
        await self._event_bus.publish(
            OrderCreatedEvent(order_id=order["id"], customer_email=dto.customer_email)
        )
        return OrderDto(id=order["id"], product_id=order["product_id"])

# modules/notifications/_internal/_service.py
class NotificationsService:
    async def handle_order_created(self, event: OrderCreatedEvent) -> None:
        await self._send_email(event.customer_email, event.order_id)

# presentation/main.py
event_bus.subscribe(OrderCreatedEvent, notification_service.handle_order_created)

---

Boundary Enforcement with import-linter

pip install import-linter

# .importlinter
[importlinter]
root_package = src

[importlinter:contract:no-internal-cross-module]
name = No module may import from another module's _internal folder
type = forbidden
source_modules =
    src.modules.orders
    src.modules.notifications
    src.modules.products
forbidden_modules =
    src.modules.orders._internal
    src.modules.notifications._internal
    src.modules.products._internal

[importlinter:contract:orders-isolation]
name = Orders module only depends on its own contracts and shared
type = layers
layers =
    src.modules.orders._internal
    src.modules.orders.contracts
    src.shared

Run in CI:

lint-imports

Output when a violation is found:

BROKEN: No module may import from another module's _internal folder
  src.modules.orders._internal._service imports from
  src.modules.notifications._internal._service

The pipeline fails. The PR cannot be merged.

---

Database: Schema per Module with SQLAlchemy

# modules/orders/_internal/_models.py
from sqlalchemy import Column, String, Integer
from sqlalchemy.orm import DeclarativeBase
from uuid import uuid4

class OrdersBase(DeclarativeBase):
    pass

class OrderModel(OrdersBase):
    __tablename__ = "orders"
    __table_args__ = {"schema": "orders"}  # schema "orders"

    id = Column(String, primary_key=True, default=lambda: str(uuid4()))
    product_id = Column(String, nullable=False)
    quantity = Column(Integer, nullable=False)

# modules/notifications/_internal/_models.py
class NotificationsBase(DeclarativeBase):
    pass

class NotificationLogModel(NotificationsBase):
    __tablename__ = "logs"
    __table_args__ = {"schema": "notifications"}  # schema "notifications"

One PostgreSQL database, two schemas. The orders module must not query the notifications schema directly.

---

Testing in Modular Monolith (Python)

Unit test within the module

# modules/orders/_internal/_test_service.py
import pytest
from unittest.mock import AsyncMock
from modules.orders.contracts.dto import CreateOrderDto
from ._service import OrderService

@pytest.fixture
def mock_repo():
    return AsyncMock()

@pytest.fixture
def mock_notifier():
    return AsyncMock()

@pytest.fixture
def service(mock_repo, mock_notifier):
    return OrderService(repo=mock_repo, notifier=mock_notifier)

@pytest.mark.anyio
async def test_create_order_sends_notification(service, mock_repo, mock_notifier):
    mock_repo.find_product_by_id.return_value = {"stock": 10}
    mock_repo.save.return_value = {"id": "order-1", "product_id": "p1"}

    await service.create_order(CreateOrderDto(product_id="p1", quantity=2))

    mock_notifier.send_order_confirmation.assert_called_once_with("order-1")

@pytest.mark.anyio
async def test_create_order_raises_when_stock_insufficient(service, mock_repo):
    mock_repo.find_product_by_id.return_value = {"stock": 0}

    with pytest.raises(ValueError, match="Insufficient stock"):
        await service.create_order(CreateOrderDto(product_id="p1", quantity=2))

Module contract test — test only through the public API

# tests/integration/test_orders_module_contract.py
import pytest
from unittest.mock import AsyncMock
from modules.orders import OrderService, CreateOrderDto, INotificationPort

@pytest.mark.anyio
async def test_orders_module_public_api():
    mock_notifier = AsyncMock(spec=INotificationPort)
    mock_repo = AsyncMock()
    mock_repo.find_product_by_id.return_value = {"stock": 10}
    mock_repo.save.return_value = {"id": "order-1", "product_id": "p1"}

    # wired like production, tested only through the public interface
    service = OrderService(repo=mock_repo, notifier=mock_notifier)
    result = await service.create_order(CreateOrderDto(product_id="p1", quantity=2))

    assert result.id == "order-1"
    mock_notifier.send_order_confirmation.assert_called_once()

Architecture test with import-linter

# tests/architecture/test_module_boundaries.py
import subprocess

def test_no_illegal_cross_module_imports():
    result = subprocess.run(
        ["lint-imports", "--config", ".importlinter"],
        capture_output=True,
        text=True,
    )
    assert result.returncode == 0, (
        f"Module boundary violations found:\n{result.stdout}"
    )

This test ensures module boundaries don't gradually erode — without relying on code review alone.

---

Comparison

Aspect	Feature-Based	Modular Monolith (Python)	Microservices
Module boundary	features/ folder convention	__init__.py + import-linter	Network boundary
Enforcement	Code review	CI pipeline	Network by default
Deployment	1 unit	1 unit	N units
Database	Shared	Shared, separate schemas	Separate per service
Contract testing	Not needed	Not needed	Required (pact-python)
Migration to microservices	Hard	Easy	N/A

---

When to Use Modular Monolith in Python

Python teams where Feature-Based has become hard to maintain due to cross-module coupling

Projects likely to be split into microservices — clear boundaries now make that much easier

Teams of 10–30 developers who still want a single deployment and fast iteration

Systems with multiple business domains that need strict separation

---

Conclusion

The main challenge of Modular Monolith in Python is that the language has no built-in mechanism to enforce module boundaries.

But the combination of __init__.py (explicit contract) and import-linter (automated CI validation) produces a system that's functionally equivalent to C#'s internal keyword. Not the compiler that rejects — but the pipeline that fails. The outcome is the same: no one can bypass the rules without it being noticed.

Add Python Protocol for structural typing without cumbersome inheritance, and you have a solid Modular Monolith foundation.

---

Closing Thoughts

The right question isn't:

> "When should we move to microservices?"

It's:

> "Are our domain boundaries clear enough to be separated?"

Modular Monolith in Python helps answer that question — with lightweight tooling and without giving up the simplicity of a single deployment.