FastAPI 知识点总结 - 第三部分：高级应用

FastAPI 学习总结 – 第三部分：高级应用

十、数据库操作（SQLAlchemy）

安装依赖

pip install sqlalchemy pymysql

数据库配置

知识点详解：

create_engine() 是 SQLAlchemy 创建数据库引擎的核心函数，它负责：

• 建立与数据库的连接
• 管理连接池
• 生成 SQL 语句

连接池参数详解：

SessionLocal 的作用：

• 每次请求创建新的数据库会话
• autocommit=False：需要手动 commit，保证数据一致性
• autoflush=False：需要手动 flush，避免意外的自动提交

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 数据库连接URL
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@localhost:3306/fastapi_db"

# 创建数据库引擎
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
pool_size=5, # 连接池大小
max_overflow=10, # 最大溢出连接数
pool_timeout=30 # 连接超时时间
)

# 创建会话工厂
SessionLocal = sessionmaker(
autocommit=False, # 不自动提交
autoflush=False, # 不自动刷新
bind=engine # 绑定引擎
)

# 创建基类
Base = declarative_base()

模型定义

知识点详解：

Column 参数详解：

relationship（关系）的本质：

• relationship() 不会在数据库中创建列
• 它是 SQLAlchemy 的"魔法"属性，访问时自动执行 JOIN 查询
• back_populates 和 backref 互为镜像，建立双向关系

一对一 vs 一对多 vs 多对多：

• 一对多：relationship("Item", back_populates="owner")
• 一对一：在一对多基础上加 uselist=False
• 多对多：需要中间表

# app/models.py
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from app.database import Base
from datetime import datetime

class User(Base):
"""用户模型"""
__tablename__ = "users"

id = Column(Integer, primary_key=True, index=True, autoincrement=True)
username = Column(String(50), nullable=False, unique=True, index=True)
email = Column(String(100), nullable=False, unique=True, index=True)
password_hash = Column(String(255), nullable=False)
created_at = Column(DateTime, default=datetime.now)

# 一对多关系：一个用户可以有多个商品
items = relationship("Item", back_populates="owner")

class Item(Base):
"""商品模型"""
__tablename__ = "items"

id = Column(Integer, primary_key=True, index=True)
name = Column(String(100), nullable=False)
description = Column(String(500))
price = Column(Integer, nullable=False)
owner_id = Column(Integer, ForeignKey("users.id")) # 外键建立关联

# 反向关系：通过 item.owner 获取用户
owner = relationship("User", back_populates="items")

创建表

# 创建所有表
Base.metadata.create_all(bind=engine)

CRUD 操作

知识点详解：

会话（Session）生命周期管理：

db.add(user) → 将对象标记为"待添加"（状态: pending）
db.commit() → 提交事务，永久保存到数据库（状态: committed）
db.refresh(user) → 从数据库重新读取最新数据（如自增ID、默认值）
db.delete(user) → 标记为待删除
db.rollback() → 回滚未提交的更改

为什么需要 db.refresh()？

• 自增 ID 是在 commit() 后由数据库生成的
• refresh() 确保获取到数据库生成的值

查询执行时机：

• .all()、.first()、.one() 才会真正执行 SQL
• 只构建查询对象时不执行

创建数据：

def create_user(db, user_data):
"""创建用户"""
user = User(
username=user_data.username,
email=user_data.email,
password_hash=hash_password(user_data.password)
)
db.add(user)
db.commit()
db.refresh(user)
return user

查询数据：

def get_user(db, user_id):
"""根据ID获取用户"""
return db.query(User).filter(User.id == user_id).first()

def get_user_by_email(db, email):
"""根据邮箱获取用户"""
return db.query(User).filter(User.email == email).first()

def get_users(db, skip=0, limit=10):
"""获取用户列表（分页）"""
return db.query(User).offset(skip).limit(limit).all()

更新数据：

def update_user(db, user_id, user_data):
"""更新用户信息"""
user = db.query(User).filter(User.id == user_id).first()
if user:
if user_data.username:
user.username = user_data.username
if user_data.email:
user.email = user_data.email
db.commit()
db.refresh(user)
return user

删除数据：

