datax-admin/app/routers/jm_job_info.py

import datetime
import croniter
import re
from typing import Optional, List
from fastapi import APIRouter
from fastapi import Depends
from sqlalchemy.orm import Session
from app import models, schemas

import app.crud as crud
from app.schemas import cron_expression
from app.utils.cron_utils import *
from utils.sx_time import sxtimeit
from utils.sx_web import web_try
from fastapi_pagination import Page, add_pagination, paginate, Params

from app import get_db



router = APIRouter(
    prefix="/jpt/jm_job_info",
    tags=["jm_job_info-定时任务管理"],
)

@router.post("/")
@web_try()
@sxtimeit
def create_jm_job_info(item: schemas.JmJobInfoCreate, db: Session = Depends(get_db)):
    jm_job_info,nodes,edges = crud.create_jm_job_info(db, item)
    job_id = jm_job_info.id
    create_jm_job_node(db, nodes, edges, job_id)
    return jm_job_info.to_dict()


def create_jm_job_node(db: Session, nodes, edges, job_id):
    uuid_node_id = {}
    if nodes is None or len(nodes) == 0:
        return
    for node in nodes:
        uuid = node['id']
        node_item = models.JmJobNode(**{
            'job_id': job_id,
            'homework_id': node['homework_id'],
            'homework_name': node['homework_name'],
            'start_point': 1,
        })
        node_item = crud.create_jm_job_node(db,node_item)
        node_id = node_item.id
        uuid_node_id.update({uuid:node_id})
    if nodes is None or len(nodes) == 0:
        return
    for edge in edges:
        edge_item = models.JmJobEdge(**{
            'job_id': job_id,
            'in_node_id': uuid_node_id[edge['source']],
            'out_node_id': uuid_node_id[edge['target']]
        })
        edge = crud.create_jm_job_edge(db,edge_item)
    return

@router.get("/")
@web_try()
@sxtimeit
def get_jm_job_infos(db: Session = Depends(get_db)):
    res_list = []
    jm_job_list = crud.get_jm_job_infos(db)
    jm_job_ids = [job.id for job in jm_job_list]
    relations = crud.get_af_ids(db,jm_job_ids, 'job')
    af_to_datax = {relation.af_id:relation.se_id for relation in relations}
    af_job_runs = crud.get_airflow_runs_by_af_job_ids(db, af_to_datax.keys())
    res = {}
    for af_job_run in af_job_runs:
        tasks = list(af_job_run.details['tasks'].values()) if len(list(af_job_run.details['tasks'].values()))>0 else []
        if len(tasks) > 0:
            task = tasks[-1]
            task.pop('log',None)
            job_id = af_to_datax[int(af_job_run.job_id)]
            log = {
                "id": af_job_run.id,
                "job_id": job_id,
                "af_job_id": int(af_job_run.job_id),
                "run_id": af_job_run.run_id,
                "trigger_time": af_job_run.start_time,
                "trigger_result": 1 if task else 0,
                "execute_time": task['start_time'] if task else 0,
                "execute_result": 1 if task and task['status'] == 'success' else 0,
                "end_time": task['end_time'] if task else 0,
            }
            if job_id in res.keys():
                res[job_id].append(log)
            else:
                res.update({job_id: [log]})
    for jm_job in jm_job_list:
        history = res[jm_job.id] if jm_job.id in res.keys() else []
        jm_job_dict = jm_job.to_dict()
        jm_job_dict.update({'history':history[0:10]})
        res_list.append(jm_job_dict)
    return res_list

@router.get("/info")
@web_try()
@sxtimeit
def get_jm_job_info(jm_job_id: int, db: Session = Depends(get_db)):
    jm_job = crud.get_jm_job_info(db,jm_job_id)
    jm_job_dict = jm_job.to_dict()
    nodes = crud.get_one_job_nodes(db,jm_job_id)
    cron_type, cron_select_type, cron_expression = jm_job_dict['cron_type'], jm_job_dict['cron_select_type'], jm_job_dict['cron_expression']
    cron_expression_dict = {}
    if cron_type == 2:
        cron_expression_dict = parsing_cron_expression(cron_expression)
        cron_expression_dict.update({
            'cron_select_type': cron_select_type,
            'cron_expression': cron_expression
        })

    jm_job_dict.update({
        'nodes': nodes,
        'cron_expression_dict': cron_expression_dict
    })
    return jm_job_dict

@router.put("/")
@web_try()
@sxtimeit
def update_jm_job_info(item: schemas.JmJobInfoUpdate, db: Session = Depends(get_db)):
    jm_job_info,nodes,edges = crud.update_jm_job_info(db, item)
    job_id = jm_job_info.id
    crud.delete_job_node(db, job_id)
    job_id = jm_job_info.id
    create_jm_job_node(db, nodes, edges, job_id)
    return jm_job_info.to_dict()

@router.delete("/")
@web_try()
@sxtimeit
def delete_jm_job_info(jm_job_id: int, db: Session = Depends(get_db)):
    return crud.delete_jm_job_info(db,jm_job_id)

@router.put("/status")
@web_try()
@sxtimeit
def update_jm_job_status(item: schemas.JmJobInfoStatusUpdate, db: Session = Depends(get_db)):
    return crud.update_jm_job_status(db,item)

@router.post("/execute/{jm_job_id}")
@web_try()
@sxtimeit
def execute_jm_job(jm_job_id: int, db: Session = Depends(get_db)):
    jm_job = crud.get_jm_job_info(db,jm_job_id)
    if jm_job.status == 0:
        raise Exception('任务已被停用')
    # 进行api调用
    return jm_job


@router.post("/cron_expression")
@web_try()
@sxtimeit
def get_cron_expression(cron_expression: schemas.CronExpression):
    print(cron_expression)
    cron = joint_cron_expression(cron_expression)
    return cron

@router.get("/cron_next_execute")
@web_try()
@sxtimeit
def get_cron_next_execute(cron_expression: str):
    execute_list = run_get_next_time(cron_expression)
    return execute_list


def run_get_next_time(cron_expression):
    now = datetime.datetime.now()
    cron_str = cron_expression.replace('?','*')
    cron = croniter.croniter(cron_str, now)
    execute_list = []
    for i in range(0, 5):
        next_time = cron.get_next(datetime.datetime).strftime("%Y-%m-%d %H:%M")
        execute_list.append(next_time)
    return execute_list