编程导航会员会员
登录/注册

用户中心项目教程 笔记 -python 后端版，使用 django ninja 框架

项目用了 Django ninja 框架实现，ninja 是基于 django 的改良，能更快更方便的去构写后端接口

构建项目

项目需要安装依赖

▼
shell
复制代码
Django==5.1.6           # 基础框架
django-ninja==1.4.3       # API框架
pydantic==2.10.4          # 数据验证
mysqlclient==2.2.5         # MySQL数据库驱动
python-dotenv==1.0.1    # 环境变量管理
django-cors-headers==4.4.0 # 跨域支持

项目初始化

  1. 构建项目
▼
shell
复制代码
django-admin startproject <项目名称 “user_center”>
  1. 创建子模块
▼
shell
复制代码
python manage.py makeapp <子模块名称-这里是“user”>
  1. 创建 Mysql 数据库，和表
▼
sql
复制代码
CREATE DATABASE `user_center` IF NOT EXISTS

USE user_center

CREATE TABLE `users` (
  `id` bigint NOT NULL AUTO_INCREMENT COMMENT 'id',
  `user_name` varchar(256) DEFAULT NULL COMMENT '用户昵称',
  `user_account` varchar(256) DEFAULT NULL COMMENT '账号',
  `avatar_url` varchar(1024) DEFAULT NULL COMMENT '用户头像',
  `gender` tinyint DEFAULT NULL COMMENT '性别',
  `user_password` varchar(512) NOT NULL COMMENT '密码',
  `phone` varchar(128) DEFAULT NULL COMMENT '电话',
  `email` varchar(512) DEFAULT NULL COMMENT '邮箱',
  `user_status` int NOT NULL DEFAULT '0' COMMENT '状态 0-正常',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  `is_delete` tinyint NOT NULL DEFAULT '0' COMMENT '是否删除',
  `user_role` int NOT NULL DEFAULT '0' COMMENT '用户角色 0-普通用户 1-管理员',
  `planet_code` varchar(512) DEFAULT NULL COMMENT '星球编号',
  `tags` varchar(1024) DEFAULT NULL COMMENT '标签列表',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=56 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='用户';
  1. django项目settings.py 里添加下方代码，连接数据库，再在 prompt 里使用 django 的 inspectdb 命令 生成对应的模型
▼
python
复制代码
#添加以下代码
DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "xxxx",  # 数据库名称
        "USER": "xxxxxx",  # 数据库用户名
        "PASSWORD": "xxxxxxx",  # 数据库密码
        "HOST": "xxxxxx",  # 数据库主机地址
        "PORT": "xxxxx",
        "TEST": {
            "MIRROR": "default",  # 关键配置：测试直接使用开发数据库
            "DEPENDENCIES": [],  # 避免数据库依赖检查
        },
    }
}

生成对应 table 的模型

▼
shell
复制代码
python manage.py inspectdb users

代码放到user 模块下的 model.py 里面

▼
python
复制代码
class Users(models.Model):
    id = models.BigAutoField(primary_key=True, db_comment="id")
    user_name = models.CharField(
        max_length=256, blank=True, null=True, db_comment="用户昵称"
    )
    user_account = models.CharField(
        max_length=256, blank=True, null=True, db_comment="账号"
    )
    avatar_url = models.CharField(
        max_length=1024, blank=True, null=True, db_comment="用户头像"
    )
    gender = models.IntegerField(blank=True, null=True, db_comment="性别")
    user_password = models.CharField(max_length=512, db_comment="密码")
    phone = models.CharField(max_length=128, blank=True, null=True, db_comment="电话")
    email = models.CharField(max_length=512, blank=True, null=True, db_comment="邮箱")
    user_status = models.IntegerField(db_comment="状态 0-正常")
    create_time = models.DateTimeField(blank=True, null=True, db_comment="创建时间")
    update_time = models.DateTimeField(blank=True, null=True, db_comment="更新时间")
    is_delete = models.IntegerField(db_comment="是否删除")
    user_role = models.IntegerField(db_comment="用户角色 0-普通用户 1-管理员")
    planet_code = models.CharField(
        max_length=512, blank=True, null=True, db_comment="星球编号"
    )
    tags = models.CharField(
        max_length=1024, blank=True, null=True, db_comment="标签列表"
    )

    class Meta:
        managed = False
        db_table = "users"
        db_table_comment = "用户"

