datax-admin/app/crud/jm_homework.py

import time
from typing import List
from app import models, schemas
from sqlalchemy.orm import Session
from sqlalchemy import func
from app.crud.constant import find_and_update
from configs.globals import g

from app.crud.jm_homework_datasource_relation import get_jm_relations


def create_jm_homework(db: Session, item: models.JmHomework):
    db.add(item)
    db.commit()
    db.refresh(item)
    return item

def get_jm_homeworks(db: Session, project_id: str):
    res: List[models.JmHomework] = db.query(models.JmHomework)\
            .filter(models.JmHomework.project_id == g.project_id)\
            .filter(models.JmHomework.status == 1)\
            .order_by(models.JmHomework.create_time.desc()).all()
    return res

def get_jm_homework_info(db: Session, homework_id: int):
    item  = db.query(models.JmHomework)\
            .filter(models.JmHomework.id == homework_id).first()
    if item.type == 'Dag':
        relations = get_jm_relations(db,homework_id)
        item.__dict__.update({"hd_relation":relations})
    return item


def get_jm_homework_by_name(db: Session, name: str):
    db_item = db.query(models.JmHomework).filter(models.JmHomework.name == func.binary(name))\
        .filter(models.JmHomework.project_id == g.project_id)\
        .filter(models.JmHomework.status != 0).first()
    return db_item


def update_jm_homework(db: Session, id: int, update_item: models.JmHomework):
    db.commit()
    db.flush()
    db.refresh(update_item)
    return update_item

def delete_jm_homework(db: Session, id: int):
    db_item = db.query(models.JmHomework).filter(models.JmHomework.id == id).first()
    if not db_item:
        raise Exception('未找到该作业')
    db_item.status = 0
    db.commit()
    db.flush()
    db.refresh(db_item)
    return db_item

def get_jm_homeworks_by_ids(db: Session, ids: List[int]):
    res: List[models.JmHomework] = db.query(models.JmHomework)\
            .filter(models.JmHomework.id.in_(ids)).all()
    return res

def get_jm_homework_by_dag_url(db: Session, dag_url: str):
    res: List[models.JmHomework] = db.query(models.JmHomework)\
        .filter(models.JmHomework.dag_url == dag_url)\
        .filter(models.JmHomework.status == 1).all()
    return res