Struktur Folder dalam Praktik: Satu Fitur, Lima Arsitektur (Python)

Di post sebelumnya, kita membahas bagaimana struktur folder berkembang dari Simple MVC hingga Microservices — dan mengapa setiap pendekatan lahir untuk menjawab masalah pada tahap sebelumnya.

Kali ini kita masuk ke ranah yang lebih konkret: kodenya seperti apa?

Strateginya: kita bangun satu fitur yang sama menggunakan kelima arsitektur, lalu lihat bagaimana cara masing-masing diuji. Dengan begitu, perbedaannya bisa dibandingkan secara langsung — apples-to-apples.

---

Skenario: Fitur Orders

Fitur yang kita bangun di semua arsitektur:

POST /orders — buat order baru

GET /orders/{id} — ambil detail order

Validasi: stok produk harus tersedia sebelum order dibuat

Notifikasi: kirim email konfirmasi setelah order berhasil

Fitur ini dipilih karena melibatkan lebih dari satu layer — ada HTTP, business logic, database, dan external service. Cukup nyata untuk menunjukkan perbedaan arsitektur, tapi tidak terlalu kompleks hingga mengaburkan poin utamanya.

Tech stack yang digunakan di semua contoh:

Kebutuhan	Library
HTTP framework	FastAPI
ORM	SQLAlchemy (async)
Unit test	pytest + anyio
Integration test	httpx AsyncClient
Mocking	unittest.mock.AsyncMock
Contract testing	pact-python

---

Arsitektur 1: Simple MVC

Struktur Folder

/controllers
  order_controller.py
/models
  order.py

Implementasi

Di Simple MVC, semua logic ada di dalam satu route function. Controller langsung melakukan validasi stok, menyimpan order ke database via SQLAlchemy, dan mengirim email — semuanya dalam satu blok kode.

# controllers/order_controller.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.db import get_session
from app.email import send_email
from app.models.order import Order, Product
from pydantic import BaseModel

router = APIRouter()

class CreateOrderRequest(BaseModel):
    product_id: str
    quantity: int
    customer_email: str

@router.post("/orders", status_code=201)
async def create_order(
    request: CreateOrderRequest,
    session: AsyncSession = Depends(get_session),
):
    product = await session.get(Product, request.product_id)
    if product is None or product.stock < request.quantity:
        raise HTTPException(status_code=400, detail="Insufficient stock")

    order = Order(
        product_id=request.product_id,
        quantity=request.quantity,
    )
    session.add(order)
    await session.commit()
    await session.refresh(order)

    await send_email(
        to=request.customer_email,
        subject=f"Order {order.id} confirmed",
    )

    return order

Kode ini mudah dibaca dan cepat ditulis. Tapi setiap kali ada logika baru — diskon, fraud check, audit log — semuanya masuk ke fungsi yang sama. Dalam beberapa sprint, fungsi ini bisa tumbuh menjadi ratusan baris.

Pengujian di Simple MVC

Karena logic dan infrastruktur bercampur dalam satu fungsi, unit test murni tidak mungkin dilakukan. Satu-satunya pilihan adalah integration test: jalankan seluruh aplikasi dan hit endpoint-nya.

Di FastAPI, ini dilakukan dengan httpx.AsyncClient dan ASGITransport:

# test_order_controller.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.mark.anyio
async def test_create_order_returns_201():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        response = await client.post("/orders", json={
            "product_id": "prod-1",
            "quantity": 2,
            "customer_email": "user@test.com",
        })

    assert response.status_code == 201
    assert "id" in response.json()

@pytest.mark.anyio
async def test_create_order_returns_400_when_stock_empty():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        response = await client.post("/orders", json={
            "product_id": "prod-no-stock",
            "quantity": 2,
            "customer_email": "user@test.com",
        })

    assert response.status_code == 400

Setiap test butuh database yang sudah terisi data produk. Test menjadi lambat karena harus melewati seluruh HTTP stack. Menguji edge case seperti "stok habis tepat saat order masuk" membutuhkan setup data yang rumit.

Cocok untuk: prototype, internal tool kecil, validasi ide dalam 1–2 minggu.

---

Arsitektur 2: Layered Architecture

Struktur Folder

/controllers
  order_controller.py
/services
  order_service.py
/repositories
  order_repository.py
/models
  order.py

Implementasi

Business logic pindah ke OrderService. Controller hanya bertugas menerima request dan mendelegasikan ke service.

# services/order_service.py
from app.repositories.order_repository import OrderRepository
from app.email import EmailSender

class OrderService:
    def __init__(self, repo: OrderRepository, email: EmailSender):
        self._repo = repo
        self._email = email

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

        order = await self._repo.save({"product_id": product_id, "quantity": quantity})

        await self._email.send(
            to=customer_email,
            subject=f"Order {order['id']} confirmed",
        )

        return order

# controllers/order_controller.py
@router.post("/orders", status_code=201)
async def create_order(
    request: CreateOrderRequest,
    service: OrderService = Depends(get_order_service),
):
    return await service.create_order(
        product_id=request.product_id,
        quantity=request.quantity,
        customer_email=request.customer_email,
    )

Pengujian di Layered Architecture

Sekarang OrderService bisa diuji secara terisolasi dengan meng-mock OrderRepository menggunakan AsyncMock — tanpa database:

# test_order_service.py
import pytest
from unittest.mock import AsyncMock
from app.services.order_service import OrderService

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

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

@pytest.fixture
def service(mock_repo, mock_email):
    return OrderService(repo=mock_repo, email=mock_email)

@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("prod-1", quantity=2, customer_email="u@test.com")

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

    await service.create_order("prod-1", quantity=2, customer_email="u@test.com")

    mock_email.send.assert_called_once_with(
        to="u@test.com", subject="Order order-1 confirmed"
    )

Ini langkah maju yang signifikan. Test berjalan cepat karena tidak menyentuh database maupun HTTP stack sama sekali. Berbagai kondisi — stok habis, email gagal — bisa diuji dengan mudah.

Cocok untuk: REST API 5–15 endpoint, tim 2–5 developer.

---

Arsitektur 3: Feature-Based Structure

Struktur Folder

/features
  /orders
    router.py
    service.py
    repository.py
    schemas.py
    models.py
    test_service.py    ← test ada di dalam folder fitur
  /notifications
    service.py
    schemas.py

Implementasi

Struktur yang sama dengan Layered, tapi diorganisir berdasarkan fitur bisnis. Semua yang berkaitan dengan Orders ada di satu folder — termasuk test-nya.

service.py menggunakan NotificationService sebagai abstraksi terpisah, bukan langsung memanggil send_email:

# features/orders/service.py
from features.notifications.service import NotificationService
from .repository import OrderRepository

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

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

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

        return order

# features/orders/router.py
from fastapi import APIRouter, Depends
from .service import OrderService
from .schemas import CreateOrderRequest

router = APIRouter(prefix="/orders")

@router.post("", status_code=201)
async def create_order(
    request: CreateOrderRequest,
    service: OrderService = Depends(),
):
    return await service.create_order(request.product_id, request.quantity)

Di main.py, router di-include:

# main.py
from features.orders.router import router as orders_router
app.include_router(orders_router)

Pengujian di Feature-Based Structure

Test file hidup di dalam features/orders/test_service.py — bukan di folder tests/ terpisah. Pytest fixtures didefinisikan langsung di file test yang sama, sehingga konteksnya terbaca dalam satu file.

