集成jupyter

app/common/security/auth.py

@@ -16,7 +16,7 @@ async def verify_super_admin(token: str = Depends(reuseable_oauth)) -> schemas.T
     token_data_str = decode_base64(token)
     token_data_dict = json.loads(token_data_str)
     token_data = schemas.TokenData(**token_data_dict)
-    if token_data.role_id != '726a51e45b4d11edbb4809c4df301a':
+    if token_data.role_id != 1:
         raise Exception("暂无权限")
     g.user_id = token_data.user_id
     g.project_id = token_data.project_id
@@ -27,7 +27,7 @@ async def verify_special(token: str = Depends(reuseable_oauth)) -> schemas.Token
     token_data_str = decode_base64(token)
     token_data_dict = json.loads(token_data_str)
     token_data = schemas.TokenData(**token_data_dict)
-    if not token_data.role_id in ['726a51e45b4d11edbb4809c4df301a','9ff183445b4d11ed87db29f50d093a']:
+    if not token_data.role_id in [1,3]:
         raise Exception("暂无权限")
     g.user_id = token_data.user_id
     g.project_id = token_data.project_id

app/crud/__init__.py

@@ -14,3 +14,4 @@ from app.crud.af_run import *
 from app.crud.relation import *
 from app.crud.roles import *
 from app.crud.data_table import *
+from app.crud.programme import *

app/crud/programme.py

@@ -0,0 +1,99 @@
+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

app/models/__init__.py

@@ -16,4 +16,5 @@ from app.models.users import *
 from app.models.project import *
 from app.models.roles import *
 from app.models.project_user_relation import *
-from app.models.data_table import *
+from app.models.data_table import *
+from app.models.programme import *

app/models/programme.py

@@ -0,0 +1,33 @@
+from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
+
+from app.models.database import BaseModel
+
+
+class Programme(BaseModel):
+    __tablename__ = "programme"
+
+    id = Column(Integer, primary_key=True, index=True)
+    # 可视化编程名称
+    name = Column(String, nullable=False)
+    # 密码（暗文）
+    password = Column(String, nullable=False)
+    # 挂载目录
+    workspace = Column(String, nullable=False)
+    # api_url
+    base_url = Column(String, nullable=False)
+    # 使用镜像
+    image = Column(String, nullable=False)
+    # 路径
+    path = Column(String, nullable=False)
+    # xx名称
+    release_name = Column(String, nullable=False)
+    # 状态
+    status = Column(Integer, nullable=False)
+    # 创建时间
+    create_time = Column(Integer, nullable=False)
+    # 用户id
+    user_id = Column(String, nullable=False)
+    # 用户名称
+    user_name = Column(String, nullable=False)
+    # 项目id
+    project_id = Column(String, nullable=False)

app/routers/auth.py

