Clean Architecture 跨語言實作對照：Node.js / Go / Python 的共通點

Clean Architecture 的同心圓是語言無關的——Entity、Use Case、Interface Adapter、Framework & Driver 這四層的分工，在任何語言裡都成立。差別只是每個語言用什麼方式表達「介面」和「依賴注入」。

這篇用同一個題目（建立訂單 use case）在三種語言裡實作，涵蓋四個面向：資料夾結構、Driven Adapter、HTTP Controller（Driving Adapter）、Unit Test。

題目：建立訂單 Use Case

業務規則：

訂單金額必須 > 0
庫存足夠才能建立
建立成功後通知用戶

這個 Use Case 需要三個 port（介面）：OrderRepository（查 / 存訂單）、InventoryRepository（查庫存）、NotificationService（發通知）。Port 定義在 domain 層，具體實作（Adapter）在 infrastructure 層。

資料夾結構對照

三種語言的資料夾結構反映同一個同心圓概念，由內到外：

Node.js / TypeScript

src/
├── domain/
│   ├── entities/
│   │   └── Order.ts            ← Entity（業務物件）
│   └── ports/
│       ├── OrderRepository.ts  ← Driven Port（介面）
│       ├── InventoryRepository.ts
│       └── NotificationService.ts
├── application/
│   └── use-cases/
│       └── CreateOrder.ts      ← Use Case（業務流程）
├── infrastructure/
│   ├── postgres/
│   │   └── PostgresOrderRepository.ts  ← Driven Adapter
│   └── email/
│       └── EmailNotificationService.ts
├── interface-adapters/
│   └── http/
│       └── OrderController.ts  ← Driving Adapter
└── main.ts                     ← Composition Root（組裝）

internal/
├── domain/
│   ├── order.go       ← Entity + Port 介面
│   └── ports.go
├── application/
│   └── create_order.go  ← Use Case
├── infrastructure/
│   └── postgres/
│       └── order_repo.go  ← Driven Adapter
└── adapter/
    └── http/
        └── order_handler.go  ← Driving Adapter
cmd/api/
└── main.go            ← Composition Root

Python

src/
├── domain/
│   ├── entities.py    ← Entity
│   └── ports.py       ← Driven Port（ABC / Protocol）
├── application/
│   └── use_cases/
│       └── create_order.py  ← Use Case
├── infrastructure/
│   └── postgres/
│       └── order_repository.py  ← Driven Adapter
├── adapters/
│   └── http/
│       └── order_router.py  ← Driving Adapter
└── main.py            ← Composition Root

共通規律：domain/ 最內層，不 import 任何其他層；infrastructure/ 和 adapters/ 最外層，知道所有技術細節；application/ 在中間，只依賴 domain 的介面。

Node.js / TypeScript

Domain + Use Case

// domain/entities/Order.ts
export class Order {
  constructor(
    public readonly id: string,
    public readonly amount: number,
    public readonly userId: string,
  ) {
    if (amount <= 0) throw new Error('Amount must be positive')
  }
}
 
// domain/ports/OrderRepository.ts
export interface OrderRepository {
  save(order: Order): Promise<void>
}
 
// domain/ports/InventoryRepository.ts
export interface InventoryRepository {
  checkAvailability(productId: string, qty: number): Promise<boolean>
}
 
// application/use-cases/CreateOrder.ts
export class CreateOrderUseCase {
  constructor(
    private orders: OrderRepository,
    private inventory: InventoryRepository,
    private notifications: NotificationService,
  ) {}
 
  async execute(cmd: CreateOrderCommand): Promise<void> {
    const available = await this.inventory.checkAvailability(cmd.productId, cmd.qty)
    if (!available) throw new Error('Insufficient inventory')
 
    const order = new Order(generateId(), cmd.amount, cmd.userId)
    await this.orders.save(order)
    await this.notifications.notify(cmd.userId, `訂單 ${order.id} 已建立`)
  }
}

Driven Adapter（PostgreSQL）

// infrastructure/postgres/PostgresOrderRepository.ts
export class PostgresOrderRepository implements OrderRepository {
  constructor(private db: Database) {}
  async save(order: Order) {
    await this.db.query('INSERT INTO orders ...', [order.id, order.amount])
  }
}

