在路線圖裡，這一天就是把資料層與應用服務層接到 Web 端入口（Adapter）上，完整走完「入站介面 → 應用服務 → 資料層」的閉環。路線圖早就把這天寫死了，不是我臨時發揮（請翻 Day 1 的清單）。

我們會沿用你在 Day 4 的分層骨架與 Day 19 的 Repository Pattern 設計，把 Web 介面當成 入站 Adapter，只依賴 Ports（介面） 而不是黏死 ORM 細節 。

今日目標

1. 用 FastAPI 建立 入站 Adapter：路由、請求/回應 Schema、錯誤對應。

2. 以 DI（依賴注入） 注入 UserService，而 UserService 只依賴 UserRepository 介面（Port）。

3. 串上 Day 10 的 Pydantic v2 資料契約 與 Day 13 的 結構化日誌，Day 14 的 重試/降級 概念在邊界錯誤時派上用場。

4. 提供 測試藍圖（pytest + TestClient），沿用 Day 11 的策略。

5. 保持與 Day 4 的目錄慣例一致，並延續 Day 8 的一鍵指令。

   （是的，這一切你都先學過了，今天只是把積木扣緊。）

專案目錄（延伸版）

延續 src/ + tests/ 與分層思維，把 Web 當作 Adapter 放在 adapters/web 下：

（Day 4 已經解釋過為什麼這樣拆比較不會長成一坨巨型 views.py）

my_project/
├─ pyproject.toml
├─ src/my_project/
│  ├─ domain/
│  │  ├─ entities.py              # 純領域實體（或 Pydantic model for domain）
│  │  └─ repositories.py          # Ports：Repository 介面
│  ├─ services/
│  │  └─ user_service.py          # 應用服務，僅依賴 Port
│  ├─ infrastructure/
│  │  └─ db/
│  │     ├─ models.py             # SQLAlchemy ORM
│  │     ├─ session.py            # engine/session/UoW
│  │     ├─ repositories.py       # Repository 實作（Adapter）
│  │     └─ repositories_cached.py# 快取裝飾（選用）
│  ├─ adapters/
│  │  └─ web/
│  │     ├─ app.py                # FastAPI app 建置與 DI
│  │     ├─ routers/
│  │     │  └─ users.py           # /v1/users* 路由
│  │     ├─ schemas.py            # API I/O 的 Pydantic v2 契約
│  │     ├─ deps.py               # Depends 提供 Session/Service/Repo 等
│  │     ├─ errors.py             # 例外轉 HTTPError，降級策略
│  │     └─ middleware.py         # 日誌/追蹤/版本/計時中介層
│  ├─ settings.py                 # pydantic-settings（Day 12）
│  └─ logging_config.py           # structlog JSON（Day 13）
└─ tests/
   ├─ unit/
   └─ integration/
      └─ test_users_api.py

API 契約：Pydantic v2 模型（入站/出站）

邊界資料該被驗證與序列化，這你在 Day 10 已經建立肌肉記憶了。

# src/my_project/adapters/web/schemas.py
from typing import Annotated, Optional
from pydantic import BaseModel, Field, EmailStr

UserId = Annotated[str, Field(min_length=1, max_length=36)]

class CreateUserIn(BaseModel):
    email: EmailStr
    name: Annotated[str, Field(min_length=1, max_length=100)]

class UserOut(BaseModel):
    id: UserId
    email: EmailStr
    name: str

class UsersPage(BaseModel):
    items: list[UserOut]
    total: int
    limit: int
    offset: int

DI 與 Lifespan：把 Repository/Service 接到 FastAPI

• Repository 具體類別仍來自 Day 19 的實作。
• Session 與 UoW 也沿用 Day 19 的集中管理。
• 設定來自 Day 12 的 pydantic-settings。
• 日誌使用 Day 13 的 structlog JSON。

# src/my_project/adapters/web/deps.py
from typing import Iterator
from fastapi import Depends
from sqlalchemy.orm import Session
from my_project.infrastructure.db.session import make_session_factory
from my_project.infrastructure.db.repositories import SqlAlchemyUserRepository
from my_project.services.user_service import UserService
from my_project.settings import AppSettings

