数据工厂系列（13）项目管理模块-项目curd开发

继续创造，加快生长！这是我参加「日新计划 6 月更文应战」的第19天，点击查看活动详情

大家好~我是小方，欢迎大家重视笋货测验笔记体完记住俾个like呀

回忆

上一期，咱们仅仅规划了项目表和写了一个新增项目的接口，这一期咱们继续开发项目根底功用

优化新增项目接口

上一期咱们有提到一个点，如果新增的项目拉取方法是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比较简单，下一期咱们试着来难一点的吧~附上进度表

• 项目地址
  • 后端：github.com/JokerChat/F…
  • 前端：github.com/JokerChat/F…
• 今日已提交代码
  • 后端：b23ae21