datax-admin/app/crud/af_task.py

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


def create_airflow_task(db: Session, item: schemas.AirflowTaskCreate):
    db_item = models.AirflowTask(**item.dict(), **{"create_time": int(time.time()), "update_time": int(time.time())})
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item


def update_airflow_task(db: Session, item_id: int, update_item: schemas.AirflowTaskUpdate):
    return update_to_db(update_item=update_item, item_id=item_id, db=db, model_cls=models.AirflowTask)


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


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