datax-admin/app/crud/project.py

import time
import uuid
from typing import List
from app import models, schemas
from sqlalchemy.orm import Session
from itsdangerous import BadSignature, SignatureExpired
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
from constants.constants import SECRET_KEY

# 生成新的project_token
def generate_item_token(project: models.Project):
    s = Serializer(SECRET_KEY)
    return s.dumps({'project_id': project.id,'type':project.type}).decode('utf-8')

# 验证Token方法
def verify_item_token(db: Session, token: str):
    s = Serializer(SECRET_KEY)
    try:
        data = s.loads(token)
    except SignatureExpired:
        raise Exception("item_token验证失败")
    except BadSignature:
        raise Exception("item_token验证失败")
    project: models.Project = db.query(models.Project).filter(models.Project.id == data['project_id']).first()
    return project

def create_project(db: Session, item: schemas.ProjectCreate):
    # 创建项目
    check_1 = db.query(models.Project).filter(models.Project.name == item.name).first()
    if check_1:
        raise Exception('项目名重复')
    project_code = str(uuid.uuid1()).replace('-','')
    db_item = models.Project(**{
        'name': item.name,
        'code': project_code,
        'type': 1,
        'create_time': int(time.time())
    })
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    project_dict = db_item.to_dict()
    # 创建超级管理员与项目的关系
    relation = models.ProjectUserRelation(**{
        'user_id': item.user_id,
        'project_id': db_item.id,
        'role_id': 1,
    })
    db.add(relation)
    db.commit()
    db.refresh(relation)
    return project_dict

def update_project(db: Session,item: schemas.ProjectUpdate):
    db_item: models.Project = db.query(models.Project).filter(models.Project.id == item.project_id).first()
    if not db_item:
        raise Exception('项目不存在')
    db_item.name = item.name
    db.commit()
    db.flush()
    db.refresh(db_item)
    return db_item

def get_project_info(db: Session, id: int):
    project: models.Project = db.query(models.Project)\
        .filter(models.Project.id == id).first()
    if not project:
        raise Exception('该项目不存在')
    project_token = generate_item_token(project)
    item = project.to_dict()
    item.update({'project_token': project_token})
    return item

def switch_project_by_user(db: Session,project_id: int,user_id: int):
    project: models.Project = db.query(models.Project)\
        .filter(models.Project.id == id).first()
    relation: models.ProjectUserRelation = db.query(models.ProjectUserRelation)\
        .filter(models.ProjectUserRelation.project_id == project_id)\
        .filter(models.ProjectUserRelation.user_id == user_id).first()
    role: models.Roles = db.query(models.Roles)\
        .filter(models.Roles.id == relation.role_id).first()
    project_token = generate_item_token(project)
    item = project.to_dict()
    item.update({
        'project_token': project_token,
        'role_id': role.id,
        'role_name': role.name,
        'role_code': role.code,
        })
    return item

def get_project_list(db: Session, user_id: int):
    relations = get_relations_by_user(db, user_id)
    project_ids = [ r.project_id for r in relations ]
    projects: List[models.Project] = db.query(models.Project)\
        .filter(models.Project.id.in_(project_ids)).all()
    res = []
    for project in projects:
        project_token = generate_item_token(project)
        item = project.to_dict()
        item.update({'project_token': project_token})
        res.append(item)
    return res

def get_relations_by_user(db: Session, user_id: int):
    relations: List[models.ProjectUserRelation] = db.query(models.ProjectUserRelation)\
        .filter(models.ProjectUserRelation.user_id == user_id).all()
    return relations

def get_relations_by_project(db: Session, project_id: int):
    relations: List[models.ProjectUserRelation] = db.query(models.ProjectUserRelation)\
        .filter(models.ProjectUserRelation.project_id == project_id).all()
    return relations