以上项目初始化完成，获得项目主模块作为项目入口，以及我们需要开发的 user 模块，并且生成了 user 模块的模型，用于后面完成业务代码的交互

项目结构

介绍一下开发完成的完整的项目结构

模块：

  • user_center 用户中心主模块，项目的配置，和入口
  • core 核心通用模块：包含一些所有模块会用到的通用配置和模型，以及方法
  • users 模块：包含一切跟用户相关的，模型以及业务代码
▼
plain
复制代码
user_center/
├── user_center/                    # 主项目目录
│   ├── __init__.py
│   ├── settings.py                 # Django项目配置
│   ├── urls.py                     # 主路由配置
│   ├── wsgi.py                     # WSGI入口
│   ├── asgi.py                     # ASGI入口
│   └──api.py                       # Ninja API入口
│
├── core/                           # 核心模块
│   ├── __init__.py
│   ├── config.py                   # 项目配置类
│   ├── constants.py                # 常量定义
│   ├── exception/                  # 异常处理
│   │   ├── __init__.py
│   │   └── business_exception.py   # 业务异常类
│   ├── exception_handler.py        # 全局异常处理器
│   ├── schemas.py                  # 通用数据校验模型
│   └── types.py                    # 类型定义
│
├── users/                          # 用户模块
│   ├── __init__.py
│   ├── models.py                   # 用户模型
│   ├── schemas.py                  # 数据校验模型
│   ├── service.py                  # 业务逻辑
│   ├── api.py                      # 用户路由
│   ├── test/                       # 测试
│   │   ├── __init__.py
│   │   └── test_user.py            # 测试用户业务函数
│   └── migrations/                 # 数据库迁移
│       ├── __init__.py
│       └── 0001_initial.py         # 初始迁移
│
├── logs/                           # 日志目录
│   └── django.log                  # 日志文件
│
├── manage.py                       # Django管理脚本
└── pytest.ini                      # pytest配置

单模块的层级

路由层，API/ROUTER 封装请求 (api.py)

  • 根据 @api 来编写 rest 风格的 api 返回默认值为 json
  • api 层 倾向于 请求本省的校验，不涉及业务逻辑本身，越少越好

数据校验层 (schema.py)

  • 对请求体和响应体中的参数，进行数据校验

服务层 (service.py)

  • service 层是对业务逻辑校验，(有可能被 api 之外调用 )

后端实现

用户注册

逻辑设计

1.用户在前端输入账户和密码、以及校验码（todo）

  1. 校验用户的账户、密码、校验密码，星球编号，是否符合要求

  2. 非空

  3. 账户长度 不小于4位

  4. 密码就 不小于8位吧

  5. 账户不能重复

  6. 账户不包含特殊字符vi. 密码和校验密码相同

  7. 星球编号不能大于 5 位数，并且编号不能重复

  8. 对密码进行加密（密码千万不要直接以明文存储到数据库中）

  9. 向数据库插入用户数据

接口设计

请求参数：用户账户，密码，校验密码

请求类型：POST (请求参数很长的时候不建议 get)

请求体：JSON 格式

返回值：用户id

实现

接口实现

  1. 在数据校验层，定义一个，user register request 类，和一个 response 类，类当中定义body对应要有的参数，这两个类是 ninja 这个框架的一个特性，可以通过 schema 来实现，对发送过来的请求对象，和响应对象 的 body，进行读取，以及进行一些定义，跟 spring 的@RequestBody 类似, 并且可以在这个类中进行数据校验，而不用放在逻辑层(Service.py)
▼
python
复制代码
from ninja import Schema
from typing import Optional


class UserRegisterRequest(Schema):
    user_account: str
    user_password: str
    check_password: str


class UserRegisterResponse(Schema):
    user_id: int
  1. 写一个对应的路由 并且通过装饰器和传参，来指定对应的定义好的 request 类和 response 类
