datax-admin/app/crud/programme.py

import time
from typing import List
from app import models, schemas
from sqlalchemy.orm import Session
import app.utils.send_util as send_util
from configs.globals import g
from configs.settings import DefaultOption, config
special_project_id = config.get('PERMISSIONS', 'special_project_id')
namespace = config.get('PROGRAMME', 'namespace')
super_image = config.get('PROGRAMME', 'super_image')
ordinary_image = config.get('PROGRAMME', 'ordinary_image')
tag = config.get('PROGRAMME', 'tag')
host = config.get('PROGRAMME', 'host')
chart = config.get('PROGRAMME', 'chart')
path_type = config.get('PROGRAMME', 'path_type')

def create_programme(db: Session, item: schemas.ProgrammeCreate):
    db_item = db.query(models.Programme).filter(models.Programme.project_id == g.project_id).first()
    if db_item:
        raise Exception("该项目已存在编程，不可重复创建")
    p_res = send_util.get_jupyter_password({"password": item.password})
    password = p_res['data']
    db_item = models.Programme(**{
        'name': item.name,
        'password': password,
        'workspace': f"workspace_{g.project_id}",
        'base_url': f"/nbss_{g.project_id}",
        'image': super_image if g.project_id == special_project_id else ordinary_image,
        'path': f"/nbss_{g.project_id}",
        'release_name': f"aihub-dag-jpt-{g.project_id}",
        'status': 0,
        'user_id': g.user_id,
        'user_name': item.user_name,
        'project_id': g.project_id,
        'create_time': int(time.time())
    })
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

def start_jupyter(db: Session, item: schemas.ProgrammeId):
    db_item: models.Programme = db.query(models.Programme).filter(models.Programme.id == item.programme_id).first()
    if not db_item:
        raise Exception("未找到该编程")
    jupyter_create_data = {
        'password': db_item.password,
        'namespace': namespace,
        'workspace': db_item.workspace,
        'base_url': db_item.base_url,
        'image': db_item.image,
        'tag': tag,
        'path': db_item.path,
        'host': host,
        'release_name': db_item.release_name,
        'chart': chart,
        'path_type': path_type
    }
    c_res = send_util.create_jupyter(jupyter_create_data)
    j_data = c_res['data'] if 'data' in c_res.keys() else None
    if not j_data:
        send_util.stop_jupyter({'namespace': namespace,'release_name': db_item.release_name})
        raise Exception("创建Jupyter失败")
    url = f"http://{j_data['host']}{j_data['base_url']}/lab"
    db_item.status = 1
    db.commit()
    db.flush()
    db.refresh(db_item)
    return url

def stop_jupyter(db: Session, item: schemas.ProgrammeId):
    db_item: models.Programme = db.query(models.Programme).filter(models.Programme.id == item.programme_id).first()
    if not db_item:
        raise Exception("未找到该编程")
    send_util.stop_jupyter({'namespace': namespace,'release_name': db_item.release_name})
    db_item.status = 0
    db.commit()
    db.flush()
    db.refresh(db_item)
    return db_item

def update_jupyter_password(db: Session, item: schemas.ProgrammeUpdate):
    db_item: models.Programme = db.query(models.Programme).filter(models.Programme.id == item.programme_id).first()
    if not db_item:
        raise Exception("未找到该编程")
    print()
    if db_item.status == 1:
        raise Exception("程序正在运行，请先停止再修改密码")
    p_res = send_util.get_jupyter_password({"password": item.password})
    password = p_res['data']
    db_item.password = password
    db.commit()
    db.flush()
    db.refresh(db_item)
    return db_item

def get_programme(db: Session):
    db_items: List[models.Programme] = db.query(models.Programme).filter(models.Programme.project_id == g.project_id).all()
    return db_items