FastAPI 高性能 Python Web 开发完全教程

关键词：FastAPI、Python、Web 开发、API、后端开发

适用读者：熟悉 Python 基础语法，希望学习现代高性能 Web API 开发的开发者。

为什么选择 FastAPI
环境搭建与第一个 API
Pydantic 数据验证
路径操作与参数处理
依赖注入系统
中间件开发
OAuth2 认证与 JWT
数据库集成：SQLAlchemy 与 asyncpg
WebSocket 支持
后台任务与 Celery 集成
API 文档自动生成
项目结构与最佳实践
性能优化
部署：Gunicorn、Uvicorn 与 Docker
总结与学习资源

1. 为什么选择 FastAPI

FastAPI 是由 Sebastián Ramírez 于 2018 年发布的现代 Python Web 框架，基于 Starlette（ASGI 工具包）和 Pydantic（数据验证库）构建。

1.1 FastAPI 的核心优势

高性能：基于 ASGI 异步框架，性能与 Node.js、Go 同一量级（远超 Flask/Django REST）
开发效率：类型提示驱动开发，自动补全、自动校验、自动文档
自动文档：内建 Swagger UI 和 ReDoc，零配置生成交互式 API 文档
依赖注入：强大且灵活的 DI 系统，方便管理数据库连接、认证等横切关注点
标准化：基于 OpenAPI 和 JSON Schema 标准

1.2 性能对比

根据 TechEmpower 基准测试，FastAPI（Uvicorn）在 Python 生态中表现最优，接近原生 Starlette 的吞吐量。与 Flask（同步 WSGI）相比，FastAPI 在高并发场景下吞吐量可提升 5-10 倍。

2. 环境搭建与第一个 API

2.1 安装

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/macOS
# venv\Scripts\activate   # Windows

# 安装 FastAPI 及 ASGI 服务器
pip install fastapi uvicorn[standard]

2.2 Hello World

创建 main.py：

from fastapi import FastAPI

app = FastAPI(
    title="My API",
    description="A sample FastAPI project",
    version="0.1.0",
)

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

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

2.3 启动服务

uvicorn main:app --reload --host 0.0.0.0 --port 8000

访问：

API：http://localhost:8000
Swagger UI：http://localhost:8000/docs
ReDoc：http://localhost:8000/redoc

2.4 路径参数与查询参数

from fastapi import FastAPI, Query, Path

app = FastAPI()

# 路径参数（必填）
@app.get("/users/{user_id}")
async def get_user(user_id: int = Path(..., title="用户ID", ge=1)):
    return {"user_id": user_id}

# 查询参数（可选）
@app.get("/search/")
async def search(
    q: str = Query(..., min_length=1, max_length=50, description="搜索关键词"),
    page: int = Query(1, ge=1, description="页码"),
    size: int = Query(20, ge=1, le=100, description="每页数量"),
):
    return {"query": q, "page": page, "size": size}

3. Pydantic 数据验证

Pydantic 是 FastAPI 的数据验证基石，通过 Python 类型提示实现声明式数据校验。

3.1 基础模型

from pydantic import BaseModel, Field, EmailStr, field_validator
from datetime import datetime
from enum import Enum

class UserRole(str, Enum):
    admin = "admin"
    editor = "editor"
    viewer = "viewer"

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=32, pattern=r"^[a-zA-Z0-9_]+$")
    email: EmailStr
    password: str = Field(..., min_length=8)
    role: UserRole = UserRole.viewer
    bio: str | None = Field(None, max_length=500)

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if not any(c.isupper() for c in v):
            raise ValueError("密码必须包含至少一个大写字母")
        if not any(c.isdigit() for c in v):
            raise ValueError("密码必须包含至少一个数字")
        return v

class UserResponse(BaseModel):
    id: int
    username: str
    email: EmailStr
    role: UserRole
    created_at: datetime

3.2 在 API 中使用

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

# 模拟数据库
fake_db: dict[int, UserResponse] = {}

