indexstore/AppIndicators/Crud.py

from sqlalchemy import or_
from sqlalchemy.orm import Session

from Utils.DataBase.SqlAlchemyUtils import set_next_id
from 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()