datax-admin/app/crud/relation.py

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

def create_relation(db: Session, se_id: int, type: str, af_id: int):
    db_item = models.Relation(**{"se_id": se_id,
                                 "type": type,
                                 "af_id": af_id})
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

def create_debug_relation(db: Session, dag_uuid: str, type: str, af_id: int):
    db_item = models.Relation(**{"dag_uuid": dag_uuid,
                                 "type": type,
                                 "af_id": af_id})
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

def get_af_id(db: Session, se_id: int, type: str):
    res: models.Relation = db.query(models.Relation)\
        .filter(models.Relation.se_id == se_id)\
        .filter(models.Relation.type == type).first()
    return res

def get_af_ids(db: Session, se_ids: List[int], type: str):
    res: List[models.Relation] = db.query(models.Relation)\
        .filter(models.Relation.se_id.in_(se_ids))\
        .filter(models.Relation.type == type).all()
    return res

def get_dag_af_id(db: Session, dag_uuid: int, type: str):
    res: models.Relation = db.query(models.Relation)\
        .filter(models.Relation.dag_uuid == dag_uuid)\
        .filter(models.Relation.type == type).first()
    return res

def delete_relation(db: Session, se_id: int, type: str):
    res: models.Relation = db.query(models.Relation)\
        .filter(models.Relation.se_id == se_id)\
        .filter(models.Relation.type == type).delete()
    return res