diff --git a/DjangoAdmin_FastAPI_EleVue/zh/struct.md b/DjangoAdmin_FastAPI_EleVue/zh/struct.md index b56c6db..d88d168 100644 --- a/DjangoAdmin_FastAPI_EleVue/zh/struct.md +++ b/DjangoAdmin_FastAPI_EleVue/zh/struct.md @@ -123,7 +123,6 @@ src/ │ └── response_body.py // 响应体工具 └── constants/ // 常量与枚举 ├── constants.py // PAGE_SIZE / DEFAULT_PASSWORD / SERIALIZE_EXCLUDE_FIELDS - ├── enums.py // 枚举定义 └── pagination.py // 分页参数 ``` diff --git a/DjangoAdmin_FastAPI_EleVue/zh/volume2/overview.md b/DjangoAdmin_FastAPI_EleVue/zh/volume2/overview.md index f6874f2..d27c6f5 100644 --- a/DjangoAdmin_FastAPI_EleVue/zh/volume2/overview.md +++ b/DjangoAdmin_FastAPI_EleVue/zh/volume2/overview.md @@ -61,9 +61,9 @@ | name | varchar(255) | 岗位名称 | | status | int | 状态:1-在用 2-停用 | | sort | int | 排序号 | -| create_user | varchar(255) | 创建人 | +| create_user | varchar(50) | 创建人 | | create_time | datetime | 创建时间 | -| update_user | varchar(255) | 更新人 | +| update_user | varchar(50) | 更新人 | | update_time | datetime | 更新时间 | | is_delete | int | 软删除标记:0-正常 1-已删除 | diff --git a/DjangoAdmin_FastAPI_EleVue/zh/volume3/base-classes.md b/DjangoAdmin_FastAPI_EleVue/zh/volume3/base-classes.md index 5c46bcc..d9aabb3 100644 --- a/DjangoAdmin_FastAPI_EleVue/zh/volume3/base-classes.md +++ b/DjangoAdmin_FastAPI_EleVue/zh/volume3/base-classes.md @@ -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='添加成功')` | diff --git a/DjangoAdmin_FastAPI_EleVue/zh/volume3/cache.md b/DjangoAdmin_FastAPI_EleVue/zh/volume3/cache.md index 451926a..a0f788a 100644 --- a/DjangoAdmin_FastAPI_EleVue/zh/volume3/cache.md +++ b/DjangoAdmin_FastAPI_EleVue/zh/volume3/cache.md @@ -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 ``` ## 缓存监控 diff --git a/DjangoAdmin_FastAPI_EleVue/zh/volume3/database-design.md b/DjangoAdmin_FastAPI_EleVue/zh/volume3/database-design.md index 6394fe1..a43b496 100644 --- a/DjangoAdmin_FastAPI_EleVue/zh/volume3/database-design.md +++ b/DjangoAdmin_FastAPI_EleVue/zh/volume3/database-design.md @@ -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` diff --git a/DjangoAdmin_FastAPI_EleVue/zh/volume3/directory.md b/DjangoAdmin_FastAPI_EleVue/zh/volume3/directory.md index 205e64d..0f12d2c 100644 --- a/DjangoAdmin_FastAPI_EleVue/zh/volume3/directory.md +++ b/DjangoAdmin_FastAPI_EleVue/zh/volume3/directory.md @@ -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/ # 工具函数 diff --git a/DjangoAdmin_FastAPI_EleVue/zh/volume3/request-flow.md b/DjangoAdmin_FastAPI_EleVue/zh/volume3/request-flow.md index b95aff6..6c34ce4 100644 --- a/DjangoAdmin_FastAPI_EleVue/zh/volume3/request-flow.md +++ b/DjangoAdmin_FastAPI_EleVue/zh/volume3/request-flow.md @@ -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 状态码差异。 ::: ## 总结 diff --git a/DjangoAdmin_FastAPI_EleVue/zh/volume3/security.md b/DjangoAdmin_FastAPI_EleVue/zh/volume3/security.md index 8bc8587..7357b2b 100644 --- a/DjangoAdmin_FastAPI_EleVue/zh/volume3/security.md +++ b/DjangoAdmin_FastAPI_EleVue/zh/volume3/security.md @@ -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 | 滑动窗口限流 | diff --git a/DjangoAdmin_FastAPI_EleVue/zh/volume4/api-cheatsheet.md b/DjangoAdmin_FastAPI_EleVue/zh/volume4/api-cheatsheet.md index bb2bdb2..1bc4b06 100644 --- a/DjangoAdmin_FastAPI_EleVue/zh/volume4/api-cheatsheet.md +++ b/DjangoAdmin_FastAPI_EleVue/zh/volume4/api-cheatsheet.md @@ -10,8 +10,8 @@ API 接口速查表,按模块分组列出所有 RESTful 接口的请求方法 | 方法 | 路径 | 说明 | 权限 | |:----|:----|:----|:----| -| GET | `/api/captcha` | 获取图形验证码 | 无需登录 | -| POST | `/api/login` | 用户登录 | 无需登录 | +| GET | `/api/v1/captcha` | 获取图形验证码 | 无需登录 | +| POST | `/api/v1/login` | 用户登录 | 无需登录 | | GET | `/api/v1/logout` | 退出登录 | 需登录 | ## 用户管理 diff --git a/DjangoAdmin_FastAPI_EleVue/zh/volume4/core/auth-flow.md b/DjangoAdmin_FastAPI_EleVue/zh/volume4/core/auth-flow.md index c09d17e..db8183f 100644 --- a/DjangoAdmin_FastAPI_EleVue/zh/volume4/core/auth-flow.md +++ b/DjangoAdmin_FastAPI_EleVue/zh/volume4/core/auth-flow.md @@ -39,7 +39,7 @@ ### 请求 ``` -GET /api/captcha +GET /api/v1/captcha ``` ### 响应 @@ -89,7 +89,7 @@ async def captcha(request: Request): ### 请求 ``` -POST /api/login +POST /api/v1/login Content-Type: application/json { diff --git a/DjangoAdmin_FastAPI_EleVue/zh/volume4/core/dict-guide.md b/DjangoAdmin_FastAPI_EleVue/zh/volume4/core/dict-guide.md index 4580fb1..78384ae 100644 --- a/DjangoAdmin_FastAPI_EleVue/zh/volume4/core/dict-guide.md +++ b/DjangoAdmin_FastAPI_EleVue/zh/volume4/core/dict-guide.md @@ -88,20 +88,41 @@ class TrainingService(BaseService[Training]): 效果:列表/详情返回数据中自动添加 `trainingMethod` 和 `trainingMethodName` 字段。 -### 方式 2:手动获取字典值 +### 方式 2:通过 dict_util 获取(推荐后端使用) ```python -from modules.dictionary.dict_item.repository import dict_item_repo +from utils.dict_util import get_dict_map, get_dict_label, get_dict_items -# 获取字典项列表 -items = dict_item_repo.get_all(dict_code='training_method', status=1) +# 获取 {value: label} 映射 +options = get_dict_map('training_method') +# 结果:{'1': '线上', '2': '线下', '3': '混合'} -# 转为 {value: label} 字典 -options = {item.value: item.label for item in items} +# 单值翻译 +label = get_dict_label('training_method', '1') +# 结果:'线上' + +# 获取字典项列表(含排序信息) +items = get_dict_items('training_method') +# 结果:[{'label': '线上', 'value': '1', 'sort': 1}, ...] +``` + +### 方式 3:通过 Repository 直接查询 + +```python +from modules.dictionary.dict.repository import dict_repo +from modules.dictionary.dict_item.repository import dictitem_repo + +# 先查字典类型获取 dict_id +dict_obj = dict_repo.get_one(code='training_method') +# 再查字典项 +items = dictitem_repo.filter_by(dict_id=dict_obj.id).all() + +# 转为 {value: name} 字典 +options = {item.value: item.name for item in items} # 结果:{'1': '线上', '2': '线下', '3': '混合'} ``` -### 方式 3:通过参数模块获取(适合系统配置) +### 方式 4:通过参数模块获取(适合系统配置) ```python from modules.param.service import param_service diff --git a/DjangoAdmin_FastAPI_EleVue/zh/volume4/core/dict.md b/DjangoAdmin_FastAPI_EleVue/zh/volume4/core/dict.md index f4b4a31..21a4f1a 100644 --- a/DjangoAdmin_FastAPI_EleVue/zh/volume4/core/dict.md +++ b/DjangoAdmin_FastAPI_EleVue/zh/volume4/core/dict.md @@ -94,18 +94,25 @@ class DictItem(base_model, base_db): ## 使用示例 -### 后端获取字典 +### 后端获取字典(推荐使用 dict_util) ```python -# 通过字典编码获取字典项列表 -from modules.dictionary.dict_item import service as dict_item_service +from utils.dict_util import get_dict_items, get_dict_map, get_dict_label -# 获取状态字典 -status_list = await dict_item_service.get_by_dict_code("sys_status") -# 返回: [{"label": "正常", "value": "1"}, {"label": "停用", "value": "2"}] +# 获取字典项列表(自动缓存,TTL 60秒兜底) +items = get_dict_items("sys_status") +# 返回: [{"label": "正常", "value": "1", "sort": 0}, {"label": "停用", "value": "2", "sort": 1}] + +# 获取 {value: label} 映射 +status_map = get_dict_map("sys_status") +# 返回: {"1": "正常", "2": "停用"} + +# 单值翻译(value 自动转字符串匹配) +label = get_dict_label("sys_status", 1, default="未知") +# 返回: "正常" ``` -### dict_util 缓存工具 +### dict_util 缓存工具详解 项目提供 `dict_util` 工具(`src/utils/dict_util.py`),内置进程内 TTL 缓存。由于字典数据几乎不变,且读写均落在同步 service 路径(无法使用请求级异步 Redis),因此采用进程内缓存而非 Redis 缓存: diff --git a/DjangoAdmin_FastAPI_EleVue/zh/volume7/service.md b/DjangoAdmin_FastAPI_EleVue/zh/volume7/service.md index f7efe65..73c0445 100644 --- a/DjangoAdmin_FastAPI_EleVue/zh/volume7/service.md +++ b/DjangoAdmin_FastAPI_EleVue/zh/volume7/service.md @@ -73,6 +73,9 @@ position_service = PositionService() | `serialize_maps` | dict | 枚举显示名映射 | `{'status': {1: '在用', 2: '停用'}}` | | `file_fields` | tuple | 文件字段 | `('cover', 'image')` | | `rich_text_fields` | tuple | 富文本字段 | `('content',)` | +| `file_dir` | str | 文件存储子目录名 | 默认从表名去掉 `DB_PREFIX` 推导 | +| `serialize_extra` | Callable | 列表/详情共用序列化钩子,返回需合并进字典的额外字段 | — | +| `detail_serialize` | Callable | 详情专用序列化钩子,仅在 `get_detail` 时合并 | — | :::tip 单例模式 每个模块在文件末尾创建 Service 实例(如 `position_service`),供 Endpoint 层使用。