88 lines
2.9 KiB
Python
88 lines
2.9 KiB
Python
from fastapi import HTTPException
|
|
from sqlalchemy import or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from . import Models, Schemas
|
|
|
|
from Utils.UniqueCoder.TimeSerialNumUtils import create_time_serial_num
|
|
|
|
|
|
def get_indicator_by_iid(db: Session, iid: str):
|
|
return db.query(Models.Indicator).filter(Models.Indicator.iid == iid).first()
|
|
|
|
|
|
def get_indicator_by_ename(db: Session, ename: str):
|
|
return db.query(Models.Indicator).filter(Models.Indicator.ename == ename).first()
|
|
|
|
|
|
def get_indicator_description_by_iid(db: Session, iid: str):
|
|
return db.query(Models.Indicator).with_entities(Models.Indicator.description).filter(Models.Indicator.iid == iid).first()
|
|
|
|
|
|
def search_indicators(db: Session, body: Schemas.IndicatorSearch, page: int, pagesize: int):
|
|
items = db.query(Models.Indicator).filter(
|
|
or_(Models.Indicator.cname.like("%{}%".format("" if body.name is None else body.name)), body.name is None),
|
|
or_(Models.Indicator.nature == body.nature, body.nature is None),
|
|
or_(Models.Indicator.category == body.category, body.category is None)
|
|
).offset((page-1)*pagesize).limit(pagesize).all()
|
|
|
|
count = db.query(Models.Indicator).filter(
|
|
or_(Models.Indicator.cname.like("%{}%".format("" if body.name is None else body.name)), body.name is None),
|
|
or_(Models.Indicator.nature == body.nature, body.nature is None),
|
|
or_(Models.Indicator.category == body.category, body.category is None)
|
|
).count()
|
|
|
|
return items, count
|
|
|
|
|
|
def create_indicator(db: Session, body: Schemas.IndicatorCreate):
|
|
indicator = body.dict()
|
|
indicator.pop("parameters")
|
|
item = Models.Indicator(**indicator)
|
|
item.iid = create_time_serial_num(prefix="IID", suffix="")
|
|
db.add(item)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
def delete_indicator(db: Session, iid: str):
|
|
data = db.query(Models.Indicator).filter_by(iid=iid).first()
|
|
if not data:
|
|
raise HTTPException(status_code=400, detail="Indicator Not Existed")
|
|
|
|
db.query(Models.Indicator).filter_by(iid=iid).delete()
|
|
db.commit()
|
|
|
|
|
|
def edit_indicator(db: Session, iid: str, body: Schemas.IndicatorEdit):
|
|
db.query(Models.Indicator).filter_by(iid=iid).update(body.dict())
|
|
db.commit()
|
|
return db.query(Models.Indicator).filter(Models.Indicator.iid == iid).first()
|
|
|
|
|
|
def create_indicator_parameter(db: Session, body: Schemas.ParameterBase, _iid: str):
|
|
|
|
item = Models.Parameters(**body.dict())
|
|
item._iid = _iid
|
|
item.pid = create_time_serial_num(prefix="PID", suffix="")
|
|
|
|
db.add(item)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
def delete_indicator_parameter(db: Session, pid: str):
|
|
db.query(Models.Parameters).filter_by(pid=pid).delete()
|
|
db.commit()
|
|
|
|
|
|
def edit_indicator_parameter(db: Session, pid: str, body: Schemas.ParameterEdit):
|
|
db.query(Models.Parameters).filter_by(pid=pid).update(body.dict())
|
|
db.commit()
|
|
|
|
|
|
def get_parameter_by_pid(db: Session, pid: str):
|
|
return db.query(Models.Parameters).filter_by(pid=pid).first()
|