Test drive: project management module - curd development project

优化新增项目接口

上一期我们有提到一个点,如果新增的项目拉取方式是http,必须配置git账号和密码,但是密码存在表里的是明文,这很不安全诶,被人知道了账号密码,不就随便拉取项目了么？所以这里我们要引入加密算法进行加密账号密码,后端存储加密后的密文,前端需要展示或者拉取项目时,再进行解密.

这里的加密算法我用的是AES加密,我们首先配置一个key和iv值,这里长度是有要求,我们需要的是16位字符串,直接写一段代码生成即可

import random, string
key = ''.join(random.choices(string.ascii_letters + string.digits, k=16))
print(key)

生成key和iv值之后,需要在config.py配置一下

在app/utils下新建aes_utils.py,这里加密原理就先不说了哈,比较复杂,有兴趣可百度搜搜相关的知识~

import base64
from Crypto.Cipher import AES
from config import Config

class AesUtils:

    @classmethod
    def add_to_16(cls, value):
        """不足16位补0"""
        while len(value) % 16 != 0:
            value += '\0'
        return str.encode(value)  # 返回bytes

    @classmethod
    def encrypt(cls, text):
        """
        AES加密
        :param text: 待加密明文
        :return:
        """
        # 初始化加密器
        aes = AES.new(cls.add_to_16(Config.AES_KEY), AES.MODE_CBC, cls.add_to_16(Config.AES_IV))
        bs = AES.block_size
        pad2 = lambda s: s + (bs - len(s) % bs) * chr(bs - len(s) % bs)  # PKS7
        encrypt_aes = aes.encrypt(pad2(text).encode())
        # 用base64转成字符串形式
        # 执行加密并转码返回bytes
        encrypted_text = str(base64.encodebytes(encrypt_aes), encoding='utf-8')
        # 和js的 结果相同 http://tool.chacuo.net/cryptaes
        return encrypted_text.replace('\n', '')  # 去除换行符

    @classmethod
    def decrypt(cls, text):
        """
        AES解密
        :param text: 待解密密文
        :return:
        """
        # 初始化解密器
        # 偏移量 16个0
        aes = AES.new(cls.add_to_16(Config.AES_KEY), AES.MODE_CBC, cls.add_to_16(Config.AES_IV))
        # 优先逆向解密base64成bytes
        base64_decrypted = base64.decodebytes(text.encode(encoding='utf-8'))
        # 执行解密并转码返回str
        decrypted_text = aes.decrypt(base64_decrypted).decode()
        unpad = lambda s: s[0:-ord(s[-1])]
        return unpad(decrypted_text)

if __name__ == '__main__':
    # 加密
    encrypt_data = AesUtils.encrypt('fang')
    print(encrypt_data)
    # 解密
    decrypt_data = AesUtils.decrypt(encrypt_data)
    print(decrypt_data)

去到project_schema.py新增项目模型里,把这个AES加密方法应用上

改造完成

编辑项目接口

还是按老套路,先来设计入参模型,编辑项目的话,跟新增项目的字段几乎一致,编辑在基础上多了个id罢了,直接继承原来的新增项目模型

class EditProject(AddProject):
    id : int=Field(..., title="项目id", description="主键id")

    @validator('id')
    def id_not_empty(cls, v):
        return ToolsSchemas.not_empty(v)

ProjectDao.py编辑项目逻辑编写

@classmethod
    @record_log
    def update_project(cls, data: EditProject, user: dict) -> None:
        """
        编辑项目
        :param data: 编辑项目模型
        :param user: 用户数据
        :return:
        """
        with Session() as session:
            project = session.query(DataFactoryProject).filter(DataFactoryProject.id == data.id,
                                                               DataFactoryProject.del_flag == 0).first()
            if project is None: raise NormalException("项目不存在")
            # 根据名称查出数据
            project_name = session.query(DataFactoryProject).filter(
                DataFactoryProject.project_name == data.project_name,
                DataFactoryProject.del_flag == 0).first()
            # 如果有数据且主键id与请求参数id不相等
            if project_name is not None and project_name.id != data.id:
                raise NormalException("项目名重复, 请重新录入！！！")
            git_project_name = session.query(DataFactoryProject).filter(
                DataFactoryProject.git_project == data.git_project,
                DataFactoryProject.del_flag == 0).first()
            if git_project_name is not None and git_project_name.id != data.id:
                raise NormalException("git项目名重复, 请重新录入！！！")
            DbUtils.update_model(project, data.dict(), user)
            session.commit()

这里要判断项目名或者git项目名是否重复,其他没了,都是些更新数据

路由函数编写

@router.post("/update", name="编辑项目")
def update_project(body: EditProject, user= Depends(Auth(Permission.LEADER))):
    try:
        ProjectDao.update_project(body, user)
        return ResponseDto(msg="编辑成功")
    except Exception as e:
        raise NormalException(str(e))

最后测试一下

入库成功

删除项目接口

删除项目,这里初期版本就只是将数据置为del_flag = 1,后续我们要加上删除本地拉取下来的项目、删除项目关联的所有造数场景…

这里入参的话,直接query参数好了,不用再定义什么入参模型

ProjectDao.py删除项目逻辑编写

    @classmethod
    @record_log
    def delete_project(cls, id: int, user: dict) -> None:
        """删除项目"""
        with Session() as session:
            project = session.query(DataFactoryProject).filter(DataFactoryProject.id == id,
                                                               DataFactoryProject.del_flag == 0).first()
            if project is None: raise NormalException("项目不存在")
            DbUtils.delete_model(project, user)
            session.commit()
            session.refresh(project)
            return project

逻辑比较简单,就更新几个字段值,上面也有说到,删除项目后需要完成一些后置的操作,所以这里我们要commit后再进行refresh,获取删除项目的信息,再进行后置的操作.

路由函数编写

测试环节

项目列表接口

这里项目列表初期版本的逻辑就跟用户列表接口一样,就单表查询,后面引入项目权限,就得根据用户权限来获取.

入参没啥好说的,直接跟用户列表接口保持一直好了,返参的话,直接参考用户列表接口来设计

ProjectDao.py项目列表逻辑编写

@classmethod
    @record_log
    def list_project(cls, page: int=1, size: int=10, search: str=None) ->(int, DataFactoryProject):
        """
        获取项目列表
        :param page: 页码
        :param size: 大小
        :param search: 搜索内容
        :return:
        """
        with Session() as session:
            filter_list = [DataFactoryProject.del_flag == 0]
            if search:
                filter_list.append(DataFactoryProject.project_name.like(f"%{search}%"))
            project = session.query(DataFactoryProject).filter(*filter_list)
            project_infos = project.order_by(desc(DataFactoryProject.update_time)).limit(size).offset((page - 1) * size).all()
            total = project.count()
        return total, project_infos

路由函数编写,跟用户列表接口写法一致,这里就不好了哈~

测试环节

总结

今天完成了项目管理模块-项目curd开发,curd比较简单,下一期我们试着来难一点的吧~附上进度表