Python Web 开发 · 第3节 · FastAPI 实战（AI 应用首选）

FastAPI 实战（AI 应用首选）

FastAPI: Modern Async Web API

FastAPI 是这几年 Python Web 圈最大的黑马：写类型提示，它自动校验参数、自动生成 Swagger 文档、原生异步。做 AI 应用后端、大模型网关，基本绕不开它。这章是本页重点。

安装与第一个应用

pip install fastapi "uvicorn[standard]"   # 装 FastAPI + ASGI 服务器

# main.py
from fastapi import FastAPI

app = FastAPI(title="我的 API", version="1.0")

@app.get("/")
def root():
    return {"msg": "你好，FastAPI"}

@app.get("/hello/{name}")
def hello(name: str, age: int = 18):
    return {"name": name, "age": age}

启动（--reload 改代码自动重启）

uvicorn main:app --reload
#  INFO:     Uvicorn running on http://127.0.0.1:8000

论自动文档：FastAPI 最爽的特性

启动后浏览器打开 http://127.0.0.1:8000/docs——Swagger UI 自动出现，所有接口列出来，能直接点 "Try it out" 发请求测。/redoc 是 ReDoc 风格的另一种文档。你写代码，它写文档，零额外成本。这就是为什么团队协作和对外提供 API 都爱 FastAPI。

路径参数、查询参数与自动校验

from fastapi import Path, Query

@app.get("/items/{item_id}")
def read_item(
    item_id: int = Path(..., ge=1, le=1000, description="商品ID，1-1000"),
    q: str | None = Query(None, max_length=50),
):
    return {"item_id": item_id, "q": q}
# item_id 传字符串？自动 422 校验错误 + 详细说明

Pydantic 校验规则大全

Field 参数	校验什么
`Field(..., gt=0)`	大于 0（数字）
`Field(..., ge=1, le=100)`	1 到 100 之间
`min_length=1, max_length=50`	字符串长度
`pattern=r"^[\w-]+$"`	正则匹配
`EmailStr` / `HttpUrl`	邮箱/URL 格式
`decimal_places=2`	小数位数

OAuth2 密码流完整流程

论OAuth2 密码流怎么走

① 前端把用户名密码 POST 到 /login；② 后端查库验密码，对就签发 JWT；③ 前端存 token，之后每个请求在 Header 带 Authorization: Bearer <token>；④ 后端 get_current_user 依赖验签取出用户。整个过程密码只传一次，之后靠 token。密码必须走 HTTPS，否则等于明文。

请求体：Pydantic 模型

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(..., min_length=1, examples=["苹果"])
    price: float = Field(..., gt=0, description="价格必须大于0")
    tax: float | None = None

@app.post("/items/")
def create_item(item: Item):
    return {"received": item.name, "price_with_tax": item.price + (item.tax or 0)}

依赖注入 Depends：FastAPI 的灵魂

论什么是依赖注入

把"公共逻辑"（拿数据库连接、验 token、分页参数）写成一个函数，别的路由用 Depends(它) 声明要用。FastAPI 自动调这个函数、把返回值注入参数。可嵌套、可复用、可覆盖测试——这是 FastAPI 工程化的核心。

from fastapi import Depends

# 依赖 1：分页参数
def pagination(page: int = 1, size: int = 10):
    return {"page": page, "size": size}

# 依赖 2：当前用户（OAuth2）
def get_current_user(token: str = Depends(oauth2_scheme)):
    user = decode_token(token)       # 验签，失败抛 401
    return user

@app.get("/me/items")
def my_items(
    page: dict = Depends(pagination),
    user: dict = Depends(get_current_user),
):
    return {"user": user["name"], "page": page}

高级依赖：yield 依赖与类依赖

# yield 依赖：在请求开始时建资源，结束时清理（比如 DB 会话）
from fastapi import Depends

async def get_db():
    db = DatabaseSession()
    try:
        yield db          # 把 session 注入路由
    finally:
        db.close()       # 请求结束自动清理

# 类依赖：把依赖写成类，自动实例化
class Pagination:
    def __init__(self, page: int = 1, size: int = 10):
        self.page = page
        self.size = size

@app.get("/items")
def list_items(p: Pagination = Depends()):
    return {"page": p.page, "size": p.size}

RESTful API 设计最佳实践

原则	怎么做
资源用名词	`/users`、`/users/1/orders`，别用 `/getUser`。
动作用 HTTP 方法	GET 查、POST 建、PUT 改、DELETE 删。
版本化	`/v1/users`，不兼容升级开 v2。
分页/筛选/排序	query 参数 `?page=1&size=10&sort=-created`。
统一错误格式	所有错误返回 `{"error": {"code": "...", "message": "..."}}`。
用状态码表达	201 新建、404 找不到、422 校验错，别全返回 200。

