datax-admin/app/crud/jm_job_info.py

import time
from typing import List
from app import models, schemas
from sqlalchemy.orm import Session

def create_jm_job_info(db: Session, jm_job_info: models.JmJobInfo):
    db.add(jm_job_info)
    db.commit()
    db.refresh(jm_job_info)
    return jm_job_info

def get_jm_job_infos(db: Session):
    res: List[models.JmJobInfo] = db.query(models.JmJobInfo)\
        .filter(models.JmJobInfo.delete_status != 0)\
        .order_by(models.JmJobInfo.id.desc()).all()
    return res

def get_jm_job_info(db: Session, jm_job_id: int):
    item = db.query(models.JmJobInfo)\
        .filter(models.JmJobInfo.id == jm_job_id)\
        .filter(models.JmJobInfo.delete_status != 0).first()
    if not item:
        raise Exception('未找到该定时任务')
    return item

def update_jm_job_info(db: Session, item: models.JmJobInfo):
    db.commit()
    db.flush()
    db.refresh(item)
    return item

def delete_jm_job_info(db: Session, jm_job_id: int):
    jm_job_info = db.query(models.JmJobInfo)\
        .filter(models.JmJobInfo.id == jm_job_id).first()
    if not jm_job_info:
        raise Exception('未找到该定时任务')
    if jm_job_info.status == 1:
        raise Exception('该任务未停用，不能删除')
    jm_job_info.delete_status = 0
    db.commit()
    db.flush()
    db.refresh(jm_job_info)
    return jm_job_info

def update_jm_job_status(db: Session, jm_job_id: int, status: int):
    jm_job_info = db.query(models.JmJobInfo)\
        .filter(models.JmJobInfo.id == jm_job_id)\
        .filter(models.JmJobInfo.delete_status != 0).first()
    if not jm_job_info:
        raise Exception('未找到该定时任务')
    jm_job_info.status = status
    db.commit()
    db.flush()
    db.refresh(jm_job_info)
    return jm_job_info