FastAPI 是一个现代、快速（高性能）的 Web 框架，用于基于 Python 3.7+ 构建 API，基于标准的 Python 类型提示。它由 Sebastián Ramírez 于 2018 年创建，迅速成为 Python 生态中最受欢迎的 Web 框架之一。
极致性能：FastAPI 基于 Starlette（用于 Web 部分）和 Pydantic（用于数据部分），是最快的 Python 框架之一，性能可与 NodeJS 和 Go 相媲美。
开发效率：
类型安全：
标准兼容：
中间件是在每个请求被处理之前和每个响应返回之前执行的代码。它可以用于：
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
import time
import logging
app = FastAPI()
logger = logging.getLogger(__name__)
# 使用装饰器创建中间件
@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"] = str(process_time)
return response
# 请求日志中间件
@app.middleware("http")
async def log_requests(request: Request, call_next):
logger.info(f"Request: {request.method} {request.url}")
response = await call_next(request)
logger.info(f"Response: {response.status_code}")
return response
依赖注入（Dependency Injection, DI）是一种设计模式，用于实现控制反转（IoC）。在 FastAPI 中，依赖注入系统允许你声明函数所需的依赖，FastAPI 会自动解析并注入这些依赖。
FastAPI 基于 ASGI（Asynchronous Server Gateway Interface）构建，原生支持异步编程。理解 Python 的 asyncio 对于充分发挥 FastAPI 性能至关重要。
import asyncio
from fastapi import FastAPI
app = FastAPI()
# 同步函数
def sync_task():
import time
time.sleep(1)
return "sync result"
# 异步函数
async def async_task():
await asyncio.sleep(1)
return "async result"
# 异步端点（推荐）
@app.get("/async")
async def async_endpoint():
result = await async_task()
return {"result": result}
# 同步端点（FastAPI 会在线程池中运行）
@app.get("/sync")
def sync_endpoint():
result = sync_task()
return {"result": result}
SQLAlchemy 是 Python 中最流行的 ORM（对象关系映射）库，FastAPI 官方推荐使用 SQLAlchemy 进行数据库操作。SQLAlchemy 2.0 版本引入了更好的异步支持和类型提示。
# 核心依赖
pip install sqlalchemy
# 异步支持
pip install sqlalchemy[asyncio]
# 数据库驱动
pip install asyncpg # PostgreSQL 异步驱动
pip install aiomysql # MySQL 异步驱动
pip install aiosqlite # SQLite 异步驱动
# 数据库迁移
pip install alembic
Pydantic 是 Python 中最流行的数据验证库，FastAPI 的核心依赖之一。它使用 Python 类型提示来进行数据验证、序列化和文档生成。
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
from enum import Enum
# 基础模型
class User(BaseModel):
id: int
username: str
email: str
is_active: bool = True # 默认值
created_at: datetime = Field(default_factory=datetime.now)
# 使用模型
user = User(id=1, username="john", email="john@example.com")
print(user.model_dump()) # 转换为字典
print(user.model_dump_json()) # 转换为 JSON 字符串
# 从字典创建
data = {"id": 2, "username": "jane", "email": "jane@example.com"}
user2 = User.model_validate(data)
# 从 JSON 创建
json_str = '{"id": 3, "username": "bob", "email": "bob@example.com"}'
user3 = User.model_validate_json(json_str)
# 安装测试依赖
pip install pytest pytest-asyncio httpx pytest-cov
在大型项目中，将所有路由定义在一个文件中会导致代码难以维护。FastAPI 提供了 APIRouter 来实现路由模块化。
# app/api/v1/endpoints/users.py
from fastapi import APIRouter, HTTPException, status
from typing import List
from pydantic import BaseModel
router = APIRouter(
prefix="/users", # 路由前缀
tags=["users"], # 文档标签
responses={404: {"description": "用户不存在"}} # 通用响应
)
class User(BaseModel):
id: int
username: str
email: str
# 模拟数据库
fake_users_db = {}
@router.get("/", response_model=List[User])
async def list_users():
"""获取所有用户列表"""
return list(fake_users_db.values())
@router.get("/{user_id}", response_model=User)
async def get_user(user_id: int):
"""根据 ID 获取用户"""
if user_id not in fake_users_db:
raise HTTPException(status_code=404, detail="用户不存在")
return fake_users_db[user_id]
@router.post("/", response_model=User, status_code=status.HTTP_201_CREATED)
async def create_user(user: User):
"""创建新用户"""
if user.id in fake_users_db:
raise HTTPException(status_code=400, detail="用户已存在")
fake_users_db[user.id] = user
return user
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int):
"""删除用户"""
if user_id not in fake_users_db:
raise HTTPException(status_code=404, detail="用户不存在")
del fake_users_db[user_id]
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
app = FastAPI()
# 连接管理器
class ConnectionManager:
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 send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.send_personal_message(f"You wrote: {data}", websocket)
await manager.broadcast(f"Client #{client_id} says: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(f"Client #{client_id} left the chat")