▼
python
复制代码
from ninja import Router, Schema
from service import UserServices
from schemas import (
    UserRegisterRequest,
    UserRegisterResponse,
)#导入请求，和响应类

router = Router()
user_service = UserServices()

#通过装饰器指定了响应格式
@router.post("/register", response=UserRegisterResponse)
def user_register(request, data: UserRegisterRequest):#通过传参定义了请求体
    # 这里不做data 是否为 none 的判断 
    # 因为框架 shcema 部分会进行判断，如果确实直接返回错误
    user_account = data.user_account
    user_password = data.user_password
    check_password = data.check_password
    #调用 服务层的函数，完成业务功能
    return user_service.user_register(user_account, user_password, check_password)

逻辑实现

在服务层 service.py 实现业务逻辑,根据上述设计描述

▼
python
复制代码
    @staticmethod
    def user_register(
        user_account: str,
        user_password: str,
        check_password: str,
        user_status: int = 0,
    ) -> int:
        """用户注册服务
        Args:
            user_account: 用户账号(4位以上字母数字组合)
            user_password: 用户密码(至少8位)
            check_password: 确认密码(需与user_password一致)
            user_status: 用户状态
        Returns:
            int: 成功返回用户ID，失败返回-1
        """
        # 1. 校验
        # 1.1 校验符合长度，不为空
        if not all(
            [
                user_account,
                user_password,
                check_password,
            ]
        ):
            return -1

        if len(user_account) < 4:
            return -1

        if len(user_password) < 8 or len(check_password) < 8:
            return -1

        # if len(planet_code) > 5:
        #     return -1

        # 1.2 账户不能包含特殊字符
        valid_pattern = r"[^a-zA-Z0-9]"
        if re.search(valid_pattern, user_account):
            return -1

        # 1.3 密码和校验密码相同
        if not user_password == check_password:
            return -1

        # 1.4 账户不能重复
        exists = User.objects.filter(user_account=user_account).exists()
        if exists:
            return -1

        # 1.5 星球编号不能重复
        # exists = User.objects.filter(planet_code=planet_code).exists()
        # if exists:
        #     return -1

        # 2. 加密
        md5 = hashlib.md5()
        salt_pass = SALT + user_password
        md5.update(salt_pass.encode("utf-8"))
        encrypt_password = md5.hexdigest()

        # 3. 插入数据
        user = User()
        user.user_account = user_account
        user.user_password = encrypt_password
        user.user_status = user_status
        user.is_delete = 0
        user.user_role = 0
        user.save()

        if user.id:
            user_dict = model_to_dict(user)
            return user.id
        else:
            return -1

用户登入

逻辑设计

  1. 校验用户账户和密码是否合法
    1. 非空
    2. 账户长度不小于 4 位
    3. 密码就不小于 8 位
    4. 账户不包含特殊字符
  2. 校验密码是否输入正确，要和数据库中的密文密码去对比
  3. 用户信息脱敏，隐藏敏感信息，防止数据库中字段泄露
  4. 我们要登记用户登录状态 session将其存在服务器上，前端会有cookie，作为标识，发送这个标识给后端，然后后端读取 cookie，与 session 进行匹配，这样就能知道用户是否有登录了

接口设计

请求参数：用户账户，密码

请求类型：POST

请求参数很长的时候不建议 get

请求体：JSON 格式

返回值：用户信息(脱敏)

实现

接口实现

  1. 跟注册一样,定义一个，user login request 类，和一个 response 类，作为数据对接和校验
▼
python
复制代码
class UserLoginRequest(Schema):
    user_account: str
    user_password: str


class UserLoginResponse(Schema):
    user_account: str
    user_name: str
    user_id: int
    avatar_url: Optional[str]
    gender: Optional[int]
    phone: Optional[str]
    email: Optional[str]
    user_status: int
    user_role: int
    planet_code: Optional[str]
  1. 在路由层完成 login 接口
▼
python
复制代码
@router.post("/login", response=Optional[UserLoginResponse])
def user_login(
    request, data: UserLoginRequest
) -> Optional[SafetyUser]:
    user_account = data.user_account
    user_password = data.user_password

    return user_service.do_login(
        request,
        user_account,
        user_password,
    )

