datax-admin/app/crud/constant.py

import time
from typing import List
from app import models, schemas
from sqlalchemy.orm import Session
from configs.settings import DefaultOption, config
python_image = config.get('TASK_IMAGES', 'python')
java_image = config.get('TASK_IMAGES', 'java')


def create_constant(db: Session,item: schemas.ConstantCreate):
    db_item = models.Constant(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

def get_constant_list(db: Session, type: str):
    if type == 'pythonImage':
        return [python_image]
    elif type == 'javaImage':
        return [java_image]
    else:
        res: List[models.Constant] = db.query(models.Constant).filter(models.Constant.type == type).all()
        return [r.value for r in res]

def find_and_update(db: Session, type: str, value: str):
    res: List[models.Constant] = db.query(models.Constant)\
        .filter(models.Constant.type == type)\
        .filter(models.Constant.value == value).all()
    if len(res) == 0:
        db_item = models.Constant(**{
            'type': type,
            'value': value
        })
        db.add(db_item)
        db.commit()
        db.refresh(db_item)

def delete_constant(db: Session, type: str, value: str):
    db_item = db.query(models.Constant)\
        .filter(models.Constant.type == type)\
        .filter(models.Constant.value == value).first()
    if not db_item:
        raise Exception('未找到该常量')
    else:
        db.query(models.Constant)\
        .filter(models.Constant.type == type)\
        .filter(models.Constant.value == value).delete()
    db.commit()
    db.flush()
    return db_item