def delete_user(db, user_id):
"""删除用户"""
user = db.query(User).filter(User.id == user_id).first()
if user:
db.delete(user)
db.commit()
return user

复杂查询

知识点详解：

and_ / or_ 组合条件：

• and_()：所有条件同时满足（相当于 SQL 的 AND）
• or_()：满足任一条件即可（相当于 SQL 的 OR）
• 可以嵌套使用，实现复杂逻辑

常见查询方法对比：

多条件查询：

from sqlalchemy import and_, or_

# 查询年龄大于18且邮箱包含example的用户
users = db.query(User).filter(
and_(
User.age > 18,
User.email.like("%example%")
)
).all()

# 查询用户名是admin或邮箱是admin@example.com的用户
users = db.query(User).filter(
or_(
User.username == "admin",
User.email == "admin@example.com"
)
).all()

排序：

# 按创建时间降序排序（最新在前）
users = db.query(User).order_by(User.created_at.desc()).all()

# 多字段排序：先按时间降序，再按用户名升序
users = db.query(User).order_by(User.created_at.desc(), User.username.asc()).all()

分页：

page = 1
page_size = 10
# 计算偏移量：(page-1) * page_size
users = db.query(User).offset((page – 1) * page_size).limit(page_size).all()
# page=1: offset(0).limit(10) 获取第1-10条
# page=2: offset(10).limit(10) 获取第11-20条

聚合函数：

from sqlalchemy import func

# 统计用户数量
count = db.query(func.count(User.id)).scalar()

# 计算平均年龄
avg_age = db.query(func.avg(User.age)).scalar()

# 分组统计：按性别统计人数
result = db.query(User.gender, func.count(User.id)).group_by(User.gender).all()

十一、安全认证

知识点详解：

JWT（JSON Web Token）认证流程：

用户登录 → 验证密码 → 生成JWT → 返回token → 后续请求携带token → 验证token → 获取用户信息

JWT 的三部分结构：

xxxxx.yyyyy.zzzzz
↑ ↑ ↑
Header Payload Signature

• Header：存储算法类型（如 HS256）
• Payload：存储声明（Claims），如 sub（用户标识）、exp（过期时间）
• Signature：签名，确保 Token 未被篡改

为什么用 sub 作为用户标识？

• sub（subject）是 JWT 标准声明
• 存储用户 email 或 user_id，便于后续通过 sub 查询用户

密码哈希的必要性：

• 不能明文存储密码（数据库泄露 = 密码泄露）
• bcrypt 是"慢哈希"，故意设计慢，抵抗暴力破解
• 同一密码每次哈希结果不同（因为有盐）

OAuth2 密码流

安装依赖：

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

创建安全模块：

# app/security.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from app.crud import get_user_by_email

# 密钥（实际应用中应该从环境变量获取）
SECRET_KEY = "your-secret-key-keep-it-safe"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 密码上下文
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2 scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_password(plain_password, hashed_password):
"""验证密码"""
return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
"""生成密码哈希"""
return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta = 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