Driving Adapter（HTTP Controller）

// interface-adapters/http/OrderController.ts
export class OrderController {
  constructor(private createOrder: CreateOrderUseCase) {}
 
  async handleCreate(req: Request, res: Response): Promise<void> {
    try {
      await this.createOrder.execute({
        productId: req.body.productId,
        qty: req.body.qty,
        amount: req.body.amount,
        userId: (req as any).user.id,  // 來自 auth middleware
      })
      res.status(201).json({ message: 'Order created' })
    } catch (err: any) {
      if (err.message === 'Insufficient inventory') {
        res.status(409).json({ error: err.message })
      } else {
        res.status(500).json({ error: 'Internal error' })
      }
    }
  }
}

Controller 只做一件事：把 HTTP 的語言（req / res）翻譯成 Use Case 的語言（Command）。業務規則在 Use Case 裡，不在 Controller 裡。

Composition Root

// main.ts
const useCase = new CreateOrderUseCase(
  new PostgresOrderRepository(db),
  new PostgresInventoryRepository(db),
  new EmailNotificationService(smtp),
)
const controller = new OrderController(useCase)
app.post('/orders', (req, res) => controller.handleCreate(req, res))

Go

Domain + Use Case

// domain/order.go
type Order struct {
    ID     string
    Amount float64
    UserID string
}
 
func NewOrder(id string, amount float64, userID string) (*Order, error) {
    if amount <= 0 {
        return nil, errors.New("amount must be positive")
    }
    return &Order{ID: id, Amount: amount, UserID: userID}, nil
}
 
// domain/ports.go
type OrderRepository interface {
    Save(ctx context.Context, order *Order) error
}
 
type InventoryRepository interface {
    CheckAvailability(ctx context.Context, productID string, qty int) (bool, error)
}
 
// application/create_order.go
type CreateOrderUseCase struct {
    orders        OrderRepository
    inventory     InventoryRepository
    notifications NotificationService
}
 
func (uc *CreateOrderUseCase) Execute(ctx context.Context, cmd CreateOrderCommand) error {
    ok, err := uc.inventory.CheckAvailability(ctx, cmd.ProductID, cmd.Qty)
    if err != nil || !ok {
        return errors.New("insufficient inventory")
    }
    order, err := domain.NewOrder(newID(), cmd.Amount, cmd.UserID)
    if err != nil {
        return err
    }
    if err := uc.orders.Save(ctx, order); err != nil {
        return err
    }
    return uc.notifications.Notify(ctx, cmd.UserID, "訂單已建立")
}

Driven Adapter（PostgreSQL）

// infrastructure/postgres/order_repo.go
type PostgresOrderRepository struct{ db *sql.DB }
 
func (r *PostgresOrderRepository) Save(ctx context.Context, order *domain.Order) error {
    _, err := r.db.ExecContext(ctx, "INSERT INTO orders ...", order.ID, order.Amount)
    return err
}

Driving Adapter（HTTP Handler）

// adapter/http/order_handler.go
type OrderHandler struct {
    createOrder *application.CreateOrderUseCase
}
 
func (h *OrderHandler) Create(w http.ResponseWriter, r *http.Request) {
    var body struct {
        ProductID string  `json:"productId"`
        Qty       int     `json:"qty"`
        Amount    float64 `json:"amount"`
    }
    json.NewDecoder(r.Body).Decode(&body)
 
    err := h.createOrder.Execute(r.Context(), application.CreateOrderCommand{
        ProductID: body.ProductID,
        Qty:       body.Qty,
        Amount:    body.Amount,
        UserID:    r.Header.Get("X-User-ID"),
    })
    if err != nil {
        http.Error(w, err.Error(), http.StatusConflict)
        return
    }
    w.WriteHeader(http.StatusCreated)
}

Go 的 implicit interface 讓 infrastructure 層不需要 implements 關鍵字——只要有對應的 method signature，就自動滿足 interface。

Python

Domain + Use Case

# domain/entities.py
from dataclasses import dataclass
 
@dataclass
class Order:
    id: str
    amount: float
    user_id: str
 
    def __post_init__(self):
        if self.amount <= 0:
            raise ValueError("Amount must be positive")
 
