changes
This commit is contained in:
parent
33035cd827
commit
7e2df42fec
|
|
@ -2,9 +2,11 @@ from fastapi import HTTPException
|
|||
from sqlalchemy import or_
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from Utils.DataBase.SqlAlchemyUtils import set_next_id
|
||||
from . 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):
|
||||
|
|
@ -15,30 +17,28 @@ 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, 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)
|
||||
|
||||
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)
|
||||
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 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)
|
||||
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, body: Schemas.IndicatorCreate):
|
||||
indicator = body.dict()
|
||||
indicator.pop("parameters")
|
||||
item = Models.Indicator(**indicator)
|
||||
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()
|
||||
|
|
@ -47,41 +47,63 @@ def create_indicator(db: Session, body: Schemas.IndicatorCreate):
|
|||
|
||||
|
||||
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()
|
||||
return True
|
||||
|
||||
|
||||
def edit_indicator(db: Session, iid: str, body: Schemas.IndicatorEdit):
|
||||
db.query(Models.Indicator).filter_by(iid=iid).update(body.dict())
|
||||
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 db.query(Models.Indicator).filter(Models.Indicator.iid == iid).first()
|
||||
return True
|
||||
|
||||
|
||||
def create_indicator_parameter(db: Session, body: Schemas.ParameterBase, _iid: str):
|
||||
|
||||
item = Models.Parameters(**body.dict())
|
||||
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()
|
||||
db.refresh(item)
|
||||
return item
|
||||
return True
|
||||
|
||||
|
||||
def delete_indicator_parameter(db: Session, pid: str):
|
||||
def delete_parameter(db: Session, pid: str):
|
||||
db.query(Models.Parameters).filter_by(pid=pid).delete()
|
||||
db.commit()
|
||||
|
||||
|
||||
def edit_indicator_parameter(db: Session, pid: str, schema: Schemas.ParameterEdit):
|
||||
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()
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ from sqlalchemy import Column, String, Enum, ForeignKey
|
|||
from sqlalchemy.orm import relationship
|
||||
from sqlalchemy.dialects.mysql import LONGTEXT
|
||||
|
||||
from .Schemas import NatureEnum, CategoryEnum
|
||||
from .Schemas import NatureEnum
|
||||
|
||||
from Utils.DataBase.SqlAlchemyUtils import Base
|
||||
|
||||
|
|
@ -10,23 +10,48 @@ from Utils.DataBase.SqlAlchemyUtils import Base
|
|||
class Indicator(Base):
|
||||
__tablename__ = "indicator"
|
||||
|
||||
iid = Column(String(32), primary_key=True, index=True)
|
||||
iid = Column(String(32), primary_key=True)
|
||||
ename = Column(String(255), index=True)
|
||||
cname = Column(String(255), index=True)
|
||||
description = Column(LONGTEXT)
|
||||
nature = Column(Enum(NatureEnum, values_callable=lambda obj: [e.value for e in obj]))
|
||||
category = Column(Enum(CategoryEnum, values_callable=lambda obj: [e.value for e in obj]))
|
||||
category_id = Column(String(12), ForeignKey("indicator_category.id"))
|
||||
|
||||
parameters = relationship("Parameters", back_populates="indicator", cascade="all, delete-orphan")
|
||||
category = relationship("IndicatorCategory", backref="indicator")
|
||||
parameters = relationship("Parameters", backref="indicator", cascade="all, delete-orphan")
|
||||
|
||||
def to_dict(self):
|
||||
_dict = {c.name: getattr(self, c.name, None) for c in self.__table__.columns}
|
||||
_dict.update({"category": self.category.__str__()})
|
||||
_dict.update({"parameters": [param.to_dict() for param in self.parameters]})
|
||||
return _dict
|
||||
|
||||
|
||||
class Parameters(Base):
|
||||
__tablename__ = "parameters"
|
||||
|
||||
pid = Column(String(32), primary_key=True, index=True)
|
||||
pid = Column(String(32), primary_key=True)
|
||||
ename = Column(String(255))
|
||||
cname = Column(String(255))
|
||||
description = Column(LONGTEXT)
|
||||
_iid = Column(String(32), ForeignKey("indicator.iid", ondelete="CASCADE"))
|
||||
|
||||
indicator = relationship("Indicator", back_populates="parameters")
|
||||
def to_dict(self):
|
||||
_dict = {c.name: getattr(self, c.name, None) for c in self.__table__.columns}
|
||||
return _dict
|
||||
|
||||
|
||||
class IndicatorCategory(Base):
|
||||
__tablename__ = "indicator_category"
|
||||
|
||||
id = Column(String(12), primary_key=True)
|
||||
title = Column(String(16), unique=True, index=True)
|
||||
|
||||
def __repr__(self):
|
||||
return self.title
|
||||
|
||||
def to_kv(self):
|
||||
_dict = dict()
|
||||
_dict.update({"key": self.title})
|
||||
_dict.update({"value": self.id})
|
||||
return _dict
|
||||
|
|
|
|||
|
|
@ -0,0 +1,157 @@
|
|||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
from . import Crud, Models, Schemas
|
||||
|
||||
from Utils.AccessControl.AccessUtil import rbac
|
||||
from Utils.DataBase.SqlAlchemyUtils import get_db, engine
|
||||
|
||||
Models.Base.metadata.create_all(bind=engine)
|
||||
|
||||
router = APIRouter(
|
||||
prefix="/api/index_store",
|
||||
dependencies=[Depends(rbac)]
|
||||
)
|
||||
|
||||
|
||||
# 新建指标
|
||||
@router.post("/create", summary='新建指标', tags=["指标"], response_model=Schemas.CreateIndicatorResBody)
|
||||
def func01(schema: Schemas.CreateIndicatorReqBody, db: Session = Depends(get_db)):
|
||||
index_obj = Crud.get_indicator_by_ename(db, ename=schema.ename)
|
||||
if index_obj:
|
||||
raise HTTPException(status_code=202, detail="Indicator Already Registered")
|
||||
|
||||
category_obj = Crud.get_category_by_id(db=db, category_id=schema.category_id)
|
||||
if not category_obj:
|
||||
raise HTTPException(status_code=202, detail="Category Not Found")
|
||||
|
||||
index_obj = Crud.create_indicator(db=db, schema=schema)
|
||||
|
||||
parameters_item = schema.parameters
|
||||
for parameter in parameters_item:
|
||||
Crud.create_parameter(db=db, schema=parameter, _iid=index_obj.iid)
|
||||
|
||||
return {"iid": index_obj.iid}
|
||||
|
||||
|
||||
# 删除指标
|
||||
@router.post("/delete/{iid}", summary='删除指标', tags=["指标"])
|
||||
def func02(iid: str, db: Session = Depends(get_db)):
|
||||
index_obj = Crud.get_indicator_by_iid(db=db, iid=iid)
|
||||
if not index_obj:
|
||||
raise HTTPException(status_code=400, detail="Indicator Not Existed")
|
||||
|
||||
Crud.delete_indicator(db=db, iid=iid)
|
||||
|
||||
return {"info": "Success"}
|
||||
|
||||
|
||||
# 编辑指标
|
||||
@router.post("/edit/{iid}", summary='编辑指标', tags=["指标"])
|
||||
def func03(iid: str, schema: Schemas.EditIndicatorReqBody, db: Session = Depends(get_db)):
|
||||
index_obj = Crud.get_indicator_by_iid(db=db, iid=iid)
|
||||
if not index_obj:
|
||||
raise HTTPException(status_code=202, detail="Indicator Not Found")
|
||||
|
||||
category_obj = Crud.get_category_by_id(db=db, category_id=schema.category_id)
|
||||
if not category_obj:
|
||||
raise HTTPException(status_code=202, detail="Category Not Found")
|
||||
|
||||
Crud.edit_indicator(db=db, schema=schema, iid=iid)
|
||||
|
||||
return {"info": "Success"}
|
||||
|
||||
|
||||
# 查看指标
|
||||
@router.get("/view/{iid}", summary='查看指标', tags=["指标"], response_model=Schemas.IndicatorFullResBody)
|
||||
def func04(iid: str, db: Session = Depends(get_db)):
|
||||
index_obj = Crud.get_indicator_by_iid(db, iid=iid)
|
||||
if index_obj is None:
|
||||
raise HTTPException(status_code=202, detail="Indicator Not Found")
|
||||
|
||||
return index_obj.to_dict()
|
||||
|
||||
|
||||
# 查询指标描述
|
||||
@router.post("/describe/{iid}", summary='查看指标描述', tags=["指标"], response_model=Schemas.IndicatorDescribeResBody)
|
||||
def func05(iid: str, db: Session = Depends(get_db)):
|
||||
index_obj = Crud.get_indicator_by_iid(db, iid=iid)
|
||||
if not index_obj:
|
||||
raise HTTPException(status_code=400, detail="Indicator Not Found")
|
||||
|
||||
return index_obj.to_dict()
|
||||
|
||||
|
||||
# 查询指标
|
||||
@router.post("/search", summary='查询指标', tags=["指标"])
|
||||
def func06(schema: Schemas.SearchIndicatorReqBody, db: Session = Depends(get_db)):
|
||||
indicators, total = Crud.search_indicators(db, schema=schema)
|
||||
|
||||
return {"items": indicators, "total": total}
|
||||
|
||||
|
||||
# 新建参数
|
||||
@router.post("/param/create/{iid}", summary='新建参数', tags=["指标参数"])
|
||||
def func07(iid: str, body: Schemas.EditParamReqBody, db: Session = Depends(get_db)):
|
||||
index_obj = Crud.get_indicator_by_iid(db=db, iid=iid)
|
||||
if not index_obj:
|
||||
raise HTTPException(status_code=400, detail="Indicator Not Found")
|
||||
|
||||
Crud.create_parameter(db=db, schema=body, _iid=iid)
|
||||
|
||||
return {"info": "Success"}
|
||||
|
||||
|
||||
@router.post("/param/edit/{pid}", summary='编辑参数', tags=["指标参数"])
|
||||
def func08(pid: str, schema: Schemas.EditParamReqBody, db: Session = Depends(get_db)):
|
||||
data = Crud.get_parameter_by_pid(db=db, pid=pid)
|
||||
if not data:
|
||||
raise HTTPException(status_code=202, detail="Parameter Not Existed")
|
||||
|
||||
Crud.edit_parameter(db=db, pid=pid, schema=schema)
|
||||
|
||||
return {"info": "Success"}
|
||||
|
||||
|
||||
# 删除参数
|
||||
@router.post("/param/delete/{pid}", summary='删除参数', tags=["指标参数"])
|
||||
def func09(pid: str, db: Session = Depends(get_db)):
|
||||
data = Crud.get_parameter_by_pid(db=db, pid=pid)
|
||||
if not data:
|
||||
raise HTTPException(status_code=202, detail="Parameter Not Existed")
|
||||
|
||||
Crud.delete_parameter(db=db, pid=pid)
|
||||
|
||||
return {"info": "Success"}
|
||||
|
||||
|
||||
@router.post("/category/create", summary="新建分类", tags=["指标分类"])
|
||||
def func10(schema: Schemas.CreateCategoryReqBody, db: Session = Depends(get_db)):
|
||||
data = Crud.get_category_by_name(db=db, title=schema.title)
|
||||
if data:
|
||||
raise HTTPException(status_code=202, detail="Category Already Existed")
|
||||
|
||||
Crud.create_category(db=db, schema=schema)
|
||||
|
||||
return {"info": "Success"}
|
||||
|
||||
|
||||
@router.post("/category/delete", summary="删除分类", tags=["指标分类"])
|
||||
def func10(schema: Schemas.DeleteCategoryReqBody, db: Session = Depends(get_db)):
|
||||
data = Crud.get_category_by_id(db=db, category_id=schema.category_id)
|
||||
if not data:
|
||||
raise HTTPException(status_code=202, detail="Category Not Existed")
|
||||
|
||||
try:
|
||||
Crud.delete_category(db=db, category_id=schema.category_id)
|
||||
except IntegrityError:
|
||||
raise HTTPException(status_code=400, detail="Cannot Delete Category")
|
||||
|
||||
return {"info": "Success"}
|
||||
|
||||
|
||||
@router.post("/category/list", summary="分类列表", tags=["指标分类"])
|
||||
def func11(db: Session = Depends(get_db)):
|
||||
category_objs = Crud.list_category(db=db)
|
||||
return [obj.to_kv() for obj in category_objs]
|
||||
|
|
@ -1,99 +0,0 @@
|
|||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
|
||||
from . import Crud, Models, Schemas
|
||||
|
||||
from Utils.AccessControl.AccessUtil import rbac
|
||||
from Utils.DataBase.SqlAlchemyUtils import get_db, engine
|
||||
|
||||
Models.Base.metadata.create_all(bind=engine)
|
||||
|
||||
router = APIRouter(
|
||||
prefix="/api/index_store",
|
||||
tags=['指标'],
|
||||
dependencies=[Depends(rbac)]
|
||||
)
|
||||
|
||||
|
||||
# 新建指标
|
||||
@router.post("/create", summary='新建指标', response_model=Schemas.Indicator)
|
||||
def func01(body: Schemas.IndicatorCreate, db: Session = Depends(get_db)):
|
||||
data = Crud.get_indicator_by_ename(db, ename=body.ename)
|
||||
|
||||
if data:
|
||||
raise HTTPException(status_code=202, detail="Indicator Already Registered")
|
||||
|
||||
indicator_item = Crud.create_indicator(db=db, body=body)
|
||||
|
||||
parameters_item = body.parameters
|
||||
for parameter in parameters_item:
|
||||
Crud.create_indicator_parameter(db=db, body=parameter, _iid=indicator_item.iid)
|
||||
|
||||
return Crud.get_indicator_by_iid(db=db, iid=indicator_item.iid)
|
||||
|
||||
|
||||
# 删除指标
|
||||
@router.post("/delete/{iid}", summary='删除指标')
|
||||
def func02(iid: str, db: Session = Depends(get_db)):
|
||||
Crud.delete_indicator(db=db, iid=iid)
|
||||
return {"info": "Success"}
|
||||
|
||||
|
||||
# 编辑指标
|
||||
@router.post("/edit/{iid}", summary='编辑指标', response_model=Schemas.Indicator)
|
||||
def func03(iid: str, body: Schemas.IndicatorEdit, db: Session = Depends(get_db)):
|
||||
db_indicator = Crud.get_indicator_by_iid(db=db, iid=iid)
|
||||
if not db_indicator:
|
||||
raise HTTPException(status_code=202, detail="Indicator Not Found")
|
||||
return Crud.edit_indicator(db=db, body=body, iid=iid)
|
||||
|
||||
|
||||
# 查看指标
|
||||
@router.get("/view/{iid}", summary='查看指标', response_model=Schemas.Indicator)
|
||||
def func04(iid: str, db: Session = Depends(get_db)):
|
||||
data = Crud.get_indicator_by_iid(db, iid=iid)
|
||||
if data is None:
|
||||
raise HTTPException(status_code=202, detail="Indicator Not Found")
|
||||
return data
|
||||
|
||||
|
||||
# 查询指标描述
|
||||
@router.post("/describe/{iid}", summary='查看指标描述')
|
||||
def func05(iid: str, db: Session = Depends(get_db)):
|
||||
indicators = Crud.get_indicator_description_by_iid(db, iid=iid)
|
||||
return indicators
|
||||
|
||||
|
||||
# 查询指标
|
||||
@router.post("/search", summary='查询指标')
|
||||
def func06(body: Schemas.IndicatorSearch, db: Session = Depends(get_db)):
|
||||
indicators, total = Crud.search_indicators(db, body=body, page=body.page, pagesize=body.pagesize)
|
||||
return {"items": indicators, "total": total}
|
||||
|
||||
|
||||
# 新建参数
|
||||
@router.post("/param/create/{iid}", summary='新建参数', response_model=Schemas.Parameter)
|
||||
def func07(iid: str, body: Schemas.ParameterEdit, db: Session = Depends(get_db)):
|
||||
return Crud.create_indicator_parameter(db=db, body=body, _iid=iid)
|
||||
|
||||
|
||||
@router.post("/param/edit/{pid}", summary='编辑参数')
|
||||
def func08(pid: str, schema: Schemas.ParameterEdit, db: Session = Depends(get_db)):
|
||||
data = Crud.get_parameter_by_pid(db=db, pid=pid)
|
||||
if not data:
|
||||
raise HTTPException(status_code=202, detail="Parameter Not Existed")
|
||||
|
||||
Crud.edit_indicator_parameter(db=db, pid=pid, schema=schema)
|
||||
return {"info": "Success"}
|
||||
|
||||
|
||||
# 删除参数
|
||||
@router.post("/param/delete/{pid}", summary='删除参数')
|
||||
def func09(pid: str, db: Session = Depends(get_db)):
|
||||
data = Crud.get_parameter_by_pid(db=db, pid=pid)
|
||||
if not data:
|
||||
raise HTTPException(status_code=202, detail="Parameter Not Existed")
|
||||
|
||||
Crud.delete_indicator_parameter(db=db, pid=pid)
|
||||
return {"info": "Success"}
|
||||
|
|
@ -5,7 +5,7 @@ from typing import List, Optional
|
|||
|
||||
# Regex
|
||||
ENameRegex = constr(regex="^[a-z0-9_]{1,}$")
|
||||
CNameRegex = constr(regex="^[\u4e00-\u9fa5a-zA-Z0-9（）]+$")
|
||||
CNameRegex = constr(regex=r"^[\u4e00-\u9fa5a-zA-Z0-9（）\/]+$")
|
||||
|
||||
|
||||
# Enums
|
||||
|
|
@ -14,84 +14,81 @@ class NatureEnum(Enum):
|
|||
enum02 = "定性"
|
||||
|
||||
|
||||
class CategoryEnum(Enum):
|
||||
enum01 = "盈利能力"
|
||||
enum02 = "收益质量"
|
||||
enum03 = "现金流量"
|
||||
enum04 = "资本结构"
|
||||
enum05 = "偿债能力"
|
||||
enum06 = "运营能力"
|
||||
enum07 = "成长能力"
|
||||
enum08 = "经营指标"
|
||||
enum09 = "资质指标"
|
||||
enum10 = "行业指标"
|
||||
enum11 = "绿色指标"
|
||||
enum12 = "司法指标"
|
||||
enum13 = "合规指标"
|
||||
enum14 = "舆情指标"
|
||||
enum15 = "其他"
|
||||
|
||||
|
||||
# Parameters
|
||||
class ParameterBase(BaseModel):
|
||||
class EditParamReqBody(BaseModel):
|
||||
ename: ENameRegex = "example_param"
|
||||
cname: CNameRegex = "示例参数"
|
||||
description: str = "参数介绍文字"
|
||||
|
||||
|
||||
class ParameterEdit(ParameterBase):
|
||||
pass
|
||||
|
||||
|
||||
class Parameter(ParameterBase):
|
||||
class ParameterResBody(BaseModel):
|
||||
pid: str
|
||||
_iid: str
|
||||
|
||||
class Config:
|
||||
orm_mode = True
|
||||
ename: ENameRegex = "example_param"
|
||||
cname: CNameRegex = "示例参数"
|
||||
description: str = "参数介绍文字"
|
||||
|
||||
|
||||
# Indicators
|
||||
class IndicatorBase(BaseModel):
|
||||
class CreateCategoryReqBody(BaseModel):
|
||||
title: str
|
||||
|
||||
|
||||
class DeleteCategoryReqBody(BaseModel):
|
||||
category_id: str
|
||||
|
||||
|
||||
class CreateIndicatorReqBody(BaseModel):
|
||||
ename: ENameRegex = "example_indicator"
|
||||
cname: CNameRegex = "示例指标"
|
||||
description: str = "指标介绍文字"
|
||||
nature: NatureEnum
|
||||
category: CategoryEnum
|
||||
category_id: str
|
||||
parameters: List[EditParamReqBody]
|
||||
|
||||
class Config:
|
||||
use_enum_values = True
|
||||
def to_dict(self):
|
||||
_dict = self.dict().copy()
|
||||
_dict.pop("parameters")
|
||||
return _dict
|
||||
|
||||
|
||||
class IndicatorCreate(IndicatorBase):
|
||||
|
||||
parameters: List[ParameterEdit]
|
||||
|
||||
|
||||
class Indicator(IndicatorBase):
|
||||
class CreateIndicatorResBody(BaseModel):
|
||||
iid: str
|
||||
parameters: List[Parameter] = []
|
||||
|
||||
class Config:
|
||||
orm_mode = True
|
||||
|
||||
|
||||
class IndicatorEdit(BaseModel):
|
||||
class EditIndicatorReqBody(BaseModel):
|
||||
cname: Optional[CNameRegex] = None
|
||||
description: Optional[str] = None
|
||||
nature: Optional[NatureEnum] = None
|
||||
category: Optional[CategoryEnum] = None
|
||||
category_id: Optional[str] = None
|
||||
|
||||
class Config:
|
||||
use_enum_values = True
|
||||
|
||||
def del_null_value(self):
|
||||
_dict = self.dict()
|
||||
for key in list(_dict.keys()):
|
||||
if not _dict.get(key):
|
||||
del _dict[key]
|
||||
return _dict
|
||||
|
||||
class IndicatorSearch(BaseModel):
|
||||
|
||||
class IndicatorFullResBody(BaseModel):
|
||||
iid: str
|
||||
ename: ENameRegex = "example_indicator"
|
||||
cname: CNameRegex = "示例指标"
|
||||
description: str = "指标介绍文字"
|
||||
nature: NatureEnum
|
||||
category: str
|
||||
category_id: str
|
||||
parameters: List[ParameterResBody] = []
|
||||
|
||||
|
||||
class IndicatorDescribeResBody(BaseModel):
|
||||
description: str
|
||||
|
||||
|
||||
class SearchIndicatorReqBody(BaseModel):
|
||||
name: Optional[str] = None
|
||||
nature: Optional[str] = None
|
||||
category: Optional[CategoryEnum] = None
|
||||
category_id: Optional[str] = None
|
||||
page: int = 1
|
||||
pagesize: int = 20
|
||||
|
||||
class Config:
|
||||
use_enum_values = True
|
||||
|
|
|
|||
|
|
@ -0,0 +1,28 @@
|
|||
from AppIndicators import Schemas
|
||||
from AppIndicators.Crud import create_category
|
||||
from Utils.DataBase.SqlAlchemyUtils import Session
|
||||
|
||||
categories = [
|
||||
{"title": "盈利能力"},
|
||||
{"title": "收益质量"},
|
||||
{"title": "现金流量"},
|
||||
{"title": "资本结构"},
|
||||
{"title": "偿债能力"},
|
||||
{"title": "运营能力"},
|
||||
{"title": "成长能力"},
|
||||
{"title": "经营指标"},
|
||||
{"title": "资质指标"},
|
||||
{"title": "行业指标"},
|
||||
{"title": "绿色指标"},
|
||||
{"title": "司法指标"},
|
||||
{"title": "合规指标"},
|
||||
{"title": "舆情指标"},
|
||||
{"title": "其他"}
|
||||
]
|
||||
|
||||
db = Session()
|
||||
for category in categories:
|
||||
schema = Schemas.CreateCategoryReqBody(**category)
|
||||
create_category(db=db, schema=schema)
|
||||
|
||||
db.close()
|
||||
|
|
@ -19,3 +19,12 @@ def get_db():
|
|||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def set_next_id(db: Session, num_len: int, prefix: str, model):
|
||||
data = db.query(model).with_entities(model.id).order_by(model.id.desc()).first()
|
||||
if data:
|
||||
num = str(int({**data}.get("id").split(prefix)[-1]) + 1)
|
||||
return prefix + "0" * (num_len - len(num)) + num
|
||||
else:
|
||||
return prefix + "0" * (num_len - 1) + "1"
|
||||
|
|
|
|||
4
main.py
4
main.py
|
|
@ -1,7 +1,7 @@
|
|||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
from AppIndicators import RoutersIndicators
|
||||
from AppIndicators import Router
|
||||
|
||||
app = FastAPI(
|
||||
title="指标仓库",
|
||||
|
|
@ -18,4 +18,4 @@ app.add_middleware(
|
|||
)
|
||||
|
||||
# 路由
|
||||
app.include_router(RoutersIndicators.router)
|
||||
app.include_router(Router.router)
|
||||
|
|
|
|||
Loading…
Reference in New Issue