= self.max_requests:
return JSONResponse(
status_code=429,
content={"detail": "请求过于频繁,请稍后再试"},
headers={"Retry-After": str(self.window)},
)
self.buckets[client_ip].append(now)
return await call_next(request)
```
注册使用:
```python
app.add_middleware(RateLimitMiddleware, max_requests=20, window=1)
```
### 纯 ASGI 中间件:更高粒度控制
如果需要完全控制 ASGI 事件(如处理 WebSocket),可以直接写纯 ASGI 中间件:
```python
from starlette.types import ASGIApp, Scope, Receive, Send
class CustomASGIMiddleware:
def __init__(self, app: ASGIApp):
self.app = app
async def __call__(self, scope: Scope, receive: Receive, send: Send):
if scope["type"] == "http":
# 仅在 HTTP 请求时干预
print(f"ASGI 中间件:{scope['method']} {scope['path']}")
await self.app(scope, receive, send)
```
## 中间件的顺序问题
中间件的执行顺序与 `add_middleware` 的调用顺序**相同**——先添加的在外层,先拦截请求、最后放行响应。
```python
# 洋葱模型:A 在最外层
app.add_middleware(MiddlewareA) # 第 1 层(最外层)
app.add_middleware(MiddlewareB) # 第 2 层
app.add_middleware(MiddlewareC) # 第 3 层(最内层,最靠近路由)
```
请求链路:`A → B → C → 路由`,响应链路:`路由 → C → B → A`。
**最佳实践排序**:
1. **最外层**:日志、耗时统计(尽量覆盖所有请求)
2. **次外层**:CORS、安全头
3. **次内层**:认证鉴权
4. **最内层**:请求数据预处理、响应格式化
## 实际场景:响应头注入
一个常见的需求是为所有响应统一注入安全头:
```python
@app.middleware("http")
async def security_headers_middleware(request: Request, call_next):
response = await call_next(request)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-XSS-Protection"] = "1; mode=block"
response.headers["Strict-Transport-Security"] = "max-age=31536000"
return response
```
## 中间件 vs 依赖注入
很多初学者会困惑:都是请求前执行的逻辑,中间件和 Depends 依赖注入有何区别?
| 维度 | 中间件 | Depends 依赖注入 |
|------|--------|-----------------|
| **作用范围** | 全局,覆盖所有路由 | 可按路由、路由分组、全局配置 |
| **执行时机** | 在路由匹配之前 | 路由匹配之后 |
| **请求信息** | 仅有 Request 对象 | 有解析后的路径/查询/请求体参数 |
| **修改响应** | 可以直接修改 Response | 不能直接修改 Response |
| **适用场景** | 日志、CORS、安全头、限流 | 鉴权、数据库会话、参数校验 |
一句话总结:**影响"所有请求"用中间件,影响"特定业务逻辑"用 Depends**。
## 小结
FastAPI 的中间件系统继承自 Starlette,简洁而强大。日常开发中:
- **优先级最高**的中间件是 CORS,前后端分离几乎必配
- 函数式 `@app.middleware("http")` 覆盖 80% 的场景
- 需要参数化或复杂逻辑时,使用类式 `BaseHTTPMiddleware`
- 牢记洋葱模型,合理安排中间件添加顺序
掌握中间件,你就掌握了 FastAPI 请求生命周期的"总开关"。
">
详解 FastAPI 中间件机制,涵盖内置中间件(CORS、HTTPSRedirect、TrustedHost)、自定义中间件的编写、请求/响应拦截、中间件执行顺序、以及实际应用场景(请求日志、耗时统计、请求限流)。
同一个请求内,对同一个依赖 FastAPI 默认会**缓存**结果。也就是说,若一个请求中多个地方依赖了同一个函数,它只会被执行一次,后续直接返回缓存值。
---
## 类作为依赖
除了函数,**任何可调用对象**(类、实例)都可以作为依赖。用类来定义依赖,结构更清晰:
```python
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/")
def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}
```
`Depends(CommonQueryParams)` 中传的参数是类本身(可调用),FastAPI 会实例化它,并把构造函数参数识别为查询参数。
### 简写形式
当依赖「类型」和「可调用对象」相同时,可以简写:
```python
# 这三种写法等价
def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)): ...
def read_items(commons: CommonQueryParams = Depends()): ...
def read_items(commons: CommonQueryParams = Depends()): ...
```
`Depends()` 不传参时,FastAPI 会自动用参数的类型注解(这里是 `CommonQueryParams`)作为依赖。
---
## 子依赖(嵌套依赖)
依赖可以层层嵌套,FastAPI 会按顺序解析整个依赖树:
```python
def query_extractor(q: str | None = None):
return q
def query_checker(q: str = Depends(query_extractor)):
if q == "admin":
raise ValueError("禁止使用保留关键字")
return q
@app.get("/search/")
def search(result: str = Depends(query_checker)):
return {"q": result}
```
执行顺序:`query_extractor` → `query_checker` → `search`。如果某一层依赖抛出异常,请求会立即中断并返回对应错误。
---
## 全局依赖与路由组依赖
有些依赖(如鉴权、CORS 校验)需要应用到整个应用或一组路由上,无需在每个函数里写 `Depends`。
### 应用级依赖
```python
async def verify_token(x_token: str = Header(...)):
if x_token != "secret-token":
raise HTTPException(status_code=400, detail="X-Token 错误")
app = FastAPI(dependencies=[Depends(verify_token)])
```
这样**所有路由**都会先执行 `verify_token`,但它的返回值不会传入路由函数。
### APIRouter 级依赖
```python
from fastapi import APIRouter
admin_router = APIRouter(dependencies=[Depends(verify_token)])
@admin_router.get("/dashboard/")
def dashboard():
return {"data": "仅管理员可见"}
app.include_router(admin_router, prefix="/admin")
```
### 单个路由的专属依赖
```python
@app.get("/items/", dependencies=[Depends(verify_token)])
def read_items():
return {"items": []}
```
`dependencies` 列表中的依赖会执行,但结果不注入函数——适合用于「拦截 / 校验」类需求。
---
## 使用 yield 的依赖(资源管理)
类似上下文管理器,依赖函数可以使用 `yield` 来在请求结束后执行清理逻辑:
```python
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users/")
def list_users(db = Depends(get_db)):
# 使用 db 查询
return db.execute("SELECT * FROM users").fetchall()
```
执行流程:
1. `get_db` 执行到 `yield db`,把 `db` 注入路由函数;
2. 路由函数执行完毕并返回响应;
3. FastAPI 继续执行 `yield` 之后的 `finally` 代码,关闭连接。
这种方式非常适合管理数据库会话、Redis 连接、文件句柄等资源。也可以配合 `with` 语句使用:
```python
def get_file():
with open("data.txt") as f:
yield f
```
---
## 实战示例:分页参数 + 鉴权
把依赖注入用到真实业务中。下面演示「分页依赖」和「当前用户依赖」的组合:
```python
from fastapi import FastAPI, Depends, HTTPException, Header
app = FastAPI()
class Pagination:
def __init__(self, page: int = 1, size: int = 10):
self.page = page
self.size = size
def get_current_user(authorization: str = Header(...)):
if not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="无效的认证信息")
token = authorization[7:]
user = verify_jwt(token) # 假设的解密函数
if not user:
raise HTTPException(status_code=401, detail="用户不存在")
return user
@app.get("/posts/")
def list_posts(
pager: Pagination = Depends(Pagination),
current_user: dict = Depends(get_current_user),
):
return {
"page": pager.page,
"size": pager.size,
"user": current_user["username"],
}
```
一个路由函数同时依赖两个组件:`Pagination` 负责解析分页参数,`get_current_user` 负责鉴权。两者互不耦合,各自独立可测试、可复用。
---
## 小结
| 用法 | 关键写法 | 适用场景 |
|------|---------|---------|
| 函数依赖 | `Depends(func)` | 抽取公共参数 / 逻辑 |
| 类依赖 | `Depends(Class)` 或 `Depends()` | 结构化参数对象 |
| 子依赖 | 依赖内部再调用 `Depends` | 多层校验、拆分逻辑 |
| 全局依赖 | `FastAPI(dependencies=[...])` | 全局拦截(鉴权、日志) |
| 路由组依赖 | `APIRouter(dependencies=[...])` | 一组接口共享前置逻辑 |
| yield 依赖 | 函数中 `yield` | 数据库会话等资源管理 |
依赖注入是 FastAPI 区别于其他框架的核心特性之一。善用 `Depends`,可以让接口代码保持「只关心业务本身」,把参数解析、权限校验、资源管理等工作交给可复用的依赖组件,显著提升代码的可维护性和可测试性。
">
详解 FastAPI 中的依赖注入机制,包括 Depends 基本用法、类作为依赖、嵌套依赖、全局依赖、yield 依赖与资源释放、路径操作专属依赖等核心知识点。
str:
if not v.isalnum():
raise ValueError('用户名只能包含字母和数字')
return v
@validator('confirm_password')
def passwords_match(cls, v: str, values: dict) -> str:
if 'password' in values and v != values['password']:
raise ValueError('两次输入的密码不一致')
return v
```
`values` 参数包含前面已验证过的字段值,因此可以在 `confirm_password` 的验证器中拿到 `password` 进行比较。
> **注意**:Pydantic V2 中 `@validator` 已改为 `@field_validator`,用法类似但参数名略有不同。
## 结合路径参数与查询参数
请求体、路径参数、查询参数可以同时存在,FastAPI 根据参数声明自动区分:
```python
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class ItemUpdate(BaseModel):
name: str = Field(..., min_length=1)
price: float = Field(..., gt=0)
is_offer: bool = False
@app.put("/items/{item_id}")
async def update_item(
item_id: int, # 路径参数
item: ItemUpdate, # 请求体(Pydantic 模型 = body)
force: bool = False, # 查询参数
):
return {
"item_id": item_id,
"force": force,
"update": item.dict()
}
```
识别规则很简单:
- 在路径中也声明的 → **路径参数**
- 是 Pydantic 模型类型的 → **请求体**
- 是简单类型(str/int/float/bool)且不在路径中 → **查询参数**
## 请求体中的额外字段
默认情况下,如果客户端多传了模型中未定义的字段,Pydantic V1 会忽略它们。如果需要严格模式(拒绝多余字段),可以在模型内部配置:
```python
from pydantic import BaseModel
class StrictItem(BaseModel):
name: str
price: float
class Config:
extra = "forbid" # 禁止额外字段
```
此时多传字段会直接返回 422 错误。
## 小结
FastAPI 结合 Pydantic 的请求体处理机制让数据校验变得极其简洁:
- 用 `BaseModel` 定义数据结构,类型即校验规则
- 用 `Field` 添加约束(长度、范围、正则)
- 用嵌套模型处理复杂 JSON 结构
- 用 `@validator` 处理跨字段校验和自定义逻辑
这套机制让开发者无需手写校验代码,同时自动产出清晰的 API 文档,显著提升开发效率和数据质量。
">
详解 FastAPI 中请求体的使用方式,Pydantic 模型的字段类型、校验规则、嵌套模型、自定义验证器以及结合路径参数和查询参数的完整实战。
= 1
le=10000, # <= 10000
)
):
return {"article_id": article_id}
```
`Path` 常用校验参数:
| 参数 | 说明 |
|------|------|
| `...` | 必填(省略号) |
| `ge` / `gt` | 大于等于 / 大于 |
| `le` / `lt` | 小于等于 / 小于 |
| `min_length` / `max_length` | 字符串最小/最大长度 |
| `regex` | 正则匹配(FastAPI 旧版) |
| `pattern` | 正则匹配(Pydantic v2) |
| `title` / `description` | 文档展示用 |
### 枚举类型的路径参数
```python
from enum import Enum
class CategoryEnum(str, Enum):
tech = "tech"
life = "life"
travel = "travel"
@app.get("/posts/{category}")
async def get_posts(category: CategoryEnum):
# 自动生成文档,且只接受这三个值
return {"category": category}
```
访问 `/posts/tech` 正常返回;访问 `/posts/music` 返回 422。
---
## 查询参数
查询参数是 URL 中 `?` 后面的键值对,如 `?page=1&size=20`。FastAPI 中,**函数参数不在路径中且不是 `Body` 类型,就会被自动识别为查询参数**。
### 基本用法
```python
@app.get("/posts")
async def list_posts(page: int = 1, size: int = 10):
return {"page": page, "size": size}
```
访问 `/posts?page=2&size=5`,`page=2, size=5`;不传参数则使用默认值。
### 可选查询参数
```python
from typing import Optional
@app.get("/search")
async def search(
keyword: Optional[str] = None,
category: Optional[str] = None,
):
result = {"keyword": keyword, "category": category}
if keyword:
result["match"] = f"搜索: {keyword}"
return result
```
`Optional[str] = None` 表示该参数可选,不传时为 `None`。
### 使用 Query 做额外校验
```python
from fastapi import Query
@app.get("/items")
async def list_items(
offset: int = Query(default=0, ge=0, description="偏移量"),
limit: int = Query(default=10, ge=1, le=100, description="每页数量"),
sort: Optional[str] = Query(default=None, pattern=r"^(asc|desc)$", description="排序方式"),
):
return {"offset": offset, "limit": limit, "sort": sort}
```
`Query` 支持与 `Path` 几乎相同的校验参数。
### 接收多个同名查询参数
```python
from typing import List
@app.get("/filter")
async def filter_items(tags: List[str] = Query(default=[])):
"""
示例: /filter?tags=python&tags=fastapi&tags=web
结果: tags = ["python", "fastapi", "web"]
"""
return {"tags": tags}
```
### 布尔类型查询参数
```python
@app.get("/products")
async def list_products(in_stock: bool = False):
"""
访问: /products?in_stock=true → True
/products?in_stock=1 → True
/products?in_stock=yes → True
/products?in_stock=on → True
/products → False(默认值)
"""
return {"in_stock": in_stock}
```
---
## 路径参数与查询参数的组合
最常见的场景是路径参数标识资源,查询参数做筛选:
```python
@app.get("/users/{user_id}/posts")
async def get_user_posts(
user_id: int = Path(..., ge=1),
page: int = Query(default=1, ge=1),
size: int = Query(default=20, ge=1, le=100),
status: Optional[str] = Query(default=None, pattern=r"^(draft|published|archived)$"),
):
"""
获取某个用户的文章列表
- **user_id**: 用户ID(路径参数)
- **page**: 页码(查询参数,默认第1页)
- **size**: 每页数量(查询参数,默认20,最大100)
- **status**: 文章状态过滤(可选)
"""
return {
"user_id": user_id,
"page": page,
"size": size,
"status": status,
}
```
请求示例:`/users/42/posts?page=2&size=5&status=published`
---
## 小结
| 特性 | 路径参数 | 查询参数 |
|------|---------|---------|
| 位置 | URL 路径中 `/users/{id}` | URL `?` 后面 `?key=val` |
| 用途 | 标识资源 | 过滤、排序、分页 |
| 必填 | 默认必填 | 默认可选 |
| 校验工具 | `Path()` | `Query()` |
| 类型 | str / int / UUID / Enum | str / int / bool / List 等 |
掌握路径参数和查询参数是 FastAPI 开发的基础,配合 `Path` 和 `Query` 的校验能力,可以在不写额外逻辑的情况下完成大部分参数校验工作,并自动生成清晰的 API 文档。
">
详解 FastAPI 中的路径参数和查询参数,包括类型校验、默认值、可选参数、枚举约束、参数校验以及两者的组合使用。
删除字符串中的空格
```python
s = " 123abc "
# 删除开头的空格
print(s.lstrip()) # "123 abc "
# 删除结尾的空格
print(s.rstrip()) # " 123 abc"
# 删除开头和结尾的空格
print(s.strip()) # "123 abc"
# 删除字符串中所有的空格
print(s.replace(' ', '')) # "123abc"
```
### 删除字符串中的所有符号,只保留数字和英文字母
```python
import re
s = "123,abc .?/&?》^_^dddA。"
# 把所有编码非\u0030-\u0039(数字)、\u0041-\u007a(英文字母)的字符替换为空字符串
rs = re.sub("([^\u0030-\u0039\u0041-\u007a])", '', s)
print(rs) # "123abcdddA"
```
### 只字符串中的保留汉字
```python
import re
s = "我爱中国🇨🇳,I love China。"
# 把所有编码非\u4e00-\u9fa5(汉字)的字符替换为空字符串
rs = re.sub("([^\u4e00-\u9fa5])", '', s)
print(rs) # "我爱中国"
```
### 对应的unicode编码范围
|说明|unicode范围|
|--|--|
|数字|\u0030-\u0039|
|汉字|\u4e00-\u9fa5|
|大写字母|\u0041-\u005a|
|小写字母|\u0061-\u007a|
|英文字母|\u0041-\u007a|
|韩文|\uAC00-\uD7AF|
|日文|\u3040-\u31FF|">
Python删除字符串中的指定符号、空格、保留指定类型字符。