逻辑实现

跟注册一样，在service.py，写个 do_login函数，根据设计描述完成业务逻辑

▼
python
复制代码
    @staticmethod
    def do_login(
        request: HttpRequest,
        user_account: str,
        user_password: str,
    ) -> Optional[SafetyUser]: #这里指定了返回类型，这个类型在下发有进行解释
        """用户登录服务
        Args:
            request: Django HttpRequest对象，用于存储session
            user_account: 用户账号(4位以上字母数字组合)
            user_password: 用户密码(至少8位)

        Returns:
            Optional[dict]: 返回脱敏后的用户数据字典，包含以下字段:
                - user_account: 账号
                - user_name: 用户名
                - user_id: 用户ID
                - avatar_url: 头像URL(可选)
                - gender: 性别(可选)
                - phone: 电话(可选)
                - email: 邮箱(可选)
                - user_status: 账户状态
                - user_role: 用户角色
                - planet_code: 星球编号(可选)
            如果登录失败返回None
        """
        # 1. 校验
        if not all([user_account, user_password]):
            return None

        if len(user_account) < 4:
            return None

        if len(user_password) < 8:
            return None

        # 账户不能包含特殊字符
        valid_pattern = r"[^a-zA-Z0-9]"
        if re.search(valid_pattern, user_account):
            return None

        # 2.校验密码和数据库中的密文对比
        # 加密
        md5 = hashlib.md5()
        salt_pass = (SALT + user_password).encode("utf8")
        md5.update(salt_pass)
        encrypt_password = md5.hexdigest()

        # 检测用户在数据库中存不存在
        if not User.objects.filter(
            user_account=user_account, user_password=encrypt_password
        ).exists():
            logger.info("user login failed user account can't match with user password")
            return None
        user = User.objects.get(
            user_account=user_account, user_password=encrypt_password
        )

        # 3. 用户数据脱敏，下方构建了一个脱敏函数
        safety_user = UserServices.convert_safety_user(user)

        # 4.记录用户登入状态
        if request != None:
            request.session[USER_LOGIN_STATE] = safety_user

        return safety_user  # type: SafetyUser

写一个 脱敏用户数据的字典类，用于定义返回的用户数据标准，以及包括哪些数据，并且保证了，数据层可以对这个类进行序列化和对接

▼
python
复制代码
from typing import TypedDict, Optional


class SafetyUser(TypedDict):
    user_account: str
    user_name: str
    user_id: int
    avatar_url: Optional[str]
    gender: Optional[int]
    phone: Optional[str]
    email: Optional[str]
    user_status: int
    user_role: int
    planet_code: Optional[str]

再写一个脱敏函数，指定返回类 为 SafetyUser 字典

▼
python
复制代码
@staticmethod
    def convert_safety_user(user: User) -> Optional[SafetyUser]:
        if user is None:
            return None
        # 1.数据脱敏, 创建一个字典，把可以传给数据层的数据，放进去，以便于数据层读取
        return {
            "user_account": user.user_account or "",
            "user_name": user.user_name or "",
            "user_id": user.id,
            "avatar_url": user.avatar_url,
            "gender": user.gender,
            "phone": user.phone,
            "email": user.email,
            "user_status": user.user_status,
            "user_role": user.user_role,
            "planet_code": user.planet_code,
        }

逻辑删除实现

Django 跟 spring 不太一样，自带 ORM，不通过第三方的 MyBatis plus 来实现，要完成逻辑删除，我们需要给对应的 User model，也就是用户表 对应的 ORM 用户类，去定义一个自定义管理器去过滤逻辑上删除的用户，并且把这个管理器作为默认管理器，当我们调用这个类的时候，获取的用户就默认被过滤了。

  1. 在 models.py 当中 添加自定义管理器来实现逻辑删除
▼
python
复制代码
# 自定义管理器，实现逻辑删除
class ActiveUserManager(models.Manager):
    #重写这个方法，过滤逻辑上删除的用户，再返回
    def get_queryset(self) -> models.QuerySet:
        return super().get_queryset().filter(is_delete=0)
  1. 在要使用自定义管理器的 model 里，把自定义管理器改成默认使用项
