工程阶段实战：把任务服务做成可配置、可测试的应用

HTTP 请求
   │
   ▼
FastAPI 路由与参数校验
   │
   ▼
TaskRepository 数据访问边界
   │
   ▼
MySQL engineering_notes.tasks

task-service/
├── app/
│   ├── __init__.py
│   ├── config.py
│   ├── db.py
│   └── main.py
├── tests/
│   └── test_api.py
├── sql/
│   └── init.sql
├── .env.example
├── .gitignore
└── requirements.txt

# requirements.txt
fastapi
uvicorn
mysql-connector-python
pytest
httpx

# .gitignore
.venv/
.env
__pycache__/
.pytest_cache/
*.log

# .env.example
DB_HOST=127.0.0.1
DB_PORT=3306
DB_USER=app_user
DB_PASSWORD=replace-me
DB_NAME=engineering_notes
LOG_LEVEL=INFO

CREATE DATABASE IF NOT EXISTS engineering_notes
  CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci;
USE engineering_notes;

CREATE TABLE IF NOT EXISTS projects (
    id BIGINT UNSIGNED PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(100) NOT NULL UNIQUE,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS tasks (
    id BIGINT UNSIGNED PRIMARY KEY AUTO_INCREMENT,
    project_id BIGINT UNSIGNED NOT NULL,
    title VARCHAR(200) NOT NULL,
    status ENUM('todo', 'doing', 'done') NOT NULL DEFAULT 'todo',
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    CONSTRAINT fk_tasks_project FOREIGN KEY (project_id)
      REFERENCES projects(id) ON DELETE CASCADE
);

INSERT IGNORE INTO projects (id, name)
VALUES (1, 'engineering-basics');

# app/config.py
import os
from dataclasses import dataclass


@dataclass(frozen=True)
class Settings:
    db_host: str
    db_port: int
    db_user: str
    db_password: str
    db_name: str


def load_settings() -> Settings:
    return Settings(
        db_host=os.environ.get("DB_HOST", "127.0.0.1"),
        db_port=int(os.environ.get("DB_PORT", "3306")),
        db_user=os.environ["DB_USER"],
        db_password=os.environ["DB_PASSWORD"],
        db_name=os.environ.get("DB_NAME", "engineering_notes"),
    )

# app/db.py
from collections.abc import Callable
from typing import Any

from mysql.connector import MySQLConnection


class TaskRepository:
    def __init__(self, connection_factory: Callable[[], MySQLConnection]) -> None:
        self.connection_factory = connection_factory

    def create(self, project_id: int, title: str) -> dict[str, Any]:
        connection = self.connection_factory()
        cursor = connection.cursor(dictionary=True)
        try:
            cursor.execute(
                "INSERT INTO tasks (project_id, title) VALUES (%s, %s)",
                (project_id, title),
            )
            task_id = int(cursor.lastrowid)
            connection.commit()
            return {"id": task_id, "project_id": project_id,
                    "title": title, "status": "todo"}
        except Exception:
            connection.rollback()
            raise
        finally:
            cursor.close()
            connection.close()

    def get(self, task_id: int) -> dict[str, Any] | None:
        connection = self.connection_factory()
        cursor = connection.cursor(dictionary=True)
        try:
            cursor.execute(
                "SELECT id, project_id, title, status FROM tasks WHERE id = %s",
                (task_id,),
            )
            return cursor.fetchone()
        finally:
            cursor.close()
            connection.close()

# app/main.py
import logging

import mysql.connector
from fastapi import FastAPI, HTTPException, status
from fastapi.responses import JSONResponse
from mysql.connector import Error as MySQLError
from pydantic import BaseModel, Field

from app.config import load_settings
from app.db import TaskRepository

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger(__name__)


class TaskCreate(BaseModel):
    project_id: int = Field(gt=0)
    title: str = Field(min_length=1, max_length=200)


class Task(BaseModel):
    id: int
    project_id: int
    title: str
    status: str


def build_repository() -> TaskRepository:
    settings = load_settings()

    def connection_factory():
        return mysql.connector.connect(
            host=settings.db_host,
            port=settings.db_port,
            user=settings.db_user,
            password=settings.db_password,
            database=settings.db_name,
            charset="utf8mb4",
        )

    return TaskRepository(connection_factory)


def create_app(repository=None) -> FastAPI:
    active_repository = repository or build_repository()
    app = FastAPI(title="Engineering Task Service")

    @app.exception_handler(MySQLError)
    def database_error_handler(request, exc):
        logger.exception("database_error path=%s", request.url.path)
        return JSONResponse(status_code=503, content={"detail": "database unavailable"})

    @app.get("/health")
    def health():
        return {"status": "ok"}

    @app.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED)
    def create_task(payload: TaskCreate):
        task = active_repository.create(payload.project_id, payload.title)
        logger.info(
            "task_created task_id=%s project_id=%s",
            task["id"], payload.project_id,
        )
        return task

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

    return app

# tests/test_api.py
from fastapi.testclient import TestClient

from app.main import create_app


class FakeRepository:
    def create(self, project_id: int, title: str):
        return {"id": 1, "project_id": project_id,
                "title": title, "status": "todo"}

    def get(self, task_id: int):
        if task_id != 1:
            return None
        return {"id": 1, "project_id": 1,
                "title": "tested", "status": "todo"}


client = TestClient(create_app(FakeRepository()))


def test_create_task():
    response = client.post("/tasks", json={"project_id": 1, "title": "tested"})
    assert response.status_code == 201
    assert response.json()["id"] == 1


def test_missing_task():
    response = client.get("/tasks/999")
    assert response.status_code == 404


def test_health():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json() == {"status": "ok"}

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt

export DB_HOST=127.0.0.1
export DB_PORT=3306
export DB_USER=app_user
export DB_PASSWORD='replace-me'
export DB_NAME=engineering_notes

pytest -q
uvicorn app.main:create_app --factory --host 127.0.0.1 --port 8000

curl -i -X POST http://127.0.0.1:8000/tasks \
  -H 'Content-Type: application/json' \
  -d '{"project_id":1,"title":"finish engineering stage"}'
curl -i http://127.0.0.1:8000/tasks/1
curl -i http://127.0.0.1:8000/health