@app.post("/users/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
    # 检查用户名是否已存在
    for u in fake_db.values():
        if u.username == user.username:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail="用户名已存在",
            )

    user_id = len(fake_db) + 1
    new_user = UserResponse(
        id=user_id,
        username=user.username,
        email=user.email,
        role=user.role,
        created_at=datetime.now(),
    )
    fake_db[user_id] = new_user
    return new_user

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    if user_id not in fake_db:
        raise HTTPException(status_code=404, detail="用户不存在")
    return fake_db[user_id]

3.3 模型嵌套与复杂验证

from pydantic import BaseModel, model_validator

class Address(BaseModel):
    street: str
    city: str
    country: str = "CN"
    zip_code: str | None = None

class OrderItem(BaseModel):
    product_id: int
    quantity: int = Field(..., gt=0)
    unit_price: float = Field(..., gt=0)

class OrderCreate(BaseModel):
    items: list[OrderItem] = Field(..., min_length=1)
    shipping_address: Address
    coupon_code: str | None = None

    @model_validator(mode="after")
    def validate_total(self) -> "OrderCreate":
        total = sum(item.quantity * item.unit_price for item in self.items)
        if total > 100000:
            raise ValueError("订单金额超过上限 ¥100,000")
        return self

4. 路径操作与参数处理

4.1 请求体、查询参数、路径参数混用

from fastapi import FastAPI, Body

app = FastAPI()

@app.put("/users/{user_id}")
async def update_user(
    user_id: int,                              # 路径参数
    notify: bool = Query(False),                # 查询参数
    user: UserCreate = Body(...),               # 请求体
    x_request_id: str | None = Header(None),    # 请求头
):
    return {
        "user_id": user_id,
        "updated": user.model_dump(),
        "notify": notify,
        "request_id": x_request_id,
    }

4.2 响应模型与过滤

class UserInDB(BaseModel):
    id: int
    username: str
    email: str
    hashed_password: str  # 敏感字段

class UserPublic(BaseModel):
    id: int
    username: str
    email: str

@app.get("/users/{user_id}", response_model=UserPublic)
async def get_user(user_id: int):
    user = get_user_from_db(user_id)
    return user  # hashed_password 会被自动过滤

4.3 文件上传

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

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

5. 依赖注入系统

FastAPI 的依赖注入是其最强大的特性之一，适合管理数据库连接、认证、配置等共享资源。

5.1 基础依赖

from fastapi import Depends, FastAPI

app = FastAPI()

async def get_db():
    db = DatabaseSession()
    try:
        yield db      # yield 之前的代码在请求前执行
    finally:
        db.close()    # yield 之后的代码在请求后执行（类似中间件）

@app.get("/items/")
async def list_items(db = Depends(get_db)):
    return db.query(Item).all()

5.2 嵌套依赖

async def get_token_header(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")

async def get_query_token(token: str = Query(...)):
    if token != "jessica":
        raise HTTPException(status_code=403, detail="No permission")

@app.get("/items/", dependencies=[Depends(get_token_header), Depends(get_query_token)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

5.3 类依赖

class CommonQueryParams:
    def __init__(
        self,
        q: str | None = None,
        skip: int = 0,
        limit: int = 100,
    ):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends()):
    return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

6. 中间件开发

6.1 自定义中间件

import time
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.perf_counter()
        response = await call_next(request)
        duration = time.perf_counter() - start_time
        response.headers["X-Process-Time"] = f"{duration:.4f}"
        print(f"{request.method} {request.url.path} - {duration:.4f}s")
        return response

app.add_middleware(TimingMiddleware)

6.2 CORS 中间件

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://myapp.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

6.3 请求日志中间件

import logging
from fastapi import Request

logger = logging.getLogger("api")

@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info(f"→ {request.method} {request.url.path} from {request.client.host}")
    response = await call_next(request)
    logger.info(f"← {request.method} {request.url.path} → {response.status_code}")
    return response

7. OAuth2 认证与 JWT

7.1 安装依赖

pip install python-jose[cryptography] passlib[bcrypt] python-multipart

7.2 完整认证实现

from datetime import datetime, timedelta
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

# 配置
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 密码哈希
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

# 用户模型
class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: str | None = None

class User(BaseModel):
    username: str
    email: str
    disabled: bool = False

class UserInDB(User):
    hashed_password: str

# 模拟数据库
fake_users_db = {
    "admin": {
        "username": "admin",
        "email": "admin@example.com",
        "hashed_password": pwd_context.hash("Admin123!"),
        "disabled": False,
    }
}

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception

    user_data = fake_users_db.get(token_data.username)
    if user_data is None:
        raise credentials_exception
    return User(**user_data)

async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="用户已被禁用")
    return current_user

# 登录接口
@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user_data = fake_users_db.get(form_data.username)
    if not user_data or not verify_password(form_data.password, user_data["hashed_password"]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(
        data={"sub": user_data["username"]},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    return {"access_token": access_token, "token_type": "bearer"}

# 受保护的接口
@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

8. 数据库集成：SQLAlchemy 与 asyncpg

8.1 安装

pip install sqlalchemy[asyncio] asyncpg alembic

8.2 数据库配置

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/mydb"

engine = create_async_engine(DATABASE_URL, echo=True, pool_size=20, max_overflow=10)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

8.3 定义模型

# models.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey, func
from sqlalchemy.orm import relationship
from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(32), unique=True, nullable=False, index=True)
    email = Column(String(128), unique=True, nullable=False)
    hashed_password = Column(String(128), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, server_default=func.now())

    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    published = Column(Boolean, default=False)
    author_id = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime, server_default=func.now())

    author = relationship("User", back_populates="posts")

8.4 依赖注入数据库会话

# deps.py
from database import AsyncSessionLocal
from sqlalchemy.ext.asyncio import AsyncSession

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

8.5 CRUD 操作

# crud.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from models import User, Post
from schemas import UserCreate, PostCreate
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

async def create_user(db: AsyncSession, user: UserCreate) -> User:
    db_user = User(
        username=user.username,
        email=user.email,
        hashed_password=pwd_context.hash(user.password),
    )
    db.add(db_user)
    await db.flush()
    await db.refresh(db_user)
    return db_user

async def get_user_by_username(db: AsyncSession, username: str) -> User | None:
    result = await db.execute(select(User).where(User.username == username))
    return result.scalar_one_or_none()

async def create_post(db: AsyncSession, post: PostCreate, author_id: int) -> Post:
    db_post = Post(**post.model_dump(), author_id=author_id)
    db.add(db_post)
    await db.flush()
    await db.refresh(db_post, ["author"])
    return db_post

async def get_posts(db: AsyncSession, skip: int = 0, limit: int = 20):
    result = await db.execute(
        select(Post)
        .options(selectinload(Post.author))
        .where(Post.published == True)
        .offset(skip)
        .limit(limit)
        .order_by(Post.created_at.desc())
    )
    return result.scalars().all()

8.6 完整路由

# main.py
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from deps import get_db
from crud import create_user, create_post, get_posts
from schemas import UserCreate, UserResponse, PostCreate, PostResponse

app = FastAPI()

@app.post("/users/", response_model=UserResponse)
async def api_create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
    return await create_user(db, user)

@app.post("/posts/", response_model=PostResponse)
async def api_create_post(
    post: PostCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_active_user),
):
    return await create_post(db, post, current_user.id)

@app.get("/posts/", response_model=list[PostResponse])
async def api_list_posts(skip: int = 0, limit: int = 20, db: AsyncSession = Depends(get_db)):
    return await get_posts(db, skip, limit)

8.7 数据库迁移（Alembic）

# 初始化
alembic init alembic

