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()

Run it:

python -m venv .venv

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

pip install fastapi uvicorn

uvicorn app_legacy:app --reload

You now have a baseline. Next, refactor to a layered structure. Create the following files and folders exactly as shown.

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")

    # Simple composition root
    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():
    # create
    r = client.post("/tasks", json={"title": "Refactor", "description": "split layers", "completed": False})
    assert r.status_code == 201
    tid = r.json()["id"]

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

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

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

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

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

    # 404 after delete
    r = client.get(f"/tasks/{tid}")
    assert r.status_code == 404

Run the refactored service and tests:

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

# In another terminal:
pytest -q

You have preserved behavior while isolating concerns. Routes are slim, the service expresses rules and fan-in logic, and repositories encapsulate SQL. This makes tests fast and future changes safer.

Clarification Table: Before vs. After

| Aspect | Legacy Monolith | Refactored Structure | Benefit |
| --- | --- | --- | --- |
| Database access | Inline sqlite 3 calls in routes | TaskRepository hides SQL | Easier to swap storage and test |
| Business rules | Scattered in endpoints | TaskService centralizes logic | Consistent validation and reuse |
| Routing | Fat endpoints | Thin handlers in routes.py | Readable HTTP layer |
| Initialization | Global state and helpers | create_app( ) composition | Deterministic startup |
| Testability | Hard to isolate logic | pytest hits stable boundaries | Safer incremental changes |

You transformed a working-but-fragile monolith into a small, layered application that is easier to reason about and extend. The refactor changed structure, not behavior, and introduced seams—a repository and a service—where Claude Code can help you iterate confidently. From here you can add caching, switch SQLite to Postgres, or plug in background jobs without touching route code. In the next section you will apply the same patterns to synchronize edits across multiple modules, letting Claude update routes, services, and tests in one guided pass while preserving contracts.

Chapter 11 – Security, Risk, and Governance in Claude Workflows