@@ -23,16 +23,16 @@ router = APIRouter(
 @sxtimeit
 def switch_project(switch: schemas.SwitchProject):
     if switch.project_id == special_project_id and super_admin_role in switch.role_ids:
-        return {"role_id": '726a51e45b4d11edbb4809c4df301a'}
+        return {"role_id": 1}
     elif switch.project_id == special_project_id:
-        return {"role_id": '9ff183445b4d11ed87db29f50d093a'}
+        return {"role_id": 3}
     else:
-        return {"role_id": '026bd8bc5b4e11ed857e6b5ec5c8d6'}
+        return {"role_id": 5}
 
 @router.post("/login",response_model=schemas.Token)
 async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
     user_id, project_id = form_data.username, form_data.password
-    token_data = {"user_id": user_id,"project_id": project_id, "role_id": "726a51e45b4d11edbb4809c4df301a"}
+    token_data = {"user_id": user_id,"project_id": project_id, "role_id": 1}
     token_data_str =json.dumps(token_data)
     access_token = encode_base64(token_data_str).replace('\n','')
     return {"access_token": access_token, "token_type": "bearer"}

app/routers/programme.py

@@ -0,0 +1,46 @@
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+from app.common.security.auth import verify_users
+from utils.sx_time import sxtimeit
+from utils.sx_web import web_try
+import app.crud as crud
+from app import schemas, get_db
+
+
+
+
+router = APIRouter(
+    prefix="/jpt/programme",
+    tags=["programme-可视化编程管理"],
+)
+
+@router.post("")
+@web_try()
+@sxtimeit
+def create_programme(item: schemas.ProgrammeCreate, token_data: schemas.TokenData = Depends(verify_users), db: Session = Depends(get_db)):
+    return crud.create_programme(db, item)
+
+@router.put("/start_jupyter")
+@web_try()
+@sxtimeit
+def start_jupyter(item: schemas.ProgrammeId, token_data: schemas.TokenData = Depends(verify_users), db: Session = Depends(get_db)):
+    return crud.start_jupyter(db, item)
+
+@router.put("/stop_jupyter")
+@web_try()
+@sxtimeit
+def stop_jupyter(item: schemas.ProgrammeId, token_data: schemas.TokenData = Depends(verify_users), db: Session = Depends(get_db)):
+    return crud.stop_jupyter(db, item)
+
+
+@router.put("/update_password")
+@web_try()
+@sxtimeit
+def update_jupyter_password(item: schemas.ProgrammeUpdate, token_data: schemas.TokenData = Depends(verify_users), db: Session = Depends(get_db)):
+    return crud.update_jupyter_password(db, item)
+
+@router.get("")
+@web_try()
+@sxtimeit
+def get_programme(token_data: schemas.TokenData = Depends(verify_users), db: Session = Depends(get_db)):
+    return crud.get_programme(db)

app/schemas/__init__.py

@@ -14,4 +14,5 @@ from app.schemas.af_task import *
 from app.schemas.af_job import *
 from app.schemas.af_run import *
 from app.schemas.dag import *
-from app.schemas.auth import *
+from app.schemas.auth import *
+from app.schemas.programme import *

app/schemas/auth.py

@@ -8,7 +8,7 @@ class Token(BaseModel):
 class TokenData(BaseModel):
     user_id: str
     project_id: str
-    role_id: str
+    role_id: int
 
 class SwitchProject(BaseModel):
     project_id: str

app/schemas/job_log.py

@@ -1,4 +1,3 @@
-from re import S
 from typing import List, Optional
 
 from pydantic import BaseModel

app/schemas/programme.py

@@ -0,0 +1,31 @@
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+class ProgrammeBase(BaseModel):
+    # 程序名称
+    name: str
+
+class ProgrammeCreate(ProgrammeBase):
+    # 密码
+    password: str
+    # 创建用户
+    user_name: str
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "name": "test",
+                "password": "test",
+                "user_name": "test",
+            }
+        }
+
+class ProgrammeId(BaseModel):
+    # 程序id
+    programme_id: int
+
+
+class ProgrammeUpdate(ProgrammeId):
+    # 密码
+    password: str

app/utils/send_util.py

@@ -2,11 +2,14 @@ from unittest import result
 import requests
 from configs.settings import config
 
-HOST = config.get('AF_BACKEND', 'host')
-PORT = config.get('AF_BACKEND', 'port')
+AF_HOST = config.get('AF_BACKEND', 'host')
+AF_PORT = config.get('AF_BACKEND', 'port')
+
+PROGRAMME_URL = config.get('PROGRAMME', 'url')
+
 
 def send_post(uri,data):
-    res = requests.post(url=f'http://{HOST}:{PORT}{uri}', json=data)
+    res = requests.post(url=f'http://{AF_HOST}:{AF_PORT}{uri}', json=data)
     result = res.json()
     if 'code' in result.keys() and result['code'] == 200:
         return res.json()
@@ -15,7 +18,7 @@ def send_post(uri,data):
         raise Exception(f'{uri}-->请求airflow失败-->{msg}')
 
 def send_submit(af_job_id):
-    res = requests.post(url=f'http://{HOST}:{PORT}/af/af_job/submit?id='+str(af_job_id))
+    res = requests.post(url=f'http://{AF_HOST}:{AF_PORT}/af/af_job/submit?id='+str(af_job_id))
     result = res.json()
     if 'code' in result.keys() and result['code'] == 200:
         return res.json()
@@ -25,7 +28,7 @@ def send_submit(af_job_id):
 
 
 def send_put(uri,path_data,data):
-    res = requests.put(url=f'http://{HOST}:{PORT}{uri}/{path_data}', json=data)
+    res = requests.put(url=f'http://{AF_HOST}:{AF_PORT}{uri}/{path_data}', json=data)
     result = res.json()
     if 'code' in result.keys() and result['code'] == 200:
         return res.json()
@@ -34,7 +37,7 @@ def send_put(uri,path_data,data):
         raise Exception(f'{uri}-->请求airflow失败-->{msg}')
 
 def send_get(uri,path_data):
-    res = requests.get(url=f'http://{HOST}:{PORT}{uri}/{path_data}')
+    res = requests.get(url=f'http://{AF_HOST}:{AF_PORT}{uri}/{path_data}')
     result = res.json()
     if 'code' in result.keys() and result['code'] == 200:
         return res.json()