# alembic/env.py 中配置异步引擎
# 生成迁移
alembic revision --autogenerate -m "create tables"

# 执行迁移
alembic upgrade head

# 回滚
alembic downgrade -1

9. WebSocket 支持

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

app = FastAPI()

class ConnectionManager:
    """管理 WebSocket 连接池"""

    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{room_id}")
async def websocket_endpoint(websocket: WebSocket, room_id: str):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"[Room {room_id}] {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"用户离开房间 {room_id}")

# 带认证的 WebSocket
@app.websocket("/ws/private")
async def private_websocket(websocket: WebSocket, token: str = Query(...)):
    user = verify_ws_token(token)
    if not user:
        await websocket.close(code=4001)
        return

    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_json()
            # 处理消息...
            await websocket.send_json({"echo": data})
    except WebSocketDisconnect:
        pass

10. 后台任务与 Celery 集成

10.1 FastAPI 内置后台任务

适合轻量级异步操作（如发送邮件、记录日志）：

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email_background(email_to: str, subject: str, body: str):
    """后台执行：不阻塞请求"""
    import time
    time.sleep(2)  # 模拟耗时操作
    print(f"Email sent to {email_to}: {subject}")

@app.post("/send-notification/")
async def send_notification(
    email: str,
    background_tasks: BackgroundTasks,
):
    background_tasks.add_task(send_email_background, email, "Welcome!", "Hello there!")
    return {"message": "通知将在后台发送"}

10.2 Celery 集成（重量级任务）

安装：

pip install celery redis

Celery 配置：

# celery_app.py
from celery import Celery

celery = Celery(
    "worker",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

celery.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="Asia/Shanghai",
    enable_utc=True,
    task_track_started=True,
    task_time_limit=300,
    worker_max_tasks_per_child=100,
)

@celery.task(bind=True, max_retries=3)
def process_data(self, data_id: int):
    try:
        # 模拟耗时任务
        import time
        time.sleep(10)
        return {"status": "completed", "data_id": data_id}
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)

在 FastAPI 中触发任务：

from celery_app import process_data

@app.post("/tasks/")
async def create_task(data_id: int):
    task = process_data.delay(data_id)
    return {"task_id": task.id, "status": "queued"}

@app.get("/tasks/{task_id}")
async def get_task_status(task_id: str):
    task = process_data.AsyncResult(task_id)
    return {
        "task_id": task_id,
        "status": task.status,
        "result": task.result if task.ready() else None,
    }

启动 Worker：

celery -A celery_app worker --loglevel=info --concurrency=4

11. API 文档自动生成

11.1 文档元数据配置

app = FastAPI(
    title="电商平台 API",
    description="""
    ## 功能概览
    - 用户注册与认证
    - 商品管理
    - 订单处理
    - 支付集成

    ## 认证方式
    使用 Bearer Token（JWT）进行认证。
    先调用 `/token` 接口获取 Token，然后在请求头中携带：
    `Authorization: Bearer <your_token>`
    """,
    version="2.0.0",
    contact={"name": "API Support", "email": "api@example.com"},
    license_info={"name": "MIT"},
)

11.2 路由标签与描述

from fastapi import APIRouter

router = APIRouter(
    prefix="/products",
    tags=["商品管理"],
    responses={404: {"description": "商品未找到"}},
)

@router.get(
    "/",
    summary="获取商品列表",
    description="分页获取已上架的商品列表，支持按类别和价格区间筛选。",
    response_model=list[ProductResponse],
)
async def list_products(
    category: str | None = Query(None, description="商品类别"),
    min_price: float | None = Query(None, ge=0, description="最低价格"),
    page: int = Query(1, ge=1),
    size: int = Query(20, ge=1, le=100),
):
    """返回商品列表，按创建时间降序排列。"""
    ...

11.3 自定义文档 UI

# 禁用默认文档，使用自定义页面
app = FastAPI(docs_url=None, redoc_url=None)