# features/orders/test_service.py
import pytest
from unittest.mock import AsyncMock
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_saves_and_notifies(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": "prod-1"}

    await service.create_order(product_id="prod-1", quantity=2)

    mock_repo.save.assert_called_once()
    mock_notifier.send_order_confirmation.assert_called_once_with("order-1")

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

    with pytest.raises(ValueError, match="Insufficient stock"):
        await service.create_order(product_id="prod-1", quantity=1)

Developer baru yang ditugaskan ke fitur Orders langsung menemukan kode dan test-nya dalam satu folder. Tidak perlu berpindah antara src/ dan tests/.

Cocok untuk: SaaS product, tim 5–20 developer, aplikasi dengan banyak fitur yang terus bertambah.

---

Arsitektur 4: Clean Architecture

Struktur Folder

/src
  /domain
    /orders
      order.py             ← pure entity (dataclass)
      order_repository.py  ← ABC sebagai port
      exceptions.py
  /application
    /orders
      create_order_use_case.py
      test_create_order_use_case.py
  /infrastructure
    /database
      sqlalchemy_order_repository.py
      test_sqlalchemy_order_repository.py
    /email
      sendgrid_notification_adapter.py
  /presentation
    /api
      order_router.py

Implementasi

Di Clean Architecture, domain layer tidak boleh tahu tentang SQLAlchemy, FastAPI, atau SendGrid. Order adalah pure Python dataclass. OrderRepository adalah Abstract Base Class yang mendefinisikan kontrak — bukan implementasi.

# domain/orders/order.py
from dataclasses import dataclass, field
from uuid import uuid4
from .exceptions import DomainException

@dataclass
class Order:
    product_id: str
    quantity: int
    id: str = field(default_factory=lambda: str(uuid4()))

    def __post_init__(self):
        if self.quantity <= 0:
            raise DomainException("Quantity must be greater than zero")

# domain/orders/order_repository.py
from abc import ABC, abstractmethod
from .order import Order

class OrderRepository(ABC):
    @abstractmethod
    async def find_product_by_id(self, product_id: str) -> dict | None:
        ...

    @abstractmethod
    async def save(self, order: Order) -> Order:
        ...

# application/orders/create_order_use_case.py
from domain.orders.order import Order
from domain.orders.order_repository import OrderRepository
from domain.orders.exceptions import DomainException

class NotificationPort(ABC):
    @abstractmethod
    async def notify(self, order_id: str) -> None:
        ...

class CreateOrderUseCase:
    def __init__(self, repo: OrderRepository, notifier: NotificationPort):
        self._repo = repo
        self._notifier = notifier

    async def execute(self, product_id: str, quantity: int) -> Order:
        product = await self._repo.find_product_by_id(product_id)
        if product is None:
            raise DomainException("Product not found")
        if product["stock"] < quantity:
            raise DomainException("Insufficient stock")

        order = Order(product_id=product_id, quantity=quantity)
        saved = await self._repo.save(order)
        await self._notifier.notify(saved.id)

        return saved

CreateOrderUseCase tidak tahu kita pakai SQLAlchemy, Tortoise ORM, atau bahkan database apapun. Ia hanya tahu OrderRepository — sebuah kontrak abstrak.

Pengujian di Clean Architecture

Ada tiga lapisan test yang masing-masing punya tujuan dan kecepatan berbeda.

Domain Entity — pure unit test, nol dependency

# domain/orders/test_order.py
import pytest
from domain.orders.order import Order
from domain.orders.exceptions import DomainException

def test_order_created_successfully():
    order = Order(product_id="prod-1", quantity=3)

    assert order.product_id == "prod-1"
    assert order.quantity == 3
    assert order.id is not None

def test_order_raises_when_quantity_zero():
    with pytest.raises(DomainException, match="Quantity must be greater than zero"):
        Order(product_id="prod-1", quantity=0)

Tidak ada AsyncMock, tidak ada fixture database. Test ini pure Python dan berjalan dalam hitungan milidetik.

Application Use Case — mock port via AsyncMock

# application/orders/test_create_order_use_case.py
import pytest
from unittest.mock import AsyncMock
from domain.orders.order import Order
from application.orders.create_order_use_case import CreateOrderUseCase

@pytest.fixture
def mock_repo():
    repo = AsyncMock()
    repo.save.side_effect = lambda order: order
    return repo

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

@pytest.fixture
def use_case(mock_repo, mock_notifier):
    return CreateOrderUseCase(repo=mock_repo, notifier=mock_notifier)

@pytest.mark.anyio
async def test_execute_saves_order_and_notifies(use_case, mock_repo, mock_notifier):
    mock_repo.find_product_by_id.return_value = {"stock": 10}

    order = await use_case.execute(product_id="prod-1", quantity=2)

    mock_repo.save.assert_called_once()
    mock_notifier.notify.assert_called_once_with(order.id)

@pytest.mark.anyio
async def test_execute_raises_when_stock_insufficient(use_case, mock_repo):
    mock_repo.find_product_by_id.return_value = {"stock": 1}

    with pytest.raises(Exception, match="Insufficient stock"):
        await use_case.execute(product_id="prod-1", quantity=5)

Use case diuji tanpa database dan tanpa HTTP stack. AsyncMock hanya perlu meng-mock ABC — bukan implementasi konkretnya.

Infrastructure — integration test via SQLite in-memory

# infrastructure/database/test_sqlalchemy_order_repository.py
import pytest
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from infrastructure.database.sqlalchemy_order_repository import SqlAlchemyOrderRepository
from domain.orders.order import Order

@pytest.fixture
async def db_session():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    async with async_session() as session:
        yield session

@pytest.mark.anyio
async def test_save_persists_order_to_database(db_session):
    repo = SqlAlchemyOrderRepository(session=db_session)
    order = Order(product_id="prod-1", quantity=2)

    saved = await repo.save(order)

    assert saved.id is not None
    assert saved.product_id == "prod-1"

Test ini satu-satunya yang menyentuh database — dan itu pun hanya SQLite in-memory. Di CI pipeline, test ini bisa dijalankan terpisah dari unit test biasa dengan marker pytest -m integration.

Keunggulan utama: domain entity dan use case bisa di-test 100% tanpa database, tanpa HTTP, tanpa library eksternal apapun. Ganti SQLAlchemy ke Tortoise ORM? Hanya sqlalchemy_order_repository.py yang berubah.

Cocok untuk: fintech, healthcare, enterprise, sistem yang akan berkembang selama 5–10 tahun.

---

Arsitektur 5: Microservices

Struktur Sistem

Di microservices, fitur Orders bukan lagi satu folder — melainkan beberapa service independen yang berjalan di proses dan container terpisah:

/order-service          ← FastAPI app
/notification-service   ← FastAPI + Celery worker
/product-service        ← FastAPI app
/api-gateway            ← nginx / Traefik

Flow antar service:

Client
  │
  ▼
[api-gateway]
  │
  ▼
[order-service] ──── HTTP GET ────► [product-service]   (cek stok)
  │
  │ publish task / event
  ▼
[RabbitMQ / Redis]  ◄── subscribe ── [notification-service]

order-service tidak langsung memanggil notification-service. Ia hanya mempublish task ke Celery atau event ke RabbitMQ. notification-service subscribe dan bereaksi secara asynchronous.

# order-service: setelah order tersimpan
from celery_app import celery

celery.send_task(
    "notification.send_order_confirmation",
    kwargs={"order_id": order.id, "customer_email": dto.customer_email},
)

Pengujian di Microservices

Unit test per service

Identik dengan Layered atau Clean Architecture secara internal. Tidak ada perbedaan — setiap service diuji sendiri menggunakan AsyncMock.

Contract testing dengan pact-python

Masalah baru muncul: bagaimana memastikan payload yang di-publish order-service sesuai dengan yang diharapkan notification-service? Keduanya berjalan di proses berbeda dan tidak bisa diuji bersamaan dengan mudah.

Jawabannya adalah contract testing:

# order-service/test_notification_contract.py
import pytest
from pact import Consumer, Provider

pact = Consumer("order-service").has_pact_with(
    Provider("notification-service"),
    pact_dir="./pacts",
    host_name="localhost",
    port=1234,
)

def test_order_service_sends_correct_payload():
    expected_body = {"order_id": "order-123", "customer_email": "user@test.com"}

    (pact
     .given("notification service is running")
     .upon_receiving("an order confirmation request")
     .with_request(method="POST", path="/notify", body=expected_body)
     .will_respond_with(status=200))

    with pact:
        import requests
        requests.post(
            f"http://localhost:1234/notify",
            json=expected_body,
        )

Contract ini di-generate sebagai file JSON dan di-publish ke Pact Broker. notification-service kemudian memverifikasi bahwa ia bisa memenuhi kontrak tersebut — tanpa order-service harus ikut berjalan.

E2E test via Docker Compose

Untuk verifikasi end-to-end, jalankan semua service menggunakan docker-compose up dan hit endpoint dari luar:

# docker-compose.test.yml
services:
  order-service:
    build: ./order-service
  notification-service:
    build: ./notification-service
  product-service:
    build: ./product-service
  rabbitmq:
    image: rabbitmq:3-management

# e2e/test_create_order_flow.py
@pytest.mark.anyio
async def test_full_order_flow():
    async with AsyncClient(base_url="http://localhost:8000") as client:
        response = await client.post("/orders", json={
            "product_id": "prod-1",
            "quantity": 2,
            "customer_email": "user@test.com",
        })

    assert response.status_code == 201
    # assert email notification was queued

Tanpa contract testing, perubahan sekecil apapun pada schema payload di order-service bisa memecah notification-service tanpa ada yang tahu — sampai sudah di production.

Cocok untuk: tim 50+ developer, platform dengan traffic tidak merata, sistem yang harus di-scale secara independen per fitur.

---

Perbandingan Side-by-Side

Aspek	Simple MVC	Layered	Feature-Based	Clean Arch	Modular Monolith	Microservices
Jumlah file untuk fitur Order	2	4	6	8+	6–8 per modul	per-service
Unit test business logic	Tidak bisa	Bisa	Bisa	Sangat mudah	Mudah (per modul)	Bisa per service
Test butuh database?	Ya (selalu)	Tidak	Tidak	Tidak	Tidak	Tidak per unit
Contract testing dibutuhkan?	Tidak	Tidak	Tidak	Tidak	Tidak	Ya (pact-python)
Kecepatan test suite	Lambat	Sedang	Sedang	Cepat	Sedang	Sedang
Ganti database	Ubah semua	Ubah repository	Ubah repository	Ubah 1 file	Ubah 1 file per modul	Per service
Onboarding developer baru	Cepat	Cepat	Sedang	Lambat	Sedang	Lambat

---

Kesimpulan

Semua arsitektur bisa di-test. Tapi cara dan kemudahannya sangat berbeda.

Simple MVC cocok saat kecepatan adalah prioritas dan sistem tidak akan berkembang jauh. Testing berat tapi acceptable untuk ukuran proyek itu.

Layered dan Feature-Based adalah sweet spot untuk sebagian besar aplikasi Python. pytest fixtures membuat mocking bersih dan terorganisir. Feature-Based menambahkan keuntungan kedekatan antara kode dan test.

Clean Architecture adalah investasi — butuh lebih banyak file dan disiplin. Tapi domain entity dan use case bisa di-test sebagai pure Python, tanpa SQLAlchemy, tanpa FastAPI, tanpa apapun. Nilainya terasa seiring waktu ketika bisnis rules makin kompleks.

Modular Monolith duduk tepat di antara Feature-Based dan Microservices — satu deployment unit, tapi batas modul ditegakkan oleh tooling bukan sekadar konvensi. Di Python, kombinasi __init__.py dan import-linter menghasilkan enforcement yang efektif. Baca lebih lanjut di post khusus Modular Monolith.

Microservices bukan tentang folder lagi, tapi tentang sistem. Ia membuka kebutuhan baru — contract testing — yang tidak ada di arsitektur monolitik manapun. pact-python adalah tooling yang perlu dipelajari saat tim mulai memecah sistem.

Pilih arsitektur yang testingnya bisa dijaga konsisten oleh tim yang ada sekarang. Bukan yang paling canggih — yang paling bisa dipertahankan.

---

Penutup

Pertanyaan yang tepat bukan:

> "Arsitektur mana yang terbaik?"

Tapi:

> "Arsitektur mana yang paling mudah diuji dan dipertahankan oleh tim saya, untuk kompleksitas yang kita hadapi sekarang?"

Kode yang bisa diuji adalah kode yang bisa dipercaya. Dan kode yang bisa dipercaya adalah fondasi dari sistem yang bisa tumbuh.