FastAPI 实战入门：用 Python 最快的框架 30 分钟搭建 REST API

1
from pydantic import BaseModel, Field
2
from datetime import datetime
3

4
class BookCreate(BaseModel):
5
    """创建图书的请求体"""
6
    title: str = Field(..., min_length=1, max_length=200, examples=["Python 编程导论"])
7
    author: str = Field(..., min_length=1, max_length=100)
8
    price: float = Field(..., gt=0, le=9999.99, description="价格，单位：元")
9
    isbn: str = Field(..., pattern=r"^\d{13}$", description="13 位 ISBN")
10
    tags: list[str] = Field(default_factory=list, max_length=10)
11

12
class BookResponse(BookCreate):
13
    """返回的图书数据"""
14
    id: int
15
    created_at: datetime
16

17
    model_config = {"from_attributes": True}  # 支持 ORM 对象转换

1
from fastapi import FastAPI, HTTPException, Query
2
from pydantic import BaseModel, Field
3
from datetime import datetime
4

5
app = FastAPI(title="书店 API", version="1.0.0")
6

7
# ---------- 模型 ----------
8

9
class BookCreate(BaseModel):
10
    title: str = Field(..., min_length=1, max_length=200)
11
    author: str = Field(..., min_length=1, max_length=100)
12
    price: float = Field(..., gt=0, le=9999.99)
13
    tags: list[str] = Field(default_factory=list)
14

15
class BookUpdate(BaseModel):
16
    title: str | None = None
17
    author: str | None = None
18
    price: float | None = Field(None, gt=0, le=9999.99)
19
    tags: list[str] | None = None
20

21
class BookResponse(BookCreate):
22
    id: int
23
    created_at: datetime
24

25
# ---------- 模拟数据库 ----------
26

27
db: dict[int, dict] = {}
28
counter = 0
29

30
# ---------- CRUD 路由 ----------
31

32
@app.post("/books", response_model=BookResponse, status_code=201)
33
async def create_book(book: BookCreate):
34
    global counter
35
    counter += 1
36
    now = datetime.now()
37
    record = {"id": counter, **book.model_dump(), "created_at": now}
38
    db[counter] = record
39
    return record
40

41
@app.get("/books", response_model=list[BookResponse])
42
async def list_books(
43
    page: int = Query(1, ge=1),
44
    size: int = Query(10, ge=1, le=100),
45
    author: str | None = None,
46
):
47
    items = list(db.values())
48
    if author:
49
        items = [b for b in items if author.lower() in b["author"].lower()]
50
    start = (page - 1) * size
51
    return items[start : start + size]
52

53
@app.get("/books/{book_id}", response_model=BookResponse)
54
async def get_book(book_id: int):
55
    if book_id not in db:
56
        raise HTTPException(status_code=404, detail=f"图书 {book_id} 不存在")
57
    return db[book_id]
58

59
@app.put("/books/{book_id}", response_model=BookResponse)
60
async def update_book(book_id: int, book: BookUpdate):
61
    if book_id not in db:
62
        raise HTTPException(status_code=404, detail=f"图书 {book_id} 不存在")
63
    # 只更新传入的字段（None 的跳过）
64
    update_data = book.model_dump(exclude_unset=True)
65
    db[book_id].update(update_data)
66
    return db[book_id]
67

68
@app.delete("/books/{book_id}", status_code=204)
69
async def delete_book(book_id: int):
70
    if book_id not in db:
71
        raise HTTPException(status_code=404, detail=f"图书 {book_id} 不存在")
72
    del db[book_id]

1
from fastapi import Depends, Header, HTTPException
2

3
# 模拟认证：从 Header 中提取 token
4
async def get_current_user(authorization: str = Header(...)):
5
    if not authorization.startswith("Bearer "):
6
        raise HTTPException(status_code=401, detail="无效的认证格式")
7
    token = authorization.replace("Bearer ", "")
8
    # 实际项目这里解析 JWT
9
    if token != "secret-token-123":
10
        raise HTTPException(status_code=401, detail="认证失败")
11
    return {"user_id": 1, "username": "admin"}
12

13
# 权限检查：依赖可以嵌套
14
async def require_admin(user: dict = Depends(get_current_user)):
15
    if user["username"] != "admin":
16
        raise HTTPException(status_code=403, detail="需要管理员权限")
17
    return user
18

19
# 使用依赖
20
@app.delete("/books/{book_id}", status_code=204)
21
async def delete_book(book_id: int, user: dict = Depends(require_admin)):
22
    """只有管理员才能删除图书"""
23
    if book_id not in db:
24
        raise HTTPException(status_code=404, detail=f"图书 {book_id} 不存在")
25
    del db[book_id]

1
import time
2
from starlette.middleware.base import BaseHTTPMiddleware
3
from starlette.requests import Request
4

5
class TimingMiddleware(BaseHTTPMiddleware):
6
    async def dispatch(self, request: Request, call_next):
7
        start = time.perf_counter()
8
        response = await call_next(request)
9
        duration = time.perf_counter() - start
10
        response.headers["X-Process-Time"] = f"{duration:.4f}"
11
        if duration > 1.0:
12
            print(f"⚠️ 慢请求: {request.method} {request.url.path} 耗时 {duration:.2f}s")
13
        return response
14

15
app.add_middleware(TimingMiddleware)

1
from fastapi import Request
2
from fastapi.responses import JSONResponse
3
from pydantic import ValidationError
4

5
@app.exception_handler(Exception)
6
async def global_exception_handler(request: Request, exc: Exception):
7
    """兜底：所有未捕获的异常"""
8
    # 生产环境应该接入日志系统（Sentry、ELK 等）
9
    print(f"❌ 未处理异常: {type(exc).__name__}: {exc}")
10
    return JSONResponse(
11
        status_code=500,
12
        content={
13
            "detail": "服务器内部错误",
14
            "type": type(exc).__name__,
15
        },
16
    )
17

18
@app.exception_handler(404)
19
async def not_found_handler(request: Request, exc: HTTPException):
20
    return JSONResponse(
21
        status_code=404,
22
        content={"detail": "资源不存在", "path": str(request.url.path)},
23
    )

1
fastapi-demo/
2
├── app/
3
│   ├── __init__.py
4
│   ├── main.py          # 入口：创建 app、注册路由
5
│   ├── config.py         # 配置管理
6
│   ├── dependencies.py   # 公共依赖（认证、数据库连接等）
7
│   ├── routers/
8
│   │   ├── __init__.py
9
│   │   ├── books.py      # 图书相关路由
10
│   │   └── users.py      # 用户相关路由
11
│   ├── models/
12
│   │   ├── __init__.py
13
│   │   └── book.py       # Pydantic 模型
14
│   └── services/
15
│       ├── __init__.py
16
│       └── book_service.py  # 业务逻辑
17
├── tests/
18
│   ├── test_books.py
19
│   └── conftest.py
20
├── requirements.txt
21
└── Dockerfile