@app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html():
    return get_swagger_ui_html(
        openapi_url=app.openapi_url,
        title="Custom API Docs",
        swagger_css_url="/static/custom-swagger.css",
    )

12. 项目结构与最佳实践

12.1 推荐目录结构

my_project/
├── alembic/                 # 数据库迁移
│   ├── versions/
│   └── env.py
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI 应用入口
│   ├── config.py            # 配置管理
│   ├── database.py          # 数据库连接
│   ├── deps.py              # 通用依赖
│   ├── models/              # SQLAlchemy 模型
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── product.py
│   ├── schemas/             # Pydantic 模型（请求/响应）
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── product.py
│   ├── api/                 # 路由模块
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── users.py
│   │   │   ├── products.py
│   │   │   └── auth.py
│   │   └── deps.py
│   ├── services/            # 业务逻辑层
│   │   ├── user_service.py
│   │   └── product_service.py
│   ├── core/                # 核心工具
│   │   ├── security.py
│   │   └── exceptions.py
│   └── utils/               # 通用工具
├── tests/                   # 测试
│   ├── conftest.py
│   ├── test_users.py
│   └── test_products.py
├── alembic.ini
├── pyproject.toml
├── Dockerfile
└── docker-compose.yml

12.2 配置管理

# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # 应用配置
    app_name: str = "My API"
    debug: bool = False

    # 数据库
    database_url: str = "postgresql+asyncpg://user:pass@localhost/db"

    # JWT
    secret_key: str = "change-me-in-production"
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30

    # Redis
    redis_url: str = "redis://localhost:6379/0"

    # CORS
    cors_origins: list[str] = ["http://localhost:3000"]

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = Settings()

12.3 全局异常处理

# core/exceptions.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

class AppException(Exception):
    def __init__(self, status_code: int, detail: str, error_code: str = "UNKNOWN"):
        self.status_code = status_code
        self.detail = detail
        self.error_code = error_code

def register_exception_handlers(app: FastAPI):
    @app.exception_handler(AppException)
    async def app_exception_handler(request: Request, exc: AppException):
        return JSONResponse(
            status_code=exc.status_code,
            content={
                "error": {
                    "code": exc.error_code,
                    "message": exc.detail,
                    "path": str(request.url),
                }
            },
        )

    @app.exception_handler(Exception)
    async def general_exception_handler(request: Request, exc: Exception):
        return JSONResponse(
            status_code=500,
            content={
                "error": {
                    "code": "INTERNAL_ERROR",
                    "message": "服务器内部错误",
                }
            },
        )

13. 性能优化

13.1 异步最佳实践

# ✅ 正确：使用 async 数据库驱动
async def get_users(db: AsyncSession):
    result = await db.execute(select(User))
    return result.scalars().all()

# ❌ 错误：在 async 函数中使用同步阻塞操作
async def bad_example():
    import time
    time.sleep(5)  # 阻塞整个事件循环！
    # 应改为：await asyncio.sleep(5)

13.2 连接池配置

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,           # 连接池大小
    max_overflow=10,        # 允许额外溢出的连接数
    pool_timeout=30,        # 获取连接超时
    pool_recycle=1800,      # 连接回收时间（秒）
    pool_pre_ping=True,     # 使用前检测连接活性
)

13.3 响应缓存

from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
import redis.asyncio as redis

app = FastAPI()

@app.on_event("startup")
async def startup():
    r = redis.from_url("redis://localhost", encoding="utf8")
    FastAPICache.init(RedisBackend(r), prefix="cache")

@app.get("/products/")
@cache(expire=300)  # 缓存 5 分钟
async def list_products(category: str = "all"):
    return await fetch_products_from_db(category)

13.4 Gzip 压缩

from fastapi.middleware.gzip import GZipMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)

14. 部署：Gunicorn、Uvicorn 与 Docker

14.1 Uvicorn 直接部署（开发/小规模）

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

14.2 Gunicorn + Uvicorn Worker（生产推荐）