@@ -45,7 +48,7 @@ def send_get(uri,path_data):
 
 # 执行任务
 def send_execute(path_data):
-    res = requests.post(url=f'http://{HOST}:{PORT}/af/af_job/{str(path_data)}/run')
+    res = requests.post(url=f'http://{AF_HOST}:{AF_PORT}/af/af_job/{str(path_data)}/run')
     result = res.json()
     if 'code' in result.keys() and result['code'] == 200:
         return res.json()
@@ -56,7 +59,7 @@ def send_execute(path_data):
 # 起停任务
 def send_pause(af_job_id, status):
     flag = True if status == 0 else False
-    res = requests.patch(url=f'http://{HOST}:{PORT}/af/af_job/{str(af_job_id)}/pause/{str(flag)}')
+    res = requests.patch(url=f'http://{AF_HOST}:{AF_PORT}/af/af_job/{str(af_job_id)}/pause/{str(flag)}')
     result = res.json()
     if 'code' in result.keys() and result['code'] == 200:
         return res.json()
@@ -66,7 +69,7 @@ def send_pause(af_job_id, status):
 
 # 删除任务
 def send_delete(uri, path_data):
-    res = requests.delete(url=f'http://{HOST}:{PORT}{uri}/{path_data}')
+    res = requests.delete(url=f'http://{AF_HOST}:{AF_PORT}{uri}/{path_data}')
     result = res.json()
     if 'code' in result.keys() and result['code'] == 200:
         return res.json()
@@ -76,7 +79,7 @@ def send_delete(uri, path_data):
 
 # 获取airflow端dag文件生成时间
 def get_job_last_parsed_time(path_data):
-    res = requests.get(url=f'http://{HOST}:{PORT}/af/af_job/{path_data}/last_parsed_time')
+    res = requests.get(url=f'http://{AF_HOST}:{AF_PORT}/af/af_job/{path_data}/last_parsed_time')
     result = res.json()
     if 'code' in result.keys() and result['code'] == 200:
         return res.json()
@@ -86,7 +89,7 @@ def get_job_last_parsed_time(path_data):
 
 # 获取job某次运行的状态
 def get_job_run_status(path_data):
-    res = requests.get(url=f'http://{HOST}:{PORT}/af/af_run/{path_data}/status')
+    res = requests.get(url=f'http://{AF_HOST}:{AF_PORT}/af/af_run/{path_data}/status')
     result = res.json()
     if 'code' in result.keys() and result['code'] == 200:
         return res.json()
@@ -96,7 +99,7 @@ def get_job_run_status(path_data):
 
 # 中间结果转存
 def data_transfer_run(source_tb: str, target_tb: str):
-    res = requests.post(url=f'http://{HOST}:{PORT}/af/af_job/000/data_transfer_run?source_tb={source_tb}&target_tb={target_tb}')
+    res = requests.post(url=f'http://{AF_HOST}:{AF_PORT}/af/af_job/000/data_transfer_run?source_tb={source_tb}&target_tb={target_tb}')
     result = res.json()
     print(result)
     if 'code' in result.keys() and result['code'] == 200:
@@ -107,7 +110,7 @@ def data_transfer_run(source_tb: str, target_tb: str):
 
 # 获取task日志
 def get_task_log(job_id: str, af_run_id: str, task_id: str):
-    res = requests.get(url=f'http://{HOST}:{PORT}/af/af_run/task_log/{job_id}/{af_run_id}/{task_id}')
+    res = requests.get(url=f'http://{AF_HOST}:{AF_PORT}/af/af_run/task_log/{job_id}/{af_run_id}/{task_id}')
     result = res.json()
     if 'code' in result.keys() and result['code'] == 200:
         return res.json()
@@ -118,10 +121,53 @@ def get_task_log(job_id: str, af_run_id: str, task_id: str):
 
 # 获取中间结果转存状态
 def get_data_transfer_run_status(af_run_id: str):
