datax-admin/app/crud/jm_job_node.py

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


def create_jm_job_node(db: Session, item: models.JmJobNode):
    db.add(item)
    db.commit()
    db.refresh(item)
    return item

def create_jm_job_edge(db: Session, item: models.JmJobEdge):
    db.add(item)
    db.commit()
    db.refresh(item)
    return item

def get_one_job_nodes(db: Session, job_id: int):
    res: List[models.JmJobNode] = db.query(models.JmJobNode)\
        .filter(models.JmJobNode.job_id == job_id).all()
    return res

def get_one_job_edges(db: Session, job_id: int):
    res: List[models.JmJobEdge] = db.query(models.JmJobEdge)\
        .filter(models.JmJobEdge.job_id == job_id).all()
    return res

def find_by_homework_and_job(db: Session, job_ids: List[int], homework_id: int):
    res: List[models.JmJobNode] = db.query(models.JmJobNode)\
                .filter(models.JmJobNode.homework_id == homework_id)\
                .filter(models.JmJobNode.job_id.in_(job_ids)).all()
    return res

def delete_job_node(db: Session, job_id: int):
    db.query(models.JmJobNode).filter(models.JmJobNode.job_id == job_id).delete()
    db.query(models.JmJobEdge).filter(models.JmJobEdge.job_id == job_id).delete()


def find_homework_by_job(db: Session, job_id: int):
    res: List[models.JmJobNode] = db.query(models.JmJobNode)\
                .filter(models.JmJobNode.job_id == job_id)\
                .filter(models.JmJobNode.start_point == 1).all()
    return res