app_legacy.py

from fastapi import FastAPI, HTTPException

from pydantic import BaseModel

import sqlite3

from typing import List, Optional

app = FastAPI(title="Tasks Legacy Monolith", version="0.1.0")

DB_PATH = "tasks.db"

class TaskIn(BaseModel):

    title: str

    description: Optional[str] = ""

    completed: bool = False

class Task(TaskIn):

    id: int

def _ensure_db():

    conn = sqlite3.connect(DB_PATH)

    cur = conn.cursor()

    cur.execute(

        "CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY, title TEXT, description TEXT, completed INTEGER)"

    )

    conn.commit()

    conn.close()

_ensure_db()

@app.get("/health")

def health():

    return {"status": "ok", "service": "legacy"}

@app.get("/tasks", response_model=List[Task])

def list_tasks():

    conn = sqlite3.connect(DB_PATH)

    cur = conn.cursor()

    cur.execute("SELECT id, title, description, completed FROM tasks ORDER BY id")

    rows = cur.fetchall()

    conn.close()

    return [

        {"id": r[0], "title": r[1], "description": r[2], "completed": bool(r[3])}

        for r in rows

    ]

@app.post("/tasks", response_model=Task, status_code=201)

def create_task(payload: TaskIn):

    if not payload.title or not payload.title.strip():

        raise HTTPException(status_code=400, detail="Title required")

    if len(payload.title) > 120:

        raise HTTPException(status_code=400, detail="Title too long")

    conn = sqlite3.connect(DB_PATH)

    cur = conn.cursor()

    cur.execute(

        "INSERT INTO tasks (title, description, completed) VALUES (?, ?, ?)",

        (payload.title.strip(), payload.description or "", int(payload.completed)),

    )

    conn.commit()

    task_id = cur.lastrowid

    cur.execute("SELECT id, title, description, completed FROM tasks WHERE id = ?", (task_id,))

    row = cur.fetchone()

    conn.close()

    return {"id": row[0], "title": row[1], "description": row[2], "completed": bool(row[3])}

@app.put("/tasks/{task_id}", response_model=Task)

def put_task(task_id: int, payload: TaskIn):

    conn = sqlite3.connect(DB_PATH)

    cur = conn.cursor()

    cur.execute("SELECT id FROM tasks WHERE id = ?", (task_id,))

    if not cur.fetchone():

        conn.close()

        raise HTTPException(status_code=404, detail="not found")

    cur.execute(

        "UPDATE tasks SET title=?, description=?, completed=? WHERE id=?",

        (payload.title.strip(), payload.description or "", int(payload.completed), task_id),

    )

    conn.commit()

    cur.execute("SELECT id, title, description, completed FROM tasks WHERE id = ?", (task_id,))

    row = cur.fetchone()

    conn.close()

    return {"id": row[0], "title": row[1], "description": row[2], "completed": bool(row[3])}

@app.delete("/tasks/{task_id}", status_code=204)

def delete_task(task_id: int):

    conn = sqlite3.connect(DB_PATH)

    cur = conn.cursor()

    cur.execute("DELETE FROM tasks WHERE id = ?", (task_id,))

    conn.commit()

    conn.close()

运行它：

python -m venv .venv

source .venv/bin/activate # Windows: .venv\Scripts\activate

pip install fastapi uvicorn

uvicorn app_legacy:app --reload

现在你已经有了一个基线版本。接下来，重构为分层结构。完全按照所示创建以下文件和文件夹。

app/

init.py

app.py

database.py

models.py

repositories.py

services.py

routes.py

tests/

test_tasks.py

requirements.txt

requirements.txt

fastapi==0.114.0

uvicorn==0.30.5

app/database.py

import sqlite3

from typing import Iterator

DB_PATH = "tasks.db"

def init_db() -> None:

    with sqlite3.connect(DB_PATH) as conn:

        cur = conn.cursor()

        cur.execute(

            "CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY, title TEXT, description TEXT, completed INTEGER)"

        )

        conn.commit()

