Python Web 框架 FastAPI 快速上手

前言

Flask 时代写 API 要自己造的轮子——参数校验、文档、类型提示——FastAPI 一次全给：写函数签名，送你文档和校验。这篇从零到部署走一遍。

一、五分钟起步

pip install "fastapi[standard]" -i https://mirrors.aliyun.com/pypi/simple/

# main.py
from fastapi import FastAPI

app = FastAPI(title="运维平台 API", version="1.0.0")

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

@app.get("/hello/{name}")
def hello(name: str):
    return {"message": f"hello, {name}"}

fastapi dev main.py              # 开发模式, 改代码热重载
# 打开 http://localhost:8000/docs        ← 自动交互式文档（Swagger UI）
# 打开 http://localhost:8000/redoc        ← 另一套文档样式

类型注解就是功能：name: str 声明后，框架自动解析路径参数、校验类型、生成文档。

二、Pydantic：请求与响应模型

from pydantic import BaseModel, Field, EmailStr
from typing import Optional

class HostCreate(BaseModel):
    name: str = Field(..., min_length=2, max_length=30, examples=["web01"])
    ip: str = Field(..., pattern=r"^(\d{1,3}\.){3}\d{1,3}$")
    port: int = Field(22, ge=1, le=65535)
    tags: list[str] = []
    owner: Optional[EmailStr] = None

class HostOut(HostCreate):
    id: int
    status: str

@app.post("/api/hosts", response_model=HostOut, status_code=201)
def create_host(host: HostCreate):
    # 走到这里, 数据一定合法（不合法根本进不来）
    saved = save_to_db(host)          # 假装入库
    return saved

非法请求的自动处理：

curl -X POST http://localhost:8000/api/hosts -H "Content-Type: application/json" \
  -d '{"name":"w","ip":"999.1.1.1","port":70000}'
# 422 Unprocessable Entity
# {"detail":[
#   {"loc":["body","name"],"msg":"String should have at least 2 characters"},
#   {"loc":["body","ip"],"msg":"String should match pattern..."},
#   {"loc":["body","port"],"msg":"Input should be at most 65535"}]}

response_model 的妙处：输出自动过滤（比如过滤密码字段）、自动转成声明的形状——文档永远和代码一致。

三、查询参数、过滤与分页

from fastapi import Query

@app.get("/api/hosts")
def list_hosts(
    page: int = Query(1, ge=1),
    size: int = Query(20, ge=1, le=100),
    tag: Optional[str] = None,
):
    return {"page": page, "size": size, "tag": tag}
# /api/hosts?page=2&size=50&tag=prod

四、依赖注入：中间逻辑的优雅归宿

鉴权、取 DB 会话、分页参数——不用每个函数复制粘贴：

from fastapi import Depends, Header, HTTPException

API_TOKEN = "secret-token"

def verify_token(authorization: str = Header(...)):
    if authorization != f"Bearer {API_TOKEN}":
        raise HTTPException(status_code=401, detail="invalid token")

def get_db():
    db = SessionLocal()
    try:
        yield db            # 用完自动走 finally 关闭
    finally:
        db.close()

@app.get("/api/hosts", dependencies=[Depends(verify_token)])
def list_hosts(db=Depends(get_db)):
    return db.query(Host).all()

依赖可以嵌套依赖，测试时用 app.dependency_overrides 替换假实现——解耦利器。

五、异步端点

import httpx

@app.get("/api/check")
async def check(url: str):
    async with httpx.AsyncClient() as client:
        resp = await client.get(url, timeout=5)
    return {"url": url, "status": resp.status_code}

# 注意: async 端点里绝不能有阻塞调用（time.sleep/同步IO）
# 要么用 await 异步库, 要么用普通 def（FastAPI 自动丢线程池）

六、后台任务与定时活

from fastapi import BackgroundTasks

def write_audit(action: str):
    with open("audit.log", "a") as f:
        f.write(f"{action} at {time.time()}\n")

@app.post("/api/hosts/{hid}/restart")
def restart_host(hid: int, bg: BackgroundTasks):
    bg.add_task(write_audit, f"restart host {hid}")   # 响应后异步执行
    return {"result": "restarting"}

# 更重的任务用 Celery/ARQ 队列, BackgroundTasks 适合轻量收尾

七、错误处理与中间件

from fastapi import Request
from fastapi.responses import JSONResponse

class BizError(Exception):
    def __init__(self, code: str, msg: str):
        self.code, self.msg = code, msg

@app.exception_handler(BizError)
async def biz_handler(request: Request, exc: BizError):
    return JSONResponse(status_code=400,
        content={"code": exc.code, "message": exc.msg})

@app.middleware("http")
async def add_timing(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    response.headers["X-Process-Time"] = f"{time.time()-start:.3f}"
    return response

八、项目结构与部署

app/
├── main.py            # FastAPI 实例与路由注册
├── deps.py            # 公共依赖
├── models.py          # Pydantic 模型
└── routers/
    ├── hosts.py
    └── users.py

# routers/hosts.py —— APIRouter 拆分路由
from fastapi import APIRouter
router = APIRouter(prefix="/api/hosts", tags=["hosts"])

@router.get("/{hid}")
def get_host(hid: int): ...

生产部署：

# 多 worker 跑生产
gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

# 或纯 uvicorn
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

前置 Nginx（反向代理篇的配置直接复用）+ systemd 拉起：

[Service]
ExecStart=/opt/app/.venv/bin/gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 127.0.0.1:8000

九、测试

from fastapi.testclient import TestClient

client = TestClient(app)

def test_create_host():
    resp = client.post("/api/hosts", json={"name": "web01", "ip": "10.0.0.1"})
    assert resp.status_code == 201
    assert resp.json()["name"] == "web01"

def test_invalid_ip():
    resp = client.post("/api/hosts", json={"name": "web01", "ip": "bad"})
    assert resp.status_code == 422

同步 TestClient + pytest，无起服务的开销，CI 里跑飞快。

十、Flask 还是 FastAPI？

维度	FastAPI	Flask
类型/校验	原生 Pydantic ✅	手写或插件
异步	原生 ✅	需要 async 视图改造
文档	自动 ✅	flask-restx 等
生态/历史	较新	极其成熟
适合	新 API 项目	传统 Web 页面/老项目

我的结论：2026 年起新 API 项目默认 FastAPI，除非团队有厚重 Flask 资产。

小结

能力	写法
校验	Pydantic 模型 + Field 约束
文档	自动 /docs，response_model 保持一致
复用逻辑	Depends 依赖注入
异步	async def + await
收尾任务	BackgroundTasks
部署	gunicorn -k UvicornWorker

本文是「Python」系列第 9 篇。

前言#

一、五分钟起步#

二、Pydantic：请求与响应模型#

三、查询参数、过滤与分页#

四、依赖注入：中间逻辑的优雅归宿#

五、异步端点#

六、后台任务与定时活#

七、错误处理与中间件#

八、项目结构与部署#

九、测试#

十、Flask 还是 FastAPI？#

小结#

前言