当前位置：首页>python>Python之FastAPI 开发知识(第五篇)

Python之FastAPI 开发知识(第五篇)

2026-03-27 11:35:05

在前面的教程中，我们深入探讨了 FastAPI 的一系列高级特性，这些特性使得 FastAPI 不仅是一个简单的 API 框架，更是一个能够构建复杂、高性能、生产级应用的全能工具。以下是核心高级特性的简要回顾：

1. 性能优化

异步数据库操作
：使用异步驱动（如 asyncpg）配合 SQLAlchemy 异步扩展，实现真正的非阻塞 I/O。
多级缓存
：内存缓存（cachetools）与分布式缓存（Redis）相结合，降低数据库负载。
Gzip 压缩
：通过 GZipMiddleware 减少网络传输量。
速率限制
：使用 slowapi 防止滥用。
连接池优化
：合理配置数据库连接池参数，平衡资源与性能。
性能监控
：中间件记录请求耗时，定位瓶颈。

2. 高级安全实践

多因素认证 (MFA)
：集成 TOTP（如 Google Authenticator）增强账户安全。
API 密钥管理
：为第三方开发者提供可过期、带权限的 API 密钥。
增强的密码哈希
：使用 passlib 配置多种哈希算法（bcrypt、argon2），支持密码升级。
CSRF 保护
：对于非 API 的传统表单，添加 CSRF 令牌验证。

3. 异步任务处理

BackgroundTasks
：轻量级后台任务，适用于简单操作。
Celery 集成
：分布式任务队列，支持重试、定时、任务状态跟踪。
APScheduler
：实现定时任务，如定期清理过期数据。

4. 测试驱动开发

单元测试
：使用 pytest 和 TestClient 编写测试，覆盖各种场景。
依赖项覆盖
：通过 app.dependency_overrides 替换真实依赖（如数据库）。
异步测试
：使用 AsyncClient 测试异步端点。
测试覆盖率
：pytest-cov 测量覆盖率。

5. GraphQL 集成

Strawberry
：将 FastAPI 与 GraphQL 结合，提供灵活的数据查询。
Query 与 Mutation
：定义 GraphQL 类型和操作，与 REST 共存。

6. 微服务架构

服务发现
：使用 Consul 注册与发现服务实例。
服务间通信
：httpx.AsyncClient 实现异步 HTTP 调用。
容错与重试
：结合 tenacity 等库增强健壮性。

这些高级特性相互配合，能够帮助你构建出高可用、可扩展、安全的现代 Web 应用。下面，我们将通过一个完整的实战项目——“实时聊天 + 用户认证”——来演示如何将这些知识付诸实践。

完整项目：实时聊天 + 用户认证

本项目实现了一个简单的聊天应用，包含以下功能：

用户注册、登录（JWT 认证）
受保护的用户信息接口
WebSocket 聊天室，支持多用户实时广播消息
WebSocket 连接需携带 JWT 令牌进行身份验证

项目结构

chat_app/
├── main.py                 # 应用入口
├── database.py             # 数据库配置
├── models.py               # SQLAlchemy 模型
├── schemas.py              # Pydantic 模型
├── auth.py                 # 认证相关（密码哈希、JWT）
├── dependencies.py         # 公共依赖项
├── routers/
│   ├── auth.py             # 认证路由
│   ├── users.py            # 用户信息路由
│   └── chat.py             # 聊天 WebSocket 路由
├── requirements.txt        # 依赖列表
└── README.md               # 项目说明

环境要求

Python 3.8+
安装依赖：pip install -r requirements.txt

依赖清单 (requirements.txt)

fastapi
uvicorn[standard]
sqlalchemy
python-jose[cryptography]
passlib[bcrypt]
python-multipart

代码实现

1. database.py - 数据库配置

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL ="sqlite:///./chat_app.db"
# SQLite 需要设置 check_same_thread=False
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread":False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

2. models.py - SQLAlchemy 模型

from sqlalchemy import Column, Integer, String
from.database import Base

classUser(Base):
    __tablename__ ="users"

id= Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)

3. schemas.py - Pydantic 模型

from pydantic import BaseModel

classUserBase(BaseModel):
    username:str

classUserCreate(UserBase):
    password:str

classUser(UserBase):
id:int
classConfig:
        orm_mode =True

classToken(BaseModel):
    access_token:str
    token_type:str

classTokenData(BaseModel):
    username:str|None=None

4. auth.py - 认证工具

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext

# 密钥应从环境变量读取，此处仅作示例
SECRET_KEY ="09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM ="HS256"
ACCESS_TOKEN_EXPIRE_MINUTES =30

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

defverify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)

defget_password_hash(password):
return pwd_context.hash(password)

defcreate_access_token(data:dict, expires_delta: timedelta |None=None):
    to_encode = data.copy()
if expires_delta:
        expire = datetime.utcnow()+ expires_delta
else:
        expire = datetime.utcnow()+ timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt

defdecode_token(token:str):
try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username:str= payload.get("sub")
if username isNone:
returnNone
return username
except JWTError:
returnNone

5. dependencies.py - 公共依赖项

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from jose import JWTError

from.import database, models, auth, schemas

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