OAuth2 + JWT 认证

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
import jwt
from datetime import datetime, timedelta

SECRET = "change-me-in-prod"
ALGO = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login")

def create_token(user_id: str):
    payload = {"sub": user_id,
               "exp": datetime.utcnow() + timedelta(hours=2)}
    return jwt.encode(payload, SECRET, algorithm=ALGO)

@app.post("/login")
def login(form: OAuth2PasswordRequestForm = Depends()):
    # 这里查数据库验密码（略），验过就发 token
    return {"access_token": create_token(form.username),
            "token_type": "bearer"}

异步：async def + async SQLAlchemy

# 异步路由：要调异步库才能真正并行
import httpx

@app.get("/fetch/{url}")
async def fetch_url(url: str):
    async with httpx.AsyncClient() as client:
        r = await client.get(url)
        return {"status": r.status_code, "len": len(r.text)}

# 异步 SQLAlchemy 2.0：async_session + select()
# async with session.execute(select(User).where(...)) as result: ...

生产部署清单（三个框架通用）

项	要做什么
DEBUG=False	关调试模式，防信息泄露。
SECRET_KEY	从环境变量读，别硬编码。
数据库迁移	部署前 `migrate` 跑一遍。
静态文件	collectstatic，交给 Nginx。
HTTPS	Let's Encrypt 证书，Nginx 终止 TLS。
进程管理	systemd/supervisor 管 Gunicorn/Uvicorn 进程。
日志	配置 logging，别只 print。
备份	数据库定时备份。

异步 SQLAlchemy 2.0 完整用法

# engine / session / 模型三件套
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy import select, update, delete

engine = create_async_engine("sqlite+aiosqlite:///./app.db")
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

# 查询：await session.execute(select(Model).where(...))
async def get_user(uid: int):
    async with SessionLocal() as session:
        result = await session.execute(select(User).where(User.id == uid))
        return result.scalar_one_or_none()

# 增：await session.add(obj); await session.commit()
# 改：await session.execute(update(User).where(...).values(...))
# 删：await session.execute(delete(User).where(...))

数据库连接池：别每个请求新开一条连接

开一条数据库连接要做 TCP 握手、认证、建会话，开销不小。如果每个 HTTP 请求都临时"连一下、用完就断"，高并发时数据库会被连接风暴压垮。连接池就是预先养着一批连接，复用它们。

论大白话：连接像"共享单车"

与其每次出门买辆新自行车（新建连接），不如门口停一排、骑完还回去（池化复用）。create_async_engine 默认就带一个异步连接池，你只要在生产把几个关键参数调对。

生产级 engine 连接池配置（MySQL/PostgreSQL）

# sqlite 是文件库不用池；下面以 PostgreSQL 异步驱动 asyncpg 为例
engine = create_async_engine(
    "postgresql+asyncpg://user:pwd@db:5432/app",
    pool_size=10,             # 常驻连接数，一般 5~20
    max_overflow=20,          # 突发时额外借的，峰值 10+20=30
    pool_recycle=1800,        # 30 分钟回收，防 MySQL 默认 8h 断开报错
    pool_pre_ping=True,        # 借连接前先 ping 一下，死掉的自动换
    echo=False,              # 生产别开 True，会把 SQL 全打印出来拖慢
)

防坑：pool_recycle 必须小于数据库的 wait_timeout

MySQL 默认 wait_timeout=8小时，一条连接闲了 8 小时服务端会悄悄断开，而池子里还以为它活着——下一个请求拿来用就报 "MySQL server has gone away"。解法就是上面的 pool_recycle=1800（半小时主动回收），再配 pool_pre_ping=True 双保险。

练习：连接池小测（点开对答案）

问：压测时发现并发一上来接口就报" too many connections"，怎么调？
答：先看 pool_size + max_overflow 之和是不是超过了数据库的 max_connections。多实例部署时，总连接数 = 实例数 × (pool_size + max_overflow)，要乘上实例数估算，别把 DB 的连接配额撑爆。

CORS 中间件：前端跨域

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],   # 前端地址，生产别写 *
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

响应模型与响应状态码

# response_model：限定返回字段，敏感字段自动过滤
class UserOut(BaseModel):
    id: int
    name: str
    # 注意：password_hash 不写在这里，就不会返回给前端

@app.post("/users", response_model=UserOut, status_code=201)
def create_user(u: UserIn):
    # 即使函数返回了带 password_hash 的 ORM 对象，response_model 也会只挑 UserOut 的字段
    return u

# tags/summary/description：让 Swagger 文档更友好
@app.get("/users/{uid}",
     tags=["users"],
     summary="按 ID 查用户",
     description="返回单个用户详情，找不到 404",
     responses={404: {"description": "用户不存在"}})
