FastAPI 完整实践：有上限的 SQLite 费用账本 · 完整源码和配置

返回分篇目录

前置：契约、依赖生命周期。完整工程位于 examples/python/p2-fastapi。示例未执行、测试未执行、依赖未安装；本页只提供完整源码、预期规格和静态检查范围。没有启动ASGI服务、访问API或调用TestClient，也没有把设计OpenAPI冒充运行结果。

完整源码和配置

FastAPI完整示例怎样复用同一份源码？本页通过include直接引用工程文件，读者下载的实现与正文展示来自同源文件；app.py、依赖、配置、输入规格、测试和README全部保留。应用工厂避免导入时产生文件写入，实际初始化发生在lifespan。

app.py（应用、模型、依赖与事务）

"""Local teaching API. Example and tests have not been executed."""
from contextlib import asynccontextmanager
from pathlib import Path
import sqlite3
from typing import Annotated, Literal

from fastapi import Depends, FastAPI, HTTPException, Path as PathParameter, Response
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ConfigDict, Field


class ExpenseInput(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    reference: Annotated[str, Field(pattern=r"^[A-Za-z0-9_-]{1,40}$")]
    category: Literal["food", "travel", "other"]
    cents: Annotated[int, Field(ge=1, le=100_000_000)]


class Expense(ExpenseInput):
    id: Annotated[int, Field(ge=1, le=2_147_483_647)]


class Error(BaseModel):
    model_config = ConfigDict(extra="forbid")
    detail: str


ERRORS = {status: {"model": Error} for status in (404, 409, 422, 503)}
ExpenseId = Annotated[int, PathParameter(ge=1, le=2_147_483_647)]


def create_app(database: str | Path = "ledger.sqlite3") -> FastAPI:
    database = str(database)
    if database == ":memory:":
        raise ValueError("Use a file: request connections must share persistent storage")

    def connect():
        # FastAPI may resume synchronous dependency/route work on different threads.
        # One connection per request; no parallel tasks or background use permitted.
        db = sqlite3.connect(database, timeout=0.2, isolation_level=None,
                             check_same_thread=False)
        db.row_factory = sqlite3.Row
        return db

    @asynccontextmanager
    async def lifespan(_app):
        # Small local schema setup blocks only startup, before requests are accepted.
        db = connect()
        try:
            db.execute("""CREATE TABLE IF NOT EXISTS expenses (
                id INTEGER PRIMARY KEY AUTOINCREMENT CHECK(id <= 2147483647),
                reference TEXT NOT NULL UNIQUE,
                category TEXT NOT NULL CHECK(category IN ('food','travel','other')),
                cents INTEGER NOT NULL CHECK(cents BETWEEN 1 AND 100000000)
            )""")
        finally:
            db.close()
        yield

    app = FastAPI(title="Bounded Expense Ledger", version="1.0.0", lifespan=lifespan)

    def get_db():
        db = connect()
        try:
            yield db
        finally:
            db.close()

    Database = Annotated[sqlite3.Connection, Depends(get_db, scope="function")]

    @app.exception_handler(RequestValidationError)
    async def invalid_request(_request, _exc):
        return JSONResponse(status_code=422, content={"detail": "invalid_request"})

    @app.exception_handler(sqlite3.OperationalError)
    async def storage_error(_request, exc):
        # Extended result codes retain the primary code in the low byte.
        if getattr(exc, "sqlite_errorcode", 0) & 0xFF in (sqlite3.SQLITE_BUSY, sqlite3.SQLITE_LOCKED):
            return JSONResponse(status_code=503, content={"detail": "storage_busy"})
        raise exc

    @app.post("/expenses", response_model=Expense, status_code=201,
              responses={k: ERRORS[k] for k in (409, 422, 503)})
    def create_expense(body: ExpenseInput, db: Database):
        db.execute("BEGIN IMMEDIATE")
        try:
            if db.execute("SELECT 1 FROM expenses WHERE reference = ?", (body.reference,)).fetchone():
                raise HTTPException(409, "duplicate_reference")
            if db.execute("SELECT COUNT(*) FROM expenses").fetchone()[0] >= 100:
                raise HTTPException(409, "ledger_full")
            cursor = db.execute("INSERT INTO expenses(reference, category, cents) VALUES (?, ?, ?)",
                                (body.reference, body.category, body.cents))
            # Validate public output before committing the mutation.
            result = Expense(id=cursor.lastrowid, **body.model_dump())
            db.execute("COMMIT")
            return result
        except Exception:
            if db.in_transaction:
                db.execute("ROLLBACK")
            raise

    @app.get("/expenses", response_model=list[Expense], responses={503: ERRORS[503]})
    def list_expenses(db: Database):
        return [dict(row) for row in db.execute("SELECT * FROM expenses ORDER BY id")]

    @app.get("/expenses/{expense_id}", response_model=Expense,
             responses={k: ERRORS[k] for k in (404, 422, 503)})
    def get_expense(expense_id: ExpenseId, db: Database):
        row = db.execute("SELECT * FROM expenses WHERE id = ?", (expense_id,)).fetchone()
        if row is None:
            raise HTTPException(404, "not_found")
        return dict(row)

    @app.delete("/expenses/{expense_id}", status_code=204, response_class=Response,
                responses={k: ERRORS[k] for k in (404, 422, 503)})
    def delete_expense(expense_id: ExpenseId, db: Database):
        db.execute("BEGIN IMMEDIATE")
        try:
            if db.execute("DELETE FROM expenses WHERE id = ?", (expense_id,)).rowcount == 0:
                raise HTTPException(404, "not_found")
            db.execute("COMMIT")
        except Exception:
            if db.in_transaction:
                db.execute("ROLLBACK")
            raise
        return Response(status_code=204)

    return app

requirements.txt（固定直接依赖）

fastapi==0.121.0
pydantic==2.12.3
uvicorn==0.38.0
httpx==0.28.1

pyproject.toml（环境与教学验证状态）

[project]
name = "p2-fastapi-ledger"
version = "1.0.0"
requires-python = ">=3.12,<3.13"
description = "Local bounded ledger teaching source; runtime not verified"
dynamic = ["dependencies"]

[tool.teaching]
contract = "openapi.json"
runtime-executed = false
tests-executed = false
dependency-file = "requirements.txt"

cases.json（正反输入规格）

{
  "status": "specification only; not executed",
  "valid": {"reference": "lunch_001", "category": "food", "cents": 1250},
  "invalid": [
    {"name": "missing_cents", "body": {"reference": "missing", "category": "food"}, "status": 422},
    {"name": "integral_float", "body": {"reference": "integral", "category": "food", "cents": 1.0}, "status": 422},
    {"name": "empty_reference", "body": {"reference": "", "category": "food", "cents": 1}, "status": 422},
    {"name": "long_reference", "body": {"reference": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", "category": "food", "cents": 1}, "status": 422},
    {"name": "zero", "body": {"reference": "zero", "category": "food", "cents": 0}, "status": 422},
    {"name": "numeric_string", "body": {"reference": "string", "category": "food", "cents": "1250"}, "status": 422},
    {"name": "boolean", "body": {"reference": "bool", "category": "food", "cents": true}, "status": 422},
    {"name": "fraction", "body": {"reference": "float", "category": "food", "cents": 12.5}, "status": 422},
    {"name": "extra", "body": {"reference": "extra", "category": "food", "cents": 1, "id": 1}, "status": 422},
    {"name": "category", "body": {"reference": "category", "category": "unknown", "cents": 1}, "status": 422},
    {"name": "reference", "body": {"reference": "has space", "category": "food", "cents": 1}, "status": 422},
    {"name": "upper_bound", "body": {"reference": "large", "category": "food", "cents": 100000001}, "status": 422}
  ]
}

test_app.py（完整测试规格，未执行）

"""Executable test specifications. Not executed in this documentation task."""
from concurrent.futures import ThreadPoolExecutor
from contextlib import ExitStack
from pathlib import Path
import json
import sqlite3
import tempfile
import unittest

from fastapi.testclient import TestClient
from app import create_app


class LedgerTests(unittest.TestCase):
    def setUp(self):
        self.directory = tempfile.TemporaryDirectory()
        self.database = Path(self.directory.name) / "ledger.sqlite3"
        self.client = TestClient(create_app(self.database))
        self.client.__enter__()
        self.cases = json.loads(Path(__file__).with_name("cases.json").read_text(encoding="utf-8"))

    def tearDown(self):
        self.client.__exit__(None, None, None)
        self.directory.cleanup()

    def post(self, reference="lunch_001"):
        return self.client.post("/expenses", json={**self.cases["valid"], "reference": reference})

    def test_lifecycle_and_persistence(self):
        created = self.post()
        self.assertEqual(created.status_code, 201)
        row = created.json()
        self.assertEqual(set(row), {"id", "reference", "category", "cents"})
        self.assertEqual(row["cents"], 1250)
        with TestClient(create_app(self.database)) as second:
            self.assertEqual(second.get(f'/expenses/{row["id"]}').json(), row)
        self.assertEqual(self.client.get("/expenses").json(), [row])
        deleted = self.client.delete(f'/expenses/{row["id"]}')
        self.assertEqual((deleted.status_code, deleted.content), (204, b""))
        self.assertEqual(self.client.get(f'/expenses/{row["id"]}').json(), {"detail": "not_found"})
        self.assertEqual(self.client.get("/expenses").json(), [])

    def test_invalid_input_never_writes(self):
        for case in self.cases["invalid"]:
            with self.subTest(case=case["name"]):
                result = self.client.post("/expenses", json=case["body"])
                self.assertEqual(result.status_code, case["status"])
                self.assertEqual(result.json(), {"detail": "invalid_request"})
        malformed = self.client.post("/expenses", content=b"{",
                                     headers={"Content-Type": "application/json"})
        self.assertEqual((malformed.status_code, malformed.json()),
                         (422, {"detail": "invalid_request"}))
        for path in ("0", "-1", "abc", "2147483648"):
            self.assertEqual(self.client.get("/expenses/" + path).status_code, 422)
        self.assertEqual(self.client.get("/expenses").json(), [])

    def test_duplicate_rollback_and_missing(self):
        self.assertEqual(self.post().status_code, 201)
        duplicate = self.post()
        self.assertEqual((duplicate.status_code, duplicate.json()), (409, {"detail": "duplicate_reference"}))
        self.assertEqual(len(self.client.get("/expenses").json()), 1)
        self.assertEqual(self.post("next").status_code, 201)
        missing = self.client.delete("/expenses/9999")
        self.assertEqual((missing.status_code, missing.json()), (404, {"detail": "not_found"}))

    def test_capacity_and_delete_releases_slot(self):
        for index in range(100):
            self.assertEqual(self.post(f"item_{index}").status_code, 201)
        overflow = self.post("overflow")
        self.assertEqual((overflow.status_code, overflow.json()), (409, {"detail": "ledger_full"}))
        first = self.client.get("/expenses").json()[0]["id"]
        self.assertEqual(self.client.delete(f"/expenses/{first}").status_code, 204)
        self.assertEqual(self.post("replacement").status_code, 201)

    def test_busy_is_503_then_recovers(self):
        db = sqlite3.connect(self.database, isolation_level=None)
        try:
            db.execute("BEGIN IMMEDIATE")
            result = self.post()
            self.assertEqual((result.status_code, result.json()), (503, {"detail": "storage_busy"}))
        finally:
            db.execute("ROLLBACK")
            db.close()
        self.assertEqual(self.post().status_code, 201)

    def test_concurrent_same_reference(self):
        # Two independent ASGI clients and request connections; no shared sqlite connection.
        with ExitStack() as stack:
            clients = [stack.enter_context(TestClient(create_app(self.database))) for _ in range(2)]
            def send(client):
                return client.post("/expenses", json=self.cases["valid"]).status_code
            with ThreadPoolExecutor(max_workers=2) as pool:
                statuses = list(pool.map(send, clients))
        self.assertEqual(statuses.count(201), 1)
        self.assertTrue(all(code in (201, 409, 503) for code in statuses))
        self.assertEqual(len(self.client.get("/expenses").json()), 1)


if __name__ == "__main__":
    unittest.main()

README.md（环境、命令与限制）

# 本地费用账本 API

示例未执行，测试未执行，依赖未安装；以下命令仅为读者自行验证的步骤。静态 AST/JSON/TOML 检查不能证明安装、ASGI 生命周期、线程、锁等待、HTTP 或持久化行为成立。

目标环境 Python 3.12，FastAPI 0.121.0、Pydantic 2.12.3、Uvicorn 0.38.0、HTTPX 0.28.1。requirements.txt 固定直接依赖，未提供已解析的传递依赖锁；不是生产版本推荐。FastAPI 0.121.0 声明 Starlette >=0.40,<0.50、Pydantic <3，选定版本在其范围内，实际组合未运行。

先阅读手写 openapi.json：3.1 设计契约先于实现创建，不是 app.openapi() 导出文件，也未比较运行时生成结果。POST /expenses 创建，GET /expenses 列表，GET /expenses/{expense_id} 读取，DELETE 同路径删除并返回空 204。费用是整数分 1–100000000；category 为 food/travel/other；reference 是 1–40 位 ASCII 字母数字下划线短横线。多余字段、字符串金额、布尔金额、浮点金额均拒绝。请求体上限应由部署层另加，本例字段约束不等于请求字节限流。

数据库最多 100 条记录，删除释放名额。参考号唯一；成功返回 201 和四字段 Expense。错误固定 {"detail":"代码"}：422 invalid_request、404 not_found、409 duplicate_reference/ledger_full、503 storage_busy。未预料到的错误仍是服务器错误，不转换成成功或空账本。

在此目录，PowerShell 可执行：

```powershell
python -m venv .venv
.\.venv\Scripts\python.exe -m pip install -r requirements.txt
.\.venv\Scripts\python.exe -m unittest -v
.\.venv\Scripts\python.exe -m uvicorn app:create_app --factory --host 127.0.0.1 --port 8000
```

Bash 使用 python3 创建环境，把其余解释器路径换成 .venv/bin/python。启动最后一条会在工作目录创建 ledger.sqlite3；父目录必须已存在且可写。只绑定回环地址；没有身份认证、授权、审计、CSRF设计或多租户隔离，不应对公网发布。停止用 Ctrl+C。读取 http://127.0.0.1:8000/docs 是读者可选步骤，本次没有启动或访问。

测试规格为 6 个 unittest 方法：创建读取删除/重启持久化、12 个非法 body、无效JSON与4个非法路径、重复回滚、100 条上限、写锁 503 后恢复、两请求相同参考号竞争。期望 unittest 成功且没有残留临时数据库；这不是实际测试结果。竞争的第二个请求可为409或503，不能承诺固定调度顺序。

每次请求由 get_db 建连接，Depends(scope="function") 的 finally 关闭；check_same_thread=False 仅解除线程身份检查，不提供并发保护。本设计每个连接仅供一个同步路由串行使用，不传入后台任务；跨请求由 SQLite BEGIN IMMEDIATE 串行化写入。事务在路由中先验证输出模型再 COMMIT，finally 只关闭不提交。测试使用真实临时文件，:memory: 每连接隔离故被拒绝；不提供外部数据库或服务绑定。

静态契约只描述成功、已知业务/校验/忙锁错误；框架404/405、未捕获500等不在其枚举错误保证内。要投入应用，应先执行测试、比较实际 OpenAPI、确认部署资源限制/数据库迁移/权限与日志脱敏，而不能把文档静态通过当作上线依据。