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-12 23:12:15 +08:00
commit 91998056ca
1841 changed files with 341325 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/auth-flow.md
+387
View File
@@ -0,0 +1,387 @@
:::tip 本章概要
登录 → Token 获取 → 权限校验 → Token 刷新 → 退出登录的完整认证链路实战说明。
:::
# 认证流程实战
本章将登录 → Token 获取 → 权限校验 → Token 刷新 → 退出登录的完整认证链路串联说明，帮助开发者理解端到端的认证机制。
## 认证架构总览
```
┌──────────┐ ①获取验证码 ┌──────────┐
│ │ ──────────────────> │ │
│ 前端 │ │ 后端 │
│ Vue3 │ ②登录(账号+密码+验证码) │ Flask │
│ │ ──────────────────> │ │
│ │ ③返回 Token │ │
│ │ <────────────────── │ │
│ │ │ │
│ │ ④携带 Token 请求 │ │
│ │ ──────────────────> │ │
│ │ ⑤返回业务数据 │ │
│ │ <────────────────── │ │
│ │ │ │
│ │ ⑥刷新 Token │ │
│ │ ──────────────────> │ │
│ │ ⑦返回新 Token │ │
│ │ <────────────────── │ │
│ │ │ │
│ │ ⑧退出登录 │ │
│ │ ──────────────────> │ │
└──────────┘ └──────────┘
```
## 第 1 步：获取验证码
登录前需先获取图形验证码，防止暴力破解。
### 请求
```
GET /api/v1/captcha
```
### 响应
```json
{
"code": 0,
"msg": "操作成功",
"data": {
"captcha": "data:image/png;base64,iVBORw0KGgo...",
"key": "captcha:a3f8b2c1d4e5"
}
}
```
- `captcha`：Base64 编码的验证码图片，前端直接赋值给 `<img>` 的 `src`
- `key`：验证码唯一标识，登录时需回传
### 前端代码示例
```typescript
// 获取验证码
const { data } = await getCaptchaApi();
captchaImg.value = data.captcha;
captchaKey.value = data.key;
```
### 后端实现
```python
# src/modules/auth/router.py
@auth_blue.route('/captcha', methods=['GET'])
def captcha():
from src.common.utils.captcha import generate_captcha
result = generate_captcha()
id_key = result['idKey']
data = result['data']
img_str = "data:image/png;base64," + data
return R.ok({"captcha": img_str, "key": id_key})
```
验证码通过 Redis 存储，Key 格式为 `captcha:{uuid}`，默认有效期 300 秒。
## 第 2 步：用户登录
携带账号、密码、验证码和 Key 发起登录请求。
### 请求
```
POST /api/v1/login
Content-Type: application/json
{
"username": "admin",
"password": "123456",
"code": "a3Kp",
"key": "captcha:a3f8b2c1d4e5"
}
```
### 登录表单字段
| 字段 | 类型 | 必填 | 说明 |
|:----|:----|:----|:----|
| `username` | str | 是 | 登录账号（1-20 字符） |
| `password` | str | 是 | 登录密码（6-128 字符） |
| `code` | str | 是 | 验证码（长度由 CAPTCHA_LENGTH 配置） |
| `key` | str | 是 | 验证码 Key |
### 成功响应
```json
{
"code": 0,
"msg": "登录成功",
"data": {
"access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...",
"must_change_password": false
},
"ok": true
}
```
- `access_token`：JWT 令牌，后续请求需携带
- `must_change_password`：是否需要修改密码（检测到默认密码时为 `true`）
### 失败响应
```json
// 验证码错误
{"code": 1, "msg": "验证码不正确", "data": null, "ok": false}
// 账号或密码错误
{"code": 1, "msg": "用户名或密码不正确", "data": null, "ok": false}
// 账号被锁定
{"code": 1, "msg": "登录失败次数过多，请280秒后再试", "data": null, "ok": false}
```
### 登录流程详解
```
1. 检查是否被锁定（IP+用户名维度）
└─ 锁定中 → 返回剩余等待秒数
2. 验证码校验（Lua 原子消费，防重放）
└─ 不正确 → 记录失败次数
3. 查询用户
└─ 不存在 → 记录失败次数
4. 比对密码（bcrypt CPU 密集）
└─ 不匹配 → 记录失败次数
5. 登录成功
├─ 清除失败记录
├─ 检测是否为默认密码
├─ 生成 JWT Token
└─ 返回 Token
```
### 防暴力破解机制
| 维度 | 策略 | 配置项 |
|:----|:----|:----|
| 失败计数 | IP+用户名维度，Redis 原子递增 | `LOGIN_FAIL_MAX=5` |
| 锁定机制 | 达到上限后锁定 | `LOGIN_LOCK_SECONDS=300` |
| 验证码防刷 | 窗口期内限制获取次数 | `CAPTCHA_RATE_LIMIT=10` |
| 验证码消费 | Lua 原子消费，防重放 | - |
## 第 3 步：携带 Token 请求业务接口
登录成功后，前端将 Token 存入 localStorage，后续所有请求自动携带。
### 请求头格式
```
Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...
```
### 前端自动携带
```typescript
// 请求拦截器（自动添加 Token）
axios.interceptors.request.use(config => {
const token = localStorage.getItem('access_token');
if (token) {
config.headers['Authorization'] = `Bearer ${token}`;
}
return config;
});
```
### Token 解析流程
```
请求进入 → login_required 中间件
├─ 提取 Authorization 头
├─ 解析 JWT Token
│ ├─ 验证签名（HS256 + JWT_SALT）
│ ├─ 验证过期时间
│ └─ 检查黑名单（Redis）
├─ 注入 g 对象
│ ├─ user_id
│ ├─ username
│ └─ realname
└─ 业务代码通过 get_user_id() 获取用户信息
```
## 第 4 步：权限校验
每个业务接口通过 `@check_permission` 装饰器校验权限。
### 权限校验流程
```
请求到达 → check_permission("sys:user:add")
├─ 用户 ID = 1（admin）→ 跳过权限校验，直接放行
└─ 其他用户
├─ 查询用户角色
├─ 查询角色权限列表
├─ 匹配权限字符串
│ ├─ 匹配 → 放行
│ └─ 不匹配 → 返回 "权限不足"
└─ 结果缓存（Redis，菜单变更时失效）
```
### 权限字符串格式
```
sys:{module}:{action}
```
示例：`sys:user:add`、`sys:role:page`、`sys:article:delete`
## 第 5 步：Token 刷新
Token 支持无感刷新，刷新时执行令牌轮换（旧 Token 立即失效）。
### 刷新请求
```
POST /api/v1/refreshToken
Authorization: Bearer <old_token>
```
### 刷新流程
```
1. 验证原令牌有效性
2. 检查是否已注销（黑名单）
3. 移除 JWT 标准字段（exp/iat/nbf），保留业务数据
4. 生成新令牌
5. 旧令牌加入黑名单
6. 返回新令牌
```
### 前端无感刷新
```typescript
// 响应拦截器：检测 Token 过期
axios.interceptors.response.use(
response => response,
async error => {
if (error.response?.status === 401) {
// 尝试刷新 Token
const newToken = await refreshToken();
if (newToken) {
// 用新 Token 重试原请求
error.config.headers['Authorization'] = `Bearer ${newToken}`;
return axios(error.config);
}
// 刷新失败，跳转登录页
router.push('/login');
}
return Promise.reject(error);
}
);
```
## 第 6 步：退出登录
退出时将当前 Token 加入 Redis 黑名单，使其立即失效。
### 请求
```
GET /api/v1/logout
Authorization: Bearer <token>
```
### 响应
```json
{
"code": 0,
"msg": "注销成功",
"data": null,
"ok": true
}
```
### 退出流程
```
1. 提取当前 Token
2. 计算 Token 的 SHA256 指纹
3. 存入 Redis 黑名单（TTL = Token 剩余有效期）
4. 记录退出日志
5. 返回成功
```
### Token 黑名单机制
- **存储方式**：Redis，Key 格式 `token:blacklist:{sha256_fingerprint}`
- **过期策略**：TTL 与 Token 剩余有效期一致，自动清理
- **故障降级**：Redis 不可用时 fail-open 放行（避免全站不可用）
## 完整时序图
```
前端 后端 Redis 数据库
│ │ │ │
│ ① GET /captcha │ │ │
│ ────────────────────> │ 生成验证码 │ │
│ │ ────────────────────> │ 存储 captcha key │
│ 返回验证码图片+key │ <──────────────────── │ │
│ <──────────────────── │ │ │
│ │ │ │
│ ② POST /login │ │ │
│ (username+pwd+code) │ │ │
│ ────────────────────> │ ① 检查锁定 │ │
│ │ ────────────────────> │ 查询 lock key │
│ │ <──────────────────── │ │
│ │ ② 验证码校验 │ │
│ │ ────────────────────> │ Lua 原子消费 │
│ │ <──────────────────── │ │
│ │ ③ 查询用户 │ │
│ │ ─────────────────────────────────────────> │
│ │ <───────────────────────────────────────── │
│ │ ④ 比对密码 │ │
│ │ ⑤ 生成 JWT Token │ │
│ 返回 Token │ │ │
│ <──────────────────── │ │ │
│ │ │ │
│ ③ GET /api/v1/user/page │ │
│ Authorization: Bearer xxx │ │
│ ────────────────────> │ 解析 Token │ │
│ │ 检查黑名单 │ │
│ │ ────────────────────> │ 查询 blacklist │
│ │ <──────────────────── │ │
│ │ 校验权限 │ │
│ │ 查询业务数据 │ │
│ │ ─────────────────────────────────────────> │
│ 返回业务数据 │ <───────────────────────────────────────── │
│ <──────────────────── │ │ │
│ │ │ │
│ ④ GET /logout │ │ │
│ ────────────────────> │ Token 加入黑名单 │ │
│ │ ────────────────────> │ SET blacklist │
│ 返回注销成功 │ <──────────────────── │ │
│ <──────────────────── │ │ │
```
## 配置项
| 环境变量 | 默认值 | 说明 |
|:----|:----|:----|
| `JWT_SALT` | 无（必填） | JWT 签名密钥，至少 16 字符 |
| `JWT_EXPIRE_MINUTES` | `20` | Token 过期时间（分钟） |
| `LOGIN_FAIL_MAX` | `5` | 登录失败次数上限 |
| `LOGIN_LOCK_SECONDS` | `300` | 登录锁定时间（秒） |
| `CAPTCHA_LENGTH` | `6` | 验证码长度 |
| `CAPTCHA_EXPIRE` | `300` | 验证码有效期（秒） |
## 总结
认证流程覆盖了从获取验证码到退出登录的完整链路，核心安全机制包括：
- bcrypt 密码哈希 + 随机盐
- JWT HS256 签名 + 过期校验
- Redis 验证码防重放
- IP+用户名维度的登录锁定
- Token 黑名单即时失效
- Redis 不可用时 fail-open 降级
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/captcha.md
+174
View File
@@ -0,0 +1,174 @@
:::tip 验证码
验证码是防止自动化攻击的重要安全手段。项目使用 Pillow 生成图片验证码，存储在 Redis 中，支持防刷限流和万能验证码（演示模式）。源码位于 `src/common/utils/captcha.py`。
:::
## 验证码流程
```
1. 前端请求 GET /api/v1/captcha
2. 后端生成随机字符 → 绘制图片（干扰线+噪点）→ Base64 编码
3. 字符存入 Redis（key=UUID，TTL=300秒）
4. 返回 {key, captcha} 给前端
5. 前端展示图片，用户输入验证码
6. 登录时后端从 Redis 取出比对，验证成功后自动删除（防重放）
```
## 配置项
| 环境变量 | 默认值 | 说明 |
|---------|--------|------|
| `CAPTCHA_LENGTH` | `6` | 验证码字符长度 |
| `CAPTCHA_EXPIRE` | `300` | 过期时间（秒） |
| `CAPTCHA_RATE_WINDOW` | `60` | 防刷窗口期（秒） |
| `CAPTCHA_RATE_LIMIT` | `10` | 窗口期内最大获取次数 |
| `CAPTCHA_CHARS` | `ABCDEFGHJKLMNPQRSTUVWXYZ2346789` | 字符集（去除易混淆字符） |
| `CAPTCHA_FONT_PATH` | `static/fonts/Vera.ttf` | 字体路径 |
| `CAPTCHA_IMAGE_WIDTH` | `160` | 图片宽度 |
| `CAPTCHA_IMAGE_HEIGHT` | `50` | 图片高度 |
| `CAPTCHA_FONT_SIZE` | `28` | 字体大小 |
| `CAPTCHA_NOISE_LINES` | `6` | 干扰线数量 |
| `CAPTCHA_NOISE_DOTS` | `30` | 噪点数量 |
| `CAPTCHA_BYPASS_CODE` | `618618` | 万能验证码（仅演示模式生效） |
## 接口说明
### 获取验证码
```
GET /api/v1/captcha
```
响应示例：
```json
{
"code": 0,
"msg": "操作成功",
"data": {
"key": "54e66743-0617-46fd-9b7e-cc1044ac1ccd",
"captcha": "data:image/png;base64,iVBORw0KGgo..."
},
"ok": true
}
```
### 验证码校验
验证码校验在登录流程中自动完成，核心逻辑在 `src/common/utils/captcha.py`：
```python
from src.common.utils.captcha import verify_captcha
# 验证成功后自动删除，防止重放攻击
is_valid = verify_captcha(key, code)
```
验证码校验采用 Lua 脚本实现原子消费，防止并发重放和暴力破解：
```python
# Lua 脚本：Redis 单线程原子执行 get -> del -> compare
_CAPTCHA_VERIFY_LUA = """
local stored = redis.call('get', KEYS[1])
if not stored then return 0 end
-- 消费验证码：一次性使用，失败也立即失效（防暴力破解/无限试错）
redis.call('del', KEYS[1])
local function norm(s)
return string.upper(tostring(s)):gsub('^%s+', ''):gsub('%s+$', '')
end
if norm(stored) == norm(ARGV[1]) then return 1 end
return 0
"""
def verify_captcha(id_key: str, user_input: str) -> bool:
"""验证验证码（原子消费：验证码无论对错一次性使用）"""
if not id_key or not user_input:
return False
redis = get_redis()
# 演示模式万能验证码
from flask import current_app
if current_app.config.get('FLASK_DEMO') and CAPTCHA_BYPASS_CODE and user_input == CAPTCHA_BYPASS_CODE:
return True
return bool(redis.eval(_CAPTCHA_VERIFY_LUA, 1, id_key, user_input))
```
:::warning 原子消费
验证码无论对错都会被删除（Lua 脚本内 get + del + compare 原子执行），同一 idKey 无法再次尝试。这既防止了重放攻击，也防止了暴力穷举。
:::
## 防刷机制
验证码获取接口内置限流保护，使用 `incr_with_expire` Lua 脚本实现原子计数：
```python
def generate_captcha():
redis = get_redis()
# 防刷：按 IP 维度限流
client_ip = get_client_ip()
rate_key = f"captcha:rate:{client_ip}"
count = redis.get(rate_key)
if count and int(count) >= CAPTCHA_RATE_LIMIT:
ttl = redis.ttl(rate_key)
remaining = ttl if ttl and ttl > 0 else CAPTCHA_RATE_WINDOW
raise BizError(msg=f"请求过于频繁，请等待 {remaining} 秒后再试")
# 生成验证码 ...
image, code = _generate_image(CAPTCHA_LENGTH)
# 存储到 Redis
id_key = secrets.token_urlsafe(24)
redis.set(id_key, code, ex=CAPTCHA_EXPIRE)
# 原子递增并设置过期（Lua 脚本，防止 incr 与 expire 之间崩溃导致 key 永久残留）
from src.core.redis_client import incr_with_expire
incr_with_expire(redis, rate_key, CAPTCHA_RATE_WINDOW)
```
限流参数：
- 按 IP 维度限流
- 窗口期内（默认60秒）最多获取10次
- 超限后返回 `BizError` 提示剩余等待秒数
- 防止恶意刷取验证码消耗服务器资源
## 万能验证码
在演示模式（`FLASK_DEMO=True`）下，万能验证码 `618618` 可绕过正常验证码校验，方便演示环境登录。
:::warning 安全提示
生产环境务必确保 `FLASK_DEMO=False`，否则万能验证码将失效。
:::
## 前端集成
```javascript
// 获取验证码
async function getCaptcha() {
const res = await axios.get('/api/v1/captcha')
captchaKey.value = res.data.data.key
captchaImg.value = res.data.data.captcha
}
// 登录时提交
const loginData = {
username: 'admin',
password: '123456',
code: userInput.value,
key: captchaKey.value
}
```
## 总结
项目的验证码方案具备以下特点：
```
1. 图片验证码：Pillow 生成，支持干扰线、噪点、自定义字体
2. Redis 存储：TTL 自动过期，Lua 原子消费防重放
3. 防刷限流：IP 维度 + incr_with_expire 原子计数，防止恶意刷取
4. 万能验证码：演示模式下可用，生产环境自动禁用
5. 字符集优化：去除易混淆字符（0/O、1/I/L 等）
6. 安全随机数：使用 secrets 模块生成验证码和唯一 ID
```
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/config.md
+146
View File
@@ -0,0 +1,146 @@
:::tip 系统配置
系统配置模块用于管理动态的键值对配置项，与数据字典类似但更偏向系统级参数。支持分组管理，可通过接口动态修改，无需重启服务。模块位于 `src/modules/configuration/`，路由位于 `src/modules/configuration/config/router.py` 和 `config_item/router.py`。
:::
## 模块结构
```
src/modules/configuration/
├── config/ # 配置分组
│ ├── models.py
│ ├── schema.py
│ ├── repository.py
│ ├── service.py
│ └── router.py
└── config_item/ # 配置项（KV）
├── models.py
├── schema.py
├── repository.py
├── service.py
└── router.py
```
## 配置模型
### 配置分组（Config）
```python
# src/modules/configuration/config/models.py
from sqlalchemy import Column, String, Integer, text
from src.common.base_db import BaseDB
from src.common.base_model import BaseModel
from config.database import DB_PREFIX
class Config(BaseModel, BaseDB):
"""配置模型类"""
__tablename__ = DB_PREFIX + "config"
__table_comment__ = "系统配置表"
# 配置名称
name = Column(String(150), nullable=False, index=True, comment="配置名称")
# 配置编码
code = Column(String(150), nullable=False, index=True, comment="配置编码")
# 配置排序
sort = Column(Integer, default=0, server_default=text('0'), comment="配置排序")
# 配置备注
note = Column(String(255), nullable=True, comment="配置备注")
def __str__(self):
"""返回配置ID作为字符串表示"""
return "配置{}".format(self.id)
```
### 配置项（ConfigItem）
```python
# src/modules/configuration/config_item/models.py
from sqlalchemy import Column, String, Integer, Text, text
from src.common.base_db import BaseDB
from src.common.base_model import BaseModel
from config.database import DB_PREFIX
class ConfigItem(BaseModel, BaseDB):
"""配置项模型类"""
__tablename__ = DB_PREFIX + "config_item"
__table_comment__ = "配置项表"
# 配置项名称
name = Column(String(150), nullable=False, index=True, comment="配置项名称")
# 配置项编码
code = Column(String(150), nullable=False, comment="配置项编码")
# 配置项值
value = Column(String(1000), nullable=True, comment="配置项值")
# 配置选项
options = Column(Text, nullable=True, comment="配置选项")
# 配置ID
config_id = Column(Integer, default=0, server_default=text('0'), index=True, comment="配置ID")
# 配置类型
type = Column(String(150), nullable=False, comment="配置类型")
# 配置状态：1-正常 2-停用
status = Column(Integer, default=1, server_default=text('1'), index=True, comment="配置状态：1-正常 2-停用")
# 配置项排序
sort = Column(Integer, default=0, server_default=text('0'), comment="配置项顺序")
# 配置项备注
note = Column(String(255), nullable=True, comment="配置项备注")
def __str__(self):
"""返回配置项ID作为字符串表示"""
return "配置项{}".format(self.id)
```
## 使用场景
系统配置适用于以下场景：
```
1. 网站基本信息（站点名称、Logo、版权信息）
2. 邮件配置（SMTP 服务器、端口、账号）
3. 短信配置（API Key、模板 ID）
4. 存储配置（OSS Bucket、上传限制）
5. 业务参数（默认密码、分页大小等）
```
## 接口说明
### 获取配置项列表
```
GET /api/v1/config/item/list?configId=1
```
### 获取配置项详情
```
GET /api/v1/config/item/detail/<id>
```
### 新增/编辑配置项
```
POST /api/v1/config/item/add
PUT /api/v1/config/item/update
```
## 与数据字典的区别
| 特性 | 数据字典 | 系统配置 |
|------|---------|---------|
| 用途 | 枚举值（下拉选项） | 系统参数（配置值） |
| 值类型 | 简单值（字符串） | 可存储较长的配置内容 |
| 典型场景 | 状态、性别、类型 | SMTP配置、站点信息 |
| 前端用法 | el-select 选项 | 配置表单 |
## 总结
系统配置模块具备以下特点：
```
1. KV 结构：配置分组 + 配置项，灵活管理各类参数
2. 动态修改：通过接口修改，无需重启服务
3. 分组管理：按功能模块分组，清晰组织配置项
4. 与字典互补：字典管理枚举值，配置管理系统参数
```
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/cors.md
+120
View File
@@ -0,0 +1,120 @@
:::tip CORS跨域
CORS（Cross-Origin Resource Sharing）跨域资源共享是前后端分离架构必须处理的问题。通过 `CORS_ALLOWED_ORIGINS` 环境变量配置允许的跨域源。中间件位于 `src/middleware/cors.py`。
:::
## 配置项
| 环境变量 | 默认值 | 说明 |
|---------|--------|------|
| `CORS_ALLOWED_ORIGINS` | 空（允许所有源） | 允许的跨域源（逗号分隔） |
| `TRUSTED_PROXY_COUNT` | `0` | 可信代理层数 |
| `TRUSTED_PROXY_IPS` | 空 | 可信代理 IP 白名单（逗号分隔） |
## 配置示例
```bash
# .env
# 允许所有源（开发环境）
CORS_ALLOWED_ORIGINS=
# 允许指定源（生产环境）
CORS_ALLOWED_ORIGINS=https://admin.example.com,https://www.example.com
# 信任一层反向代理（Nginx）
TRUSTED_PROXY_COUNT=1
TRUSTED_PROXY_IPS=127.0.0.1
```
:::warning 生产环境
生产环境务必配置 `CORS_ALLOWED_ORIGINS` 为具体域名，避免允许所有源带来的安全风险。
:::
## CORS 中间件注册
Flask 使用 `flask-cors` 扩展实现 CORS 支持：
```python
# src/middleware/cors.py
from flask_cors import CORS
from config.app import CORS_ALLOWED_ORIGINS
def register_cors(app):
if CORS_ALLOWED_ORIGINS:
origins = [o.strip() for o in CORS_ALLOWED_ORIGINS.split(',') if o.strip()]
CORS(app, origins=origins, supports_credentials=True)
else:
# 未配置时允许所有源但不携带凭证
CORS(app, origins="*")
```
## 中间件顺序
CORS 中间件在 `src/app.py` 中注册为最外层：
```python
# 最先注册→最外层：确保限流等直接返回的响应也带 CORS 头
register_cors(app)
```
中间件执行顺序（从外到内）：
```
CORS → 限流 → 上传体积限制 → 操作日志 → DB会话 → Redis → 业务逻辑
```
## 代理头处理
当应用部署在 Nginx 等反向代理后，需要配置代理信任：
```bash
# 信任一层代理
TRUSTED_PROXY_COUNT=1
TRUSTED_PROXY_IPS=127.0.0.1
```
`TRUSTED_PROXY_COUNT` 说明：
| 值 | 说明 |
|----|------|
| `0` | 不信任 X-Forwarded-For，使用 request.remote_addr |
| `1` | 信任一层反向代理 |
| `2` | 信任两层代理 |
:::tip 安全提示
仅当应用部署在反向代理后才设置 `TRUSTED_PROXY_COUNT>=1`。如果应用被直接暴露（如 Docker bridge 映射端口），设置代理信任会导致客户端伪造 XFF 绕过限流。
:::
## Flask-CORS 配置选项
```python
from flask_cors import CORS
CORS(app,
origins=["https://admin.example.com"], # 允许的源
methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"], # 允许的方法
allow_headers=["Content-Type", "Authorization"], # 允许的头
supports_credentials=True, # 允许携带凭证
max_age=3600, # 预检请求缓存时间（秒）
)
```
| 参数 | 说明 | 默认值 |
|------|------|--------|
| `origins` | 允许的源 | `*` |
| `methods` | 允许的方法 | `GET, HEAD, POST, OPTIONS` |
| `allow_headers` | 允许的头 | `*` |
| `supports_credentials` | 允许携带凭证 | `False` |
| `max_age` | 预检请求缓存时间 | `None` |
## 总结
CORS 跨域配置具备以下特点：
```
1. 环境变量配置：通过 CORS_ALLOWED_ORIGINS 灵活配置允许的源
2. 开发/生产分离：开发环境允许所有源，生产环境指定域名
3. 代理头支持：TRUSTED_PROXY_COUNT 配置代理信任层数
4. 安全默认：未配置时允许所有源但不携带凭证
5. 最外层注册：确保所有响应都带 CORS 头
```
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/demo-mode.md
+128
View File
@@ -0,0 +1,128 @@
:::tip 演示模式
演示模式用于在线体验环境，开启后所有写操作（新增、编辑、删除等）将被拦截，返回"演示环境，暂无操作权限"提示。通过 `FLASK_DEMO` 环境变量控制。装饰器位于 `src/middleware/demo.py`。
:::
## 配置项
| 环境变量 | 默认值 | 说明 |
|---------|--------|------|
| `FLASK_DEMO` | `False` | 是否开启演示模式 |
```bash
# .env
# 开启演示模式
FLASK_DEMO=True
# 关闭演示模式（生产环境）
FLASK_DEMO=False
```
## check_demo 装饰器
`src/middleware/demo.py` 中的 `@check_demo` 装饰器用于拦截写操作：
```python
from src.middleware.demo import check_demo
@position_blue.route('/add', methods=['POST'])
@check_permission('sys:position:add')
@check_demo
def add():
from src.modules.system.position.service import add_position
from src.modules.system.position.schema import PositionForm
data = parse_request_body(request, PositionForm)
return add_position(data)
```
装饰器实现：
```python
from functools import wraps
from flask import current_app
from src.core.response import R
def check_demo(func):
"""演示模式拦截装饰器，放在 @check_permission 之后、路由函数之前"""
@wraps(func)
def wrapper(*args, **kwargs):
if current_app.config.get('FLASK_DEMO'):
return R.failed("演示环境，暂无操作权限")
return func(*args, **kwargs)
return wrapper
```
## 装饰器顺序
`@check_demo` 必须放在 `@check_permission` 之后：
```python
# 正确顺序
@position_blue.route('/add', methods=['POST'])
@check_permission('sys:position:add') # 先校验权限
@check_demo # 再检查演示模式
def add():
...
```
:::warning 顺序重要
如果 `@check_demo` 放在 `@check_permission` 之前，权限校验会被跳过，可能导致安全问题。
:::
## 万能验证码
演示模式下万能验证码 `618618` 可绕过正常验证码校验，方便演示环境登录：
```python
# src/common/utils/captcha.py
def verify_captcha(key: str, code: str) -> bool:
# 演示模式下检查万能验证码
if current_app.config.get('FLASK_DEMO') and code == CAPTCHA_BYPASS_CODE:
return True
# 正常校验逻辑...
```
## 适用场景
```
1. 在线体验：让用户无需注册即可体验系统功能
2. 演示环境：销售或产品演示时防止误操作
3. 公开测试：开放测试环境但保护数据安全
```
## 受保护的操作
演示模式拦截以下操作（带 `@check_demo` 装饰器的端点）：
```
- 新增（add）
- 编辑（update）
- 删除（delete / batchDelete）
- 状态设置（status）
- 重置密码（resetPwd）
- 导入（import）
- 授权（auth）
- 生成代码（generate）
```
不受影响的操作：
```
- 登录（login）
- 获取验证码（captcha）
- 查询列表（list）
- 查询详情（detail）
- 导出（export）
```
## 总结
演示模式具备以下特点：
```
1. 环境变量控制：FLASK_DEMO=True 开启，False 关闭
2. 写操作拦截：@check_demo 装饰器拦截所有写操作
3. 万能验证码：演示模式下可用，方便登录体验
4. 安全顺序：必须放在 @check_permission 之后
5. 只读体验：查询操作不受影响，用户可正常浏览数据
```
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/dict-guide.md
+232
View File
@@ -0,0 +1,232 @@
:::tip 本章概要
数据字典的完整使用流程，从新增字典类型到在代码中使用字典数据的实操指南。
:::
# 数据字典使用指南
数据字典是项目的基础配置功能，用于管理枚举类型的下拉选项。本章从「如何新增字典」到「如何在代码中使用」的完整实操流程。
## 数据字典结构
数据字典由两层组成：
```
字典类型（dict） 字典项（dict_item）
├── sys_user_status ├── 1 → 正常
├── sys_user_gender ├── 2 → 停用
├── article_type └── 3 → 删除
└── ...
```
| 层级 | 表 | 说明 |
|:----|:----|:----|
| 字典类型 | `flask_dict` | 字典分类（如「用户状态」「文章类型」） |
| 字典项 | `flask_dict_item` | 字典的具体选项（如「1-正常」「2-停用」） |
## 实操：新增一个字典类型
### 场景
需要为「培训方式」模块添加一个下拉选项：`1-线上 2-线下 3-混合`。
### 第 1 步：在后台添加字典类型
登录管理后台，进入「系统管理 → 数据字典」：
1. 点击「新增」按钮
2. 填写信息：
| 字段 | 值 |
|:----|:----|
| 字典名称 | 培训方式 |
| 字典编码 | training_method |
| 状态 | 正常 |
3. 保存
### 第 2 步：添加字典项
在字典类型列表中点击「培训方式」，进入字典项管理：
1. 添加字典项：
| 字典值 | 字典标签 | 排序 |
|:----|:----|:----|
| 1 | 线上 | 1 |
| 2 | 线下 | 2 |
| 3 | 混合 | 3 |
2. 保存
### 第 3 步：在模型中关联字典
在模块的 `service.py` 中配置 `serialize_maps`：
```python
class TrainingService(BaseService):
# ... 其他配置 ...
# 枚举显示名映射：字段名 → 字典编码
serialize_maps = {
'training_method': 'training_method',
'status': 'training_status',
}
```
配置后，列表和详情接口会自动将 `training_method=1` 转换为 `training_method_name='线上'`。
## 在后端代码中使用
### 方式 1：通过 serialize_maps 自动转换（推荐）
在 `service.py` 中配置 `serialize_maps`，基类自动处理：
```python
class TrainingService(BaseService):
serialize_maps = {'training_method': 'training_method'}
```
效果：列表/详情返回数据中自动添加 `trainingMethod` 和 `trainingMethodName` 字段。
### 方式 2：手动获取字典值
```python
from src.modules.dictionary.dict_item.repository import DictItemRepository
# 获取字典项列表
items = DictItemRepository.get_all(dict_code='training_method', status=1)
# 转为 {value: label} 字典
options = {item.value: item.label for item in items}
# 结果：{'1': '线上', '2': '线下', '3': '混合'}
```
### 方式 3：通过参数模块获取（适合系统配置）
```python
from src.modules.param.service import get_param_value
# 获取参数值
timeout = get_param_value('REQUEST_TIMEOUT', '30')
```
## 在前端代码中使用
### 方式 1：字典下拉组件
```vue
<template>
<el-select v-model="form.trainingMethod" placeholder="请选择培训方式">
<el-option
v-for="item in dictOptions"
:key="item.value"
:label="item.label"
:value="item.value"
/>
</el-select>
</template>
<script setup>
import { ref, onMounted } from 'vue';
import { getDictItems } from '@/api/system/dictionary';
const dictOptions = ref([]);
onMounted(async () => {
const { data } = await getDictItems('training_method');
dictOptions.value = data;
});
</script>
```
### 方式 2：表格列显示字典文本
在 `columns.ts` 中使用 `render` 函数：
```typescript
import { h } from 'vue';
import { ElTag } from 'element-plus';
export const columns = [
{
label: '培训方式',
prop: 'trainingMethod',
render(record) {
const map = { 1: '线上', 2: '线下', 3: '混合' };
const typeMap = { 1: 'success', 2: 'warning', 3: 'info' };
return h(ElTag, { type: typeMap[record.row.trainingMethod] }, {
default: () => map[record.row.trainingMethod] || '-'
});
},
},
];
```
### 方式 3：搜索表单下拉
在 `querySchemas.ts` 中配置：
```typescript
export const schemas: FormSchema[] = [
{
field: 'trainingMethod',
component: 'Select',
label: '培训方式',
componentProps: {
placeholder: '请选择培训方式',
clearable: true,
options: [
{ label: '线上', value: '1' },
{ label: '线下', value: '2' },
{ label: '混合', value: '3' },
],
},
},
];
```
## 代码生成器与字典
代码生成器会自动识别字段注释中的枚举格式（`1-线上 2-线下 3-混合`），并：
1. 自动解析 `choices` 列表
2. 自动生成 `dict_code`（格式：`{table_name}_{field_name}`）
3. 自动配置 `serialize_maps`
4. 前端自动生成下拉选择框
生成后需在后台手动创建对应的字典类型和字典项。
## 字典缓存
项目使用进程内缓存存储字典数据：
| 层级 | 说明 | 失效策略 |
|:----|:----|:----|
| L1 进程内缓存 | Python 字典，TTL 60 秒 | 自动过期 |
字典数据修改时自动清除相关缓存。
## 字典管理 API
| 方法 | 路径 | 说明 |
|:----|:----|:----|
| GET | `/api/v1/dict/page` | 字典类型分页 |
| POST | `/api/v1/dict/add` | 新增字典类型 |
| PUT | `/api/v1/dict/update` | 更新字典类型 |
| DELETE | `/api/v1/dict/delete/<id>` | 删除字典类型 |
| GET | `/api/v1/dictItem/page` | 字典项分页 |
| POST | `/api/v1/dictItem/add` | 新增字典项 |
| PUT | `/api/v1/dictItem/update` | 更新字典项 |
| DELETE | `/api/v1/dictItem/delete/<id>` | 删除字典项 |
## 最佳实践
1. **字典编码命名**：使用 `模块_字段` 格式，如 `user_status`、`article_type`
2. **字典值使用整数**：便于排序和比较，避免使用字符串
3. **字典标签简洁**：控制在 4 个字以内，适合表格列显示
4. **状态字段统一**：`1-正常 2-停用` 全项目保持一致
5. **生成后补充字典**：代码生成器创建的 `dict_code` 需在后台手动创建对应数据
## 总结
数据字典通过「字典类型 + 字典项」两层结构管理枚举选项。后端通过 `serialize_maps` 自动转换，前端通过 API 获取下拉数据。代码生成器会自动识别注释中的枚举格式并生成对应配置。
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/dict.md
+194
View File
@@ -0,0 +1,194 @@
:::tip 数据字典
数据字典用于管理系统中的枚举值和配置项，如状态、类型、性别等。通过字典编码统一管理，前端可通过 `useDictStore` 获取字典选项，避免硬编码。后端读取工具位于 `src/common/utils/dict_util.py`，内置进程内 TTL 缓存。
:::
## 模块结构
```
src/modules/dictionary/
├── dict/ # 字典类型（分组）
│ ├── models.py
│ ├── schema.py
│ ├── repository.py
│ ├── service.py
│ └── router.py
└── dict_item/ # 字典项（具体值）
├── models.py
├── schema.py
├── repository.py
├── service.py
└── router.py
```
## 字典模型
### 字典类型（Dict）
```python
# src/modules/dictionary/dict/models.py
from sqlalchemy import Column, String, Integer, text
from src.common.base_db import BaseDB
from src.common.base_model import BaseModel
from config.database import DB_PREFIX
class Dict(BaseModel, BaseDB):
"""字典模型类"""
__tablename__ = DB_PREFIX + "dict"
__table_comment__ = "数据字典表"
# 字典名称
name = Column(String(150), nullable=False, index=True, comment="字典名称")
# 字典编码
code = Column(String(150), nullable=False, index=True, comment="字典编码")
# 字典排序
sort = Column(Integer, default=0, server_default=text('0'), comment="字典排序")
# 字典备注
note = Column(String(255), nullable=True, comment="字典备注")
def __str__(self):
"""返回字典ID作为字符串表示"""
return "字典{}".format(self.id)
```
### 字典项（DictItem）
```python
# src/modules/dictionary/dict_item/models.py
from sqlalchemy import Column, String, Integer, text
from src.common.base_db import BaseDB
from src.common.base_model import BaseModel
from config.database import DB_PREFIX
class DictItem(BaseModel, BaseDB):
"""字典项模型类"""
__tablename__ = DB_PREFIX + "dict_item"
__table_comment__ = "字典项表"
# 字典项名称
name = Column(String(150), nullable=False, index=True, comment="字典项名称")
# 字典项值
value = Column(String(150), nullable=False, comment="字典项值")
# 字典ID
dict_id = Column(Integer, default=0, server_default=text('0'), index=True, comment="字典ID")
# 字典项排序
sort = Column(Integer, default=0, server_default=text('0'), comment="字典项顺序")
# 字典项备注
note = Column(String(255), nullable=True, comment="字典项备注")
def __str__(self):
"""返回字典项ID作为字符串表示"""
return "字典项{}".format(self.id)
```
## 使用示例
### 后端获取字典
```python
# 通过字典编码获取字典项列表
from src.modules.dictionary.dict_item.service import get_by_dict_code
# 获取状态字典
status_list = get_by_dict_code("sys_status")
# 返回: [{"label": "正常", "value": "1"}, {"label": "停用", "value": "2"}]
```
### dict_util 缓存工具
项目提供 `dict_util` 工具（`src/common/utils/dict_util.py`），内置进程内 TTL 缓存。由于字典数据几乎不变，且读写均落在同步 service 路径，因此采用进程内缓存而非 Redis 缓存：
```python
from src.common.utils.dict_util import get_dict_items, get_dict_map, get_dict_label
# 获取字典项列表（自动缓存，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="未知")
# 返回: "正常"
```
缓存实现细节：
```python
# 进程内字典缓存：{code: (过期时间戳, 字典项列表)}
_dict_cache: dict = {}
_CACHE_TTL = 60 # 秒
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
def invalidate_dict(code):
"""清除指定字典编码的缓存（字典/字典项增删改时由 service 主动调用）"""
if code:
_dict_cache.pop(code, None)
def invalidate_dicts_by_ids(dict_ids):
"""按字典ID批量清除缓存：字典项变更后据 dict_id 反查 code 再失效"""
```
:::warning value 类型
字典项的 `value` 字段为字符串类型，`get_dict_label()` 内部会自动将查询值转为字符串匹配。前端使用时注意类型转换。
:::
### 前端使用
前端通过 `useDictStore` 获取字典数据：
```javascript
import { useDictStore } from '@/store/dict'
const dictStore = useDictStore()
// 获取字典选项
const statusOptions = dictStore.getDict('sys_status')
// [{label: "正常", value: "1"}, {label: "停用", value: "2"}]
// 在 el-select 中使用
<el-select v-model="form.status">
<el-option
v-for="item in statusOptions"
:key="item.value"
:label="item.label"
:value="item.value"
/>
</el-select>
```
## 常用字典示例
| 字典编码 | 说明 | 字典项 |
|---------|------|--------|
| `sys_status` | 系统状态 | 正常(1)、停用(2) |
| `sys_gender` | 性别 | 男(1)、女(2) |
| `operation_type` | 操作类型 | 新增(1)、修改(2)、删除(3) 等 |
| `menu_type` | 菜单类型 | 菜单(0)、节点(1) |
## 总结
数据字典模块具备以下特点：
```
1. 两级结构：字典类型（分组）+ 字典项（具体值），清晰管理
2. 进程内缓存：dict_util 内置 TTL 缓存（60秒兜底），变更时主动失效
3. 前端集成：useDictStore 统一管理，el-select 直接绑定
4. 值为字符串：字典项 value 统一为字符串，get_dict_label 自动转字符串匹配
5. 避免硬编码：枚举值统一走字典，修改无需改代码
6. 三种取值：get_dict_items（列表）、get_dict_map（映射）、get_dict_label（单值翻译）
```
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/exception.md
+197
View File
@@ -0,0 +1,197 @@
:::tip 全局异常与响应
项目定义了统一的响应格式和全局异常处理机制。所有 API 返回 `{code, data, msg, ok}` 标准格式，异常被捕获后也转换为相同格式。响应函数位于 `src/core/response.py`，异常类型位于 `src/core/exceptions.py`。
:::
## 统一响应格式
`src/core/response.py` 提供统一的响应生成函数：
```python
from src.core.response import R
# 成功响应
R.ok(data={"id": 1}, msg="操作成功")
# {"code": 0, "data": {"id": 1}, "msg": "操作成功", "ok": true}
# 成功响应（带额外字段）
R.ok(data=list, count=total)
# {"code": 0, "data": [...], "msg": "操作成功", "ok": true, "count": 100}
# 失败响应
R.failed(msg="参数错误")
# {"code": 1, "data": null, "msg": "参数错误", "ok": false}
```
### 响应函数说明
| 函数 | 说明 | code | ok |
|------|------|------|------|
| `R.ok()` | 成功响应 | 0 | true |
| `R.failed()` | 失败响应 | 1 | false |
| `R.page()` | 分页响应 | 0 | true |
| `R.response()` | 通用响应 | 自定义 | 自定义 |
函数签名：
```python
def ok(data=None, msg="操作成功", code=0, **kwargs):
"""生成成功响应，支持通过 kwargs 添加额外字段（如 count）"""
return jsonify({"code": code, "data": data, "msg": msg, "ok": True, **kwargs})
def failed(msg="操作失败", code=1, data=None, **kwargs):
"""生成失败响应"""
return jsonify({"code": code, "data": data, "msg": msg, "ok": False, **kwargs})
def page(data: list, total: int, current: int, size: int, msg="操作成功", code=0):
"""分页响应：封装 {records, total, size, current, pages} 标准分页结构"""
pages = (total + size - 1) // size if size > 0 else 0
return jsonify({
"code": code,
"data": {"records": data, "total": total, "size": size, "current": current, "pages": pages},
"msg": msg,
"ok": True
})
def response(data=None, msg="操作成功", code=0, success=True, **kwargs):
"""通用响应生成器（高级用法），允许完全自定义字段"""
return jsonify({"code": code, "data": data, "msg": msg, "ok": success, **kwargs})
```
### 分页响应
```python
R.page(data=records, total=100, current=1, size=10)
# {
# "code": 0,
# "data": {
# "records": [...],
# "total": 100,
# "size": 10,
# "current": 1,
# "pages": 10
# },
# "msg": "操作成功",
# "ok": true
# }
```
## 异常类型
### BizError
业务异常，业务代码可主动抛出：
```python
# src/core/exceptions.py
class BizError(Exception):
def __init__(self, msg: str = "操作失败", code: int = 1):
self.code = code
self.msg = msg
super().__init__(msg)
```
使用示例：
```python
from src.core.exceptions import BizError
def some_business_logic():
if some_condition:
raise BizError(msg="数据不存在")
```
### ValidationError
参数验证异常，由 WTForms 校验失败时抛出：
```python
from wtforms import ValidationError
class PositionForm(BaseForm):
name = StringField('岗位名称', validators=[...])
def validate_name(self, field):
if some_condition:
raise ValidationError('岗位名称已存在')
```
## 全局异常处理器
`src/app.py` 注册了全局异常处理器，捕获所有异常并转换为标准格式：
```python
def register_exception(app):
# 业务异常
@app.errorhandler(BizError)
def biz_error_handler(exc):
return R.failed(msg=exc.msg, code=exc.code)
# 参数验证异常
@app.errorhandler(ValidationError)
def validation_error_handler(exc):
return R.failed(msg=str(exc))
# HTTP 异常（404/405 等）
@app.errorhandler(404)
def not_found_handler(exc):
return R.failed(msg="请求的资源不存在")
@app.errorhandler(405)
def method_not_allowed_handler(exc):
return R.failed(msg="请求方法不允许")
# 未捕获异常
@app.errorhandler(Exception)
def exception_handler(exc):
# 记录异常日志
app.logger.error(f"未捕获异常: {exc}", exc_info=True)
return R.failed(msg="服务器内部错误")
```
:::tip HTTP 200
所有异常响应均返回 HTTP 200 状态码，业务状态通过 `code` 字段区分。这是前后端分离架构的常见做法，避免浏览器拦截非200响应。
:::
## 状态码常量
```python
# src/core/response.py
class Codes:
SUCCESS = 0
FAILED = 1
UNAUTHORIZED = 401 # 未认证
FORBIDDEN = 403 # 无权限
NOT_FOUND = 404 # 资源不存在
VALIDATE_ERROR = 422 # 验证错误
SERVER_ERROR = 500 # 服务器错误
```
## 异常处理流程
```
请求进入
↓
路由匹配
↓
视图函数执行
├─ 正常返回 → R.ok() / R.failed()
├─ 抛出 BizError → biz_error_handler → R.failed()
├─ 抛出 ValidationError → validation_error_handler → R.failed()
├─ 抛出 HTTP 异常 → http_exception_handler → R.failed()
└─ 抛出其他异常 → exception_handler → R.failed()
↓
返回 JSON 响应
```
## 总结
全局异常与响应模块具备以下特点：
```
1. 统一格式：所有 API 返回 {code, data, msg, ok} 标准格式
2. 全局捕获：业务异常、验证异常、HTTP异常、未捕获异常
3. HTTP 200：所有响应均返回 200，业务状态通过 code 区分
4. 错误聚合：多个验证错误用 / 分隔，便于前端展示
5. 安全隐藏：未捕获异常返回通用提示，不暴露内部错误信息
6. 日志记录：未捕获异常会记录详细日志，便于排查问题
```
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/idempotent.md
+120
View File
@@ -0,0 +1,120 @@
:::tip 防重复提交
防重复提交是保障数据一致性的重要措施。通过前端防抖和后端唯一性校验双重保护，防止用户重复点击导致的数据重复提交。后端唯一性校验通过 WTForms 自定义校验器查询数据库实现。
:::
## 前端防抖
### useLockFn
前端提供 `useLockFn` 钩子函数，实现请求级别的防重复提交：
```javascript
import { useLockFn } from '@/utils/lock'
// 包装提交函数，防止重复点击
const handleSubmit = useLockFn(async () => {
await axios.post('/api/v1/level/add', formData)
ElMessage.success('添加成功')
})
```
`useLockFn` 原理：
```javascript
function useLockFn(fn) {
let isLocked = false
return async (...args) => {
if (isLocked) return
isLocked = true
try {
return await fn(...args)
} finally {
isLocked = false
}
}
}
```
### 按钮 loading
ElementPlus 的按钮组件支持 `loading` 属性，请求期间禁用按钮：
```vue
<el-button
type="primary"
:loading="submitting"
@click="handleSubmit"
>
提交
</el-button>
<script setup>
const submitting = ref(false)
async function handleSubmit() {
submitting.value = true
try {
await axios.post('/api/v1/level/add', formData)
ElMessage.success('添加成功')
} finally {
submitting.value = false
}
}
</script>
```
## 后端幂等性
### 唯一性约束
通过数据库唯一索引和 WTForms 验证器保证数据唯一性：
```python
from wtforms import StringField
from wtforms.validators import DataRequired, Length, ValidationError
class LevelForm(BaseForm):
name = StringField('职级名称', validators=[
DataRequired(message='职级名称不能为空'),
Length(min=1, max=150, message='职级名称长度必须在1-150个字符之间')
])
def validate_name(self, field):
# 查询数据库校验唯一性
exclude_id = self.id.data if self.id.data else None
if LevelRepository.exists_by([Level.name == field.data], exclude_id):
raise ValidationError('职级名称已存在')
```
### 软删除 + 唯一索引
软删除场景下，唯一索引需要特殊处理：
```sql
-- MySQL: 唯一索引包含 is_delete 字段
ALTER TABLE flask_level ADD UNIQUE INDEX uk_name (name, is_delete);
-- PostgreSQL: 部分唯一索引
CREATE UNIQUE INDEX uk_name ON flask_level (name) WHERE is_delete = 0;
```
## 最佳实践
```
1. 前端防抖：useLockFn 或按钮 loading，防止用户重复点击
2. 后端校验：WTForms 验证器 + 数据库唯一索引，双重保障
3. 幂等设计：相同请求多次执行结果一致
4. 错误提示：重复提交时返回明确的错误提示
```
## 总结
防重复提交方案具备以下特点：
```
1. 前端防抖：useLockFn 函数级防抖，按钮 loading 状态控制
2. 后端校验：WTForms 验证器查询数据库校验唯一性
3. 数据库约束：唯一索引兜底，防止并发重复插入
4. 双重保护：前端防抖 + 后端校验，确保数据一致性
5. 用户体验：重复点击时按钮禁用，返回明确提示
```
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/jwt.md
+417
View File
@@ -0,0 +1,417 @@
:::tip JWT认证
JWT（JSON Web Token）是项目的核心认证机制。用户登录成功后获取 Token，后续请求通过 `Authorization: Bearer <token>` 头携带 Token 进行身份验证。项目内置了 Token 生成、解析、刷新和黑名单管理功能，源码位于 `src/core/jwt.py`。
:::
## 认证流程
```
1. 前端发送 POST /api/v1/login（username + password + captcha）
2. 后端校验验证码、用户名密码
3. 登录成功 → create_token() 生成 JWT
4. 前端存储 Token，后续请求携带 Authorization: Bearer <token>
5. login_required 中间件解析 Token → 注入 g 变量
6. 退出登录 → Token 加入 Redis 黑名单立即失效
```
## 配置项
| 环境变量 | 默认值 | 说明 |
|---------|--------|------|
| `JWT_SALT` | 无（必填） | JWT 签名密钥，至少16字符 |
| `JWT_EXPIRE_MINUTES` | `20` | Token 过期时间（分钟） |
:::warning 密钥安全
`JWT_SALT` 必须通过环境变量设置，启动时会校验密钥强度（至少16字符），强度不足将拒绝启动。生成方式：
```bash
python -c "import secrets; print(secrets.token_urlsafe(48))"
```
:::
## Token 生成
`src/core/jwt.py` 中的 `create_token()` 函数负责生成 JWT：
```python
from src.core.jwt import create_token
# 登录成功后生成 Token
access_token = create_token({
"userId": user.id,
"username": user.username,
"realname": user.realname or ''
})
```
函数签名与核心逻辑：
```python
def create_token(payload, timeout=None):
"""
生成JWT令牌
参数:
payload (dict): JWT载荷数据，通常包含userId、username等信息
timeout (int): 令牌过期时间（分钟），默认使用DEFAULT_TIMEOUT_MINUTES（来自JWT_EXPIRE_MINUTES）
返回:
str: 生成的JWT令牌字符串
"""
if timeout is None:
timeout = DEFAULT_TIMEOUT_MINUTES
payload_copy = payload.copy()
current_time = datetime.datetime.now(tz=datetime.timezone.utc)
payload_copy['exp'] = current_time + datetime.timedelta(minutes=timeout)
payload_copy['iat'] = current_time
headers = {"typ": "JWT", "alg": JWT_ALGORITHM}
token = jwt.encode(
payload=payload_copy,
key=JWT_SALT,
algorithm=JWT_ALGORITHM,
headers=headers
)
return token
```
Token 结构包含三部分：
```python
# Header
{
"typ": "JWT",
"alg": "HS256"
}
# Payload（自动添加 exp 和 iat）
{
"userId": 1,
"username": "admin",
"realname": "管理员",
"exp": "2026-09-05T12:00:00Z", # 过期时间
"iat": "2026-09-05T11:40:00Z" # 签发时间
}
```
## Token 解析验证
`parse_payload()` 函数验证 Token 签名和过期状态，返回标准格式结果：
```python
from src.core.jwt import parse_payload
result = parse_payload(token)
if result['code'] == 0:
user_id = result['data']['userId']
username = result['data']['username']
else:
error_msg = result['msg']
# "token已失效，请重新登录" / "token认证失败" / "非法的token"
```
核心实现：
```python
def parse_payload(token):
"""解析验证JWT令牌，返回 {code, data, msg} 标准格式"""
result = {"code": 0, "data": None, "msg": "操作成功"}
if not token:
result['code'] = 1
result['msg'] = "token不能为空"
return result
try:
verified_payload = jwt.decode(token, JWT_SALT, algorithms=[JWT_ALGORITHM])
result['data'] = verified_payload
except exceptions.ExpiredSignatureError:
result['code'] = 1
result['msg'] = "token已失效，请重新登录"
except exceptions.DecodeError:
result['code'] = 1
result['msg'] = "token认证失败，无效的令牌格式"
except exceptions.InvalidTokenError:
result['code'] = 1
result['msg'] = "非法的token，请检查令牌有效性"
return result
```
异常处理说明：
| 异常类型 | 提示信息 | 说明 |
|---------|---------|------|
| `ExpiredSignatureError` | token已失效，请重新登录 | Token 过期 |
| `DecodeError` | token认证失败，无效的令牌格式 | 格式错误或签名无效 |
| `InvalidTokenError` | 非法的token | 算法不匹配等 |
## 登录接口
登录接口 `POST /api/v1/login` 接收 `LoginForm` 参数：
```python
# src/modules/auth/schema.py
class LoginForm(BaseForm):
username = StringField('登录账号', validators=[
DataRequired(message='登录账号不能为空'),
Length(min=1, max=20, message='登录账号长度必须在1-20个字符之间')
])
password = StringField('登录密码', validators=[
DataRequired(message='登录密码不能为空'),
Length(min=6, max=128, message='登录密码长度必须在6-128个字符之间')
])
code = StringField('验证码', validators=[
DataRequired(message='验证码不能为空'),
Length(min=CAPTCHA_LENGTH, max=CAPTCHA_LENGTH, message='验证码长度必须为{}个字符'.format(CAPTCHA_LENGTH))
])
key = StringField('KEY值', validators=[
DataRequired(message='KEY值不能为空')
])
```
登录逻辑位于 `src/modules/auth/service.py`，完整流程包括锁定检查、验证码校验、密码比对、默认密码检测：
```python
def login(data):
ip = get_client_ip()
redis = get_redis()
# 1. 检查是否被锁定（IP+用户名维度）
locked, remaining = _is_locked(redis, ip, data['username'])
if locked:
return R.failed(f"登录失败次数过多，请{remaining}秒后再试")
# 2. 验证码校验（Lua 原子消费，防重放）
if not verify_captcha(data['key'], data['code']):
return R.failed("验证码不正确")
# 3. 查询用户
user = UserRepository.first_by({'username': data['username'], 'status': 1})
if not user:
_record_fail(redis, ip, data['username'])
return R.failed("用户名或密码不正确")
# 4. 比对密码（bcrypt CPU密集）
password_ok = verify_password(data['password'], user.salt or '', user.password or '')
if not password_ok:
_record_fail(redis, ip, data['username'])
return R.failed("用户名或密码不正确")
# 5. 登录成功，清除失败记录
_clear_fail(redis, ip, data['username'])
# 6. 检测是否为默认密码（仅提示，不阻断登录）
is_default_pwd = bool(user.password and verify_password(
DEFAULT_PASSWORD, user.salt or '', user.password or ''))
# 7. 生成 Token
access_token = create_token({
"userId": user.id,
"username": user.username,
"realname": user.realname or ''
})
return R.ok(msg="登录成功", data={
"access_token": access_token,
"must_change_password": is_default_pwd
})
```
登录成功返回：
```json
{
"code": 0,
"msg": "登录成功",
"data": {
"access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9..."
},
"ok": true
}
```
## Token 刷新
Token 支持无感刷新，刷新时会执行令牌轮换（旧 Token 立即失效）：
```python
from src.core.jwt import refresh_token
result = refresh_token(old_token, extend_minutes=30)
if result['code'] == 0:
new_token = result['data']['access_token']
expires_in = result['data']['expires_in'] # 秒
```
刷新核心逻辑：
```python
def refresh_token(token, extend_minutes=None):
# 1. 验证原令牌有效性
result = parse_payload(token)
if result['code'] != 0:
return {"code": 1, "data": None, "msg": f"令牌无效，无法刷新: {result['msg']}"}
# 2. 已注销的令牌禁止刷新（防止攻击者用已注销 token 无限轮换）
if is_token_blacklisted(token):
return {"code": 1, "data": None, "msg": "令牌已注销，无法刷新，请重新登录"}
# 3. 移除JWT标准字段，保留业务数据
payload = result['data']
clean_payload = {k: v for k, v in payload.items() if k not in ['exp', 'iat', 'nbf']}
# 4. 生成新令牌
new_token = create_token(clean_payload, timeout=extend_minutes)
# 5. 令牌轮换：将旧令牌加入黑名单
add_token_to_blacklist(token)
return {
"code": 0,
"data": {"access_token": new_token, "expires_in": extend_minutes * 60},
"msg": "令牌刷新成功"
}
```
:::tip 令牌轮换
刷新后旧 Token 会被加入黑名单，防止旧 Token 在有效期内继续使用。已注销的 Token 不允许刷新，防止攻击者利用已注销令牌无限轮换出新令牌。
:::
## Token 黑名单
退出登录时将 Token 加入 Redis 黑名单，使其立即失效：
```python
from src.core.jwt import add_token_to_blacklist, is_token_blacklisted
# 退出登录：加入黑名单
add_token_to_blacklist(token)
# 中间件校验：检查是否已注销
if is_token_blacklisted(token):
# 拒绝访问
```
黑名单实现细节：
```python
# 黑名单 Key 前缀
_TOKEN_BLACKLIST_PREFIX = 'token:blacklist:'
# SHA256 指纹存储，不将完整 Token 存入 Redis
def _token_fingerprint(token: str) -> str:
return hashlib.sha256(token.encode('utf-8')).hexdigest()
def add_token_to_blacklist(token: str) -> bool:
"""退出登录时调用，TTL 与 Token 剩余有效期一致"""
expiration = get_token_expiration(token)
ttl_seconds = int((expiration - datetime.datetime.now(tz=datetime.timezone.utc)).total_seconds())
fingerprint = _token_fingerprint(token)
key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}'
redis.set(key, '1', ex=ttl_seconds)
def is_token_blacklisted(token: str) -> bool:
"""检查 Token 是否已注销，Redis 不可用时 fail-open 放行"""
try:
fingerprint = _token_fingerprint(token)
key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}'
return redis.exists(key) > 0
except Exception as e:
# fail-open：Redis 故障时放行，避免全站不可用
_log_blacklist_check_alert(e)
return False
```
黑名单特性：
- 使用 SHA256 指纹存储，不将完整 Token 存入 Redis
- TTL 与 Token 剩余有效期一致，过期自动清理
- Redis 不可用时采用 fail-open 策略放行（避免全站不可用）
- 告警节流：Redis 故障期间首次失败记 ERROR，随后每 60 秒至多一条 ERROR，避免日志刷屏
## 请求中的 Token 提取
```python
from src.core.jwt import get_access_token, parse_token
# 从请求头提取 Token
success, token, msg = get_access_token()
# 完整解析（提取 + 验证）
result = parse_token()
# result['code'] == 0 表示成功
# result['data']['userId'], result['data']['username']
```
`get_access_token()` 支持 Bearer 前缀大小写不敏感（RFC 7235）：
```python
def get_access_token():
"""从请求头中提取 access_token，返回 (success, token, message)"""
from flask import request
auth_header = request.headers.get('Authorization', '')
stripped = auth_header.strip()
if stripped.lower().startswith('bearer '):
access_token = stripped[7:].strip()
else:
access_token = stripped
return True, access_token, 'success'
```
## login_required 中间件
`src/middleware/auth.py` 中的 `login_required` 是全局认证中间件，应用于所有 v1 路由：
```python
# 认证成功后注入 g 变量
from flask import g
g.user_id = userId
g.username = username
g.realname = realname
```
业务代码通过 `src/core/security.py` 获取当前用户信息：
```python
from src.core.security import get_user_id, get_username, get_realname
user_id = get_user_id()
username = get_username()
realname = get_realname()
```
## 前端集成
前端登录后将 Token 存入 localStorage，每次请求自动携带：
```javascript
// 请求拦截器
axios.interceptors.request.use(config => {
const token = localStorage.getItem('access_token')
if (token) {
config.headers['Authorization'] = `Bearer ${token}`
}
return config
})
// 响应拦截器：检测 Token 过期
axios.interceptors.response.use(response => {
if (response.data.code === 401) {
// Token 过期，跳转登录页
router.push('/login')
}
return response
})
```
## 总结
项目的 JWT 认证方案具备以下特点：
```
1. 安全性：HS256 算法 + 强密钥校验（至少16字符），启动时 fail-close
2. 令牌轮换：刷新时旧 Token 立即加入黑名单，防止重放攻击
3. 黑名单管理：SHA256 指纹存储，TTL 自动清理，Redis 故障时 fail-open 降级
4. 防暴力破解：登录失败计数 + 账户锁定（IP+用户名维度）
5. 全局统一：login_required 中间件 + g 变量注入，业务代码零侵入
```
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/login-log.md
+122
View File
@@ -0,0 +1,122 @@
:::tip 登录日志
登录日志记录每次登录尝试（成功/失败），包含登录账号、IP地址、浏览器信息、登录时间等。支持 IP 地理位置解析（ip2region）和 User-Agent 解析。登录逻辑位于 `src/modules/auth/service.py`，日志记录位于 `src/modules/login_log/service.py`。
:::
## 记录时机
登录日志在以下时机自动记录：
```
1. 登录成功 → 记录成功日志（status=0）
2. 登录失败（密码错误）→ 记录失败日志（status=1）
3. 登录失败（用户不存在）→ 记录失败日志（status=1）
4. 登录失败（账户被锁定）→ 记录失败日志（status=1）
```
## 日志字段
| 字段 | 说明 |
|------|------|
| `username` | 登录账号 |
| `ip` | 客户端 IP 地址 |
| `ip_location` | IP 地理位置（ip2region 解析） |
| `browser` | 浏览器类型 |
| `os` | 操作系统 |
| `status` | 登录状态：0-成功 1-失败 |
| `message` | 提示信息 |
| `login_time` | 登录时间 |
## IP 地理位置解析
项目集成 ip2region 库，自动将 IP 解析为地理位置：
```python
# src/common/utils/ip2region.py
from src.common.utils.ip2region import get_ip_location
location = get_ip_location("116.25.146.177")
# 返回: "广东省深圳市"
```
:::tip ip2region
ip2region 是离线 IP 地理位置解析库，数据文件存储在 `static/ip2region.xdb`，无需网络请求，解析速度极快。
:::
## User-Agent 解析
项目解析请求头中的 User-Agent，提取浏览器和操作系统信息：
```python
# src/common/utils/user_agent.py
from src.common.utils.user_agent import parse_ua
ua_info = parse_ua(user_agent_string)
# 返回: {"browser": "Chrome 120", "os": "Windows 10"}
```
## 查询接口
```
GET /api/v1/login/log/list?pageNo=1&pageSize=10
GET /api/v1/login/log/detail/<id>
```
## 防暴力破解
登录日志配合登录失败锁定机制，防止暴力破解。锁定逻辑按 IP+用户名 维度：
```python
# src/modules/auth/service.py
_LOGIN_FAIL_PREFIX = 'login:fail:'
_LOGIN_LOCK_PREFIX = 'login:lock:'
def _fail_key(ip: str, username: str) -> str:
"""登录失败计数Key：按IP+用户名维度"""
return f'{_LOGIN_FAIL_PREFIX}{ip}:{username}'
def _lock_key(ip: str, username: str) -> str:
"""登录锁定Key"""
return f'{_LOGIN_LOCK_PREFIX}{ip}:{username}'
def _is_locked(redis, ip: str, username: str) -> tuple:
"""检查是否处于锁定状态，返回 (是否锁定, 剩余秒数)"""
key = _lock_key(ip, username)
ttl = redis.ttl(key)
if ttl and ttl > 0:
return True, ttl
return False, 0
def _record_fail(redis, ip: str, username: str):
"""记录一次登录失败，达到上限则锁定"""
key = _fail_key(ip, username)
# 原子 incr + 首次设置过期（Lua 脚本，防止崩溃窗口内 key 永久残留）
count = incr_with_expire(redis, key, LOGIN_LOCK_SECONDS)
if count >= LOGIN_FAIL_MAX:
lock_k = _lock_key(ip, username)
redis.set(lock_k, '1', ex=LOGIN_LOCK_SECONDS)
def _clear_fail(redis, ip: str, username: str):
"""登录成功后清除失败记录"""
fail_k = _fail_key(ip, username)
lock_k = _lock_key(ip, username)
redis.delete(fail_k, lock_k)
```
配置项：
| 环境变量 | 默认值 | 说明 |
|---------|--------|------|
| `LOGIN_FAIL_MAX` | `5` | 登录失败次数上限 |
| `LOGIN_LOCK_SECONDS` | `300` | 锁定时间（秒） |
## 总结
登录日志模块具备以下特点：
```
1. 自动记录：登录成功/失败自动记录，无需手动埋点
2. IP 解析：ip2region 离线解析，速度快、无网络依赖
3. UA 解析：自动提取浏览器和操作系统信息
4. 防暴力破解：失败计数 + 账户锁定（IP+用户名维度）
5. 安全审计：完整记录登录行为，便于安全审计
```
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/operation-log.md
+173
View File
@@ -0,0 +1,173 @@
:::tip 操作日志
操作日志中间件自动拦截写请求（POST/PUT/DELETE/PATCH）并记录操作日志，无需手动埋点。日志落库通过后台线程异步执行，不阻塞响应。源码位于 `src/middleware/operation_log.py`。
:::
## 工作原理
```
1. 请求进入 → operation_log 判断是否需要记录
2. 仅拦截写请求（POST/PUT/DELETE/PATCH），排除登录/验证码等路径
3. 记录请求参数、响应结果、耗时、操作类型
4. 日志落库通过后台线程异步执行
5. 不阻塞主请求，对性能影响极小
```
## 排除路径
以下路径不记录操作日志：
```python
_EXCLUDE_PATHS = (
'/api/v1/login',
'/api/v1/logout',
'/api/v1/captcha',
'/api/v1/upload',
'/api/v1/loginlog',
'/api/v1/operationlog',
'/api/v1/index/menu',
'/api/v1/index/user',
)
```
## 操作类型自动推断
中间件根据 URL 路径末段自动推断操作类型，映射表定义在 `_TYPE_MAP` 中：
```python
_TYPE_MAP = {
'add': 1, # 新增
'update': 2, # 修改
'delete': 3, # 删除
'batchdelete': 4, # 批量删除
'status': 5, # 状态设置
'resetpwd': 6, # 重置密码
'updatepassword': 7, # 修改密码
'import': 8, # 导入
'export': 9, # 导出
'kickout': 10, # 强退
'generate': 11, # 生成代码
'clear': 12, # 清空数据
'auth': 13, # 授权
'refresh': 14, # 缓存
'openaccount': 15, # 开账号
}
def _infer_type(request) -> int:
"""从 URL 路径末段推断操作类型，匹配不到返回 0（其它）"""
path = request.path.rstrip('/')
last_segment = path.split('/')[-1].lower()
# 跳过纯数字路径参数（如 /delete/123 → 取 delete）
if last_segment.isdigit():
parts = path.split('/')
last_segment = parts[-2].lower() if len(parts) >= 2 else ''
return _TYPE_MAP.get(last_segment, 0)
```
| 路径末段 | 操作类型 | 类型ID |
|---------|---------|--------|
| `add` | 新增 | 1 |
| `update` | 修改 | 2 |
| `delete` | 删除 | 3 |
| `batchdelete` | 批量删除 | 4 |
| `status` | 状态设置 | 5 |
| `resetpwd` | 重置密码 | 6 |
| `updatepassword` | 修改密码 | 7 |
| `import` | 导入 | 8 |
| `export` | 导出 | 9 |
| `kickout` | 强退 | 10 |
| `generate` | 生成代码 | 11 |
| `clear` | 清空数据 | 12 |
| `auth` | 授权 | 13 |
| `refresh` | 缓存 | 14 |
| `openaccount` | 开账号 | 15 |
| 其它 | 其它 | 0 |
## 日志字段
| 字段 | 说明 |
|------|------|
| `title` | 操作标题（从路由装饰器获取） |
| `type` | 操作类型（自动推断） |
| `method` | 路由方法名 |
| `url` | 请求路径 |
| `param` | 请求参数（截取前2000字符） |
| `result` | 响应结果（截取前2000字符） |
| `consume_time` | 耗时（毫秒） |
| `status` | 状态：0-成功 1-失败 |
| `error` | 错误信息 |
| `create_user` | 操作人 |
## 查询接口
```
GET /api/v1/oper/log/list?pageNo=1&pageSize=10
GET /api/v1/oper/log/detail/<id>
```
## 性能优化
```
1. 后台任务：日志落库通过后台线程异步执行，不阻塞响应
2. 前缀解析：仅解析响应体前 8KB，避免大响应体全量解析
3. 长度截断：参数和结果截取前 2000 字符，防止撑爆日志表
4. 路径排除：登录/验证码等高频路径不记录
5. 路径段边界匹配：排除前缀按 path == prefix 或 prefix + '/' 匹配，避免子串误伤
```
## 中间件实现
```python
# src/middleware/operation_log.py
from flask import request, g
import time
import threading
def register_operation_log(app):
"""注册操作日志中间件"""
@app.before_request
def _start_timer():
g.start_time = time.time()
@app.after_request
def _log_operation(response):
# 仅记录写请求
if request.method not in ('POST', 'PUT', 'DELETE', 'PATCH'):
return response
# 排除特定路径
if request.path in _EXCLUDE_PATHS:
return response
# 计算耗时
consume_time = int((time.time() - g.start_time) * 1000)
# 获取响应内容
result = response.get_data(as_text=True)[:2000]
# 异步记录日志
def _save_log():
# 保存日志到数据库
pass
thread = threading.Thread(target=_save_log)
thread.start()
return response
```
:::tip 后台线程
Flask 使用后台线程异步记录操作日志，不阻塞主请求。日志记录失败不会影响正常业务流程。
:::
## 总结
操作日志模块具备以下特点：
```
1. 自动记录：中间件拦截写请求，无需手动埋点
2. 类型推断：根据 URL 路径自动推断操作类型
3. 异步落库：后台线程执行，不阻塞主请求
4. 性能友好：前缀解析 + 长度截断 + 路径排除
5. 完整信息：记录参数、结果、耗时、操作人等
```
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/rate-limit.md
+145
View File
@@ -0,0 +1,145 @@
:::tip 请求限流
请求限流中间件基于 Redis 实现 IP 级别的滑动窗口限流，防止恶意请求和 DoS 攻击。默认关闭，可通过环境变量开启。源码位于 `src/middleware/rate_limit.py`，滑动窗口 Lua 脚本位于 `src/core/redis_client.py`。
:::
## 配置项
| 环境变量 | 默认值 | 说明 |
|---------|--------|------|
| `RATE_LIMIT_ENABLED` | `False` | 是否启用限流 |
| `RATE_LIMIT_LIMIT` | `100` | 单窗口内每个 IP 最大请求数 |
| `RATE_LIMIT_WINDOW_SECONDS` | `60` | 统计窗口时长（秒） |
| `RATE_LIMIT_KEY_PREFIX` | `rl:` | Redis 计数 key 前缀 |
:::warning 生产环境建议
生产环境建议开启限流（`RATE_LIMIT_ENABLED=True`），根据实际流量调整 `RATE_LIMIT_LIMIT` 和 `RATE_LIMIT_WINDOW_SECONDS`。
:::
## 工作原理
```
1. 请求进入 → 检查 RATE_LIMIT_ENABLED 开关
2. 获取客户端 IP（支持 X-Forwarded-For 代理头）
3. 使用滑动窗口算法计算当前窗口请求数
4. 超过限制 → 返回 {"code":1, "msg":"请求过于频繁，请 X 秒后再试"}
5. 未超过 → 放行请求
```
## 滑动窗口算法
采用滑动窗口计数器，按当前窗口与上一窗口的计数时间加权，消除固定窗口边界处的 2x 突发问题。
### 中间件实现
```python
# src/middleware/rate_limit.py
from flask import request, g
import time
def register_rate_limit(app):
"""注册限流中间件"""
@app.before_request
def rate_limit():
"""按客户端 IP 做滑动窗口限流，超限直接返回统一 200 + code 响应"""
from flask import current_app
if not current_app.config.get('RATE_LIMIT_ENABLED'):
return
# CORS 预检请求不计数，避免浏览器预检占用限流额度
if request.method == 'OPTIONS':
return
# Redis 未就绪时降级放行
redis = getattr(g, 'redis', None)
if redis is None:
return
client_ip = get_client_ip()
now = time.time()
window_index = int(now // RATE_LIMIT_WINDOW_SECONDS)
current_key = f"{RATE_LIMIT_KEY_PREFIX}{client_ip}:{window_index}"
prev_key = f"{RATE_LIMIT_KEY_PREFIX}{client_ip}:{window_index - 1}"
# 原子滑动窗口计数
try:
count = sliding_window_incr(
redis, current_key, prev_key, RATE_LIMIT_WINDOW_SECONDS, window_index, now)
except Exception as e:
app.logger.warning(f"限流计数异常，降级放行: {e}")
return
# 直接返回而非抛异常
if count > RATE_LIMIT_LIMIT:
try:
ttl = redis.ttl(current_key)
except Exception:
ttl = RATE_LIMIT_WINDOW_SECONDS
from src.core.response import R
return R.failed(msg=f"请求过于频繁，请 {ttl} 秒后再试")
```
### Lua 脚本
滑动窗口的核心在 `src/core/redis_client.py` 中的 Lua 脚本：
```python
_SLIDING_WINDOW_LUA = """
local cur = redis.call('incr', KEYS[1])
if cur == 1 then
redis.call('expire', KEYS[1], ARGV[1])
end
local prev = tonumber(redis.call('get', KEYS[2]) or '0')
local elapsed = tonumber(ARGV[3]) - tonumber(ARGV[2]) * tonumber(ARGV[1])
local f = elapsed / tonumber(ARGV[1])
if f < 0 then f = 0 elseif f > 1 then f = 1 end
local effective = prev * (1 - f) + cur
return effective
"""
def sliding_window_incr(redis, current_key, prev_key,
window_seconds, window_index, now) -> float:
"""滑动窗口限流计数：返回当前滑动窗口内的加权请求数（近似精确）"""
return redis.eval(
_SLIDING_WINDOW_LUA, 2, current_key, prev_key,
window_seconds, window_index, now)
```
算法原理：当前窗口计数 `cur` + 上一窗口计数 `prev` 按时间比例加权（`prev * (1 - f)`），窗口边界处请求数平滑过渡，消除固定窗口在边界整体重置导致的 2x 突发。
## 降级策略
```
1. Redis 未连接：直接放行，不阻塞请求
2. Redis 运行期故障：降级放行，记录警告日志
3. 限流计数异常：降级放行，避免全站不可用
```
:::tip fail-open 策略
限流采用 fail-open 策略：Redis 不可用时放行所有请求，优先保证服务可用性。这是安全与可用性的权衡。
:::
## 中间件位置
限流中间件在 `src/app.py` 中注册为最外层（CORS 内层）：
```python
# 最外层：限流（被限流的请求不进入 DB 会话与操作日志）
register_rate_limit(app)
# CORS（确保限流响应也带 CORS 头）
register_cors(app)
```
## 总结
请求限流模块具备以下特点：
```
1. 滑动窗口：消除固定窗口边界处的 2x 突发问题
2. IP 维度：按客户端 IP 限流，支持代理头
3. 默认关闭：opt-in 模式，按需开启
4. 降级策略：Redis 故障时 fail-open 放行
5. 最外层拦截：限流请求不进入 DB 会话和操作日志
6. Lua 原子操作：计数和过期设置原子执行，避免并发问题
```
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/rbac.md
+240
View File
@@ -0,0 +1,240 @@
:::tip RBAC权限
项目采用基于角色的访问控制（RBAC）模型，通过用户-角色-菜单三层关系实现细粒度的权限管理。权限控制贯穿前后端：后端通过 `@check_permission` 装饰器校验（`src/core/permission.py`），前端通过 `v-perm` 指令控制按钮显隐。
:::
## 权限模型
```
用户(User) ──N:N── 角色(Role) ──N:N── 菜单(Menu)
│
type=0: 菜单（页面路由）
type=1: 节点（权限按钮）
```
数据库表关系：
| 关联表 | 说明 |
|--------|------|
| `flask_user_role` | 用户-角色关联表 |
| `flask_role_menu` | 角色-菜单关联表 |
| `flask_menu` | 菜单表（含权限节点） |
## 权限节点格式
权限节点采用 `sys:module:action` 三段式命名：
```
sys:user:list 用户列表
sys:user:add 用户新增
sys:user:update 用户编辑
sys:user:delete 用户删除
sys:user:resetPwd 重置密码
sys:role:list 角色列表
sys:role:add 角色新增
sys:menu:list 菜单列表
sys:dept:list 部门列表
sys:level:list 职级列表
sys:position:list 岗位列表
```
## 菜单模型
```python
# src/modules/system/menu/models.py
# ============================================================
# 菜单模型
# ============================================================
class Menu(BaseModel, BaseDB):
"""菜单模型类"""
# ============================================================
# 表名配置
# ============================================================
__tablename__ = DB_PREFIX + "menu"
__table_comment__ = "菜单表"
# ============================================================
# 字段定义
# ============================================================
# 菜单名称
name = Column(String(150), nullable=False, index=True, comment="菜单名称")
# 菜单图标
icon = Column(String(50), nullable=True, comment="菜单图标")
# 菜单图标（备用）
icon2 = Column(String(50), nullable=True, comment="菜单图标")
# 上级ID
parent_id = Column(Integer, default=0, server_default=text('0'), index=True, comment="上级ID")
# 菜单路径
path = Column(String(255), nullable=True, comment="菜单路径")
# 菜单组件
component = Column(String(255), nullable=True, comment="菜单组件")
# 权限节点
permission = Column(String(150), nullable=True, comment="权限节点")
# 打开方式：0-组件 1-内链 2-外链
target = Column(Integer, default=1, server_default=text('1'), comment="打开方式：0-组件 1-内链 2-外链")
# 菜单类型：0-菜单 1-节点
type = Column(Integer, default=0, server_default=text('0'), index=True, comment="菜单类型：0-菜单 1-节点")
# 状态选项
status = Column(Integer, default=0, server_default=text('0'), index=True, comment="菜单状态：0-正常 1-停用")
# 是否可见：0-可见 1-不可见
hide = Column(Integer, default=1, server_default=text('1'), comment="是否可见：0-可见 1-不可见")
# 排序
sort = Column(Integer, default=0, server_default=text('0'), comment="排序")
# 备注
note = Column(String(255), nullable=True, comment="备注")
# ============================================================
# 内置方法
# ============================================================
def __str__(self):
"""返回菜单ID作为字符串表示"""
return "菜单{}".format(self.id)
```
菜单类型说明：
| type | 说明 | 用途 |
|------|------|------|
| `0` | 菜单 | 页面路由，前端生成菜单导航 |
| `1` | 节点 | 权限按钮，控制操作权限 |
## 后端权限校验
### check_permission 装饰器
`src/core/permission.py` 中的 `@check_permission` 装饰器用于端点级权限校验：
```python
from src.core.permission import check_permission
@position_blue.route('/add', methods=['POST'])
@check_permission('sys:position:add')
def add():
from src.modules.system.position.service import add_position
from src.modules.system.position.schema import PositionForm
data = parse_request_body(request, PositionForm)
return add_position(data)
```
校验逻辑（完整实现）：
```python
from functools import wraps
from flask import request, g
from src.core.response import R
from src.core.security import get_user_id
def check_permission(permission: str):
"""节点权限鉴权装饰器，userId==1（管理员）直接放行"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 获取当前用户ID
user_id = get_user_id()
# 管理员免检
if user_id == 1:
return func(*args, **kwargs)
# 获取用户权限列表
from src.modules.system.menu.service import get_permissions_list
permission_list = get_permissions_list(user_id)
# 检查权限
if permission not in permission_list:
return R.failed("权限不足")
return func(*args, **kwargs)
return wrapper
return decorator
```
:::warning Request 提取
装饰器通过 `flask.request` 获取当前请求对象，无需显式传递 request 参数。
:::
:::tip 管理员免检
用户ID为1的管理员账号（admin）自动跳过所有权限校验，拥有全部权限。
:::
### 权限缓存
权限列表通过 Redis 缓存，提高查询性能：
```python
# 获取用户的权限节点列表（带 Redis 缓存）
def get_permissions_list(user_id: int) -> list:
# 1. 先查 Redis 缓存
cache_key = f"user:permissions:{user_id}"
cached = redis.get(cache_key)
if cached:
return json.loads(cached)
# 2. 缓存未命中 → 查 DB（user → role → menu where type=1）
permission_list = _query_permissions_from_db(user_id)
# 3. 写入缓存
redis.set(cache_key, json.dumps(permission_list), ex=3600)
return permission_list
```
## 前端权限控制
### v-perm 指令
前端使用自定义 `v-perm` 指令控制按钮显隐：
```html
<!-- 只有拥有 sys:user:add 权限的用户才能看到此按钮 -->
<el-button v-perm="'sys:user:add'" type="primary">新增</el-button>
<!-- 多个权限，满足其一即可 -->
<el-button v-perm="['sys:user:update', 'sys:user:delete']">操作</el-button>
```
### 路由守卫
前端路由守卫根据用户菜单权限动态生成路由，无权限的页面不会出现在导航菜单中。
## 权限数据查询
权限列表通过角色-菜单关联查询：
```python
# 获取用户的权限节点列表（带 Redis 缓存）
def get_permissions_list(user_id: int) -> list:
# 1. 先查 Redis 缓存
# 2. 缓存未命中 → 查 DB（user → role → menu where type=1）
# 3. 返回 permission 字段列表
pass
```
## 权限缓存失效
当用户权限变更时，需要清除对应的缓存：
```python
def clear_permission_cache(user_id: int):
"""清除用户权限缓存"""
cache_key = f"user:permissions:{user_id}"
redis.delete(cache_key)
```
权限变更场景：
1. 用户角色变更
2. 角色菜单变更
3. 菜单权限节点变更
## 总结
项目的 RBAC 权限方案具备以下特点：
```
1. 三层模型：用户-角色-菜单，支持灵活的角色分配
2. 节点权限：sys:module:action 格式，细粒度控制每个操作
3. 前后端统一：后端 @check_permission + 前端 v-perm 指令
4. 管理员免检：userId=1 自动跳过权限校验
5. 权限缓存：Redis 缓存权限列表，提高查询性能
6. 菜单路由：type=0 菜单生成前端路由，type=1 节点控制按钮权限
7. 懒加载：装饰器内懒加载业务 service，避免 core → 业务模块顶层循环依赖
```
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/transaction-decorator.md
+172
View File
@@ -0,0 +1,172 @@
# 事务装饰器
除了 AOP 中间件自动管理事务外，项目还提供了事务装饰器，用于特殊场景下的事务控制。
## @transactional 装饰器
`@transactional` 装饰器用于需要显式控制事务的函数：
```python
from src.middleware.transaction import transactional
@transactional
def complex_operation(data):
"""复杂操作（需要显式事务控制）"""
# 所有数据库操作在同一个事务中
user = User(**data)
user.save()
# 如果抛出异常，整个事务会回滚
if not check_condition(user):
raise BizError("条件不满足")
# 其他操作
# ...
return user
```
### 装饰器实现
```python
from functools import wraps
from src.core.database import db
def transactional(func):
"""事务装饰器：确保函数在事务中执行"""
@wraps(func)
def wrapper(*args, **kwargs):
try:
result = func(*args, **kwargs)
db.session.commit()
return result
except Exception as e:
db.session.rollback()
raise
return wrapper
```
### 使用场景
1. **定时任务**：定时任务不经过 HTTP 请求，不触发中间件
2. **批量导入**：大量数据导入需要显式控制事务边界
3. **跨模块操作**：多个模块的操作需要在同一个事务中
## @transactional_response 装饰器
`@transactional_response` 装饰器用于返回 R 对象的函数，根据响应 code 自动决定提交或回滚：
```python
from src.middleware.transaction import transactional_response
@transactional_response
def add_position(data):
"""新增岗位"""
position = Position(**data)
position.save()
# 返回成功响应时自动提交
return R.ok(msg="新增成功")
# 返回失败响应时自动回滚
# return R.failed("新增失败")
```
### 装饰器实现
```python
from functools import wraps
from src.core.database import db
def transactional_response(func):
"""事务响应装饰器：根据响应 code 自动提交或回滚"""
@wraps(func)
def wrapper(*args, **kwargs):
try:
result = func(*args, **kwargs)
# 检查响应是否为 R 对象
if hasattr(result, 'get_json'):
body = result.get_json()
if body and body.get('code') == 0:
db.session.commit()
else:
db.session.rollback()
else:
# 非 R 对象，默认提交
db.session.commit()
return result
except Exception as e:
db.session.rollback()
raise
return wrapper
```
### 使用场景
1. **Service 层方法**：Service 层方法返回 R 对象
2. **需要精细控制**：需要根据业务逻辑决定是否提交
## TransactionContext 上下文管理器
`TransactionContext` 上下文管理器用于需要立即提交的场景：
```python
from src.middleware.transaction import TransactionContext
def import_data(file):
"""导入数据（需要立即提交）"""
with TransactionContext():
# 在此上下文中的操作会立即提交
for row in parse_excel(file):
position = Position(**row)
position.save()
# 上下文结束后自动提交
return R.ok(msg="导入成功")
```
### 实现原理
```python
from contextlib import contextmanager
from src.core.database import db
@contextmanager
def TransactionContext():
"""事务上下文管理器：立即提交"""
try:
yield
db.session.commit()
except Exception as e:
db.session.rollback()
raise
```
### 使用场景
1. **批量导入**：大量数据导入时，每批次立即提交
2. **分步操作**：需要分步执行，每步成功后立即提交
3. **外部调用**：调用外部服务前需要确保数据已持久化
## 装饰器选择指南
| 场景 | 推荐方式 | 说明 |
|:-----|:---------|:-----|
| 标准 CRUD | AOP 中间件 | 自动管理，无需额外代码 |
| 定时任务 | `@transactional` | 不经过 HTTP 请求 |
| 批量导入 | `TransactionContext` | 需要立即提交 |
| Service 层 | `@transactional_response` | 根据响应 code 决定 |
| 复杂业务 | `@transactional` | 需要显式控制 |
## 注意事项
:::warning 重要提示
1. **不要混用多种方式**：同一函数不要同时使用 AOP 中间件和装饰器
2. **装饰器顺序**：如果同时使用多个装饰器，事务装饰器应该在最外层
3. **异常处理**：装饰器会自动捕获异常并回滚，但业务代码仍应处理特定异常
4. **嵌套事务**：SQLAlchemy 不支持真正的嵌套事务，嵌套调用会使用同一个事务
:::
## 总结
事务装饰器提供了灵活的事务控制方式，适用于特殊场景。`@transactional` 用于需要显式控制事务的函数，`@transactional_response` 用于返回 R 对象的函数，`TransactionContext` 用于需要立即提交的场景。选择合适的事务管理方式，可以简化业务代码，提高代码可维护性。
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/transaction.md
+209
View File
@@ -0,0 +1,209 @@
# AOP 事务管理
项目采用 AOP（面向切面编程）风格的事务管理机制，通过中间件自动管理数据库事务，业务代码无需手动调用 `commit` 或 `rollback`。
## 设计理念
传统的事务管理方式需要在业务代码中显式调用 `commit` 和 `rollback`，这种方式存在以下问题：
1. **代码侵入性强**：业务代码需要关注事务细节
2. **容易遗漏**：忘记调用 `rollback` 可能导致数据不一致
3. **重复代码**：每个写操作都需要相同的事务处理逻辑
AOP 事务管理通过中间件自动处理事务，业务代码只需关注业务逻辑。
## 实现原理
事务管理通过 `src/middleware/transaction.py` 实现：
```python
from flask import request, g
from src.core.database import db
def register_transaction(app):
"""注册 AOP 事务管理中间件"""
@app.before_request
def _transaction_begin():
"""请求开始前：标记事务活跃"""
g.transaction_active = True
@app.after_request
def _transaction_commit(response):
"""请求结束后：根据响应状态决定提交或回滚"""
if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
if response.status_code == 200:
try:
body = response.get_json()
if body and body.get('code') == 0:
# 业务成功：提交事务
db.session.commit()
else:
# 业务失败：回滚事务
db.session.rollback()
except Exception:
db.session.rollback()
else:
# HTTP 非 200：回滚事务
db.session.rollback()
return response
@app.teardown_request
def _transaction_rollback(exception):
"""请求异常时：回滚事务"""
if exception:
db.session.rollback()
# 移除会话
db.session.remove()
```
## 事务触发规则
| HTTP 方法 | 响应状态 | 业务 code | 事务操作 |
|:----------|:---------|:----------|:---------|
| POST | 200 | 0（成功） | commit |
| POST | 200 | 1（失败） | rollback |
| PUT | 200 | 0（成功） | commit |
| PUT | 200 | 1（失败） | rollback |
| DELETE | 200 | 0（成功） | commit |
| DELETE | 200 | 1（失败） | rollback |
| PATCH | 200 | 0（成功） | commit |
| PATCH | 200 | 1（失败） | rollback |
| GET | 任意 | 任意 | 不处理（只读） |
| 任意 | 非 200 | 任意 | rollback |
| 任意 | 异常 | 任意 | rollback |
## 业务代码示例
### 标准 CRUD 操作
```python
from src.modules.system.position.models import Position
def add_position(data):
"""新增岗位"""
# 创建模型实例
position = Position()
position.name = data['name']
position.status = data['status']
position.sort = data['sort']
# 保存到数据库（仅 flush，不 commit）
position.save()
# 返回成功响应
return R.ok(msg="新增成功")
```
### 批量操作
```python
def batch_delete_positions(id_list):
"""批量删除岗位"""
# 批量软删除（仅 flush，不 commit）
PositionRepository.batch_delete(id_list)
# 返回成功响应
return R.ok(msg="删除成功")
```
### 多表操作
```python
def transfer_user(user_id, new_dept_id):
"""转移用户部门"""
# 更新用户部门
user = UserRepository.get_by_id(user_id)
user.dept_id = new_dept_id
user.save()
# 更新部门人数统计
dept = DeptRepository.get_by_id(new_dept_id)
dept.user_count = DeptRepository.count_users(new_dept_id)
dept.save()
# 两个操作在同一个事务中，要么都成功，要么都回滚
return R.ok(msg="转移成功")
```
## BaseDB.save() 方法
`BaseDB` 基类提供的 `save()` 方法只执行 `flush`，不执行 `commit`：
```python
class BaseDB:
def save(self):
"""保存模型到数据库（仅 flush，不 commit）"""
db.session.add(self)
db.session.flush()
return self
```
`flush` 的作用：
1. 将 SQL 语句发送到数据库执行
2. 获取自增 ID 等数据库生成的值
3. 但不提交事务，事务由中间件统一管理
## 特殊情况处理
### 需要立即提交的场景
某些特殊场景需要立即提交事务，可以使用 `TransactionContext` 上下文管理器：
```python
from src.middleware.transaction import TransactionContext
def import_data(file):
"""导入数据（需要立即提交）"""
with TransactionContext():
# 在此上下文中的操作会立即提交
for row in parse_excel(file):
position = Position(**row)
position.save()
# 上下文结束后自动提交
return R.ok(msg="导入成功")
```
### 需要手动回滚的场景
某些业务场景需要手动回滚：
```python
from src.core.database import db
def complex_operation():
"""复杂操作（需要手动回滚）"""
try:
# 执行一系列操作
user = User(...)
user.save()
# 检查业务条件
if not check_condition(user):
# 手动回滚
db.session.rollback()
return R.failed("条件不满足")
# 继续其他操作
# ...
return R.ok(msg="操作成功")
except Exception as e:
# 异常时手动回滚
db.session.rollback()
raise
```
## 注意事项
:::warning 重要提示
1. **不要手动调用 `db.session.commit()`**：事务由中间件统一管理，手动 commit 可能导致事务提前提交
2. **不要手动调用 `db.session.rollback()`**：除非在 `try-except` 块中需要手动回滚
3. **只读操作使用 GET 方法**：GET 请求不会触发事务处理
4. **写操作必须返回 R 对象**：中间件通过 `code` 字段判断是否需要提交
:::
## 总结
AOP 事务管理通过中间件自动处理数据库事务，业务代码无需手动调用 `commit` 或 `rollback`。`save()` 方法只执行 `flush`，事务由中间件在请求结束时统一提交或回滚。这种设计简化了业务代码，提高了代码可维护性，同时保证了数据一致性。
DjangoAdmin_Flask_NaiveVue/zh/volume4/core/validator.md
+225
View File
@@ -0,0 +1,225 @@
:::tip 参数验证
项目使用 WTForms 进行请求参数验证。所有请求体参数通过 WTForms 表单类定义，框架自动解析和验证 JSON 请求体，验证失败返回标准格式错误信息。BaseForm 基类位于 `src/common/forms/base_form.py`。
:::
## 基本用法
### 定义表单类
```python
# src/modules/system/position/schema.py
# +======================================================================
# | 模块: 岗位表单验证
# | 说明: 岗位创建/编辑/状态更新的请求数据校验
# +======================================================================
from wtforms import StringField, IntegerField
from wtforms.validators import DataRequired, Length, NumberRange
from src.common.forms.base_form import BaseForm
# ============================================================
# 岗位表单类
# ============================================================
class PositionForm(BaseForm):
"""岗位创建/编辑表单"""
name = StringField('岗位名称', validators=[
DataRequired(message='岗位名称不能为空'),
Length(min=1, max=150, message='岗位名称长度必须在1-150个字符之间')
])
status = IntegerField('岗位状态', validators=[
DataRequired(message='岗位状态不能为空'),
NumberRange(min=1, max=2, message='岗位状态必须是1或2')
])
sort = IntegerField('岗位排序', validators=[
DataRequired(message='岗位排序不能为空'),
NumberRange(min=0, max=99999, message='岗位排序必须在0-99999之间')
])
# ============================================================
# 设置状态表单类
# ============================================================
class PositionStatusForm(BaseForm):
"""岗位状态更新表单"""
id = IntegerField('岗位ID', validators=[
DataRequired(message='岗位ID不能为空'),
NumberRange(min=1, message='岗位ID必须大于0')
])
status = IntegerField('岗位状态', validators=[
DataRequired(message='岗位状态不能为空'),
NumberRange(min=1, max=2, message='岗位状态必须是1或2')
])
```
### Router 使用
```python
@position_blue.route('/add', methods=['POST'])
@check_permission('sys:position:add')
def add():
from src.modules.system.position.service import add_position
from src.modules.system.position.schema import PositionForm
data = parse_request_body(request, PositionForm)
return add_position(data)
```
:::tip 自动验证
`parse_request_body` 函数在调用 Service 函数前自动完成参数验证。验证失败时返回标准格式错误：
```json
{"code": 1, "data": null, "msg": "岗位名称不能为空"}
```
:::
## Validator 约束
WTForms 提供丰富的字段约束：
| Validator | 说明 | 示例 |
|------|------|------|
| `DataRequired` | 必填字段 | `DataRequired(message='不能为空')` |
| `Length` | 字符串长度 | `Length(min=1, max=150)` |
| `NumberRange` | 数值范围 | `NumberRange(min=0, max=99999)` |
| `Optional` | 可选字段 | `Optional()` |
| `Email` | 邮箱格式 | `Email(message='邮箱格式不正确')` |
| `URL` | URL 格式 | `URL(message='URL格式不正确')` |
| `Regexp` | 正则匹配 | `Regexp(r'^[a-z]+$')` |
## 自定义验证器
使用 `validate_` 方法添加自定义验证逻辑：
```python
from wtforms import StringField
from wtforms.validators import DataRequired, Length, ValidationError
class UserForm(BaseForm):
username = StringField('用户名', validators=[
DataRequired(message='用户名不能为空'),
Length(min=1, max=50, message='用户名长度必须在1-50个字符之间')
])
email = StringField('邮箱', validators=[
DataRequired(message='邮箱不能为空'),
Length(max=100, message='邮箱长度不能超过100个字符')
])
def validate_username(self, field):
if not field.data.isalnum():
raise ValidationError('用户名只能包含字母和数字')
def validate_email(self, field):
if '@' not in field.data:
raise ValidationError('邮箱格式不正确')
```
## BaseForm 基类
所有创建/编辑表单应继承 `BaseForm`，自动获得 `id` 字段和空字符串处理：
```python
# src/common/forms/base_form.py
from wtforms import Form, IntegerField
from wtforms.validators import Optional
class BaseForm(Form):
"""
表单基类
所有创建/编辑表单应继承此类，自动获得：
1. id 字段：Optional，前端传空字符串时自动转为 None
"""
id = IntegerField('ID', validators=[Optional()])
```
:::warning 前端空字符串
前端某些组件（如 el-select）在未选择时可能传空字符串 `""`，`parse_request_body` 函数会自动将其转为 `None`，避免整数字段解析报错。
:::
## 唯一性校验
在 WTForms 验证器中查询数据库校验唯一性：
```python
from wtforms import StringField
from wtforms.validators import DataRequired, Length, ValidationError
class PositionForm(BaseForm):
name = StringField('岗位名称', validators=[
DataRequired(message='岗位名称不能为空'),
Length(min=1, max=150, message='岗位名称长度必须在1-150个字符之间')
])
def validate_name(self, field):
from src.modules.system.position.repository import PositionRepository
exclude_id = self.id.data if self.id.data else None
if PositionRepository.exists_by([Position.name == field.data], exclude_id):
raise ValidationError('岗位名称已存在')
```
## 登录表单示例
```python
# src/modules/auth/schema.py
# +======================================================================
# | 模块: 登录表单验证
# | 说明: 用户登录的请求数据校验
# +======================================================================
from wtforms import StringField
from wtforms.validators import DataRequired, Length
from config.captcha import CAPTCHA_LENGTH
# ============================================================
# 登录表单类
# ============================================================
class LoginForm(BaseForm):
"""登录表单验证类"""
username = StringField('登录账号', validators=[
DataRequired(message='登录账号不能为空'),
Length(min=1, max=20, message='登录账号长度必须在1-20个字符之间')
])
password = StringField('登录密码', validators=[
DataRequired(message='登录密码不能为空'),
Length(min=6, max=128, message='登录密码长度必须在6-128个字符之间')
])
code = StringField('验证码', validators=[
DataRequired(message='验证码不能为空'),
Length(min=CAPTCHA_LENGTH, max=CAPTCHA_LENGTH, message='验证码长度必须为{}个字符'.format(CAPTCHA_LENGTH))
])
key = StringField('KEY值', validators=[
DataRequired(message='KEY值不能为空')
])
```
## 错误响应格式
验证失败时，`parse_request_body` 函数捕获验证错误并返回标准格式：
```json
{
"code": 1,
"data": null,
"msg": "岗位名称不能为空",
"ok": false
}
```
多个错误返回第一个错误信息。
## 总结
参数验证模块具备以下特点：
```
1. WTForms：类型安全、自动解析 JSON 请求体
2. 丰富约束：DataRequired、Length、NumberRange、Optional、Email、URL、Regexp
3. 自定义验证：validate_ 方法实现复杂校验逻辑
4. BaseForm 基类：统一 id 字段处理，兼容前端空字符串
5. 唯一性校验：验证器中查询数据库，确保字段唯一性
6. 标准错误：统一错误响应格式
```