第四章架构设计 - 核心基类体系

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):
    """
    基础模型抽象类
    ==============
    功能：提供所有模型的公共字段，如主键、创建人、创建时间等
    特点：定义为抽象类，不会在数据库中创建对应的表
    用途：被其他业务模型继承，实现字段复用

    继承此基类的模型会自动拥有以下6个公共字段：
    - id: 自增主键
    - create_user: 创建人
    - create_time: 创建时间
    - update_user: 更新人
    - update_time: 更新时间
    - is_delete: 逻辑删除标识

    注意：公共字段会自动排列在业务字段之后，使数据库表结构更清晰
    """

    # ==================== 公共字段定义 ====================

    # 主键ID字段
    # 自增主键，每个模型的唯一标识符
    id = models.AutoField(
        auto_created=True,  # 自动创建主键字段
        primary_key=True,  # 设置为主键
        serialize=False,  # 不参与序列化（避免在API响应中暴露）
        verbose_name='主键ID',  # 字段中文名称
        db_comment='主键ID'  # 数据库字段注释
    )

    # 创建人字段
    # 记录数据由谁创建，通常关联用户表的用户名或ID
    # 例如：admin, zhangsan, system
    create_user = models.CharField(
        null=True,  # 数据库允许为NULL
        max_length=50,  # 最大长度50个字符
        default=None,  # 默认值为None
        db_default=None,  # 数据库层默认值
        verbose_name='创建人',  # 字段中文名称
        db_comment='创建人'  # 数据库字段注释
    )

    # 创建时间字段
    # 自动记录数据创建的时间戳
    # auto_now_add: 仅在创建时自动设置当前时间，后续更新不会改变
    create_time = models.DateTimeField(
        null=True,  # 数据库允许为NULL
        auto_now_add=True,  # 创建时自动设为当前时间
        verbose_name="创建时间",  # 字段中文名称
        max_length=11,  # 时间戳长度
        db_comment='创建时间'  # 数据库字段注释
    )

    # 更新人字段
    # 记录数据最后由谁更新，通常关联用户表的用户名或ID
    # null=True: 允许数据库为NULL（新创建记录时可能没有更新人）
    update_user = models.CharField(
        null=True,  # 数据库允许为NULL
        max_length=50,  # 最大长度50个字符
        default=None,  # 默认值为None
        db_default=None,  # 数据库层默认值
        verbose_name='更新人',  # 字段中文名称
        db_comment='更新人'  # 数据库字段注释
    )

    # 更新时间字段
    # 自动记录数据最后更新的时间戳
    # auto_now: 每次保存（更新）时自动设为当前时间
    update_time = models.DateTimeField(
        null=True,  # 数据库允许为NULL
        auto_now=True,  # 每次保存时自动更新为当前时间
        verbose_name="更新时间",  # 字段中文名称
        max_length=11,  # 时间戳长度
        db_comment='更新时间'  # 数据库字段注释
    )

    # 逻辑删除标识
    # 用于软删除，避免物理删除数据造成的数据丢失
    # 0：正常（默认值）| 1：已删除
    is_delete = models.BooleanField(
        default=0,  # 默认为0，表示正常
        db_default=0,  # 数据库层默认值
        verbose_name="逻辑删除",  # 字段中文名称
        db_comment='逻辑删除：0-正常 1-已删除'  # 数据库字段注释
    )

    class Meta:
        """
        元数据配置
        """
        # 定义为抽象模型类
        # abstract = True 表示这个模型是抽象的，不会在数据库中创建对应的表
        # 子类模型会继承这些字段，并在各自对应的表中创建这些列
        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 个公共字段，无需手动定义。

文件位置：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响应生成

🔐 认证与安全

📊 数据与配置

📝 日志与监控

🛡️ 防护与治理

📁 文件处理

🔧 核心工具

🧩 辅助工具

第四章架构设计 - 核心基类体系

4.1 基类总览

4.2 BaseModel 模型基类

设计目标

通用字段

字段说明

字段顺序重排

使用示例

设计目标

核心实现

认证流程

使用方式

4.4 PermissionRequired 权限混入类

设计目标

核心实现

权限验证流程

权限获取逻辑

使用方式

4.5 R 统一响应类

设计目标

响应格式规范

核心实现

使用示例

状态码常量

4.6 基类协作关系

请求处理中的基类协作

各基类职责总结

Become a sponsor

企业级中台解决方案

企业级中台解决方案

第四章 架构设计 - 核心基类体系 ​

4.1 基类总览 ​

4.2 BaseModel 模型基类 ​

设计目标 ​

通用字段 ​

字段说明 ​

字段顺序重排 ​

使用示例 ​

4.3 check_login 认证装饰器 ​

设计目标 ​

核心实现 ​

认证流程 ​

使用方式 ​

4.4 PermissionRequired 权限混入类 ​

设计目标 ​

核心实现 ​

权限验证流程 ​

权限获取逻辑 ​

使用方式 ​

4.5 R 统一响应类 ​

设计目标 ​

响应格式规范 ​

核心实现 ​

使用示例 ​

状态码常量 ​

4.6 基类协作关系 ​

请求处理中的基类协作 ​

各基类职责总结 ​

第四章架构设计 - 核心基类体系

4.1 基类总览

4.2 BaseModel 模型基类

设计目标

通用字段

字段说明

字段顺序重排

使用示例

4.3 check_login 认证装饰器

设计目标

核心实现

认证流程

使用方式

4.4 PermissionRequired 权限混入类

设计目标

核心实现

权限验证流程

权限获取逻辑

使用方式

4.5 R 统一响应类

设计目标

响应格式规范

核心实现

使用示例

状态码常量

4.6 基类协作关系

请求处理中的基类协作

各基类职责总结