▼
python
复制代码
class Users(models.Model):
    # ... 现有字段 ...
    
    # 添加自定义管理器
    objects = ActiveUserManager()  # 默认只返回未删除用户
    all_objects = models.Manager()  # 原始管理器

    class Meta:
        managed = False
        db_table = "users"
        db_table_comment = "用户"
  1. 在调用的时候 默认过滤逻辑删除用户
▼
python
复制代码
user.objects.all() # 返回逻辑删除用户合集

用户登出

逻辑设计

  1. 校验请求体中的，USER_LOGIN_STATE, 是否 存在，
  2. 如果是，移除登录状态返回

接口设计

请求参数： 请求带着 SESSION

请求类型：POST

请求体 无

返回值 数值 1

实现

接口实现

逻辑实现

用户管理搜索

逻辑设计

  1. 必须鉴权利，管理员才可以删除
  2. 查询用户
    1. 允许根据用户名搜查
    2. 如果用户名为空返回所以用户信息

接口设计

请求参数：用户账户 (非必需字段)

请求类型：GET

返回值：用户信息(脱敏)

实现

接口实现

返回数据类，可以复用那个UserLoginResponse 用户登入返回类，user_name 参数为非必须字段，通过 typing 来指定。再写个 is_admin 函数来鉴权，如果不是管理员返回空数组, 如果是 通过 UserSevices 当中的 list 条件，查找对应的用户

▼
python
复制代码
@router.get("/search", response=List[UserLoginResponse])
def search_user(request, user_name: Optional[str] = None) -> List[SafetyUser]:
    #1.鉴权
    if not is_admin(request):
        return []
    #2.查询符合要求的用户
    users = UserServices.list(user_name)

    return users

def is_admin(request):
    #这里 Django，会自动从前端发来的请求中的，
    #cookie 找到 session_id,再到数据库中找到匹配session_id
    #对应的用户数据,返回给我们
    safety_user = request.session[config.user_login_state]
    #判断用户是否登录，以及是否为管理员
    if safety_user == None or safety_user["user_role"] != config.admin_role:
        return False

    return True

逻辑实现

▼
python
复制代码
@staticmethod
def list(user_name: Optional[str]) -> List[SafetyUser]:
    """
    根据用户名模糊查询用户列表
    Args:
        user_name: 要查询的用户名(支持模糊匹配)
    Returns:
        List[SafetyUser]: 脱敏后的用户信息列表，可能为空列表
    """
    # 1.检查是否为None或纯空格
    if not user_name or user_name.isspace():
        # 1.1. 如果纯空，获取所以用户信息
        users = User.objects.all()
    else:
        # 2. 查询符合条件的，icontains 模糊查询，不分大小写
        users = User.objects.filter(user_name__icontains=user_name)  #
    # 3. 数据脱敏
    return [UserServices.convert_safety_user(user) for user in users]

用户管理删除

逻辑设计

  1. 鉴权
  2. 通过用户 id 去查询到要删除的用户
  3. 实现逻辑删除

接口设计

请求参数： 用户 id

请求类型：POST

请求体：JSON

请求返回值: 用户 id

实现

接口实现

  1. 对返回数据类型，进行一下定义，创建DeleteResponse 类
▼
python
复制代码
class DeleteResponse(Schema):
    response: bool
  1. 实现接口函数和逻辑
▼
python
复制代码
@router.get("/delete", response=DeleteResponse)
def delete_user(request, user_id: int) -> Dict[str, bool]:
    #1. 鉴权
    if not is_admin(request):
        return {"response": False}
    #2. 数据校验，确保有效id
    if user_id <= 0:
        return {"response": False}
    #3. 删除用户
    return {"response": UserServices.delete_user(user_id)}

逻辑实现

获取对应 id 的用户，并且删除，然后返回布尔值，确认删除

▼
python
复制代码
    @staticmethod
    def delete_user(user_id) -> bool:
        try:
            return User.objects.get(id=user_id).delete()[0] > 0
        except User.DoesNotExist:
            logger.warning(f"尝试删除不存在的用户ID: {user_id}")
            return False

字体转换

