AI Learn
登录注册
English中文

正在学习

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

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

练习题

What is the purpose of the _ensure_db() function in the legacy FastAPI application?

A. To create a new task
B. To ensure the database table exists
C. To connect to the database
D. To delete a task

Which endpoint is responsible for returning the health status of the legacy FastAPI application?

A. /tasks
B. /health
C. /create
D. /status

What are the valid fields in the TaskIn model?

A. id
B. title
C. description
D. completed

Which of the following are valid HTTP methods used in the legacy FastAPI application endpoints?

A. GET
B. POST
C. PUT
D. DELETE

The create_task endpoint in the legacy FastAPI application raises an HTTPException if the title is empty or exceeds 120 characters.

The delete_task endpoint in the legacy FastAPI application returns a response body after deleting a task.

The database connection path is defined by the variable ___.

The put_task endpoint updates a task's details using the ___ method.

Explain the purpose of the list_tasks endpoint in the legacy FastAPI application.

What is the role of the init_db() function in the layered structure of the FastAPI application?

Which of the following best describes the purpose of the get_conn() function in the layered structure?

A. To execute SQL queries
B. To initialize the database
C. To provide a database connection iterator
D. To commit changes to the database

What are the key benefits of refactoring the FastAPI application to a layered structure? (Select all that apply)

A. Isolates data access
B. Moves rules to the service layer
C. Keeps I/O at the edges
D. Introduces a minimal startup path

In the refactored layered structure, which component is responsible for handling database connection details and SQL queries?

A. The routes.py file
B. The TaskRepository class in repositories.py
C. The services.py file
D. The app.py file

Which of the following are benefits of moving business rules and validation checks to a service layer in a FastAPI application? (Select all that apply)

A. Makes routes thinner and more focused on HTTP handling
B. Allows for easier testing of core business logic
C. Reduces the need for database connections
D. Enables reuse of business logic across multiple routes
E. Improves performance by reducing network latency

In the refactored structure, the routes.py file should contain all the SQL queries for task operations.

The _____ class in repositories.py should be used to encapsulate all database operations for tasks, following the principle of isolating data access.

Explain how the refactored structure improves maintainability compared to the legacy monolith.

登录后解锁笔记、知识点解析、AI 问答

立即登录