-    res = requests.get(url=f'http://{HOST}:{PORT}/af/af_run/data_transfer_log/{af_run_id}')
+    res = requests.get(url=f'http://{AF_HOST}:{AF_PORT}/af/af_run/data_transfer_log/{af_run_id}')
+    result = res.json()
+    if 'code' in result.keys() and result['code'] == 200:
+        return res.json()
+    else:
+        msg = result['msg'] if 'msg' in result.keys() else result
+        raise Exception(f'获取中间结果转存状态,请求airflow失败-->{msg}')
+
+
+# 创建jupyter
+def create_jupyter(data):
+    res = requests.post(url=f'http://{PROGRAMME_URL}/helm/ops/start', json=data)
+    result = res.json()
+    if 'code' in result.keys() and result['code'] == 200:
+        return res.json()
+    else:
+        stop_jupyter({'namespace': data['namespace'],'release_name': data['release_name']})
+        msg = result['msg'] if 'msg' in result.keys() else result
+        raise Exception(f'创建jupyter,请求jupyter端失败-->{msg}')
+
+
+# 获取jupyter密码
+def get_jupyter_password(data):
+    res = requests.get(url=f'http://{PROGRAMME_URL}/helm/password', json=data)
+    result = res.json()
+    if 'code' in result.keys() and result['code'] == 200:
+        return res.json()
+    else:
+        msg = result['msg'] if 'msg' in result.keys() else result
+        raise Exception(f'获取jupyter密码,请求jupyter端失败-->{msg}')
+
+# 停止jupyter
+def stop_jupyter(data):
+    res = requests.post(url=f'http://{PROGRAMME_URL}/helm/ops/stop', json=data)
+    result = res.json()
+    if 'code' in result.keys() and result['code'] == 200:
+        return res.json()
+    else:
+        msg = result['msg'] if 'msg' in result.keys() else result
+        raise Exception(f'停止jupyter,请求jupyter端失败-->{msg}')
+
+# 更新jupyter
+def update_jupyter(data):
+    res = requests.post(url=f'http://{PROGRAMME_URL}/helm/ops/upgrade', json=data)
     result = res.json()
     if 'code' in result.keys() and result['code'] == 200:
         return res.json()
     else:
         msg = result['msg'] if 'msg' in result.keys() else result
-        raise Exception(f'获取中间结果转存状态,请求airflow失败-->{msg}')
+        raise Exception(f'更新jupyter,请求jupyter端失败-->{msg}')

data/data.sql

@@ -441,7 +441,28 @@ CREATE TABLE `data_table` (
   `create_time` bigint NOT NULL COMMENT '创建时间',
   `user_id` bigint NOT NULL COMMENT '创建人、分享人',
   PRIMARY KEY (`id`)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3 COMMENT='表';
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='表';
+
+-- ----------------------------
+-- Table structure for programme
+-- ----------------------------
+DROP TABLE IF EXISTS `programme`;
+CREATE TABLE `programme` (
+  `id` bigint NOT NULL AUTO_INCREMENT,
+  `name` varchar(50) NOT NULL COMMENT '名称',
+  `password` varchar(300) NOT NULL COMMENT '密码（暗文）',
+  `workspace` varchar(50) NOT NULL COMMENT '挂载目录',
+  `base_url` varchar(50) NOT NULL COMMENT 'api_url',
+  `image` varchar(100) NOT NULL COMMENT '使用镜像',
+  `path` varchar(50) NOT NULL COMMENT '路径',
+  `release_name` varchar(50) NOT NULL COMMENT '名称',
+  `status` tinyint NOT NULL COMMENT '状态',
+  `create_time` int NOT NULL COMMENT '创建时间',
+  `user_id` varchar(50) NOT NULL COMMENT '用户id',
+  `user_name` varchar(50) NOT NULL COMMENT '用户名称',
+  `project_id` varchar(50) NOT NULL COMMENT '项目id',
+  PRIMARY KEY (`id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4  COMMENT='可视化编程';
 
 
 SET FOREIGN_KEY_CHECKS = 1;

development.ini

@@ -75,3 +75,13 @@ kerberos_config = {
 [PERMISSIONS]
 super_admin_role = TENANT_ADMIN
 special_project_id = test
+
+[PROGRAMME]
+url = aihub-dag.sxkj.com
+namespace = airflow
+super_image = SXKJ:32775/jupyterlab
+ordinary_image = SXKJ:32775/jupyterlab0
+tag = sxkj
+host = jupyterlab.sxkj.com
+chart = aihub-dag-jupyter.tgz
+path_type = ImplementationSpecific

server.py

@@ -16,6 +16,7 @@ import app.routers.code_check as router_code_check
 import app.routers.jm_job_info as router_jm_job_info
 import app.routers.jm_job_log as router_jm_job_log
 import app.routers.auth as router_auth
+import app.routers.programme as router_programme
 from app.routers.run import router_af_run
 from app.routers.job import router_af_job
 from app.routers.task import router_af_task
@@ -44,6 +45,7 @@ app.add_middleware(
 app.add_middleware(GlobalsMiddleware)
 
 app.include_router(router_auth.router)
+app.include_router(router_programme.router)
 app.include_router(router_jjds.router)
 app.include_router(router_constants.router)
 app.include_router(router_job_info.router)