def get_conn() -> Iterator[sqlite3.Connection]:

    return sqlite3.connect(DB_PATH)

app/models.py

from pydantic import BaseModel, Field

from typing import Optional

class TaskCreate(BaseModel):

    title: str = Field(min_length=1, max_length=120)

    description: Optional[str] = ""

    completed: bool = False

class Task(TaskCreate):

    id: int

class TaskUpdate(BaseModel):

    title: Optional[str] = None

    description: Optional[str] = None

    completed: Optional[bool] = None

app/repositories.py

import sqlite3

from typing import List, Optional, Dict, Any

class TaskRepository:

    def __init__(self, conn: sqlite3.Connection):

        self.conn = conn

    def list(self) -> List[Dict[str, Any]]:

        cur = self.conn.cursor()

        cur.execute("SELECT id, title, description, completed FROM tasks ORDER BY id")

        return [

            {"id": r[0], "title": r[1], "description": r[2], "completed": bool(r[3])}

            for r in cur.fetchall()

        ]

    def get(self, task_id: int) -> Optional[Dict[str, Any]]:

        cur = self.conn.cursor()

        cur.execute("SELECT id, title, description, completed FROM tasks WHERE id=?", (task_id,))

        r = cur.fetchone()

        if not r:

            return None

        return {"id": r[0], "title": r[1], "description": r[2], "completed": bool(r[3])}

    def create(self, title: str, description: str, completed: bool) -> Dict[str, Any]:

        cur = self.conn.cursor()

        cur.execute(

            "INSERT INTO tasks (title, description, completed) VALUES (?, ?, ?)",

            (title, description, int(completed)),

        )

        self.conn.commit()

        return self.get(cur.lastrowid)

def put(self, task_id: int, title: str, description: str, completed: bool) -> Optional[Dict[str, Any]]:
    cur = self.conn.cursor()
    cur.execute("UPDATE tasks SET title=?, description=?, completed=? WHERE id=?",
                (title, description, int(completed), task_id))
    self.conn.commit()
    return self.get(task_id)

def delete(self, task_id: int) -> None:
    cur = self.conn.cursor()
    cur.execute("DELETE FROM tasks WHERE id=?", (task_id,))
    self.conn.commit()

app/services.py

from typing import Dict, Any, Optional

from .repositories import TaskRepository
from .models import TaskCreate, TaskUpdate

class TaskService:
    def __init__(self, repo: TaskRepository):
        self.repo = repo

    def list_tasks(self):
        return self.repo.list()

    def create_task(self, payload: TaskCreate):
        title = payload.title.strip()
        desc = (payload.description or "").strip()
        return self.repo.create(title, desc, payload.completed)

    def get_task(self, task_id: int) -> Optional[Dict[str, Any]]:
        return self.repo.get(task_id)

    def put_task(self, task_id: int, payload: TaskCreate):
        title = payload.title.strip()
        desc = (payload.description or "").strip()
        return self.repo.put(task_id, title, desc, payload.completed)

    def patch_task(self, task_id: int, payload: TaskUpdate):
        current = self.repo.get(task_id)
        if not current:
            return None
        title = current["title"] if payload.title is None else payload.title.strip()
        desc = current["description"] if payload.description is None else (payload.description or "").strip()
        completed = current["completed"] if payload.completed is None else bool(payload.completed)
        return self.repo.put(task_id, title, desc, completed)

    def delete_task(self, task_id: int):
        self.repo.delete(task_id)

app/routes.py

from fastapi import APIRouter, HTTPException
from typing import List
from .models import Task, TaskCreate, TaskUpdate
from .services import TaskService

router = APIRouter()

