🏗️ 架构设计：Project Knowledge Base/后端框架/FastAPI 架构设计

FastAPI

Python 生态中性能最高的现代 API 框架，async-first + Pydantic 类型验证 + 自动 OpenAPI 文档，2026 年 38% Python 开发者使用，AI/ML 后端的默认选择。

🎯 为什么需要它

Flask 同步阻塞，高并发下性能差；Django 太重，API 开发要写大量样板代码。FastAPI 用类型提示一个声明搞定验证+序列化+文档，async 原生支持每秒处理 2 万请求，写 API 的体验直接起飞。

✅ 核心优势

• 极致性能：15,000-20,000 RPS（Flask 的 5-7 倍，Django REST 的 3-4 倍）
• 类型驱动开发：一个 Python 类型声明 = 自动验证 + 自动序列化 + 自动文档
• 自动 API 文档：内置 Swagger UI + ReDoc，零配置生成交互式文档
• async 原生：基于 ASGI，一个 worker 处理 I/O 并发 = 8-10 个 Django worker
• 依赖注入：内置 DI 系统，优雅管理数据库连接、认证、配置
• AI/ML 生态核心：Hugging Face、LangChain、BentoML 都推荐 FastAPI 作为模型服务框架
• 标准兼容：完整支持 OpenAPI 3.1、JSON Schema、OAuth2、JWT

⚡ 性能表现

注意：带数据库的实际场景中差距缩小，但 FastAPI 的异步优势在 I/O 密集型场景（API 调用、文件读写）中依然显著。

🚀 快速上手

安装

pip install fastapi uvicorn[standard]
# 或使用 uv（推荐）
uv pip install fastapi uvicorn[standard]

最小可运行示例

main.py：

from fastapi import FastAPI

app = FastAPI(
    title="My API",
    description="我的第一个 FastAPI 应用",
    version="1.0.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):
    return {"item_id": item_id, "q": q}

运行：

uvicorn main:app --reload
# 访问 http://127.0.0.1:8000/docs 查看 Swagger UI
# 访问 http://127.0.0.1:8000/redoc 查看 ReDoc

项目结构（推荐）

my-api/
├── app/
│   ├── __init__.py
│   ├── main.py              # 应用入口
│   ├── config.py            # 配置
│   ├── database.py          # 数据库连接
│   ├── dependencies.py      # 公共依赖
│   ├── models/              # 数据库模型
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/             # Pydantic 模型
│   │   ├── __init__.py
│   │   └── user.py
│   ├── routers/             # 路由模块
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── items.py
│   ├── services/            # 业务逻辑
│   │   └── user_service.py
│   └── utils/               # 工具函数
│       └── security.py
├── tests/
├── requirements.txt
└── Dockerfile

📦 常用功能速查

Pydantic 模型（请求/响应）

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

# 请求模型
class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50, examples=["john"])
    email: EmailStr
    password: str = Field(..., min_length=8)
    age: Optional[int] = Field(None, ge=0, le=150)

# 响应模型（隐藏密码）
class UserResponse(BaseModel):
    id: int
    username: str
    email: EmailStr
    created_at: datetime

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

# 使用
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    # user 已自动验证，不合法会返回 422
    db_user = await save_user(user)
    return db_user

路径参数 & 查询参数

from fastapi import Path, Query

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., title="Item ID", ge=1),
    q: str = Query(None, max_length=50, regex="^fixed"),
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100)
):
    return {"item_id": item_id, "q": q, "skip": skip, "limit": limit}

请求体

from fastapi import Body

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Item = Body(..., examples=[{
        "name": "Foo",
        "price": 42.0,
        "is_offer": True
    }]),
    user: User = Body(...),
    importance: int = Body(gt=0)
):
    return {"item_id": item_id, "item": item, "user": user}

依赖注入

from fastapi import Depends
from sqlalchemy.orm import Session

# 数据库依赖
async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 分页参数复用
class Pagination:
    def __init__(self, skip: int = 0, limit: int = 100):
        self.skip = skip
        self.limit = limit

# 认证依赖
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
) -> User:
    payload = verify_token(token)
    user = db.query(User).filter(User.username == payload["sub"]).first()
    if not user:
        raise HTTPException(status_code=401, detail="用户不存在")
    return user

# 使用
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

路由分组

from fastapi import APIRouter

router = APIRouter(
    prefix="/api/v1/items",
    tags=["items"],
    responses={404: {"description": "Not found"}}
)

@router.get("/", response_model=list[ItemResponse])
async def list_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return db.query(Item).offset(skip).limit(limit).all()

@router.get("/{item_id}", response_model=ItemResponse)
async def get_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

# 在 main.py 中注册
app.include_router(router)

OAuth2 + JWT 认证

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=401,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"}
        )
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

中间件

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

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

# 自定义中间件
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = f"{process_time:.4f}"
    return response

后台任务

