Modular Monolith dalam Praktik: Batas Modul yang Ditegakkan (Python)

Di post sebelumnya, kita membangun fitur Orders menggunakan lima arsitektur — dari Simple MVC hingga Microservices. Setiap arsitektur diuji dan dibandingkan secara langsung.

Tapi ada satu arsitektur yang belum dibahas: Modular Monolith.

Python tidak memiliki internal keyword seperti C# atau enforced module access seperti Java. Batas modul di Python — secara default — hanya bergantung pada konvensi dan disiplin tim. Ini bisa menjadi masalah seiring bertumbuhnya codebase.

Post ini menunjukkan bagaimana __init__.py dan import-linter bisa menjadi pengganti yang efektif.

---

Masalah yang Dipecahkan

Di Feature-Based Structure, tidak ada yang mencegah ini:

# features/orders/service.py
from features.notifications._internal._service import NotificationsService
# langsung ke class internal modul lain — Python tidak protes

Satu developer yang terburu-buru, satu import yang "seharusnya sementara", dan coupling tersembunyi mulai menumpuk. Python tidak memiliki mekanisme bawaan untuk mencegahnya.

Modular Monolith di Python menggunakan dua lapisan perlindungan:

__init__.py — mendefinisikan secara eksplisit apa yang bisa diakses dari luar

import-linter — memvalidasi di CI bahwa tidak ada import yang melanggar aturan

---

Skenario: Fitur Orders

Sama seperti seri sebelumnya:

POST /orders — buat order baru

GET /orders/{id} — ambil detail order

Validasi: stok produk harus tersedia

Notifikasi: kirim email setelah order berhasil

---

Struktur Folder

/src
  /modules
    /orders
      __init__.py               ← public API modul orders
      /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 modul notifications
      /contracts
        ports.py                ← INotificationService (Protocol)
      /_internal
        _service.py
        _adapter.py             ← implementasi 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, wiring semua modul

Konvensi yang berlaku:

Prefix _ pada nama folder dan file menandakan private/internal

Modul lain hanya boleh mengimpor dari modules/*/(__init__.py)

Ditegakkan oleh import-linter di CI

---

Public API via __init__.py

__init__.py adalah kontrak resmi modul — apa yang di-export di sini adalah apa yang boleh digunakan modul lain.

# 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",  # diekspos untuk DI wiring di main.py
]

# BENAR — dari public API
from modules.orders import CreateOrderDto, OrderService

# SALAH — langsung ke internal (import-linter akan menolak di CI)
from modules.orders._internal._repository import OrderRepository

---

Python Protocol sebagai Interface

Python tidak memiliki keyword interface. Sebagai gantinya, kita menggunakan Protocol dari modul typing — ini memungkinkan structural typing tanpa explicit inheritance.

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

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

Setiap class yang memiliki method send_order_confirmation(order_id: str) secara otomatis dianggap mengimplementasikan INotificationPort — tanpa perlu implements atau inheritance. Ini adalah duck typing yang type-safe.

---

Implementasi Orders Module

# 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 tidak tahu NotificationService dari modul notifications. Ia hanya tahu INotificationPort — sebuah Protocol yang didefinisikan di kontrak modul orders sendiri.

---

Komunikasi Antar Modul

Pola 1 — Dependency Injection via Protocol

Modul notifications mengimplementasikan INotificationPort dari modul orders. Wiring dilakukan di 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

# wiring di satu tempat
notification_service = NotificationsService()
notification_adapter = NotificationAdapter(notification_service)
order_service = OrderService(repo=order_repository, notifier=notification_adapter)

Pola 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)

---

Penegakan Batas dengan 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

Jalankan di CI:

lint-imports

Output jika ada violation:

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

Pipeline gagal. PR tidak bisa di-merge.

---

Database: Schema per Modul dengan SQLAlchemy

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

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"

Satu database PostgreSQL, dua schema: orders dan notifications. Modul orders tidak boleh query ke schema notifications secara langsung.

---

Testing di Modular Monolith (Python)

Unit test dalam modul

# modules/orders/_internal/_test_service.py
import pytest
from unittest.mock import AsyncMock
from modules.orders.contracts.dto import CreateOrderDto
from modules.orders.contracts.events import OrderCreatedEvent
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 lewat 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"}

    # wiring seperti di production, test lewat public interface saja
    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 dengan 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}"
    )

Test ini memastikan batas modul tidak erosi seiring waktu — tanpa mengandalkan code review saja.

---

Perbandingan

Aspek	Feature-Based	Modular Monolith (Python)	Microservices
Batas modul	Konvensi features/	__init__.py + import-linter	Network boundary
Penegakan	Code review	CI pipeline	Network by default
Deployment	1 unit	1 unit	N unit
Database	Shared	Shared, schema terpisah	Terpisah per service
Contract testing	Tidak perlu	Tidak perlu	Wajib (pact-python)
Migrasi ke microservices	Sulit	Mudah	N/A

---

Kapan Menggunakan Modular Monolith di Python

Tim Python yang sudah merasakan Feature-Based mulai sulit di-maintain

Project yang kemungkinan akan dipecah ke microservices — batas yang sudah jelas memudahkan pemisahan

Tim dengan 10–30 developer yang masih ingin single deployment dan fast iteration

Sistem dengan banyak domain bisnis yang perlu dipisahkan secara ketat

---

Kesimpulan

Tantangan utama Modular Monolith di Python adalah: Python tidak memiliki mekanisme bawaan untuk memaksa batas modul.

Tapi kombinasi __init__.py (kontrak eksplisit) dan import-linter (validasi otomatis di CI) menghasilkan sistem yang secara fungsional setara dengan internal keyword di C#. Bukan compiler yang menolak, tapi pipeline yang gagal — dan hasilnya sama: tidak ada yang bisa bypass aturan tanpa disadari.

Ditambah Python Protocol untuk structural typing tanpa inheritance yang rumit, dan kamu memiliki fondasi Modular Monolith yang solid.

---

Penutup

Pertanyaan yang tepat bukan:

> "Kapan kita harus beralih ke microservices?"

Tapi:

> "Apakah batas-batas domain bisnis kita sudah cukup jelas untuk dipisahkan?"

Modular Monolith di Python membantu menjawab pertanyaan itu — dengan tooling yang ringan dan tanpa meninggalkan kemudahan single deployment.