datax-admin/app/crud/constant.py

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


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):
    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