pip install gunicorn
gunicorn app.main:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000 \
    --timeout 120 \
    --access-logfile - \
    --error-logfile -

Worker 数量经验公式：workers = 2 × CPU核心数 + 1

14.3 Docker 部署

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential libpq-dev \
    && rm -rf /var/lib/apt/lists/*

# 安装 Python 依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 创建非 root 用户
RUN useradd -m appuser
USER appuser

EXPOSE 8000

CMD ["gunicorn", "app.main:app", \
     "--workers", "4", \
     "--worker-class", "uvicorn.workers.UvicornWorker", \
     "--bind", "0.0.0.0:8000"]

14.4 Docker Compose 完整编排

# docker-compose.yml
version: "3.9"

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://postgres:password@db:5432/mydb
      - REDIS_URL=redis://redis:6379/0
      - SECRET_KEY=${SECRET_KEY}
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started
    restart: unless-stopped

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: mydb
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: password
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data

  celery_worker:
    build: .
    command: celery -A celery_app worker --loglevel=info
    environment:
      - DATABASE_URL=postgresql+asyncpg://postgres:password@db:5432/mydb
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - db
      - redis

volumes:
  postgres_data:
  redis_data:

14.5 健康检查端点

@app.get("/health")
async def health_check():
    return {"status": "healthy", "version": "2.0.0"}

@app.get("/health/db")
async def db_health(db: AsyncSession = Depends(get_db)):
    try:
        await db.execute(text("SELECT 1"))
        return {"status": "healthy", "database": "connected"}
    except Exception:
        raise HTTPException(status_code=503, detail="数据库连接失败")

15. 总结与学习资源

15.1 FastAPI 核心要点回顾

特性	说明
类型提示驱动	用 Python 类型提示定义参数、响应，自动生成文档
Pydantic 校验	声明式数据验证，自动处理序列化/反序列化
依赖注入	优雅管理数据库连接、认证、配置等
异步优先	原生 async/await 支持，高并发性能优秀
自动文档	Swagger UI + ReDoc，零配置即可用
OAuth2/JWT	内建安全工具，快速实现认证授权

15.2 学习路线

入门：FastAPI 官方教程（有中文版）
进阶：Full Stack FastAPI Template
实战：构建一个完整的 CRUD 应用（用户 + 文章 + 评论）
深入：阅读 Starlette 源码，理解 ASGI 协议
部署：学习 Docker、Kubernetes、CI/CD 流程

15.3 推荐学习资源

FastAPI 官方文档 — 最权威的参考资料
Pydantic 官方文档 — 数据验证深入学习
SQLAlchemy 2.0 文档 — ORM 使用指南
TestDriven.io FastAPI 课程 — 实战驱动
tiangolo 的 GitHub — 作者的项目示例

15.4 常见陷阱

在 async 函数中调用同步阻塞代码：会阻塞事件循环，用 run_in_executor 包装
忘记 await：异步函数不 await 等于没调用，拿到的是协程对象
数据库 session 泄漏：确保在依赖中用 try/finally 或 async with 管理
生产环境用 --reload：这是开发选项，生产应使用 Gunicorn + 多 Worker
硬编码密钥：使用环境变量或 pydantic-settings 管理敏感配置

总结

FastAPI 是 Python Web 开发生态中的一颗新星——它将类型安全、自动文档、高性能异步处理和优雅的开发体验完美结合。无论你是构建微服务、REST API 还是全栈应用，FastAPI 都是 2024 年 Python Web 开发的最佳选择之一。

核心记住：

Pydantic 模型是你的数据契约，定义好就等于做好了一半
依赖注入是 FastAPI 的灵魂，善用它管理共享资源
异步优先，但不要在 async 函数中阻塞
生产部署用 Gunicorn + Uvicorn Worker + Docker
文档自动生成是你的超能力，善加利用

Happy coding with FastAPI! 🚀

FastAPI高性能Python Web开发完全教程

教程简介