from fastapi import BackgroundTasks

def send_email(email: str, message: str):
    # 耗时操作
    ...

@app.post("/send-notification")
async def send_notification(
    email: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(send_email, email, "欢迎注册！")
    return {"message": "通知已发送"}

WebSocket

from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        print("客户端断开连接")

异常处理

from fastapi import HTTPException
from fastapi.responses import JSONResponse

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something"}
    )

数据库集成（SQLAlchemy async）

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

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"

engine = create_async_engine(DATABASE_URL)
async_session = async_sessionmaker(engine, class_=AsyncSession)

class Base(DeclarativeBase):
    pass

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

# 路由中使用
@app.get("/users", response_model=list[UserResponse])
async def list_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

测试

from fastapi.testclient import TestClient
import pytest

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, FastAPI!"}

def test_create_user():
    response = client.post("/users", json={
        "username": "testuser",
        "email": "test@example.com",
        "password": "strongpassword123"
    })
    assert response.status_code == 201
    assert response.json()["username"] == "testuser"

📋 关键配置

环境变量管理

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My API"
    debug: bool = False
    database_url: str
    secret_key: str
    allowed_origins: list[str] = ["http://localhost:3000"]

    model_config = {"env_file": ".env"}

settings = Settings()

# 作为依赖注入
def get_settings():
    return settings

Docker 部署

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

生产环境运行

# Uvicorn 多 worker
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

# 或使用 Gunicorn + Uvicorn worker
gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

📦 适用场景

适合：

• 高性能 REST API（微服务、后端服务）
• AI/ML 模型服务（推理 API、RAG 系统、LLM 编排层）
• 实时 API（WebSocket、SSE、流式响应）
• 数据密集型应用（数据处理管道、ETL 服务）
• 与 React/Vue/Angular 前端配合的 API 后端
• 快速原型开发（自动文档 + 类型验证 = 极速迭代）

不适合：

• 全栈 Web 应用（需要模板渲染、Session 管理 → Django 更合适）
• 需要内置 Admin 面板的应用（Django Admin 无可替代）
• 团队完全不熟悉 async 概念（误用 async 会阻塞事件循环）
• 需要成熟 ORM + 数据库迁移的应用（Django ORM 更成熟）

⚠️ 已知坑 & 注意事项

• async 误用：在 async def 中调用同步阻塞函数（如 time.sleep、同步数据库驱动）会阻塞整个事件循环，用 run_in_executor 或改用异步库
• Pydantic v1 → v2 迁移：@validator → @field_validator，Config → model_config，旧代码需要更新
• 依赖注入生命周期：yield 依赖在请求结束后才执行清理，注意数据库连接池管理
• CORS 配置：默认不允许跨域，必须显式配置 CORSMiddleware，否则前端请求会被拒绝
• OpenAPI 文档暴露：生产环境考虑限制 /docs 和 /redoc 的访问，或通过环境变量关闭
• 文件上传：UploadFile 是异步的，但读取大文件时注意内存占用
• 数据库驱动选择：必须用异步驱动（asyncpg、aiomysql），不能用同步驱动（psycopg2、pymysql）

🆚 竞品对比

选择建议：

• Python API / AI 后端 → FastAPI（首选）
• 全栈 Web 应用 → Django
• 轻量原型 / ML 端点 → Flask
• JavaScript 全栈 → Express / Hono
• 极致性能 → Go Gin / Rust Actix

🌍 生态 & 社区

• 维护状态：极活跃，当前最新 v0.136.1（2026-04-23），频繁发布
• 文档质量：极优秀，官方教程是 Python 框架文档的标杆，多语言翻译
• 周边生态：
  • SQLAlchemy — 推荐的 ORM（异步支持）
  • Pydantic v2 — 数据验证核心
  • Uvicorn / Gunicorn — ASGI 服务器
  • httpx — 异步 HTTP 客户端
  • asyncpg / aiomysql — 异步数据库驱动
  • FastAPI Users — 用户认证模块
  • FastAPI Utils — 常用工具集
  • Strawberry — GraphQL 集成
• 社区活跃度：91k+ Stars，PyPI 月下载 900 万+，Discord/GitHub 社区极活跃

💡 引入评估

结论：强烈推荐 — FastAPI 是 2026 年 Python API 开发的事实标准。async-first 设计、Pydantic 类型验证、自动文档生成让它在开发效率和运行性能上全面超越 Flask 和 Django REST。AI/ML 后端的首选框架。唯一需要评估的是：如果你需要完整的全栈框架（ORM + Admin + Auth），Django 仍是更好的选择。

推荐引入版本：0.136.1（当前最新稳定版）

🔗 相关链接

• Python Pydantic Starlette ASGI REST API 微服务 AI后端
• 官方文档
• GitHub 仓库
• PyPI
• Releases
• Discord
• FastAPI Best Practices

📝 个人备注

（留白，供后续补充实际使用心得）