app_旧版.py (1)

# app_legacy.py (1)
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="未找到")
        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="未找到")
        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="未找到")
        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="未找到")
        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="任务重构版", 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