defget_db():
    db = database.SessionLocal()
try:
yield db
finally:
        db.close()

defget_current_user(
    token:str= Depends(oauth2_scheme),
    db: Session = Depends(get_db)
)-> models.User:
    username = auth.decode_token(token)
if username isNone:
raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate":"Bearer"},
)
    user = db.query(models.User).filter(models.User.username == username).first()
if user isNone:
raise HTTPException(status_code=404, detail="User not found")
return user

6. routers/auth.py - 认证路由

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from..import schemas, models, auth, dependencies

router = APIRouter(prefix="/auth", tags=["authentication"])

@router.post("/register", response_model=schemas.User)
defregister(user: schemas.UserCreate, db: Session = Depends(dependencies.get_db)):
    db_user = db.query(models.User).filter(models.User.username == user.username).first()
if db_user:
raise HTTPException(status_code=400, detail="Username already registered")
    hashed = auth.get_password_hash(user.password)
    new_user = models.User(username=user.username, hashed_password=hashed)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
return new_user

@router.post("/token", response_model=schemas.Token)
deflogin(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(dependencies.get_db)):
    user = db.query(models.User).filter(models.User.username == form_data.username).first()
ifnot user ornot auth.verify_password(form_data.password, user.hashed_password):
raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token = auth.create_access_token(data={"sub": user.username})
return{"access_token": access_token,"token_type":"bearer"}

7. routers/users.py - 用户信息路由

from fastapi import APIRouter, Depends
from..import schemas, dependencies

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/me", response_model=schemas.User)
defread_users_me(current_user = Depends(dependencies.get_current_user)):
return current_user

8. routers/chat.py - 聊天 WebSocket 路由

from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from typing import Dict
from..auth import decode_token

router = APIRouter(prefix="/chat", tags=["chat"])

classConnectionManager:
def__init__(self):
        self.active_connections: Dict[str, WebSocket]={}# username -> websocket

asyncdefconnect(self, websocket: WebSocket, username:str):
await websocket.accept()
        self.active_connections[username]= websocket

defdisconnect(self, username:str):
        self.active_connections.pop(username,None)

asyncdefsend_personal(self, message:str, username:str):
if username in self.active_connections:
await self.active_connections[username].send_text(message)

asyncdefbroadcast(self, message:str, exclude_username:str=None):
for username, conn in self.active_connections.items():
if username != exclude_username:
await conn.send_text(message)

manager = ConnectionManager()

@router.websocket("/ws")
asyncdefwebsocket_endpoint(websocket: WebSocket):
# 从查询参数获取 token
    token = websocket.query_params.get("token")
ifnot token:
await websocket.close(code=1008)
return

    username = decode_token(token)
ifnot username:
await websocket.close(code=1008)
return

await manager.connect(websocket, username)
try:
whileTrue:
            data =await websocket.receive_text()
# 广播消息给所有其他用户
await manager.broadcast(f"{username}: {data}", exclude_username=username)
except WebSocketDisconnect:
        manager.disconnect(username)
await manager.broadcast(f"User {username} left the chat")

9. main.py - 应用入口

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from.database import engine
from.models import Base
from.routers import auth, users, chat

# 创建数据库表
Base.metadata.create_all(bind=engine)

app = FastAPI(title="Chat App with JWT & WebSocket")

# 配置 CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],# 生产环境应限制
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 包含路由
app.include_router(auth.router)
app.include_router(users.router)
app.include_router(chat.router)

@app.get("/")
defroot():
return{"message":"Welcome to Chat App API"}

10. README.md - 项目说明

# Chat App with FastAPI

一个简单的实时聊天应用，包含用户认证和 WebSocket 通信。

## 功能

- 用户注册
- 用户登录（获取 JWT 令牌）
- 获取当前用户信息
- WebSocket 聊天室（需令牌验证）

## 快速开始

1. 克隆仓库
2. 安装依赖：`pip install -r requirements.txt`
3. 运行服务：`uvicorn chat_app.main:app --reload`
4. 访问 API 文档：http://localhost:8000/docs
5. 使用 WebSocket 客户端连接 `ws://localhost:8000/chat/ws?token=<your_jwt_token>` 进行聊天

## API 端点

- `POST /auth/register` - 注册新用户
- `POST /auth/token` - 登录获取令牌
- `GET /users/me` - 获取当前用户信息（需要认证）
- WebSocket `/chat/ws` - 聊天室连接（需提供 token 查询参数）

运行与测试

启动服务：
```
uvicorn chat_app.main:app --reload
```
注册用户：
使用 /auth/register 端点，POST JSON：
```
{"username":"alice","password":"secret"}
```

登录获取令牌：
POST /auth/token，表单数据：

username: alice
password: secret

返回类似：

{"access_token":"eyJhbGc...","token_type":"bearer"}

WebSocket 连接：
使用 WebSocket 客户端（如浏览器 JavaScript、Postman、wscat）连接到：
```
ws://localhost:8000/chat/ws?token=eyJhbGc...
```
连接成功后，可以发送消息，所有其他在线用户将收到广播。
测试受保护路由：
在请求头中添加 Authorization: Bearer <token> 访问 /users/me。