_settings = AppSettings()  # Day 12
_SessionFactory = make_session_factory(_settings.db_url)

def get_session() -> Iterator[Session]:
    with _SessionFactory() as s:
        yield s

def get_user_repo(session: Session = Depends(get_session)) -> SqlAlchemyUserRepository:
    return SqlAlchemyUserRepository(session)

def get_user_service(repo: SqlAlchemyUserRepository = Depends(get_user_repo)) -> UserService:
    return UserService(users=repo)

# src/my_project/adapters/web/app.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from .middleware import setup_middlewares
from .errors import setup_exception_handlers
from .routers import users as users_router
from my_project.logging_config import setup_logging

@asynccontextmanager
async def lifespan(app: FastAPI):
    setup_logging()        # Day 13：JSON 結構化日誌
    yield                  # 可在此處理連線池 warm-up 或 metrics 啟動

def create_app() -> FastAPI:
    app = FastAPI(title="Awesome API", version="1.0.0", lifespan=lifespan)
    setup_middlewares(app)
    setup_exception_handlers(app)
    app.include_router(users_router.router, prefix="/v1", tags=["users"])
    return app

app = create_app()

路由與用例對接：只跟 Service 說話

UserService 在 Day 19 已經長大了。這裡 API 層只把 HTTP 請求轉成用例呼叫，再把回應包成 schema。

# src/my_project/adapters/web/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status, Query
from ..schemas import CreateUserIn, UserOut, UsersPage
from my_project.services.user_service import UserService
from ..deps import get_user_service

router = APIRouter()