def mount_routes(service: TaskService) -> APIRouter:
    @router.get("/health")
    def health():
        return {"status": "ok", "service": "refactored"}

    @router.get("/tasks", response_model=List[Task])
    def list_tasks():
        return service.list_tasks()

    @router.post("/tasks", response_model=Task, status_code=201)
    def create_task(payload: TaskCreate):
        return service.create_task(payload)

    @router.get("/tasks/{task_id}", response_model=Task)
    def get_task(task_id: int):
        task = service.get_task(task_id)
        if not task:
            raise HTTPException(status_code=404, detail="not found")
        return task

    @router.put("/tasks/{task_id}", response_model=Task)
    def put_task(task_id: int, payload: TaskCreate):
        if not service.get_task(task_id):
            raise HTTPException(status_code=404, detail="not found")
        return service.put_task(task_id, payload)

    @router.patch("/tasks/{task_id}", response_model=Task)
    def patch_task(task_id: int, payload: TaskUpdate):
        updated = service.patch_task(task_id, payload)
        if not updated:
            raise HTTPException(status_code=404, detail="not found")
        return updated

    @router.delete("/tasks/{task_id}", status_code=204)
    def delete_task(task_id: int):
        if not service.get_task(task_id):
            raise HTTPException(status_code=404, detail="not found")
        service.delete_task(task_id)

    return router

app/app.py

from fastapi import FastAPI
from .database import init_db, get_conn
from .repositories import TaskRepository
from .services import TaskService
from .routes import mount_routes

def create_app() -> FastAPI:
    init_db()
    app = FastAPI(title="Tasks Refactored", version="1.0.0")

    # 简单的组合根
    conn = get_conn()
    repo = TaskRepository(conn)
    service = TaskService(repo)
    app.include_router(mount_routes(service))

    return app

app = create_app()

tests/test_tasks.py

from fastapi.testclient import TestClient
from app.app import app

client = TestClient(app)

def test_health():
    r = client.get("/health")
    assert r.status_code == 200
    assert r.json()["status"] == "ok"

def test_create_list_get_put_patch_delete():
    # 创建
    r = client.post("/tasks", json={"title": "Refactor", "description": "split layers", "completed": False})
    assert r.status_code == 201
    tid = r.json()["id"]

    # 列表
    r = client.get("/tasks")
    assert r.status_code == 200
    assert any(t["id"] == tid for t in r.json())

    # 获取
    r = client.get(f"/tasks/{tid}")
    assert r.status_code == 200
    assert r.json()["title"] == "Refactor"

    # 更新
    r = client.put(f"/tasks/{tid}", json={"title": "Refactor All", "description": "", "completed": True})
    assert r.status_code == 200
    assert r.json()["completed"] is True

    # 部分更新
    r = client.patch(f"/tasks/{tid}", json={"description": "done"})
    assert r.status_code == 200
    assert r.json()["description"] == "done"

    # 删除
    r = client.delete(f"/tasks/{tid}")
    assert r.status_code == 204

    # 删除后返回 404
    r = client.get(f"/tasks/{tid}")
    assert r.status_code == 404

运行重构后的服务和测试：

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt pytest
uvicorn app.app:app --reload

# 在另一个终端中：
pytest -q

你在隔离关注点的同时保留了原有行为。路由层简洁，服务层表达业务规则和聚合逻辑，仓储层封装 SQL。这使得测试更快，也让未来的变更更加安全。

对比表：重构前 vs.

之后

你将一个可用但脆弱的单体应用，转变为一个易于理解和扩展的、规模较小的分层应用。这次重构改变了结构，而非行为，并引入了接缝——一个仓储和一个服务——在这些位置 Claude Code 可以帮助你自信地进行迭代。从这里开始，你可以添加缓存、将 SQLite 切换为 Postgres，或者接入后台任务，而无需改动路由代码。在下一节中，你将把同样的模式应用于跨多个模块同步编辑，让 Claude 在一次引导式的操作中同时更新路由、服务和测试，同时保持接口契约不变。

第十一章 – Claude 工作流中的安全、风险与治理