classFormDB(DBBase):''' Form Collection '''def__init__(self)->None:super().__init__('form')defindex(self)->None:''' To make collection's index Indexs: - `case` - `pid` '''self.create_index([('case',1),])self.create_index([('pid',1),])defadd_by_case(self,case:str,pid:str,uid:str,data:dict[str,Any])->dict[str,Any]:''' Add data by case Args: case (str): Case name. pid (str): Project id. uid (str): User id. data (dict): The data to insert / update. Returns: Return the inserted / updated data. '''_data={}forkindata:_data[f'data.{k}']=data[k]returnself.find_one_and_update({'case':case,'pid':pid,'uid':uid},{'$set':_data},upsert=True,return_document=ReturnDocument.AFTER,)
defadd_by_case(self,case:str,pid:str,uid:str,data:dict[str,Any])->dict[str,Any]:''' Add data by case Args: case (str): Case name. pid (str): Project id. uid (str): User id. data (dict): The data to insert / update. Returns: Return the inserted / updated data. '''_data={}forkindata:_data[f'data.{k}']=data[k]returnself.find_one_and_update({'case':case,'pid':pid,'uid':uid},{'$set':_data},upsert=True,return_document=ReturnDocument.AFTER,)
classFormTrafficFeeMappingDB(DBBase):''' Form traffic fee mapping Collection '''def__init__(self)->None:super().__init__('form_traffic_fee_mapping')defsave(self,pid:str,data:dict[str,Any])->dict[str,Any]:# pylint: disable=arguments-differ''' Save location / fee data Args: pid (str): Project id. data (dict): The data to insert / update. Format: `{'{location}': {fee}, ...}` Returns: Return the inserted / updated data. '''returnself.find_one_and_update({'_id':pid},{'$set':{'data':data}},upsert=True,return_document=ReturnDocument.AFTER,)
defsave(self,pid:str,data:dict[str,Any])->dict[str,Any]:# pylint: disable=arguments-differ''' Save location / fee data Args: pid (str): Project id. data (dict): The data to insert / update. Format: `{'{location}': {fee}, ...}` Returns: Return the inserted / updated data. '''returnself.find_one_and_update({'_id':pid},{'$set':{'data':data}},upsert=True,return_document=ReturnDocument.AFTER,)