datax-admin/app/services/jm_job.py

from app import crud, models
from app.utils.send_util import *
from app.utils.utils import get_cmd_parameter
from sqlalchemy.orm import Session

type_dict = {
    "Java": "java",
    "Python": "python",
    "Dag": "sparks"
}

def jm_job_create_task(jm_homework: models.JmHomework, db: Session):
    af_task = {
        "name": jm_homework.name,
        "file_urls": [jm_homework.dag_url] if jm_homework.type == "Dag" else [jm_homework.script_file],
        "script": "",
        "cmd": jm_homework.execute_command if jm_homework.type != "Dag" else "",
        "cmd_parameters": "",
        "envs": {},
        "run_image": jm_homework.image_url if jm_homework.type != "Dag" else "",
        "task_type": type_dict[jm_homework.type],
        "user_id": 0,
    }
    res = send_post('/jpt/af_task', af_task)
    af_task = res['data']
    crud.create_relation(db ,jm_homework.id, jm_homework.type, af_task['id'])
    return af_task

def jm_job_update_task(jm_homework: models.JmHomework, db: Session):
    relation = crud.get_af_id(db, jm_homework.id, jm_homework.type)
    af_task = {
        "name": jm_homework.name,
        "file_urls": [jm_homework.dag_url] if jm_homework.type == "Dag" else [jm_homework.script_file],
        "script": "",
        "cmd": jm_homework.execute_command if jm_homework.type != "Dag" else "",
        "cmd_parameters": "",
        "envs": {},
        "run_image": jm_homework.image_url if jm_homework.type != "Dag" else "",
        "task_type": type_dict[jm_homework.type],
        "user_id": 0,
    }
    res = send_put('/jpt/af_task', relation.af_id, af_task)
    af_task = res['data']

    return af_task