FastAPI — 现代异步 Web 框架

FastAPI 是目前最流行的 Python Web 框架之一，结合了高性能、自动文档、类型安全三大优势。

为什么选 FastAPI

特性	FastAPI	Flask	Django
性能	⭐⭐⭐（异步）	⭐⭐	⭐⭐
自动 API 文档	✅ 内置	❌	❌
类型安全	✅ Pydantic	❌	❌
学习曲线	低	极低	高
适合场景	API 服务	小型应用	全栈 Web

快速开始

pip install fastapi uvicorn[standard]

# main.py
from fastapi import FastAPI

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

@app.get("/")
async def root():
    return {"message": "Hello, FastAPI!"}

@app.get("/items/{item_id}")
async def get_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "query": q}

uvicorn main:app --reload
# 访问 http://localhost:8000/docs 查看自动生成的 Swagger UI

Pydantic 数据模型

from fastapi import FastAPI
from pydantic import BaseModel, Field, EmailStr
from datetime import datetime
from typing import Optional

app = FastAPI()

class UserCreate(BaseModel):
    name: str = Field(..., min_length=2, max_length=50)
    email: EmailStr
    age: int = Field(..., ge=0, le=150)
    bio: Optional[str] = None

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime

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

@app.post("/users/", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    # FastAPI 自动验证请求体，类型不对直接返回 422
    return {"id": 1, "created_at": datetime.now(), **user.model_dump()}

依赖注入系统

FastAPI 的依赖注入是其最强大的特性：

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 数据库会话依赖
def get_db():
    db = SessionLocal()
    try:
        yield db          # 提供给路由函数
    finally:
        db.close()        # 请求结束后自动清理

# 认证依赖
async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = verify_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的认证凭据",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

# 权限依赖（依赖的依赖）
async def require_admin(current_user = Depends(get_current_user)):
    if not current_user.is_admin:
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return current_user

@app.get("/admin/users")
async def list_users(
    db = Depends(get_db),
    admin = Depends(require_admin)
):
    return db.query(User).all()

路由组织

# routers/users.py
from fastapi import APIRouter, Depends

router = APIRouter(prefix="/users", tags=["用户管理"])

@router.get("/", summary="获取用户列表")
async def list_users():
    ...

@router.get("/{user_id}", summary="获取单个用户")
async def get_user(user_id: int):
    ...

@router.post("/", summary="创建用户", status_code=201)
async def create_user(user: UserCreate):
    ...

# main.py
from fastapi import FastAPI
from routers import users, items, auth

app = FastAPI()
app.include_router(users.router)
app.include_router(items.router)
app.include_router(auth.router, prefix="/auth")

中间件

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
import time

app = FastAPI()

# CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 自定义中间件：请求耗时日志
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start
    response.headers["X-Process-Time"] = str(elapsed)
    print(f"{request.method} {request.url} — {elapsed:.3f}s")
    return response

异常处理

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# 自定义异常类
class AppException(Exception):
    def __init__(self, code: str, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code

# 全局异常处理器
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"code": exc.code, "message": exc.message}
    )

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id <= 0:
        raise AppException("INVALID_ID", "ID 必须为正整数")
    if item_id > 100:
        raise HTTPException(status_code=404, detail="商品不存在")
    return {"item_id": item_id}

后台任务

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(email: str, message: str):
    # 耗时操作，在后台执行
    import time
    time.sleep(2)
    print(f"发送邮件到 {email}: {message}")

@app.post("/register")
async def register(
    email: str,
    background_tasks: BackgroundTasks
):
    # 立即返回响应，邮件在后台发送
    background_tasks.add_task(send_email, email, "欢迎注册！")
    return {"message": "注册成功，欢迎邮件发送中"}

WebSocket

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active: List[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active.append(ws)

    def disconnect(self, ws: WebSocket):
        self.active.remove(ws)

    async def broadcast(self, message: str):
        for ws in self.active:
            await ws.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(ws: WebSocket, client_id: str):
    await manager.connect(ws)
    try:
        while True:
            data = await ws.receive_text()
            await manager.broadcast(f"{client_id}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(ws)
        await manager.broadcast(f"{client_id} 离开了聊天室")

完整项目结构

my_api/
├── app/
│   ├── __init__.py
│   ├── main.py          # FastAPI 实例，挂载路由
│   ├── config.py        # 配置（pydantic-settings）
│   ├── database.py      # 数据库连接
│   ├── models/          # SQLAlchemy 模型
│   │   ├── user.py
│   │   └── item.py
│   ├── schemas/         # Pydantic 模型
│   │   ├── user.py
│   │   └── item.py
│   ├── routers/         # 路由
│   │   ├── users.py
│   │   └── items.py
│   ├── services/        # 业务逻辑
│   │   └── user_service.py
│   └── dependencies.py  # 公共依赖
├── tests/
├── alembic/             # 数据库迁移
├── pyproject.toml
└── Dockerfile

生产部署

# 生产环境用 gunicorn + uvicorn workers
gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker