datax-admin/app/routers/job.py

from fastapi import APIRouter, Depends
from fastapi_pagination import paginate, Params
from sqlalchemy.orm import Session

from app import schemas, get_db, crud
from app.crud import create_airflow_job_submit
from utils import web_try, sxtimeit

router_af_job = APIRouter(
    prefix="/jpt/af_job",
    tags=["airflow_job-任务管理"],
)


@router_af_job.get("/")
@web_try()
@sxtimeit
def get_af_jobs(params: Params = Depends(), db: Session = Depends(get_db)):
    return paginate(crud.get_airflow_jobs(db), params)


@router_af_job.post("/")
@web_try()
@sxtimeit
def add_af_job(item: schemas.AirflowJobCreate, db: Session = Depends(get_db)):
    return crud.create_airflow_job(db, item)


@router_af_job.put("/{item_id}")
@web_try()
@sxtimeit
def update_af_job(item_id: int, update_item: schemas.AirflowJobUpdate, db: Session = Depends(get_db)):
    return crud.update_airflow_job(db=db, item_id=item_id, update_item=update_item)


@router_af_job.post("/submit")
@web_try()
@sxtimeit
def add_dag_submit(id: int, db: Session = Depends(get_db)):
    item = crud.get_airflow_job_once(db, id)
    create_airflow_job_submit(schemas.AirflowJob(**item.to_dict()))
    # return crud.create_airflow_job(item)