datax-admin/app/crud/job_info.py

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


def create_job_info(db: Session, item: schemas.JobInfoCreate):
    create_time: int = int(time.time())
    db_item = models.JobInfo(**item.dict(), **{
        'trigger_status': 0,
        'create_time': create_time,
        'update_time': create_time,
        'user_id': 1, # TODO 用户
        'trigger_status': 0,
        'delete_status': 1,
    })
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item



def get_job_infos(db: Session, skip: int = 0, limit: int = 20):
    res: List[models.JobInfo] = db.query(models.JobInfo).filter(models.JobInfo.delete_status == 1).all()  # TODO: 排序
    return res



def update_job_info(db: Session, id: int, update_item: schemas.JobInfoUpdate):
    db_item = db.query(models.JobInfo).filter(models.JobInfo.id == id).first()
    if not db_item:
        raise Exception('未找到该任务')
    update_dict = update_item.dict(exclude_unset=True)
    for k, v in update_dict.items():
        setattr(db_item, k, v)
    db_item.update_time = int(time.time())
    db.commit()
    db.flush()
    db.refresh(db_item)
    return db_item

def delete_job_info(db: Session, job_id: int):
    job_item = db.query(models.JobInfo).filter(models.JobInfo.id == job_id).first()
    if not job_item:
        raise Exception('未找到该任务')
    job_item.delete_status = 0
    db.commit()
    db.flush()
    db.refresh(job_item)
    return job_item