datax-admin/app/crud/job_log.py

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

def create_job_log(db: Session, item: schemas.JobLogCreate):
    db_item = models.JobLog(**item.dict(), **{"trigger_code": 2, "handle_code": 2})
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

def get_job_logs(db: Session):
    res: List[models.JobLog] = db.query(models.JobLog).all()
    return res

def get_job_log_once(db: Session, id: int):
    res: models.JobLog = db.query(models.JobLog).filter(models.JobLog.id == id).first()
    return res