async def get_current_user(token: str = Depends(oauth2_scheme), db=Depends(get_db)):
"""获取当前用户"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
email: str = payload.get("sub")
if email is None:
raise credentials_exception
except JWTError:
raise credentials_exception

user = get_user_by_email(db, email=email)
if user is None:
raise credentials_exception
return user

登录接口：

# app/api/auth.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from datetime import timedelta
from app.security import create_access_token, verify_password, ACCESS_TOKEN_EXPIRE_MINUTES
from app.crud import get_user_by_email

router = APIRouter(tags=["认证"])

@router.post("/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db=Depends(get_db)):
"""获取访问令牌"""
user = get_user_by_email(db, email=form_data.username)
if not user or not verify_password(form_data.password, user.password_hash):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)

access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.email}, expires_delta=access_token_expires
)

return {"access_token": access_token, "token_type": "bearer"}

保护接口：

from app.security import get_current_user
from app.models import User

@router.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
"""获取当前用户信息"""
return {"email": current_user.email, "username": current_user.username}

API 密钥认证

知识点详解：

API Key vs JWT：

• API Key：简单，适合服务端到服务端
• JWT：支持过期时间、可携带更多信息、适合跨域

auto_error=False 的作用：

• 默认为 True：请求头缺失时自动抛出 403 错误
• 设置为 False：请求头缺失时返回 None，由我们自己处理逻辑

from fastapi import Security, HTTPException
from fastapi.security import APIKeyHeader

# 从环境变量获取API密钥
API_KEY = "your-api-key"

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

async def get_api_key(api_key: str = Security(api_key_header)):
"""验证API密钥"""
if api_key != API_KEY:
raise HTTPException(status_code=403, detail="Invalid API Key")
return api_key

@app.get("/protected")
async def protected_route(api_key: str = Depends(get_api_key)):
"""受保护的接口"""
return {"message": "This is a protected route"}

十二、中间件

知识点详解：

中间件执行流程：

请求 → 中间件1 → 中间件2 → … → 路由处理 → 中间件2 → 中间件1 → 响应

中间件 vs 依赖注入：

• 中间件：全局影响所有请求，可以修改请求/响应
• 依赖注入：按需使用，只能处理特定逻辑

自定义中间件

from fastapi import FastAPI, Request
import time

app = FastAPI()

@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

CORS 中间件

知识点详解：

CORS（跨域资源共享）原理：

• 浏览器的同源策略限制 Ajax 只能访问同源资源
• CORS 通过响应头告诉浏览器是否允许跨域请求

参数详解：

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 允许的来源
allow_credentials=True, # 允许携带凭证
allow_methods=["*"], # 允许的HTTP方法
allow_headers=["*"], # 允许的请求头
expose_headers=["X-Process-Time"], # 暴露给前端的响应头
max_age=600, # 预检请求缓存时间
)

日志中间件

@app.middleware("http")
async def log_requests(request: Request, call_next):
"""记录请求日志"""
print(f"请求方法: {request.method}")
print(f"请求路径: {request.url}")
print(f"请求头: {dict(request.headers)}")

response = await call_next(request)

print(f"响应状态码: {response.status_code}")
return response

十三、背景任务

知识点详解：

BackgroundTasks vs Celery：

为什么用 background_tasks.add_task() 而不是直接调用？

• 直接调用会阻塞，直到函数执行完毕才返回响应
• add_task() 将任务添加到队列，立即返回响应

基本用法

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_log(message: str):
"""写入日志"""
with open("log.txt", mode="a") as log:
log.write(f"{message}\\n")

def send_email(email: str, message: str):
"""发送邮件（模拟）"""
print(f"Sending email to {email}: {message}")

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
"""发送通知"""
# 添加后台任务
background_tasks.add_task(write_log, f"Notification sent to {email}")
background_tasks.add_task(send_email, email, "Hello from FastAPI!")

return {"message": "Notification sent"}

带依赖的背景任务

知识点详解：

依赖注入与后台任务的冲突：

问题：后台任务在请求结束后执行，但 get_db 依赖在请求结束时关闭连接。

解决方案：需要将 db 对象传递给后台任务，而不是依赖注入的 db。

def get_db():
"""获取数据库连接"""
db = SessionLocal()
try:
yield db
finally:
db.close()

def process_data(db, data_id: int):
"""处理数据"""
# 执行耗时操作
item = db.query(Item).filter(Item.id == data_id).first()
if item:
item.processed = True
db.commit()

@app.post("/process/{data_id}")
async def process_item(
data_id: int,
background_tasks: BackgroundTasks,
db=Depends(get_db)
):
"""处理数据（后台任务）"""
background_tasks.add_task(process_data, db, data_id)
return {"message": "Processing started"}

十四、静态文件服务

知识点详解：

app.mount() vs app.add_api_route()：

• mount()：挂载整个 ASGI 应用（如 StaticFiles），处理所有路径
• add_api_route()：添加单个路由

为什么叫 “mount”？

• 类似 Linux 挂载点，将某个路径绑定到指定应用
• /static 下的所有请求都由 StaticFiles 处理

基本配置

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# 挂载静态文件目录
app.mount("/static", StaticFiles(directory="static"), name="static")

目录结构：

project/
├── main.py
└── static/
├── images/
│ └── logo.png
├── css/
│ └── style.css
└── js/
└── app.js

访问示例：

• http://localhost:8000/static/images/logo.png
• http://localhost:8000/static/css/style.css

HTML 模板

知识点详解：

为什么 TemplateResponse 需要 request 对象？

• Jinja2 模板可能需要生成 URL（如 url_for()）
• 传入 request 才能使用 url_for() 和 session 等

Jinja2 模板变量：

• {{ variable }}：输出变量值
• {% if %} … {% endif %}：条件判断
• {% for %} … {% endfor %}：循环

from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates

app = FastAPI()
templates = Jinja2Templates(directory="templates")

@app.get("/")
async def read_root(request: Request):
"""渲染首页"""
return templates.TemplateResponse("index.html", {"request": request, "name": "FastAPI"})

模板文件 templates/index.html：

<!DOCTYPE html>
<html>
<head>
<title>FastAPI</title>
<link rel="stylesheet" href="/static/css/style.css">
</head>
<body>
<h1>Hello, {{ name }}!</h1>
<img src="/static/images/logo.png" alt="Logo">
</body>
</html>

十五、异步支持

知识点详解：

同步 vs 异步的本质区别：

什么时候用 async/await：

• I/O 操作：数据库查询、HTTP 请求、文件读写
• 不需要 CPU 一直计算

什么时候用同步（不用 async）：

• CPU 密集型：图片处理、数据计算
• 如果用了 async 但没有 await，代码会立即返回，没有意义

异步路径操作

from fastapi import FastAPI

app = FastAPI()

@app.get("/items")
async def read_items():
"""异步获取商品列表"""
# 模拟异步操作
items = await fetch_items_from_database()
return items

async def fetch_items_from_database():
"""从数据库异步获取数据"""
# 实际应用中使用异步数据库驱动
return [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}]

异步依赖

知识点详解：

同步 vs 异步依赖注入：

async def get_db():
"""异步获取数据库连接"""
db = AsyncSessionLocal()
try:
yield db
finally:
await db.close()

@app.get("/users")
async def read_users(db=Depends(get_db)):
"""异步获取用户列表"""
users = await db.execute(select(User))
return users.scalars().all()

异步 HTTP 请求

知识点详解：

为什么用 httpx 而不是 requests？

• requests：同步库，会阻塞
• httpx：支持异步，完美配合 FastAPI

async with 的作用：

• 自动管理资源（连接）
• 进入时创建，退出时关闭

import httpx

@app.get("/fetch-data")
async def fetch_external_data():
"""异步获取外部API数据"""
async with httpx.AsyncClient() as client:
response = await client.get("https://api.example.com/data")
return response.json()

十六、测试

知识点详解：

为什么用 pytest？

• 简洁的语法：assert 而不是 self.assertEqual()
• 丰富的 fixture 支持
• 强大的参数化功能

TestClient 的注意事项：

• TestClient 是同步的，即使路由是 async def
• 如果想测试真正的异步，使用 httpx.AsyncClient

测试金字塔：

/\\
/ \\ E2E 测试（少量）
/—-\\
/ \\ 集成测试（中等）
/——–\\
/ \\ 单元测试（大量）
/____________\\

使用 pytest

安装依赖：

pip install pytest httpx

测试文件 tests/test_main.py：

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_root():
"""测试根路径"""
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello World"}

def test_create_user():
"""测试创建用户"""
response = client.post(
"/users",
json={"username": "testuser", "email": "test@example.com", "password": "123456"}
)
assert response.status_code == 200
data = response.json()
assert data["username"] == "testuser"
assert data["email"] == "test@example.com"
assert "password" not in data # 确保密码不在响应中

def test_read_items():
"""测试获取商品列表"""
response = client.get("/items?skip=0&limit=10")
assert response.status_code == 200
assert isinstance(response.json(), list)

运行测试：

pytest tests/ -v

测试认证接口

知识点详解：

Bearer Token 认证流程：

def test_login():
"""测试登录"""
# 先创建用户
client.post(
"/users",
json={"username": "admin", "email": "admin@example.com", "password": "secret"}
)

# 测试登录
response = client.post(
"/token",
data={"username": "admin@example.com", "password": "secret"}
)
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert data["token_type"] == "bearer"

def test_protected_route():
"""测试受保护的接口"""
# 获取token
response = client.post(
"/token",
data={"username": "admin@example.com", "password": "secret"}
)
token = response.json()["access_token"]

# 访问受保护的接口
response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
assert response.json()["email"] == "admin@example.com"

十七、项目结构

知识点详解：

分层架构的优点：

各层职责详解：

推荐结构

project/
├── main.py # 应用入口
├── requirements.txt # 依赖配置
├── .env # 环境变量
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI实例配置
│ ├── api/ # 路由接口
│ │ ├── __init__.py
│ │ ├── users.py # 用户相关接口
│ │ ├── items.py # 商品相关接口
│ │ └── auth.py # 认证接口
│ ├── models/ # 数据库模型
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── item.py
│ ├── schemas/ # Pydantic模型
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── item.py
│ ├── crud/ # 数据操作
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── item.py
│ ├── security.py # 安全相关
│ └── database.py # 数据库配置
├── tests/ # 测试文件
│ ├── __init__.py
│ ├── test_users.py
│ └── test_items.py
└── static/ # 静态文件
├── css/
├── js/
└── images/

模块职责说明

十八、部署

知识点详解：

Uvicorn vs Gunicorn：

进程数计算公式：

• CPU 密集型：2 * CPU 核心数 + 1
• I/O 密集型：4 * CPU 核心数 + 1

使用 Uvicorn

# 开发模式（代码变更自动重载）
uvicorn main:app –host 0.0.0.0 –port 8000 –reload

# 生产模式
uvicorn main:app –host 0.0.0.0 –port 8000

使用 Gunicorn + Uvicorn

知识点详解：

为什么生产环境用 Gunicorn？

• 自动管理多个 Worker 进程
• 处理 Worker 崩溃重启
• 更稳定的进程管理

# 安装依赖
pip install gunicorn

# 启动服务（4个worker进程）
gunicorn main:app –workers 4 –worker-class uvicorn.workers.UvicornWorker –bind 0.0.0.0:8000

使用 Docker

知识点详解：

Docker 部署的优势：

• 环境一致性：开发、测试、生产环境相同
• 隔离性：应用之间互不影响
• 快速部署：一键启动

镜像选择建议：

• python:3.11-slim：生产推荐，体积小
• python:3.11：体积大，包含完整工具

Dockerfile：

FROM python:3.11-slim

WORKDIR /app

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

COPY . .

CMD ["uvicorn", "main:app", "–host", "0.0.0.0", "–port", "8000"]

构建和运行：

# 构建镜像
docker build -t fastapi-app .

# 运行容器
docker run -p 8000:8000 fastapi-app

使用 Nginx 反向代理

知识点详解：

为什么用 Nginx？

用户请求 → Nginx (80/443) → FastAPI (8000)

• SSL/TLS 终结：HTTPS 加密在 Nginx 处理
• 静态文件服务：Nginx 比 Python 更高效
• 负载均衡：多 FastAPI 实例分担压力
• 请求限流：保护后端服务

关键配置解释：

Nginx配置文件 /etc/nginx/sites-available/fastapi：

server {
listen 80;
server_name example.com;

location / {
proxy_pass http://127.0.0.1:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
}

总结

FastAPI 核心优势回顾：

适合场景：

• RESTful API 开发
• 微服务架构
• 实时应用（WebSocket）
• 数据处理后端

学习路径建议：

第一阶段：基础
├── 路由定义（@app.get/post/put/delete）
├── 参数获取（路径、查询、请求体）
└── Pydantic 数据验证

第二阶段：数据库
├── SQLAlchemy 模型定义
├── CRUD 操作
└── 关系映射（一对多、多对多）

第三阶段：安全
├── JWT 认证流程
├── OAuth2 密码流
└── 密码哈希（bcrypt）

第四阶段：高级
├── 异步编程（async/await）
├── 中间件
├── 背景任务
└── 测试（pytest）

第五阶段：部署
├── Uvicorn/Gunicorn
├── Docker 容器化
└── Nginx 反向代理

快速上手建议：