# domain/ports.py
from abc import ABC, abstractmethod
 
class OrderRepository(ABC):
    @abstractmethod
    async def save(self, order: Order) -> None: ...
 
class InventoryRepository(ABC):
    @abstractmethod
    async def check_availability(self, product_id: str, qty: int) -> bool: ...
 
# application/use_cases/create_order.py
class CreateOrderUseCase:
    def __init__(self, orders, inventory, notifications):
        self._orders = orders
        self._inventory = inventory
        self._notifications = notifications
 
    async def execute(self, cmd: CreateOrderCommand) -> None:
        available = await self._inventory.check_availability(cmd.product_id, cmd.qty)
        if not available:
            raise ValueError("Insufficient inventory")
        order = Order(id=new_id(), amount=cmd.amount, user_id=cmd.user_id)
        await self._orders.save(order)
        await self._notifications.notify(cmd.user_id, "訂單已建立")

Driven Adapter（PostgreSQL）

# infrastructure/postgres/order_repository.py
class PostgresOrderRepository(OrderRepository):
    def __init__(self, pool):
        self._pool = pool
 
    async def save(self, order: Order) -> None:
        async with self._pool.acquire() as conn:
            await conn.execute("INSERT INTO orders ...", order.id, order.amount)

Driving Adapter（FastAPI Router）

# adapters/http/order_router.py
from fastapi import APIRouter, Depends, HTTPException
 
router = APIRouter()
 
@router.post("/orders", status_code=201)
async def create_order(
    body: CreateOrderRequest,
    use_case: CreateOrderUseCase = Depends(get_create_order_use_case),
    current_user = Depends(get_current_user),
):
    try:
        await use_case.execute(CreateOrderCommand(
            product_id=body.product_id,
            qty=body.qty,
            amount=body.amount,
            user_id=current_user.id,
        ))
    except ValueError as e:
        raise HTTPException(status_code=409, detail=str(e))

FastAPI 的 Depends 是輕量的 DI 機制——get_create_order_use_case 在 main.py 裡組裝好真實的 adapter，測試時可以用 app.dependency_overrides 替換成 in-memory 版本。

Unit Test：不起資料庫就能跑

Clean Architecture 最重要的承諾是：業務邏輯的測試不需要任何 infrastructure。把 Postgres adapter 換成 in-memory 實作，Use Case 測試就能獨立跑。

Node.js

// tests/application/CreateOrder.test.ts
class InMemoryOrderRepository implements OrderRepository {
  orders: Order[] = []
  async save(order: Order) { this.orders.push(order) }
}
 
class StubInventoryRepository implements InventoryRepository {
  constructor(private available: boolean) {}
  async checkAvailability() { return this.available }
}
 
class SpyNotificationService implements NotificationService {
  notified: string[] = []
  async notify(userId: string) { this.notified.push(userId) }
}
 
describe('CreateOrderUseCase', () => {
  it('creates order when inventory is available', async () => {
    const orders = new InMemoryOrderRepository()
    const useCase = new CreateOrderUseCase(
      orders,
      new StubInventoryRepository(true),
      new SpyNotificationService(),
    )
 
    await useCase.execute({ productId: 'p1', qty: 2, amount: 500, userId: 'u1' })
 
    expect(orders.orders).toHaveLength(1)
  })
 
  it('throws when inventory is insufficient', async () => {
    const useCase = new CreateOrderUseCase(
      new InMemoryOrderRepository(),
      new StubInventoryRepository(false),
      new SpyNotificationService(),
    )
 
    await expect(
      useCase.execute({ productId: 'p1', qty: 10, amount: 500, userId: 'u1' })
    ).rejects.toThrow('Insufficient inventory')
  })
})

Go

// application/create_order_test.go
type inMemoryOrderRepo struct {
    orders []*domain.Order
}
func (r *inMemoryOrderRepo) Save(_ context.Context, o *domain.Order) error {
    r.orders = append(r.orders, o)
    return nil
}
 
type stubInventory struct{ available bool }
func (s *stubInventory) CheckAvailability(_ context.Context, _ string, _ int) (bool, error) {
    return s.available, nil
}
 
