datax-admin/app/routers/jm_job_log.py

from typing import Optional
from fastapi import APIRouter

from fastapi import Depends
from sqlalchemy.orm import Session
from app import schemas

import app.crud as crud
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_log",
    tags=["jm_job_log-定时任务日志管理"],
)


@router.get("/")
@web_try()
@sxtimeit
def get_job_logs(job_id: int = None, db: Session = Depends(get_db)):
    jm_job_list = []
    if job_id is not None:
        jm_job_list = [crud.get_jm_job_info(db, job_id)]
    else:
        jm_job_list = crud.get_jm_job_infos(db)
    jm_job_list = crud.get_jm_job_infos(db)
    id_to_job = {job.id:job for job in jm_job_list}
    relations = crud.get_af_ids(db,id_to_job.keys(), '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,
                "job_name": id_to_job[job_id].name,
                "job_type": id_to_job[job_id].type,
                "job_tag": id_to_job[job_id].tag,
                "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,
            }
            res.append(log)
    return res

@router.get("/logs")
@web_try()
@sxtimeit
def get_job_logs(run_id: str, job_id: int, db: Session = Depends(get_db)):
    af_job_run = crud.get_airflow_run_once(db, run_id, job_id)
    tasks = list(af_job_run.details['tasks'].values()) if len(list(af_job_run.details['tasks'].values()))>0 else []
    res = []
    for task in tasks:
        log = {
            "id": af_job_run.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,
            "log": task['log'] if task else None
        }
        res.append(log)
    return res