datax-admin/app/crud/jm_homework.py

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

from app.crud.jm_homework_datasource_relation import create_jm_hd_relation, get_jm_relations


def create_jm_homework(db: Session, item: schemas.JmHomeworkCreate):
    jm_homework_create = item.dict()
    db_item = db.query(models.JmHomework).filter(models.JmHomework.name == jm_homework_create['name'])\
        .filter(models.JmHomework.status != 0).first()
    if db_item:
        raise Exception('JmHomework name already exists')
    relation_list = jm_homework_create.pop('relation_list')
    if jm_homework_create['type'] != 'DAG':
        for relation in relation_list:
            create_jm_hd_relation(db, schemas.JmHomeworkDatasourceRelationCreate(**relation))
    tag = jm_homework_create['tag']
    find_and_update(db, '作业分类', tag)
    create_time: int = int(time.time())
    db_item = models.JmHomework(**jm_homework_create,**{
        'create_time': create_time,
        'update_time': create_time,
        'status': 1
    })
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

def get_jm_homeworks(db: Session, project_id: str):
    res: List[models.JmHomework] = db.query(models.JmHomework)\
            .filter(models.JmHomework.project_id == project_id)\
            .filter(models.JmHomework.status == 1).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 update_jm_homework(db: Session, id: int, update_item: schemas.JmHomeworkUpdate):
    db_item = db.query(models.JmHomework).filter(models.JmHomework.id == id).first()
    if not db_item:
        raise Exception('JmHomework not found')
    db_name_item = db.query(models.JmHomework).filter(models.JmHomework.name == update_item.name)\
        .filter(models.JmHomework.status != 0).first()
    if db_name_item:
        raise Exception('JmHomework name already exists')
    update_dict = update_item.dict(exclude_unset=True)
    tag = update_dict.tag
    find_and_update(db, '作业分类', tag)
    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_jm_homework(db: Session, id: int):
    db_item = db.query(models.JmHomework).filter(models.JmHomework.id == id).first()
    if not db_item:
        raise Exception('JmHomework not found')
    db_item.status = 0
    db.commit()
    db.flush()
    db.refresh(db_item)
    return db_item