505 lines
13 KiB
Markdown
505 lines
13 KiB
Markdown
# 第四章 架构设计 - 核心基类体系
|
||
|
||
## 4.1 基类总览
|
||
|
||
核心基类体系由四大基类组成，覆盖从数据模型到请求校验再到响应输出的完整链路。子类只需声明差异点，即可获得标准化的增删改查能力。
|
||
|
||
```
|
||
BaseModel (application/models.py)
|
||
|
|
||
+-- 通用字段: id / create_user / create_time / update_user / update_time / is_delete
|
||
+-- 字段顺序重排: class_prepared 信号
|
||
|
|
||
v
|
||
Example(BaseModel) <-- 业务模型继承
|
||
|
||
PermissionRequired (middleware/permission_middleware.py)
|
||
|
|
||
+-- RBAC权限验证: has_permission()
|
||
+-- 超级管理员放行: user_id == 1
|
||
+-- 无权限处理: handle_no_permission()
|
||
|
|
||
v
|
||
ExamplePageView(PermissionRequired, View) <-- 视图类混入
|
||
|
||
check_login (middleware/login_middleware.py)
|
||
|
|
||
+-- JWT认证: parse_payload()
|
||
+-- 忽略URL: /login, /captcha
|
||
+-- 认证失败: R.failed(code=401)
|
||
|
|
||
v
|
||
@method_decorator(check_login, name="get") <-- 视图方法装饰器
|
||
|
||
R (utils/R.py)
|
||
|
|
||
+-- ok(): 成功响应
|
||
+-- failed(): 失败响应
|
||
+-- response(): 通用响应
|
||
|
|
||
v
|
||
JsonResponse <-- Django HTTP响应
|
||
```
|
||
|
||
## 4.2 BaseModel 模型基类
|
||
|
||
**文件位置**：`application/models.py`
|
||
|
||
### 设计目标
|
||
|
||
为所有业务模型提供统一的通用字段和字段顺序管理，消除重复定义。
|
||
|
||
### 通用字段
|
||
|
||
```python
|
||
class BaseModel(models.Model):
|
||
"""基础模型抽象类"""
|
||
|
||
# 主键ID字段
|
||
id = models.AutoField(
|
||
auto_created=True,
|
||
primary_key=True,
|
||
serialize=False,
|
||
verbose_name='主键ID',
|
||
db_comment='主键ID'
|
||
)
|
||
|
||
# 创建人字段
|
||
create_user = models.CharField(
|
||
null=True,
|
||
max_length=50,
|
||
default=None,
|
||
db_default=None,
|
||
verbose_name='创建人',
|
||
db_comment='创建人'
|
||
)
|
||
|
||
# 创建时间字段
|
||
create_time = models.DateTimeField(
|
||
null=True,
|
||
auto_now_add=True,
|
||
verbose_name="创建时间",
|
||
max_length=11,
|
||
db_comment='创建时间'
|
||
)
|
||
|
||
# 更新人字段
|
||
update_user = models.CharField(
|
||
null=True,
|
||
max_length=50,
|
||
default=None,
|
||
db_default=None,
|
||
verbose_name='更新人',
|
||
db_comment='更新人'
|
||
)
|
||
|
||
# 更新时间字段
|
||
update_time = models.DateTimeField(
|
||
null=True,
|
||
auto_now=True,
|
||
verbose_name="更新时间",
|
||
max_length=11,
|
||
db_comment='更新时间'
|
||
)
|
||
|
||
# 逻辑删除标识
|
||
is_delete = models.BooleanField(
|
||
default=0,
|
||
db_default=0,
|
||
verbose_name="逻辑删除",
|
||
db_comment='逻辑删除：0-正常 1-已删除'
|
||
)
|
||
|
||
class Meta:
|
||
abstract = True # 抽象模型，不创建数据库表
|
||
```
|
||
|
||
### 字段说明
|
||
|
||
| 字段 | 类型 | 说明 |
|
||
|------|------|------|
|
||
| `id` | AutoField, PK | 主键自增 |
|
||
| `create_user` | CharField(50) | 创建人（用户名） |
|
||
| `create_time` | DateTimeField | 创建时间（auto_now_add） |
|
||
| `update_user` | CharField(50) | 更新人（用户名） |
|
||
| `update_time` | DateTimeField | 更新时间（auto_now） |
|
||
| `is_delete` | BooleanField | 软删除标识：0=正常，1=已删除 |
|
||
|
||
### 字段顺序重排
|
||
|
||
通过 `class_prepared` 信号，自动将公共字段移到业务字段之后：
|
||
|
||
```python
|
||
BASE_FIELD_NAMES = ['id', 'create_user', 'create_time',
|
||
'update_user', 'update_time', 'is_delete']
|
||
|
||
def _reorder_fields(sender, **kwargs):
|
||
"""信号处理函数：重排模型字段顺序"""
|
||
if not hasattr(sender, '_meta') or sender._meta.abstract:
|
||
return
|
||
|
||
# 分离公共字段和业务字段
|
||
base_fields = []
|
||
business_fields = []
|
||
|
||
for field in sender._meta.local_fields:
|
||
if field.name in BASE_FIELD_NAMES:
|
||
base_fields.append(field)
|
||
else:
|
||
business_fields.append(field)
|
||
|
||
# 重新排列：业务字段在前，公共字段在后
|
||
sender._meta.local_fields = business_fields + base_fields
|
||
|
||
# 注册信号处理器
|
||
class_prepared.connect(_reorder_fields)
|
||
```
|
||
|
||
### 使用示例
|
||
|
||
```python
|
||
from application.models import BaseModel
|
||
from utils.common import get_table_name
|
||
|
||
class Example(BaseModel):
|
||
"""案例模型类"""
|
||
name = models.CharField(max_length=100, verbose_name="案例名称")
|
||
status = models.IntegerField(verbose_name="状态")
|
||
|
||
class Meta:
|
||
db_table = get_table_name('example')
|
||
ordering = ("sort",)
|
||
```
|
||
|
||
继承 BaseModel 后，Example 模型自动拥有 6 个公共字段，无需手动定义。
|
||
|
||
## 4.3 check_login 认证装饰器
|
||
|
||
**文件位置**：`middleware/login_middleware.py`
|
||
|
||
### 设计目标
|
||
|
||
提供基于 JWT 的用户登录状态验证装饰器，保护需要登录才能访问的视图函数。
|
||
|
||
### 核心实现
|
||
|
||
```python
|
||
from utils import R
|
||
from utils.jwt import parse_payload
|
||
|
||
def check_login(func):
|
||
"""JWT登录认证装饰器"""
|
||
|
||
def wrapper(request, *args, **kwargs):
|
||
# 定义不需要认证的URL列表
|
||
ignoreURL = ['/login', '/captcha']
|
||
|
||
# 判断当前请求是否需要验证登录
|
||
if request.path not in ignoreURL:
|
||
# 从请求头中获取token值
|
||
access_token = request.headers['Authorization']
|
||
|
||
# 移除"Bearer "前缀
|
||
access_token = access_token.replace('Bearer ', "")
|
||
|
||
# JWT解密验证
|
||
result = parse_payload(access_token)
|
||
|
||
# token验证失败处理
|
||
code = result['code']
|
||
if code != 0:
|
||
return R.failed(code=401, msg=result['msg'])
|
||
|
||
# 认证通过，继续执行原视图函数
|
||
return func(request, *args, **kwargs)
|
||
|
||
return wrapper
|
||
```
|
||
|
||
### 认证流程
|
||
|
||
```
|
||
1. 检查请求路径是否在忽略列表中
|
||
+-- 是 -> 直接执行视图函数
|
||
+-- 否 -> 继续验证
|
||
|
||
2. 从请求头获取 Authorization 值
|
||
+-- 移除 "Bearer " 前缀
|
||
+-- 获取纯净的 token 字符串
|
||
|
||
3. 调用 parse_payload() 验证 token
|
||
+-- 检查 Token 是否在黑名单中
|
||
+-- 解密验证签名和有效期
|
||
+-- 返回验证结果
|
||
|
||
4. 根据验证结果决定是否放行
|
||
+-- code == 0 -> 放行
|
||
+-- code != 0 -> 返回 401 错误
|
||
```
|
||
|
||
### 使用方式
|
||
|
||
```python
|
||
from django.utils.decorators import method_decorator
|
||
from middleware.login_middleware import check_login
|
||
|
||
# 在类视图中使用
|
||
@method_decorator(check_login, name="get")
|
||
class ExamplePageView(View):
|
||
def get(self, request):
|
||
pass
|
||
|
||
# 在函数视图中使用
|
||
@check_login
|
||
def user_info(request):
|
||
pass
|
||
```
|
||
|
||
## 4.4 PermissionRequired 权限混入类
|
||
|
||
**文件位置**：`middleware/permission_middleware.py`
|
||
|
||
### 设计目标
|
||
|
||
提供基于 RBAC 的细粒度权限控制，通过检查用户是否拥有指定权限节点来决定是否允许访问。
|
||
|
||
### 核心实现
|
||
|
||
```python
|
||
from django.contrib.auth.mixins import PermissionRequiredMixin
|
||
from utils import R
|
||
from utils.security import get_user_id
|
||
|
||
class PermissionRequired(PermissionRequiredMixin):
|
||
"""自定义权限控制混入类"""
|
||
|
||
def has_permission(self):
|
||
"""检查用户是否拥有所需权限"""
|
||
# 获取所需权限节点列表
|
||
permissions = self.get_permission_required()
|
||
|
||
# 获取当前登录用户ID
|
||
user_id = get_user_id(self.request)
|
||
|
||
# 超级管理员（ID=1）自动放行
|
||
if user_id and user_id != 1:
|
||
from application.menu import services
|
||
# 获取用户权限节点
|
||
permission_list = services.get_user_permissions(user_id)
|
||
|
||
# 遍历所需权限列表
|
||
for permission in permissions:
|
||
if permission not in permission_list:
|
||
return False
|
||
|
||
return True
|
||
|
||
def handle_no_permission(self):
|
||
"""无权限访问时的处理函数"""
|
||
return R.failed("暂无操作权限", 401)
|
||
```
|
||
|
||
### 权限验证流程
|
||
|
||
```
|
||
1. 获取视图所需的权限节点列表
|
||
permission_required = ("sys:example:page",)
|
||
|
||
2. 获取当前登录用户ID
|
||
user_id = get_user_id(request)
|
||
|
||
3. 判断是否为超级管理员
|
||
+-- user_id == 1 -> 直接放行
|
||
+-- user_id != 1 -> 继续验证
|
||
|
||
4. 获取用户的实际权限列表
|
||
permission_list = services.get_user_permissions(user_id)
|
||
|
||
5. 比对所需权限是否在用户权限列表中
|
||
+-- 全部存在 -> 放行
|
||
+-- 有缺失 -> 返回 401 错误
|
||
```
|
||
|
||
### 权限获取逻辑
|
||
|
||
```python
|
||
# application/menu/services.py
|
||
def get_user_permissions(user_id):
|
||
"""获取用户权限节点列表"""
|
||
if user_id == 1:
|
||
# 超级管理员拥有所有权限
|
||
menu_list = Menu.objects.filter(
|
||
is_delete=False, type=1, status=0
|
||
).values()
|
||
return [item['permission'] for item in menu_list]
|
||
else:
|
||
# 普通用户从角色关联中获取权限
|
||
sql = '''
|
||
SELECT m.* FROM django_menu AS m
|
||
INNER JOIN django_role_menu AS rm ON m.id=rm.menu_id
|
||
INNER JOIN django_user_role AS ur ON ur.role_id=rm.role_id
|
||
WHERE ur.user_id=%s
|
||
AND (m.type=1 OR (m.type=0 AND m.permission!=''))
|
||
AND m.status=0 AND m.is_delete=0
|
||
'''
|
||
menu_list = Menu.objects.raw(sql, [user_id])
|
||
return [item.permission for item in menu_list]
|
||
```
|
||
|
||
### 使用方式
|
||
|
||
```python
|
||
from django.views import View
|
||
from middleware.permission_middleware import PermissionRequired
|
||
|
||
class ExamplePageView(PermissionRequired, View):
|
||
"""案例分页数据查询视图"""
|
||
|
||
# 声明所需权限节点
|
||
permission_required = ("sys:example:page",)
|
||
|
||
def get(self, request):
|
||
pass
|
||
```
|
||
|
||
## 4.5 R 统一响应类
|
||
|
||
**文件位置**：`utils/R.py`
|
||
|
||
### 设计目标
|
||
|
||
提供系统统一的 JSON 响应格式，确保所有 API 接口返回格式一致。
|
||
|
||
### 响应格式规范
|
||
|
||
```json
|
||
{
|
||
"code": 0, // 状态码：0-成功，1-失败，401-未授权
|
||
"data": {}, // 响应数据
|
||
"msg": "操作成功", // 提示信息
|
||
"ok": true // 操作是否成功
|
||
}
|
||
```
|
||
|
||
### 核心实现
|
||
|
||
```python
|
||
from django.http import JsonResponse
|
||
|
||
def ok(data=None, msg="操作成功", code=0, **kwargs):
|
||
"""生成成功响应"""
|
||
response_body = {
|
||
"code": code,
|
||
"data": data,
|
||
"msg": msg,
|
||
"ok": True
|
||
}
|
||
if kwargs:
|
||
response_body.update(kwargs)
|
||
return JsonResponse(data=response_body)
|
||
|
||
def failed(msg="操作失败", code=1, data=None, **kwargs):
|
||
"""生成失败响应"""
|
||
response_body = {
|
||
"code": code,
|
||
"data": data,
|
||
"msg": msg,
|
||
"ok": False
|
||
}
|
||
if kwargs:
|
||
response_body.update(kwargs)
|
||
return JsonResponse(data=response_body)
|
||
|
||
def response(data=None, msg="操作成功", code=0, success=True, **kwargs):
|
||
"""通用响应生成器"""
|
||
response_body = {
|
||
"code": code,
|
||
"data": data,
|
||
"msg": msg,
|
||
"ok": success
|
||
}
|
||
if kwargs:
|
||
response_body.update(kwargs)
|
||
return JsonResponse(data=response_body)
|
||
```
|
||
|
||
### 使用示例
|
||
|
||
```python
|
||
from utils import R
|
||
|
||
# 成功响应
|
||
return R.ok(data={"id": 1, "name": "张三"})
|
||
# {"code": 0, "data": {"id": 1, "name": "张三"}, "msg": "操作成功", "ok": true}
|
||
|
||
# 分页响应
|
||
return R.ok(data=page_data, total=100, pages=10)
|
||
# {"code": 0, "data": {...}, "msg": "操作成功", "ok": true, "total": 100, "pages": 10}
|
||
|
||
# 失败响应
|
||
return R.failed(msg="用户名或密码错误")
|
||
# {"code": 1, "data": null, "msg": "用户名或密码错误", "ok": false}
|
||
|
||
# 认证失败响应
|
||
return R.failed(code=401, msg="token已失效")
|
||
# {"code": 401, "data": null, "msg": "token已失效", "ok": false}
|
||
```
|
||
|
||
### 状态码常量
|
||
|
||
```python
|
||
class Codes:
|
||
"""常用状态码定义"""
|
||
SUCCESS = 0
|
||
FAILED = 1
|
||
UNAUTHORIZED = 401 # 未认证
|
||
FORBIDDEN = 403 # 无权限
|
||
NOT_FOUND = 404 # 资源不存在
|
||
VALIDATE_ERROR = 422 # 验证错误
|
||
SERVER_ERROR = 500 # 服务器错误
|
||
```
|
||
|
||
## 4.6 基类协作关系
|
||
|
||
### 请求处理中的基类协作
|
||
|
||
```
|
||
HTTP Request
|
||
|
|
||
v
|
||
@check_login (认证装饰器)
|
||
|
|
||
+-- 验证JWT Token
|
||
+-- 失败 -> R.failed(code=401)
|
||
|
|
||
v
|
||
PermissionRequired (权限混入类)
|
||
|
|
||
+-- 验证权限节点
|
||
+-- 失败 -> R.failed("暂无操作权限", 401)
|
||
|
|
||
v
|
||
View.get/post (视图方法)
|
||
|
|
||
+-- 调用 Service 函数
|
||
|
|
||
v
|
||
Service函数
|
||
|
|
||
+-- 使用 BaseModel 子类进行数据库操作
|
||
+-- 成功 -> R.ok(data=...)
|
||
+-- 失败 -> R.failed(msg=...)
|
||
|
|
||
v
|
||
JsonResponse (HTTP响应)
|
||
```
|
||
|
||
### 各基类职责总结
|
||
|
||
| 基类 | 文件位置 | 核心职责 |
|
||
|------|----------|----------|
|
||
| `BaseModel` | `application/models.py` | 提供6个通用字段，字段顺序重排 |
|
||
| `check_login` | `middleware/login_middleware.py` | JWT认证，验证登录状态 |
|
||
| `PermissionRequired` | `middleware/permission_middleware.py` | RBAC权限控制，细粒度权限验证 |
|
||
| `R` | `utils/R.py` | 统一响应格式，JSON响应生成 |
|