djangoadmin/djangoadmin-wiki
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化更新

Browse Source
This commit is contained in:
gitadmin
2026-09-14 10:33:29 +08:00
parent e8b5d55603
commit 11c46addfa
61 changed files with 1025 additions and 896 deletions
Download Patch File Download Diff File Expand all files Collapse all files
DjangoAdmin_FastAPI_NaiveVue/zh/volume3/base-classes.md
+20 -22
View File
@@ -163,27 +163,25 @@ def __init_subclass__(cls, **kwargs):
### 序列化方法
```python
# ============================================================
# 子类钩子：自动将 __table_comment__ 写入 __table_args__
# ============================================================
def __init_subclass__(cls, **kwargs):
super().__init_subclass__(**kwargs)
comment = getattr(cls, '__table_comment__', None)
if not comment:
return
# 仅对有 __tablename__ 的具体模型操作（跳过纯抽象中间层）
if not getattr(cls, '__tablename__', None):
return
existing = getattr(cls, '__table_args__', {})
if isinstance(existing, dict):
merged = {**existing}
elif isinstance(existing, tuple) and existing:
merged = dict(existing[-1]) if isinstance(existing[-1], dict) else {}
else:
merged = {}
if 'comment' not in merged:
merged['comment'] = comment
cls.__table_args__ = merged
# src/core/base_model.py
def to_dict(self) -> Dict[str, Any]:
"""将模型实例转换为字典"""
result = {}
for column in self.__table__.columns:
if column.name in SERIALIZE_EXCLUDE_FIELDS:
continue
value = getattr(self, column.name)
if isinstance(value, datetime):
if value.tzinfo is None:
# MySQL DATETIME 无时区信息，读回为 naive UTC，统一按 UTC 转本地
value = value.replace(tzinfo=timezone.utc)
value = value.astimezone().strftime('%Y-%m-%d %H:%M:%S')
elif isinstance(value, date):
value = value.strftime('%Y-%m-%d')
# 字段名转驼峰
camel_key = convert_camel_case(column.name)
result[camel_key] = value
return result
```
关键设计点：
@@ -416,7 +414,7 @@ Pydantic Schema 可能用 camelCase（如 `parentId`），而模型列是 snake_
| 方法 | 说明 | 返回值 |
|:---|:---|:---|
| `get_page(request)` | 分页查询 | `R.page(records, total, size, current, pages)` |
| `get_page(request)` | 分页查询 | `R.page(data, total, current, size)` |
| `get_list(request)` | 不分页查询 | `R.ok(data=[...])` |
| `get_detail(obj_id)` | 详情查询 | `dict` 或 `None` |
| `add(request, data)` | 新增记录 | `R.ok(msg='添加成功')` |
DjangoAdmin_FastAPI_NaiveVue/zh/volume3/cache.md
+128 -85
View File
@@ -46,107 +46,142 @@
| **容量限制** | 受进程内存限制 | 受 Redis 内存限制 |
| **适用场景** | 高频读取、低变更数据 | 跨进程共享、分布式场景 |
## L1 缓存：进程内字典缓存
## 进程内字典缓存
### 设计决策
`dict_util` 采用**纯进程内 TTL 缓存**（无 Redis L2），原因：
1. 字典数据几乎不变，读写均落在同步 service 路径，无法使用请求级异步 Redis
2. 字典/字典项增删改时由 service 主动失效，TTL 仅作兜底
3. 多 worker 部署下各进程缓存最多在 TTL 内短暂不一致，属可接受范围
### 实现原理
`dict_util` 模块实现了带 TTL 的进程内字典缓存。数据首次查询时从 Redis 加载到进程内存，后续查询直接从内存读取，TTL 过期后自动重新加载。
```python
# src/utils/dict_util.py
import logging
import time
import json
from typing import Dict
# 进程内缓存：{cache_key: (data, expire_timestamp)}
_cache: Dict[str, tuple] = {}
from modules.dictionary.dict.models import Dict
from modules.dictionary.dict_item.models import DictItem
from modules.dictionary.dict.repository import dict_repo
from modules.dictionary.dict_item.repository import dictitem_repo
CACHE_TTL = 300 # 默认 5 分钟
logger = logging.getLogger(__name__)
# 字典缓存 TTL（秒）：兜底过期时间，实际变更靠主动失效即时生效
_CACHE_TTL = 60
# 进程内字典缓存：{code: (过期时间戳, 字典项列表)}
_dict_cache: dict = {}
def get_dict_map(dict_code: str) -> Dict[str, str]:
"""获取数据字典的 {value: label} 映射（进程内缓存 + Redis 二级缓存）"""
cache_key = f"dict:{dict_code}"
def get_dict_items(code):
"""按字典编码读取字典项列表 [{label, value, sort}, ...]，字典不存在返回空列表"""
now = time.time()
# L1: 进程内缓存命中
if cache_key in _cache:
data, expire_at = _cache[cache_key]
if now < expire_at:
return data
del _cache[cache_key]
# L2: Redis 缓存
import asyncio
from core.redis_client import get_redis
redis = get_redis()
cached = asyncio.get_event_loop().run_until_complete(redis.get(cache_key))
if cached:
data = json.loads(cached)
_cache[cache_key] = (data, now + CACHE_TTL)
return data
# 数据库查询（L3 回源）
from modules.dictionary.dict_item.repository import dict_item_repo
items = dict_item_repo.get_all(dict_code=dict_code)
data = {str(item.value): item.label for item in items}
# 写回两级缓存
_cache[cache_key] = (data, now + CACHE_TTL)
asyncio.get_event_loop().run_until_complete(
redis.set(cache_key, json.dumps(data, ensure_ascii=False), ex=CACHE_TTL)
)
return data
hit = _dict_cache.get(code)
if hit and hit[0] > now:
return hit[1]
items = _load_dict_items(code)
_dict_cache[code] = (now + _CACHE_TTL, items)
return items
def invalidate_dict_cache(dict_code: str):
"""失效指定字典的缓存（L1 + L2）"""
cache_key = f"dict:{dict_code}"
def get_dict_map(code):
"""按字典编码读取 {value: label} 映射（value 为字典项存储值，字符串类型）"""
return {item['value']: item['label'] for item in get_dict_items(code)}
# 清除 L1（进程内缓存）
_cache.pop(cache_key, None)
# 清除 L2（Redis 缓存）
from core.redis_client import get_redis
import asyncio
redis = get_redis()
asyncio.get_event_loop().run_until_complete(redis.delete(cache_key))
def get_dict_label(code, value, default=None):
"""取单个字典项名称：value 自动转字符串匹配存储值，查不到返回 default"""
return get_dict_map(code).get(str(value), default)
def invalidate_dict(code):
"""清除指定字典编码的缓存（None/未缓存时安全无操作）"""
if code:
_dict_cache.pop(code, None)
def invalidate_dicts_by_ids(dict_ids):
"""按字典ID批量清除缓存：字典项变更后据 dict_id 反查 code 再失效"""
if not dict_ids:
return
try:
rows = dict_repo.filter(Dict.id.in_(dict_ids), Dict.is_delete == 0).all()
for row in rows:
invalidate_dict(row.code)
except Exception as e:
logger.warning("字典缓存失效失败（将按 TTL 自动过期）: %s", e)
def _load_dict_items(code):
"""从数据库读取字典项（未删除、按 sort+id 升序），字典不存在返回空列表"""
dict_obj = dict_repo.get_one(code=code)
if not dict_obj:
return []
rows = dictitem_repo.filter_by(dict_id=dict_obj.id).all()
rows.sort(key=lambda item: (item.sort, item.id))
return [{'label': item.name, 'value': item.value, 'sort': item.sort} for item in rows]
```
### 公开接口
| 函数 | 签名 | 说明 |
|:---|:---|:---|
| `get_dict_items` | `(code) → [{label, value, sort}]` | 字典项列表（供前端下拉） |
| `get_dict_map` | `(code) → {value: label}` | 值→显示名映射（供后端翻译） |
| `get_dict_label` | `(code, value, default) → str` | 单值翻译 |
| `invalidate_dict` | `(code) → None` | 按字典编码清除缓存 |
| `invalidate_dicts_by_ids` | `(dict_ids) → None` | 按字典 ID 批量清除缓存 |
### 使用场景
| 场景 | 调用方式 | 说明 |
|:---|:---|:---|
| 枚举显示名 | `get_dict_map('status_type')` | 序列化时将值转为显示名 |
| 表单下拉 | `get_dict_map('gender')` | 前端获取字典项列表 |
| 表单下拉 | `get_dict_items('gender')` | 前端获取字典项列表 |
| 单值翻译 | `get_dict_label('status', 1)` | 将 1 转为 "正常" |
| 业务校验 | `value in get_dict_map('xxx')` | 校验值是否在字典范围内 |
### 缓存失效
字典数据变更时（新增/编辑/删除字典项），主动清除两级缓存：
字典数据变更时（新增/编辑/删除字典项），通过 `invalidate_dicts_by_ids` 按字典 ID 批量清除：
```python
# src/modules/dictionary/dict_item/service.py
class DictItemService(BaseService[DictItem]):
def add(self, request, data) -> R:
repo = dictitem_repo
model = DictItem
def add(self, request, data):
"""新增字典项，新增后失效所属字典的缓存"""
result = super().add(request, data)
if result.code == 0:
invalidate_dict_cache(data.dict_code) # 新增成功后失效缓存
if data.dictId:
invalidate_dicts_by_ids({data.dictId})
return result
def update(self, request, data) -> R:
def update(self, request, data):
"""更新字典项，更新后失效新旧所属字典的缓存"""
old = self.repo.get_by_id(data.id) if data.id else None
result = super().update(request, data)
if result.code == 0:
invalidate_dict_cache(data.dict_code) # 更新成功后失效缓存
dict_ids = {data.dictId} if data.dictId else set()
if old and old.dict_id:
dict_ids.add(old.dict_id)
invalidate_dicts_by_ids(dict_ids)
return result
def delete(self, ids) -> R:
# 删除前先获取 dict_code（删除后无法从 data 中获取）
item = self.repo.get_by_id(ids)
dict_code = item.dict_code if item else None
def delete(self, ids):
"""删除字典项，删除后失效所属字典的缓存"""
from utils.request import parse_id_list
id_list = parse_id_list(str(ids))
dict_ids = set()
if id_list:
for row in dictitem_repo.filter(DictItem.id.in_(id_list), DictItem.is_delete == 0).all():
if row.dict_id:
dict_ids.add(row.dict_id)
result = super().delete(ids)
if result.code == 0 and dict_code:
invalidate_dict_cache(dict_code) # 删除成功后失效缓存
invalidate_dicts_by_ids(dict_ids)
return result
```
@@ -178,11 +213,15 @@ def get_redis() -> Redis:
| **JWT 黑名单** | String | `jwt:blacklist:{token_hash}` | Token 剩余有效期 | 登出时将 Token 加入黑名单 |
| **登录失败锁** | String | `login:lock:{username}` | 600 秒 | 连续 5 次失败后锁定 |
| **登录失败计数** | String (Lua) | `login:fail:{username}` | 300 秒 | 原子递增 + 首次设置 TTL |
| **滑动窗口限流** | String (Lua) | `rate:{ip}:{window_index}` | 2 倍窗口时长 | 滑动窗口计数器 |
| **数据字典缓存** | String (JSON) | `dict:{dict_code}` | 300 秒 | 字典项 {value: label} 映射 |
| **权限列表缓存** | String (JSON) | `perm:user:{user_id}` | 变更时失效 | 用户权限字符串列表 |
| **在线用户** | String | `online:user:{user_id}` | Token 有效期 | 在线用户 Token 存储 |
| **验证码** | String | `captcha:{uuid}` | 120 秒 | 图形验证码文本 |
| **滑动窗口限流** | String (Lua) | `rl:{ip}:{window_index}` | 2 倍窗口时长 | 滑动窗口计数器（前缀可配置） |
| **权限列表缓存** | String (JSON) | `perm:list:{user_id}:{version}` | 7200 秒 | 用户权限字符串列表（版本号策略） |
| **菜单缓存** | String (JSON) | `perm:menu:{user_id}:{version}` | 7200 秒 | 用户菜单树（版本号策略） |
| **用户有效状态** | String | `user:active:{user_id}` | 300 秒 | login_required 每请求校验 |
| **验证码** | String | `captcha:{uuid}` | 300 秒 | 图形验证码文本 |
:::tip 字典缓存不在 Redis 中
数据字典（`dict_util`）采用纯进程内 TTL 缓存，不经过 Redis。原因是字典读取落在同步 service 路径，无法使用请求级异步 Redis 客户端。详见上方「进程内字典缓存」章节。
:::
### Lua 脚本
@@ -243,14 +282,17 @@ return effective
数据变更时主动清除相关缓存，保证数据一致性：
```python
# 字典项变更 → 清除字典缓存
invalidate_dict_cache(dict_code)
# 字典项变更 → 按字典 ID 批量清除字典缓存
invalidate_dicts_by_ids(dict_ids)
# 菜单变更 → 递增全局版本号，所有用户权限缓存自然失效
await invalidate_menu_change()
# 用户角色变更 → 删除该用户的权限和菜单缓存
await invalidate_user_perms(user_id)
# 角色菜单变更 → 清除该角色下所有用户的权限缓存
invalidate_perm_cache(user_id)
# 用户角色变更 → 清除该用户的权限缓存
invalidate_perm_cache(user_id)
await invalidate_role_perms(role_id)
```
### TTL 自动过期
@@ -258,26 +300,27 @@ invalidate_perm_cache(user_id)
对于允许短暂不一致的数据，使用 TTL 自动过期：
```
数据字典缓存：TTL 300 秒（5 分钟）
数据字典缓存：TTL 60 秒（1 分钟，兜底过期，实际变更靠主动失效即时生效）
登录失败计数：TTL 300 秒（5 分钟）
滑动窗口限流：TTL 2 倍窗口时长
验证码：TTL 120 秒（2 分钟）
验证码：TTL 300 秒（5 分钟）
```
### 缓存穿透防护
:::warning 缓存穿透
当查询一个不存在的数据时，缓存和数据库都不会命中，导致每次请求都穿透到数据库。对于字典查询，如果 `dict_code` 不存在，应缓存空结果（空字典），避免反复查询数据库。
当查询一个不存在的字典编码时，缓存和数据库都不会命中，导致每次请求都穿透到数据库。`dict_util` 通过缓存空列表来防护：即使 `_load_dict_items` 返回空列表，也写入缓存，避免反复查询数据库。
:::
```python
def get_dict_map(dict_code: str) -> Dict[str, str]:
# ... 缓存查询逻辑 ...
items = dict_item_repo.get_all(dict_code=dict_code)
data = {str(item.value): item.label for item in items}
# 即使 data 为空字典，也写入缓存，防止穿透
_cache[cache_key] = (data, now + CACHE_TTL)
return data
def get_dict_items(code):
now = time.time()
hit = _dict_cache.get(code)
if hit and hit[0] > now:
return hit[1]
items = _load_dict_items(code) # 字典不存在时返回 []
_dict_cache[code] = (now + _CACHE_TTL, items) # 空列表也缓存，防止穿透
return items
```
## 缓存监控
DjangoAdmin_FastAPI_NaiveVue/zh/volume3/database-design.md
+4 -4
View File
@@ -55,7 +55,7 @@
| code | VARCHAR(100) | NOT NULL, UNIQUE | 角色编码 |
| status | INT | DEFAULT 1 | 状态：1=正常 2=禁用 |
| sort | INT | DEFAULT 0 | 排序 |
| remark | VARCHAR(500) | NULLABLE | 备注 |
| note | VARCHAR(255) | NULLABLE | 备注 |
| + 通用字段 | | | |
#### 菜单表 `fastapi_menu`
@@ -139,7 +139,7 @@
| name | VARCHAR(200) | NOT NULL | 字典名称 |
| code | VARCHAR(200) | NOT NULL, UNIQUE | 字典编码 |
| status | INT | DEFAULT 1 | 状态 |
| remark | VARCHAR(500) | NULLABLE | 备注 |
| note | VARCHAR(255) | NULLABLE | 备注 |
| + 通用字段 | | | |
#### 字典项表 `fastapi_dict_item`
@@ -152,7 +152,7 @@
| value | VARCHAR(200) | NOT NULL | 字典值 |
| sort | INT | DEFAULT 0 | 排序 |
| status | INT | DEFAULT 1 | 状态 |
| remark | VARCHAR(500) | NULLABLE | 备注 |
| note | VARCHAR(255) | NULLABLE | 备注 |
| + 通用字段 | | | |
### 内容管理模块
@@ -243,7 +243,7 @@
| misfire_policy | INT | DEFAULT 3 | 执行策略 |
| concurrent | INT | DEFAULT 0 | 是否并发 |
| status | INT | DEFAULT 1 | 状态：1=运行 2=暂停 |
| remark | VARCHAR(500) | NULLABLE | 备注 |
| note | VARCHAR(255) | NULLABLE | 备注 |
| + 通用字段 | | | |
#### 任务日志表 `fastapi_job_log`
DjangoAdmin_FastAPI_NaiveVue/zh/volume3/directory.md
+2 -4
View File
@@ -36,8 +36,7 @@ src/
│ ├── base_db.py # 数据库操作基类：save() 持久化方法
│ ├── base_repository.py # 仓库基类：通用 CRUD + 软删除 + 分页
│ ├── base_service.py # 服务基类：模板方法 CRUD + 差异点钩子
│ ├── base_schemas.py # Schema 基类：BaseForm（含可选 id 字段）
│ └── middleware.py # 中间件注册入口
│ └── base_schemas.py # Schema 基类：BaseForm（含可选 id 字段）
│
├── middleware/ # HTTP 中间件层
│ ├── __init__.py # 统一导出
@@ -45,7 +44,7 @@ src/
│ ├── rate_limit.py # 滑动窗口限流中间件
│ ├── db_session.py # 数据库会话中间件（commit/rollback）
│ ├── operation_log.py # 操作日志中间件
│ ├── upload_size.py # 上传体积限制中间件
│ ├── upload_size_limit.py # 上传体积限制中间件
│ └── logger.py # 请求日志钩子
│
├── api/ # 路由装配层
@@ -125,7 +124,6 @@ src/
├── constants/ # 常量与枚举
│ ├── __init__.py # 统一导出
│ ├── constants.py # 通用常量：PAGE_SIZE / DEFAULT_PASSWORD / SERIALIZE_EXCLUDE_FIELDS
│ ├── enums.py # 枚举定义
│ └── pagination.py # 分页参数解析
│
├── utils/ # 工具函数
DjangoAdmin_FastAPI_NaiveVue/zh/volume3/extensibility.md
+1 -1
View File
@@ -272,7 +272,7 @@ def _normalize_driver(driver: str) -> str:
DB_DRIVER = _normalize_driver(os.getenv('DB_DRIVER', 'mysql'))
DB_HOST = os.getenv('DB_HOST', '127.0.0.1')
DB_PORT = int(os.getenv('DB_PORT', '3306'))
DB_DATABASE = os.getenv('DB_DATABASE', 'djangoadmin.fastapi.elevue')
DB_DATABASE = os.getenv('DB_DATABASE', 'djangoadmin.fastapi.naivevue')
DB_USERNAME = os.getenv('DB_USERNAME', 'root')
DB_PASSWORD = os.getenv('DB_PASSWORD', '')
DjangoAdmin_FastAPI_NaiveVue/zh/volume3/overview.md
+2 -2
View File
@@ -8,7 +8,7 @@
```
┌─────────────────────────────────────────────────────────────────────────┐
│ Browser (Vue3 + ElementPlus) │
│ Browser (Vue3 + NaiveUI) │
│ Axios HTTP Request / Vite Dev Proxy │
└──────────────────────────────┬──────────────────────────────────────────┘
│
@@ -191,7 +191,7 @@ async def _combined_lifespan(app: FastAPI):
```
┌──────────────────────────────────────────────────────────────┐
│ Frontend (Vue3) │
│ ElementPlus · Pinia · Vue Router · Axios │
│ NaiveUI · Pinia · Vue Router · Axios │
└──────────────────────────┬───────────────────────────────────┘
│ HTTP / JSON
┌──────────────────────────┴───────────────────────────────────┐
DjangoAdmin_FastAPI_NaiveVue/zh/volume3/request-flow.md
+115 -88
View File
@@ -5,7 +5,7 @@
## 请求链路总览
:::tip 链路概览
一个典型的 API 请求经过以下阶段：浏览器 → Nginx → CORS → 限流 → Redis 注入 → DB 会话 → 操作日志 → Token 续签 → 路由匹配 → JWT 认证 → RBAC 鉴权 → Pydantic 校验 → Endpoint → Service → Repository → DB → 响应序列化 → 中间件反向链 → 浏览器
一个典型的 API 请求经过以下阶段：浏览器 → Nginx → CORS → 限流 → 上传体积限制 → 请求计时 → Token 续签 → 操作日志 → DB 会话 → Redis 注入 → 路由匹配 → JWT 认证 → RBAC 鉴权 → Pydantic 校验 → Endpoint → Service → Repository → DB → 响应序列化 → 中间件反向链 → 浏览器
:::
以下以 `POST /api/v1/position/add` 添加岗位为例，逐步拆解完整链路。
@@ -52,28 +52,19 @@ FastAPI 的中间件按注册顺序形成洋葱模型。请求从最外层进入
### 3. CORS 跨域中间件（最外层）
```python
# src/middleware/cors.py
register_cors(app) # 注册 CORS 中间件
# src/middleware/cors.py — 最后注册，位于洋葱最外层
register_cors(app)
```
- 检查 `Origin` 头是否在允许列表中
- 预检请求（OPTIONS）直接返回 CORS 响应头
- 普通请求附加 `Access-Control-Allow-*` 响应头
- 确保限流等直接返回的响应也带 CORS 头
### 4. 上传体积限制中间件
### 4. 请求限流中间件
```python
# src/middleware/upload_size.py
app.middleware("http")(upload_size_limit_middleware)
```
- 按 `Content-Length` 提前拒绝超大请求体
- 避免超大 multipart 落临时盘后再由业务层校验
### 5. 请求限流中间件
```python
# src/middleware/rate_limit.py
# src/middleware/rate_limit.py — 倒数第二注册，仅 CORS 内层
async def rate_limit_middleware(request: Request, call_next):
if not RATE_LIMIT_ENABLED:
return await call_next(request)
@@ -92,29 +83,73 @@ async def rate_limit_middleware(request: Request, call_next):
- 基于客户端 IP 的滑动窗口计数器
- 使用 Redis Lua 脚本保证原子性
- 超限直接返回统一 JSON 响应，不进入后续链路
- 超限直接返回 HTTP 200 + `{"code": 1}` 统一 JSON 响应，不进入后续链路
- 被限流的请求不进入 DB 会话与操作日志，避免 DoS 攻击时日志表与 DB 连接不降反增
- Redis 不可用时降级放行（fail-open）
- CORS 预检请求不计数
### 6. Redis 上下文注入中间件
### 5. 上传体积限制中间件
```python
# src/core/redis_client.py
async def redis_middleware(request: Request, call_next):
# 从应用状态中获取 Redis 客户端实例
redis = getattr(request.app.state, 'redis', None)
if redis is not None:
# 将 Redis 客户端注入到 contextvars，供业务层通过依赖注入获取
_redis_ctx.set(redis)
return await call_next(request)
# src/middleware/upload_size_limit.py — 按 Content-Length 提前拒绝超大请求体
app.middleware("http")(upload_size_limit_middleware)
```
- 将 Redis 异步客户端注入 `contextvars`
- 后续业务代码通过 `get_redis()` 获取当前请求的 Redis 客户端
- 按 `Content-Length` 提前拒绝超大请求体
- 避免超大 multipart 落临时盘后再由业务层校验
- 位于限流内层，已被限流放行的请求才检查体积
### 7. 数据库会话中间件
### 6. 请求计时中间件
```python
# src/middleware/db_session.py
# src/core/app.py — 先注册→内层，统计业务耗时
@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(round(process_time, 5))
return response
```
- 记录请求处理耗时（不含限流/CORS 等外层开销）
- 通过 `X-Process-Time` 响应头返回
### 7. Token 续签中间件
```python
# src/core/app.py — 续签 token 写入响应头
async def token_refresh_middleware(request, call_next):
response = await call_next(request)
refresh_token = getattr(request.state, '_refresh_token', None)
if refresh_token:
response.headers['X-Refresh-Token'] = refresh_token
return response
```
- 检查是否需要续签 Token
- 续签后的 Token 通过 `X-Refresh-Token` 响应头返回
### 8. 操作日志中间件
```python
# src/middleware/operation_log.py — 在 DB 会话外层，记录写操作日志
async def operation_log_middleware(request: Request, call_next):
response = await call_next(request)
if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
# 记录操作日志到 fastapi_operation_log 表
# 包括：操作人、操作IP、请求路径、请求参数、响应状态、耗时等
...
return response
```
- 在 DB 会话外层，记录写操作日志
- 包含请求方法、URL、参数、响应码、耗时、操作人等信息
### 9. 数据库会话中间件
```python
# src/middleware/db_session.py — 后注册→外层，DB 会话包裹业务逻辑
async def db_session_middleware(request: Request, call_next):
# 创建数据库会话
db = SessionLocal()
@@ -144,52 +179,22 @@ async def db_session_middleware(request: Request, call_next):
- 写操作根据响应体 `code` 字段自动 commit/rollback
- 异常时自动 rollback，finally 中关闭会话
### 8. 操作日志中间件
### 10. Redis 上下文注入中间件（最内层）
```python
# src/middleware/operation_log.py
async def operation_log_middleware(request: Request, call_next):
response = await call_next(request)
if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
# 记录操作日志到 fastapi_operation_log 表
# 包括：操作人、操作IP、请求路径、请求参数、响应状态、耗时等
...
return response
# src/core/redis_client.py — 先注册→最内层：先注入 Redis 再开 DB 会话
async def redis_middleware(request: Request, call_next):
# 从应用状态中获取 Redis 客户端实例
redis = getattr(request.app.state, 'redis', None)
if redis is not None:
# 将 Redis 客户端注入到 contextvars，供业务层通过依赖注入获取
_redis_ctx.set(redis)
return await call_next(request)
```
- 在 DB 会话外层，记录写操作日志
- 包含请求方法、URL、参数、响应码、耗时、操作人等信息
### 9. Token 续签中间件
```python
# src/core/app.py
async def token_refresh_middleware(request: Request, call_next):
response = await call_next(request)
refresh_token = getattr(request.state, '_refresh_token', None)
if refresh_token:
response.headers['X-Refresh-Token'] = refresh_token
return response
```
- 最外层中间件，检查是否需要续签 Token
- 续签后的 Token 通过 `X-Refresh-Token` 响应头返回
### 10. 请求计时中间件
```python
# src/core/app.py
@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(round(process_time, 5))
return response
```
- 记录请求处理耗时
- 通过 `X-Process-Time` 响应头返回
- 将 Redis 异步客户端注入 `contextvars`
- 后续业务代码通过 `get_redis()` 获取当前请求的 Redis 客户端
- 位于最内层，确保 DB 会话和操作日志都能使用 Redis
## 第三阶段：路由与认证
@@ -301,15 +306,25 @@ def permission_required(permission: str):
```python
# src/core/access_decorators.py
def check_demo(func):
"""演示模式拦截装饰器，放在 @permission_required 之后、路由函数之前"""
if asyncio.iscoroutinefunction(func):
@wraps(func)
async def async_wrapper(*args, **kwargs):
if FASTAPI_DEMO:
return R.failed("演示环境，暂无操作权限")
return await func(*args, **kwargs)
return async_wrapper
@wraps(func)
async def wrapper(*args, **kwargs):
def sync_wrapper(*args, **kwargs):
if FASTAPI_DEMO:
return R.failed("演示环境，暂无操作权限")
return await func(*args, **kwargs)
return wrapper
return func(*args, **kwargs)
return sync_wrapper
```
- `FASTAPI_DEMO=True` 时，所有写操作返回失败响应
- 根据被装饰函数类型自动选择同步/异步包装
- 保护演示环境数据不被修改
## 第四阶段：参数校验
@@ -440,7 +455,7 @@ R.ok(msg='添加成功')
## 完整链路流程图
```
请求链路（自上而下）
请求链路（自上而下，洋葱模型外层 → 内层）
[浏览器]
│ POST /api/v1/position/add
@@ -448,29 +463,41 @@ R.ok(msg='添加成功')
[Nginx] ── 反向代理转发
│
▼
[CORS 中间件] ── 跨域校验通过
[CORS 中间件] ── 跨域校验通过（最外层）
│
▼
[限流中间件] ── 滑动窗口计数（Redis）
│ ├─ 超限 → 直接返回 429
[限流中间件] ── 滑动窗口计数（Redis Lua）
│ ├─ 超限 → 直接返回 HTTP 200 + {code:1, msg:"请求过于频繁"}
│ └─ 通过 ▼
▼
[上传体积限制] ── 按 Content-Length 拦截超大请求
│
▼
[请求计时] ── 记录起始时间，响应时附加 X-Process-Time
│
▼
[Token 续签] ── 检查是否需要续签，响应时附加 X-Refresh-Token
│
▼
[操作日志中间件] ── 异步记录写操作（不影响主流程）
│
▼
[DB 会话中间件] ── 创建 SessionLocal()，注入 contextvars
│
▼
[操作日志中间件] ── 异步记录写操作（不影响主流程）
[Redis 注入中间件] ── 将 Redis 客户端注入 contextvars（最内层）
│
▼
[路由层] ── 匹配 /api/v1/position/add
│
▼
[认证依赖 login_required] ── JWT 验证 + 黑名单检查（Redis）
│ ├─ 失败 → 返回 401
│ ├─ 失败 → 返回 HTTP 200 + {code:401, msg:"..."}
│ └─ 通过 → 解析 user_id/username 注入 request.state ▼
▼
[Endpoint 层]
├─ @permission_required ── 权限校验（同步直查DB / 异步Redis缓存）
├─ @check_demo ── 演示模式拦截
├─ @permission_required ── 权限校验（Redis 缓存）
└─ 调用 Service ▼
▼
[Service 层]
@@ -485,14 +512,14 @@ R.ok(msg='添加成功')
▼
[数据库] ── 执行 INSERT，返回自增 ID
│
▼ 逐层返回响应
[Repository] → [Service] → [Endpoint] → [Router] → [OpLog] → [DBSession]
▼ 逐层返回响应（内层 → 外层）
[Redis] → [DBSession] → [OpLog] → [Token续签] → [请求计时] → [上传体积] → [限流] → [CORS]
│
├─ 业务成功（code=0）→ commit 提交事务
└─ 业务失败（code≠0）→ rollback 回滚事务
├─ 业务成功（code=0）→ DB 会话 commit 提交事务
└─ 业务失败（code≠0）→ DB 会话 rollback 回滚事务
│
▼
[限流] → [CORS] → [Nginx]
[Nginx]
│
▼
[浏览器] ← JSON 响应（{ code, data, msg, ok }）
@@ -502,7 +529,7 @@ R.ok(msg='添加成功')
当请求链路中发生异常时，全局异常处理器统一捕获并返回标准响应：
| 异常类型 | 触发场景 | 响应 |
| 异常类型 | 触发场景 | 响应（均为 HTTP 200） |
|:---|:---|:---|
| `AuthorizationException` | JWT 过期 / 无权限 | `{"code": 401, "msg": "..."}` |
| `BusinessException` | 业务逻辑错误 | `{"code": 1, "msg": "..."}` |
@@ -511,7 +538,7 @@ R.ok(msg='添加成功')
| `Exception` | 未捕获异常 | `{"code": 1, "msg": "服务器内部错误"}` |
:::warning 异常处理原则
所有异常均返回 HTTP 200 状态码，通过 `code` 字段区分业务成功（0）与失败（1）。这是前后端约定，前端统一按 `code` 判断，无需处理 HTTP 状态码差异。
所有异常均返回 **HTTP 200** 状态码，通过响应体 `code` 字段区分业务成功（0）与失败（1/401 等）。注意：`AuthorizationException` 的 `code=401` 是响应体中的业务码，不是 HTTP 状态码。这是前后端约定，前端统一按 `code` 判断，无需处理 HTTP 状态码差异。
:::
## 总结
DjangoAdmin_FastAPI_NaiveVue/zh/volume3/security.md
+63 -27
View File
@@ -471,37 +471,73 @@ def permission_required(permission: str):
### 权限缓存
权限缓存采用「用户级缓存 + 全局版本号」双层策略（`src/utils/perm_cache.py`）：
```python
# src/modules/system/menu/service.py
async def get_permissions_list(user_id: int) -> list:
"""获取用户权限列表（异步，走 Redis 缓存）"""
cache_key = f"perm:user:{user_id}"
# src/utils/perm_cache.py
_VERSION_KEY = 'perm:version' # 全局版本号
_PERM_PREFIX = 'perm:list:' # 用户权限列表前缀
_MENU_PREFIX = 'perm:menu:' # 用户菜单树前缀
_CACHE_TTL = 7200 # 缓存过期时间（2小时）
async def get_cached_permissions(user_id: int):
"""获取缓存的用户权限列表，未命中返回 None"""
redis = get_redis()
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
# 从数据库查询
permissions = query_user_permissions(user_id)
await redis.set(cache_key, json.dumps(permissions), ex=3600)
return permissions
def get_permissions_list_sync(user_id: int) -> list:
"""获取用户权限列表（同步，直查DB，供线程池场景使用）"""
return query_user_permissions(user_id)
def invalidate_perm_cache(user_id: int):
"""失效用户权限缓存（角色/菜单变更时调用）"""
from core.redis_client import get_redis
import asyncio
ver = await _get_version()
key = f'{_PERM_PREFIX}{user_id}:{ver}'
data = await redis.get(key)
if data is not None:
return json.loads(data)
return None
async def set_cached_permissions(user_id: int, perm_list: list):
"""缓存用户权限列表（空列表不缓存，避免缓存空壳后无法感知后续数据变更）"""
if not perm_list:
return
redis = get_redis()
asyncio.get_event_loop().run_until_complete(
redis.delete(f"perm:user:{user_id}")
)
ver = await _get_version()
key = f'{_PERM_PREFIX}{user_id}:{ver}'
await redis.set(key, json.dumps(perm_list), ex=_CACHE_TTL)
```
缓存失效策略：
```python
# src/utils/perm_cache.py
async def invalidate_menu_change():
"""菜单增删改时调用，递增全局版本号使所有用户缓存自然失效"""
await _incr_version()
async def invalidate_user_perms(user_id: int):
"""用户角色变更时调用，删除该用户的权限和菜单缓存"""
redis = get_redis()
ver = await _get_version()
keys = [
f'{_PERM_PREFIX}{user_id}:{ver}',
f'{_MENU_PREFIX}{user_id}:{ver}',
]
await redis.delete(*keys)
async def invalidate_role_perms(role_id: int):
"""角色菜单变更时调用，清理该角色下所有用户的权限和菜单缓存"""
# 查询该角色下所有用户 ID
user_ids = await asyncio.to_thread(_query_user_ids, role_id)
if not user_ids:
return
redis = get_redis()
ver = await _get_version()
keys = []
for uid in user_ids:
keys.append(f'{_PERM_PREFIX}{uid}:{ver}')
keys.append(f'{_MENU_PREFIX}{uid}:{ver}')
if keys:
await redis.delete(*keys)
```
:::tip 版本号策略的优势
传统方案在菜单变更时需要遍历所有用户删除缓存，版本号方案只需递增一个全局 key，所有用户的旧版本缓存自然失效（key 中包含版本号，读取时用新版本号拼 key，旧 key 不再被访问，等 TTL 过期自动清除）。
:::
### check_demo 装饰器
```python
@@ -600,7 +636,7 @@ def sanitize_html(content: str) -> str:
### 上传体积限制
```python
# src/middleware/upload_size.py
# src/middleware/upload_size_limit.py
async def upload_size_limit_middleware(request: Request, call_next):
content_length = request.headers.get('content-length')
if content_length and int(content_length) > MAX_UPLOAD_SIZE:
@@ -695,7 +731,7 @@ async def operation_log_middleware(request: Request, call_next):
| 配置项 | 环境变量 | 默认值 | 说明 |
|:---|:---|:---|:---|
| JWT 密钥 | `JWT_SALT` | 必须配置 | HS256 签名密钥 |
| JWT 过期时间 | `JWT_EXPIRE_MINUTES` | 120 | Token 有效期（分钟） |
| JWT 过期时间 | `JWT_EXPIRE_MINUTES` | 20 | Token 有效期（分钟） |
| 密码错误锁定 | — | 5 次/10 分钟 | 连续失败后锁定 |
| 验证码有效期 | — | 120 秒 | 验证码 Redis TTL |
| 限流开关 | `RATE_LIMIT_ENABLED` | False | 滑动窗口限流 |
DjangoAdmin_FastAPI_NaiveVue/zh/volume3/tech-stack.md
+3 -3
View File
@@ -63,13 +63,13 @@ SQLAlchemy 是 Python 生态中最成熟、功能最完备的 ORM 框架，2.0
## 前端技术栈
### Vue3 + Vite + ElementPlus
### Vue3 + Vite + NaiveUI
| 组件 | 版本 | 选型理由 |
|:---|:---|:---|
| **Vue3** | 3.x | Composition API、更好的 TypeScript 支持、Teleport/Suspense 等新特性 |
| **Vite** | 5.x | 极速 HMR、ESM 原生支持、开箱即用的 TypeScript 支持 |
| **ElementPlus** | 2.x | Vue3 生态最成熟的 UI 组件库，企业级后台管理系统首选 |
| **NaiveUI** | 2.x | Vue3 生态功能完备的 UI 组件库，企业级后台管理系统首选 |
| **Pinia** | 2.x | Vue3 官方推荐的状态管理方案，替代 Vuex |
| **Vue Router** | 4.x | Vue3 官方路由，支持动态路由、路由守卫 |
| **Axios** | 1.x | HTTP 客户端，支持请求/响应拦截器、取消请求 |
@@ -229,4 +229,4 @@ Redis 在中承担多个关键职责：JWT 令牌黑名单、登录失败锁定
## 总结
项目的技术选型以「高性能、类型安全、可扩展」为核心原则。FastAPI + SQLAlchemy 2.0 + Pydantic v2 构成后端铁三角，Vue3 + Vite + ElementPlus 提供现代化前端体验，Redis 承担缓存与中间件职责。各组件版本经过充分测试验证，形成稳定的技术底座。选型时优先考虑组件的独立性和可替换性，避免强耦合，为未来的架构演进预留空间。
项目的技术选型以「高性能、类型安全、可扩展」为核心原则。FastAPI + SQLAlchemy 2.0 + Pydantic v2 构成后端铁三角，Vue3 + Vite + NaiveUI 提供现代化前端体验，Redis 承担缓存与中间件职责。各组件版本经过充分测试验证，形成稳定的技术底座。选型时优先考虑组件的独立性和可替换性，避免强耦合，为未来的架构演进预留空间。