python 的代码一般都是用蛇形字体，但是我们前端JS 一般都用 camel，那么这边我们需要对，发过来的请求，和返回的请求做一些处理，可以看我之前写的 schema 字体转换文章

常量，以及一些隐私变量的存储方式

settings.py 会有一些数据库的用户名，密码，还有一些参数，和常量在项目中经常被调用，可以根据这篇文章的方式去处理来分离开

通用返回接口

逻辑设计

目的：给对象补充一些信息，告诉前端业务层面上是成功还是失败

200 , 404 , 500, 502, 503，定义自己的信息

  1. 定义自己的通用返回类
  2. 所有返回结果都会使用这个同样返回类来返回信息
▼
json
复制代码
//成功
{
  "code" : 0
  "data" : {
    "username": "Anson",
    "email": "xxx@xx.com",
    "address": "1234"

  }
  "messgae" : "ok"
}
失败
{
  "code" : 5001,
  "data" : null,
  "messgae" : "用户操作异常,xxx",
  "description": "参数不符合要求"
}

代码实现

构建通用返回类

构建我们的通用返回类，这边我本来一开始的想法，是跟着鱼皮，java 一样的写法，在当前所有放回类，外面去包一层，把当前返回信息 包在 data 这个属性里面就可以了，但是因为我用的 ninja 框架要用通过定义 schema 类，去返回信息，因此如果 我定义一个通用返回类，去包装所有返回值，这样做确实可以，但是 在 api 函数层，所有函数的返回类型 都会变成这个 ResponseBase, 在 swagger doc 也会 以 ResponseBase 来显示，这样带来的的问题是 可读性比较差，因为 ResponseBase 的 定义是这样的

▼
python
复制代码
class ResponseBase(Schema):
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
    code: Optional[int]
    data: Any
    message: Optional[str]
    description: Optional[str]

如果 所有接口的返回类被定义成这个样子，可读性是很差的，并不知道每个接口具体返回什么，

因此我这边使用的方式是把上方这个通用返回类作为父类，每个接口 的返回模板 通过继承这个父类，子类对 data 类型进行从新定义的方式去实现

▼
python
复制代码
class UserLoginResponseData(ToCamel, SafetyUser):
    user_name: str = Field(..., alias="username")
    user_id: int = Field(..., alias="id")


class UserLoginResponse(ResponseBase):
    data: Optional[UserLoginResponseData]

上方是我的 登录接口的实现方式，我先定义了一个 用户登录数据返回的模板UserLoginResponseData，再定义了一个接口应该返回给前端的模版类UserLoginResponse，在这个返回给前端的模版类中，我继承了通用返回类，并且把其中的data 属性 定义为 登录数据返回的模板类 data: Optional[UserLoginResponseData]

通过这种方式 我实现了 通用返回类，我把所有接口都通过 这个方式，从构了一下。

构建通用返回函数

至此所有接口都被从新定义成，通用返回类格式，但是又有着自己的数据属性，然而 api当中的函数要返回符合格式的结构，还需要我们在 ResponseBase 中构建一个通用函数，让每个接口返回模版类都可以调用

接口成功返回的函数

▼
python
复制代码
    # 定义类型变量，限定为ResponseBase及其子类
    T = TypeVar("T", bound="ResponseBase")

    @classmethod
    def success(cls: Type[T], data: Any, description: str = "") -> T:
        return cls(code=0, data=data, message="ok", description=description)

通过这个封装函数，去返回对应的模版类

构建全局异常处理

除了成功的返回 但我们还有异常的请求返回，这个相对复杂一些，因为之前在我们编写业务的时候，有很多异常的请求，比如密码不符合要求，用户名重复， 我们用 return -1 和 none 去返回 ，但是这样可读性也很差不规范，并且这些异常是我们业务中定义的异常不是代码 bug 产生的，因此现在我们要去定义自己的一个异常类 BusinessException，以及一个全局异常处理器 统一处理这些 BusinessException ，也就是说我们的业务中，出现不合规的情况，我们就raise 这个， BusinessException，然后 在通过 全局异常处理器去捕捉，然后返回信息给前端。这边我一开始也搞不清楚，只是跟着鱼皮的思路一步步写下来，边写边捋逻辑。才比较清晰了

  1. 定义这个 BusinessException