def get_user(uid: int): ...

后台任务 BackgroundTasks

from fastapi import BackgroundTasks

def send_welcome_email(email: str):
    # 发邮件慢，放后台跑，不阻塞响应
    print(f"给 {email} 发欢迎邮件（异步）")

@app.post("/signup")
def signup(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_welcome_email, email)
    return {"ok": True}     # 立即返回，邮件在后台慢慢发

WebSocket 实时通信

@app.websocket("/ws/chat")
async def websocket_endpoint(websocket, username: str):
    await websocket.accept()
    await websocket.send_text(f"{username} 已连接")
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"{username}: {data}")

自定义中间件

# @app.middleware("http")：在请求前后加逻辑
import time

@app.middleware("http")
async def log_request_time(request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    cost = time.perf_counter() - start
    response.headers["X-Process-Time"] = str(cost)
    print(f"{request.method} {request.url.path} {cost:.3f}s")
    return response

测试与部署

# 用 TestClient + pytest 写测试
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_hello():
    r = client.get("/hello/小明?age=20")
    assert r.status_code == 200
    assert r.json() == {"name": "小明", "age": 20}

def test_create_task():
    r = client.post("/tasks", json={"title": "测试任务"})
    assert r.status_code == 201
    assert r.json()["title"] == "测试任务"

def test_validation_error():
    r = client.post("/items/", json={"name": "", "price": -5})
    assert r.status_code == 422      # Pydantic 自动校验失败

# 跑测试：pytest -v

生产部署：Gunicorn 管 Uvicorn worker

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000
# 前面套 Nginx，HTTPS 证书用 certbot 申请
# worker 数建议 = 2 * CPU 核数 + 1

请求体进阶：嵌套模型、文件上传、表单

from pydantic import BaseModel
from fastapi import File, UploadFile, Form

# 嵌套模型：订单里有用户，用户又是个模型
class User(BaseModel):
    name: str
    email: str

class Order(BaseModel):
    item: str
    price: float
    buyer: User               # 嵌套

@app.post("/orders")
def create_order(order: Order):
    return {"buyer_email": order.buyer.email}

# 文件上传
@app.post("/upload")
def upload_file(file: UploadFile = File(...)):
    content = file.file.read()
    return {"filename": file.filename, "size": len(content)}

# 表单字段（application/x-www-form-urlencoded）
@app.post("/login-form")
def login_form(username: str = Form(...), password: str = Form(...)):
    return {"username": username}

静态文件与 JSON 兼容

from fastapi.staticfiles import StaticFiles

# 把 static/ 目录挂到 /static 路径
app.mount("/static", StaticFiles(directory="static"), name="static")

# JSON 兼容编码器：遇到 datetime/UUID 等自动转字符串
from fastapi.encoders import jsonable_encoder
data = jsonable_encoder({"now": datetime.utcnow()})

完整案例：AI 聊天 API（OpenAI 兼容 + SSE 流式）

这是 FastAPI 最典型的杀手场景：写一个 OpenAI 兼容的聊天接口，支持流式输出（SSE）、多轮对话、JWT 认证。

# ai_chat.py —— 依赖 pip install fastapi uvicorn openai python-multipart
from fastapi import FastAPI, Depends
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from openai import AsyncOpenAI

app = FastAPI(title="AI Chat")
client = AsyncOpenAI()          # 读 OPENAI_API_KEY 环境变量

class ChatRequest(BaseModel):
    message: str
    history: list[dict] = []     # 多轮对话历史

@app.post("/v1/chat")
async def chat(req: ChatRequest):
    messages = req.history + [{"role": "user", "content": req.message}]

    async def stream():
        # 真实用法：async with client.chat.completions.create(...) as s:
        #     async for chunk in s: yield f"data: {chunk}\n\n"
        yield "data: " + "[流式返回模型回答]" + "\n\n"

    # SSE：Content-Type 必须是 text/event-stream
    return StreamingResponse(stream(), media_type="text/event-stream")

用 curl 测试

curl -N -X POST http://localhost:8000/v1/chat \
  -H "Content-Type: application/json" \
  -d '{"message":"用一句话解释什么是GIL"}'
# data: [流式返回模型回答]

带认证和聊天记录存储的完整版（核心思路）

# 1. 用户先 POST /login 拿 JWT
# 2. 之后每个请求带 Authorization: Bearer <token>
# 3. 后端 get_current_user 依赖验签，取出 user_id
# 4. 把对话存 SQLite/Redis，下次带上 history 就能多轮

class ChatMessage(Base):
    __tablename__ = "chat_messages"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int]
    role: Mapped[str]         # "user" / "assistant"
    content: Mapped[str]
    created: Mapped[datetime] = mapped_column(default=datetime.utcnow)