@router.post("/users", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def create_user(payload: CreateUserIn, svc: UserService = Depends(get_user_service)):
    # id 決策可由 service 生成（uuid）或由 domain 規則決定
    user = svc.register_new_user(email=payload.email, name=payload.name)
    if user is None:
        raise HTTPException(status_code=409, detail="Email already exists")
    return UserOut.model_validate(user.__dict__)

@router.get("/users/{user_id}", response_model=UserOut)
def get_user(user_id: str, svc: UserService = Depends(get_user_service)):
    u = svc.get_user(user_id)
    if not u:
        raise HTTPException(status_code=404, detail="User not found")
    return UserOut.model_validate(u.__dict__)

@router.get("/users", response_model=UsersPage)
def list_users(
    limit: int = Query(50, ge=1, le=200),
    offset: int = Query(0, ge=0),
    svc: UserService = Depends(get_user_service),
):
    items, total = svc.list_users(limit=limit, offset=offset)
    return UsersPage(
        items=[UserOut.model_validate(i.__dict__) for i in items],
        total=total, limit=limit, offset=offset
    )

註：register_new_user、get_user、list_users 這些用例只是把 Day 19 的 service 小小延伸，語意清楚、側重 business 用詞，完全不帶 ORM 與 HTTP 細節。

例外與降級：把失敗設計成一等公民

• 針對「暫時性錯誤」保守重試，把 Day 14 的策略包在 Repository 或外層裝飾器即可。
• API 層統一把 domain/service 例外轉成 HTTP 狀態碼，並輸出結構化日誌。

# src/my_project/adapters/web/errors.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import structlog

log = structlog.get_logger()

class DomainError(Exception): ...
class ConflictError(DomainError): ...
class NotFoundError(DomainError): ...

def setup_exception_handlers(app: FastAPI):
    @app.exception_handler(ConflictError)
    async def conflict_handler(_: Request, exc: ConflictError):
        log.info("conflict", reason=str(exc))
        return JSONResponse({"detail": str(exc)}, status_code=409)

    @app.exception_handler(NotFoundError)
    async def notfound_handler(_: Request, exc: NotFoundError):
        log.info("not_found", reason=str(exc))
        return JSONResponse({"detail": str(exc)}, status_code=404)

    @app.exception_handler(Exception)
    async def unhandled_handler(_: Request, exc: Exception):
        # 降級/預設訊息，避免把內部堆疊灑給使用者
        log.error("unhandled_error", reason=str(exc))
        return JSONResponse({"detail": "internal error"}, status_code=500)

中介層：請求追蹤、計時與 JSON Log

Day 13 提醒過，雲端與容器化世界最好走 JSON，方便平台解析與聚合。

# src/my_project/adapters/web/middleware.py
import time, uuid, structlog
from fastapi import FastAPI, Request

def setup_middlewares(app: FastAPI):
    log = structlog.get_logger()

    @app.middleware("http")
    async def tracing(request: Request, call_next):
        rid = request.headers.get("x-request-id", str(uuid.uuid4()))
        t0 = time.perf_counter()
        resp = await call_next(request)
        ms = round((time.perf_counter() - t0) * 1000, 2)
        log.info(
            "http_access",
            method=request.method, path=request.url.path, status=resp.status_code,
            ms=ms, request_id=rid
        )
        resp.headers["x-request-id"] = rid
        return resp

啟動與一鍵指令（Hatch scripts）

延續 Day 8，把「起服務、檢查、測試」包進 scripts。

# pyproject.toml 片段
[project.optional-dependencies]
api = ["fastapi>=0.115", "uvicorn>=0.30"]

[tool.hatch.envs.api]
features = ["api", "dev"]  # dev: pytest/ruff/mypy 見 Day 6/8 設定
[tool.hatch.envs.api.scripts]
serve = "uvicorn my_project.adapters.web.app:app --reload --port 8000"
check = ["ruff .", "mypy src/"]
test = "pytest -q"

測試策略：TestClient、fixtures 與覆蓋率

Day 11 的 pytest 藍圖可以原封不動拿來用，這裡示範 API 端到端一小段。

# tests/integration/test_users_api.py
from fastapi.testclient import TestClient
from my_project.adapters.web.app import create_app

def test_create_and_get_user(monkeypatch):
    app = create_app()
    client = TestClient(app)

    # 建立
    r = client.post("/v1/users", json={"email": "a@x.com", "name": "Alice"})
    assert r.status_code == 201, r.text
    uid = r.json()["id"]

    # 取得
    r = client.get(f"/v1/users/{uid}")
    assert r.status_code == 200
    assert r.json()["email"] == "a@x.com"

如果你想走「純 service + fake repo」的單元測試，就照 Day 11 的 unit/integration 分層，讓 unit 不碰 HTTP，integration 才進 TestClient。

效能與可靠性補強（選用）

• 快取：把 Day 18 的 Cache-Aside 裝飾 Repository，API 不必知道快取存在。
• 重試：對 OperationalError 類可恢復錯誤套上 Day 14 的 tenacity，避免短暫抖動打到 API 使用者臉上。
• 序列化：API 回應可切到 orjson（FastAPI Response class 或全域 JSONEncoder），對高 QPS 蠻有感（Day 15）。
• 非同步：若路由多是 I/O 密集，請把 Repository/Session 換成 async 版本（Day 16/19 已點到）。

對照檢查表（Definition of Done）

• API 只依賴 Service/Port，不碰 ORM 或 SQL。
• Pydantic v2 契約明確：入站驗證、出站模型固定。
• 結構化日誌與請求計時上線，輸出 JSON。
• 例外對映與保守重試，必要時降級。
• pytest + TestClient 的 integration 測試通過，覆蓋率收斂到 CI。
• Hatch scripts 一鍵啟動/檢查/測試。

小結

六邊形架構不是宗教儀式，它只是在逼你把邊界切清楚：Web 是入站 Adapter，Repository 是出站 Adapter，Service/Port 才是中間那圈有價值的規則。

這樣設計的副作用叫「自由」：你可以在不動用例的前提下替換 FastAPI 成別的 Web 框架、把 ORM 換掉、把快取抽掉，甚至把資料庫從 SQLite 換成 Postgres 而不讓控制器哭出來。

你不需要喜歡它，但之後會感謝它。當你在壓測、排錯、或被半夜叫醒的時候，會的。

下一步，該把這條鏈接上部署與運維那邊了。你知道路線圖上 Day 21 到 Day 27 的內容，別裝不知道。