109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
|
from sqlalchemy import or_
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
|
||
|
|
from Utils.DataBase.IndexStore.SqlAlchemyUtils import set_next_id
|
||
|
|
from IndexStore.AppIndicators import Models, Schemas
|
||
|
|
|
||
|
|
from Utils.UniqueCoder.TimeSerialNumUtils import create_time_serial_num
|
||
|
|
from .Models import IndicatorCategory
|
||
|
|
|
||
|
|
|
||
|
|
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 search_indicators(db: Session, schema: Schemas.SearchIndicatorReqBody):
|
||
|
|
|
||
|
|
page = 1 if schema.page < 1 else (100 if schema.page > 100 else schema.page)
|
||
|
|
pagesize = 5 if schema.pagesize < 5 else (20 if schema.pagesize > 20 else schema.pagesize)
|
||
|
|
|
||
|
|
items = db.query(Models.Indicator).filter(
|
||
|
|
or_(Models.Indicator.cname.like("%{}%".format("" if schema.name is None else schema.name)), schema.name is None),
|
||
|
|
or_(Models.Indicator.nature == schema.nature, schema.nature is None),
|
||
|
|
or_(Models.Indicator.category_id == schema.category_id, schema.category_id is None)
|
||
|
|
).offset((page-1)*pagesize).limit(pagesize).all()
|
||
|
|
|
||
|
|
count = db.query(Models.Indicator).filter(
|
||
|
|
or_(Models.Indicator.cname.like("%{}%".format("" if schema.name is None else schema.name)), schema.name is None),
|
||
|
|
or_(Models.Indicator.nature == schema.nature, schema.nature is None),
|
||
|
|
or_(Models.Indicator.category_id == schema.category_id, schema.category_id is None)
|
||
|
|
).count()
|
||
|
|
|
||
|
|
return items, count
|
||
|
|
|
||
|
|
|
||
|
|
def create_indicator(db: Session, schema: Schemas.CreateIndicatorReqBody):
|
||
|
|
item = Models.Indicator(**schema.to_dict())
|
||
|
|
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):
|
||
|
|
db.query(Models.Indicator).filter_by(iid=iid).delete()
|
||
|
|
db.commit()
|
||
|
|
return True
|
||
|
|
|
||
|
|
|
||
|
|
def edit_indicator(db: Session, iid: str, schema: Schemas.EditIndicatorReqBody):
|
||
|
|
db.query(Models.Indicator).filter_by(iid=iid).update(schema.del_null_value())
|
||
|
|
db.commit()
|
||
|
|
return True
|
||
|
|
|
||
|
|
|
||
|
|
def create_parameter(db: Session, schema: Schemas.EditParamReqBody, _iid: str):
|
||
|
|
item = Models.Parameters(**schema.dict())
|
||
|
|
item._iid = _iid
|
||
|
|
item.pid = create_time_serial_num(prefix="PID", suffix="")
|
||
|
|
db.add(item)
|
||
|
|
db.commit()
|
||
|
|
return True
|
||
|
|
|
||
|
|
|
||
|
|
def delete_parameter(db: Session, pid: str):
|
||
|
|
db.query(Models.Parameters).filter_by(pid=pid).delete()
|
||
|
|
db.commit()
|
||
|
|
|
||
|
|
|
||
|
|
def edit_parameter(db: Session, pid: str, schema: Schemas.EditParamReqBody):
|
||
|
|
db.query(Models.Parameters).filter_by(pid=pid).update(schema.dict())
|
||
|
|
db.commit()
|
||
|
|
|
||
|
|
|
||
|
|
def get_parameter_by_pid(db: Session, pid: str):
|
||
|
|
return db.query(Models.Parameters).filter_by(pid=pid).first()
|
||
|
|
|
||
|
|
|
||
|
|
def create_category(db: Session, schema: Schemas.CreateCategoryReqBody):
|
||
|
|
item = IndicatorCategory(**schema.dict())
|
||
|
|
item.id = set_next_id(db=db, num_len=3, prefix="CATEGORY", model=IndicatorCategory)
|
||
|
|
db.add(item)
|
||
|
|
db.commit()
|
||
|
|
return True
|
||
|
|
|
||
|
|
|
||
|
|
def get_category_by_name(db: Session, title: str):
|
||
|
|
item = db.query(Models.IndicatorCategory).filter_by(title=title).first()
|
||
|
|
return item
|
||
|
|
|
||
|
|
|
||
|
|
def get_category_by_id(db: Session, category_id: str):
|
||
|
|
item = db.query(Models.IndicatorCategory).filter_by(id=category_id).first()
|
||
|
|
return item
|
||
|
|
|
||
|
|
|
||
|
|
def delete_category(db: Session, category_id: str):
|
||
|
|
db.query(Models.IndicatorCategory).filter_by(id=category_id).delete()
|
||
|
|
db.commit()
|
||
|
|
return True
|
||
|
|
|
||
|
|
|
||
|
|
def list_category(db: Session):
|
||
|
|
return db.query(Models.IndicatorCategory).order_by(Models.IndicatorCategory.id.asc()).all()
|