# 查询历史：SELECT * FROM chat_messages WHERE user_id=? ORDER BY created
# 拼成 OpenAI 格式的 messages 数组传给模型

SSE vs WebSocket

SSE（Server-Sent Events）是服务器→客户端单向推送，用普通 HTTP，自动重连，适合大模型流式输出；WebSocket 是双向，适合聊天、协同编辑。AI 流式用 SSE 就够，别上来就 WebSocket。FastAPI 返回 SSE 用 StreamingResponse(media_type="text/event-stream")。

完整案例：任务管理 RESTful API（SQLAlchemy 2.0 异步）

# requirements: fastapi uvicorn sqlalchemy[asyncio] aiosqlite
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

DATABASE_URL = "sqlite+aiosqlite:///./tasks.db"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase): pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    done: Mapped[bool] = mapped_column(default=False)

class TaskIn(BaseModel):
    title: str

class TaskOut(BaseModel):
    id: int
    title: str
    done: bool
    class Config:
        from_attributes = True

app = FastAPI()

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.post("/tasks", response_model=TaskOut)
async def create_task(t: TaskIn, db: AsyncSession = Depends(get_db)):
    task = Task(title=t.title)
    db.add(task); await db.commit(); await db.refresh(task)
    return task

@app.get("/tasks", response_model=list[TaskOut])
async def list_tasks(done: bool | None = None,
                 db: AsyncSession = Depends(get_db)):
    stmt = select(Task)
    if done is not None:
        stmt = stmt.where(Task.done == done)
    result = await db.execute(stmt)
    return result.scalars().all()

@app.get("/tasks/{task_id}", response_model=TaskOut)
async def get_task(task_id: int, db: AsyncSession = Depends(get_db)):
    task = await db.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="任务不存在")
    return task

@app.put("/tasks/{task_id}", response_model=TaskOut)
async def update_task(task_id: int, t: TaskIn,
                   db: AsyncSession = Depends(get_db)):
    task = await db.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404)
    task.title = t.title
    await db.commit()
    return task

@app.delete("/tasks/{task_id}", status_code=204)
async def delete_task(task_id: int, db: AsyncSession = Depends(get_db)):
    task = await db.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404)
    await db.delete(task)
    await db.commit()

用 curl 测一轮 CRUD

curl -X POST http://localhost:8000/tasks -H "Content-Type: application/json" -d '{"title":"买菜"}'
# {"id":1,"title":"买菜","done":false}

curl http://localhost:8000/tasks
# [{"id":1,"title":"买菜","done":false}]

curl http://localhost:8000:8000/docs   # Swagger UI 里点着测更爽

async def 里别调阻塞库

FastAPI 写了 async def，但里面塞 requests.get()、同步 SQLAlchemy、time.sleep()——整个事件循环被卡住，异步性能全废。要么用异步库（aiohttp/async SQLAlchemy/httpx.AsyncClient），要么干脆用普通 def 让 FastAPI 自动丢线程池跑。判断标准：这个函数会"等"吗？会等就必须异步版本。

本章面试题 · FastAPI

1.（概念题）FastAPI 的依赖注入怎么工作？Depends 的函数什么时候执行？

查看答案

答案：路由调用前，FastAPI 解析签名里所有 Depends(x)，先执行 x()（支持嵌套），把返回值注入路由参数。可缓存（Depends(x, use_cache=True) 同一请求内只跑一次）。这是 FastAPI 工程化的核心。

2.（概念题）FastAPI 为什么比 Flask 快？什么是 Pydantic 2.x？

查看答案

答案：① 原生 ASGI 异步，单事件循环高并发；② Pydantic 2.x 用 Rust 写校验内核，比 1.x 快 5-50 倍；③ 类型提示在启动期一次性生成校验代码，运行时零反射开销。但"快"主要体现在 IO 密集，CPU 密集仍受 GIL 限制。

3.（概念题）JWT 和 Session 有什么区别？为什么 API 都用 JWT？

查看答案

答案：Session 把状态存服务端（Redis/DB），靠 cookie 里的 sessionid 关联；JWT 把用户信息签名后塞进 token，服务端不存状态，靠验签自证。API/微服务/跨端更爱 JWT：无状态、易水平扩展。代价是 JWT 发出去后难"注销"，要靠短过期 + 黑名单。

4.（代码题）下面路由定义有什么问题？async def get_user(uid: str)

查看答案

答案：uid 应该是 int——声明成 str，FastAPI 不会自动转 int，查库时还得手动转；写成 uid: int 还能自动 422 校验。类型提示既是文档也是校验器，别浪费。

← 上一节：Flask 3.x 实战 Python Web 开发 · 课程总览下一节：Django 5.x 实战 →