▼
python
复制代码
from typing import Optional
from core.constants import ErrorCode


class BusinessException(Exception):

    def __init__(
        self,
        message: Optional[str] = None,
        code: Optional[int] = None,
        description: Optional[str] = None,
        error_code: Optional[ErrorCode] = None,
    ) -> None:
        """
        初始化业务异常

        参数:
            message: 错误信息
            code: 错误码
            description: 错误描述
            error_code: 错误码枚举值
        """
        if error_code is not None:
            super().__init__(error_code.message)
            self.code = error_code.code
            self.message = error_code.message
            self.description = description
        else:
            super().__init__(message)
            self.code = code
            self.message = message
            self.description = description

    def __str__(self):
        return f"[{self.code}] {super().__str__()}: {self.description}"

因为这是一个 自定义Exception 首先必须继承 Exception 类，然后 我们再把自己想要添加的属性加到这个类当中，因为我们希望延续上面写的的通用返回类一样的模板，因此 这里 我们必须也要有，code，message，description，这些信息，因为这个类带着这些信息，后面的全局异常处理器，才能从中获取对应的信息并且返回给前端，但是除此 我们还定义了一个 error_code属性 ，这是一个枚举类,

  1. 因为现在我们需要自定义错误代码和返回信息，而一般对应的 code 都是有对应的 message，而枚举类正好可以一组组的去捆绑 code,message,description 的信息，让我们更方便的去调用
▼
python
复制代码
from enum import Enum


class ErrorCode(Enum):
    # 成功
    SUCCESS = (0, "ok", "")
    # 通用错误
    PARAMS_ERROR = (40000, "请求参数错误", "")
    NULL_ERROR = (40001, "请求数据为空", "")
    NOT_LOGIN = (40100, "未登录", "")
    NO_AUTH = (40101, "无权限", "")
    USER_NOT_EXIST = (40102, "密码不正确", "")
    ALREADY_LOGOUT = (40300, "已经登出", "")
    SYSTEM_ERROR = (50000, "系统内部异常", "")

    def __init__(self, code: int, message: str, description: str):
        self.code = code
        self.message = message
        self.description = description
  1. 定义完异常类以后 我们 就可以在 service.py ，对于 不合规的 ，请求 去抛出 我们这个 BusinessException 了，比如 user_register 函数
▼
python
复制代码
@staticmethod
    def user_register(
        user_account: str,
        user_password: str,
        check_password: str,
        planet_code: str,
        user_status: int = 0,
    ) -> int:
        """用户注册服务
        Args:
            user_account: 用户账号(4位以上字母数字组合)
            user_password: 用户密码(至少8位)
            check_password: 确认密码(需与user_password一致)
            planet_code: 星球编号
            user_status: 用户状态
        Returns:
            int: 成功返回用户ID，失败返回-1
        """
        # 1. 校验
        # 1.1 校验符合长度，不为空
        if not all([user_account, user_password, check_password, planet_code]):
            raise BusinessException(
                error_code=ErrorCode.PARAMS_ERROR, description="参数为空"
            )

        if len(user_account) < 4:
            raise BusinessException(
                error_code=ErrorCode.PARAMS_ERROR, description="用户账号过短"
            )

        if len(user_password) < 8 or len(check_password) < 8:
            raise BusinessException(
                error_code=ErrorCode.PARAMS_ERROR, description="用户密码过短"
            )

        if len(planet_code) > 5:
            raise BusinessException(
                error_code=ErrorCode.PARAMS_ERROR, description="星球编号过长"
            )

        # 1.2 账户不能包含特殊字符
        valid_pattern = r"[^a-zA-Z0-9]"
        if re.search(valid_pattern, user_account):
            raise BusinessException(
                error_code=ErrorCode.PARAMS_ERROR, description="用户名含特色字符"
            )

        # 1.3 密码和校验密码相同
        if not user_password == check_password:
            raise BusinessException(
                error_code=ErrorCode.PARAMS_ERROR, description="密码和校验密码不一致"
            )

        # 1.4 账户不能重复
        exists = User.objects.filter(user_account=user_account).exists()
        if exists:
            raise BusinessException(
                error_code=ErrorCode.PARAMS_ERROR, description="重复用户名"
            )

        # 1.5 星球编号不能重复
        exists = User.objects.filter(planet_code=planet_code).exists()
        if exists:
            raise BusinessException(
                error_code=ErrorCode.PARAMS_ERROR, description="重复星球编号"
            )

        # 2. 加密
        md5 = hashlib.md5()
        salt_pass = SALT + user_password
        md5.update(salt_pass.encode("utf-8"))
        encrypt_password = md5.hexdigest()

        # 3. 插入数据
        user = User()
        user.user_account = user_account
        user.user_password = encrypt_password
        user.user_status = user_status
        user.is_delete = 0
        user.user_role = 0
        user.planet_code = planet_code
        user.create_time = timezone.now()
        user.update_time = timezone.now()
        user.save()

        if user.id:
            return user.id
        else:
            return -1
  1. 最后我们要写一个全局异常处理器去捕获，这次抛出的异常，当异常发生，我们捕获异常信息，返回给前端，不让程序中断，这里 ninja 框架自带了这个@api.exception_handler(xxxException)，这个装饰器，我们只要使用这个装饰器，构写一个函数捕获这个异常，再在， ResponseBase 中 写一个返回信息的封装函数即可