func TestCreateOrder_Success(t *testing.T) {
    repo := &inMemoryOrderRepo{}
    uc := &CreateOrderUseCase{
        orders:        repo,
        inventory:     &stubInventory{available: true},
        notifications: &stubNotification{},
    }
 
    err := uc.Execute(context.Background(), CreateOrderCommand{
        ProductID: "p1", Qty: 2, Amount: 500, UserID: "u1",
    })
 
    assert.NoError(t, err)
    assert.Len(t, repo.orders, 1)
}
 
func TestCreateOrder_InsufficientInventory(t *testing.T) {
    uc := &CreateOrderUseCase{
        orders:        &inMemoryOrderRepo{},
        inventory:     &stubInventory{available: false},
        notifications: &stubNotification{},
    }
 
    err := uc.Execute(context.Background(), CreateOrderCommand{
        ProductID: "p1", Qty: 10, Amount: 500, UserID: "u1",
    })
 
    assert.ErrorContains(t, err, "insufficient inventory")
}

Python

# tests/application/test_create_order.py
import pytest
from domain.entities import Order
from application.use_cases.create_order import CreateOrderUseCase, CreateOrderCommand
 
class InMemoryOrderRepository:
    def __init__(self): self.orders: list[Order] = []
    async def save(self, order: Order) -> None: self.orders.append(order)
 
class StubInventoryRepository:
    def __init__(self, available: bool): self._available = available
    async def check_availability(self, *_) -> bool: return self._available
 
class SpyNotificationService:
    def __init__(self): self.notified: list[str] = []
    async def notify(self, user_id: str, _: str) -> None: self.notified.append(user_id)
 
@pytest.mark.asyncio
async def test_creates_order_when_inventory_available():
    repo = InMemoryOrderRepository()
    use_case = CreateOrderUseCase(
        orders=repo,
        inventory=StubInventoryRepository(available=True),
        notifications=SpyNotificationService(),
    )
 
    await use_case.execute(CreateOrderCommand(
        product_id="p1", qty=2, amount=500.0, user_id="u1"
    ))
 
    assert len(repo.orders) == 1
 
@pytest.mark.asyncio
async def test_raises_when_inventory_insufficient():
    use_case = CreateOrderUseCase(
        orders=InMemoryOrderRepository(),
        inventory=StubInventoryRepository(available=False),
        notifications=SpyNotificationService(),
    )
 
    with pytest.raises(ValueError, match="Insufficient inventory"):
        await use_case.execute(CreateOrderCommand(
            product_id="p1", qty=10, amount=500.0, user_id="u1"
        ))

測試裡完全沒有資料庫、HTTP、外部服務——跑一個測試的時間是毫秒，不是秒。這就是 port / adapter 分離的主要收益。

三種語言的共通骨架

語法不同，但以下四件事在三種語言裡完全一樣：

Port 定義在 domain 層，不 import 任何 infrastructure：OrderRepository 介面只描述「我需要什麼操作」，不知道是 PostgreSQL 還是 MongoDB。

Use Case 只依賴 port：CreateOrderUseCase 接收介面，不知道具體實作——測試傳 in-memory mock，生產傳 Postgres 實作，Use Case 的 code 一字不動。

Controller 只做翻譯：HTTP request → Use Case Command；Use Case 結果 / 錯誤 → HTTP response。業務規則不住在 Controller 裡。

組裝在 Composition Root：只有程式進入點（main / app factory）知道「用哪個 adapter」。其他所有地方只看到介面，不知道具體實作是什麼。

這個骨架在任何有介面概念的語言裡都能實現，語法是表面差異，架構決策是共通的。

Terry Yao's Blog

目錄

Clean Architecture 跨語言實作對照：Node.js / Go / Python 的共通點

題目：建立訂單 Use Case

資料夾結構對照

Node.js / TypeScript

Domain + Use Case

Driven Adapter（PostgreSQL）

Driving Adapter（HTTP Controller）

Composition Root

Go

Domain + Use Case

Driven Adapter（PostgreSQL）

Driving Adapter（HTTP Handler）

Python

Domain + Use Case

Driven Adapter（PostgreSQL）

Driving Adapter（FastAPI Router）

Unit Test：不起資料庫就能跑

Node.js

Go

Python

三種語言的共通骨架

關係圖譜