▼
python
复制代码
from ninja import NinjaAPI
from .exception.business_exception import BusinessException
from users.schemas import ResponseBase
from core.constants import ErrorCode
import logging

logger = logging.getLogger("django")


def exception_handler(api: NinjaAPI):
    @api.exception_handler(BusinessException)
    def handle_business_exception(request, exc: BusinessException):
        logger.error(
            f"BusinessException occurred: {exc.code} - {exc.message}-{exc.description}",
            exc_info=True,
        )
        response = ResponseBase.from_error_code(
            code=exc.code, description=exc.description, message=exc.message
        )
        return api.create_response(request, response, status=400)

    @api.exception_handler(Exception)
    def handle_exception(request, exc: Exception):
        logger.error(f"Exception occurred: {str(exc)}", exc_info=True)
        response = ResponseBase.from_error_code(error_code=ErrorCode.PARAMS_ERROR)
        return api.create_response(request, response, status=500)

在 ResponseBase类中，写一个from_error_code的函数，对数据进行封装，这里为什么 我直接以字典返回的原因是因为，@api.exception_handler 这个装饰器，只允许返回 HTTP_RESPONSE, 然后构造 ninja 中 直接构造 response 的函数create_response，中的 Response不接受 schema 所有这边我只能用字典作为返回。

▼
python
复制代码
    @staticmethod
    def from_error_code(
        code: Optional[int] = 00000,
        description: Optional[str] = "",
        message: Optional[str] = "",
        error_code: Optional[ErrorCode] = None,
    ) -> ExceptionResponse:
        if error_code != None:
            return {
                "code": error_code.code,
                "message": error_code.message,
                "description": error_code.description,
            }

        return {"code": code, "message": message, "description": description}
Python
用户中心项目
项目
0个评论
全部评论
点击登录，快来和大家讨论吧～
表情
图片
暂无评论
Anson
Anson
等级vip
内容推荐
更多
耄耋专属AI Loop全自动编码器技术笔记
4
Django 是什么？
2
Django 的设计哲学有哪些？
3
什么是 MVC 模式？
2
开源我的 Vibe Coding 工作流，已有人靠它把项目做完了
7
作者分享
更多
AI 热点监控平台项目 VibeCoding 学习 -未完成 后续要持续更新
3
Python 后端项目中，写单元测试中 MagicMock 的使用指南
2
python 项目中的配置和一些常量 写在.env 做隐私保护隔离
1
Python Django 添加自定义管理器-实现逻辑删除
4
Python Django Ninja 框架 snake-case蛇形 转换 camel-case 驼峰
5
关联教程
更多
精选
SpringBoot + React 企业核心用户中心系统（新手必学）

SpringBoot + React 企业核心用户中心系统（新手必学）

¥19.9¥69
会员免费
下载 APP