75 lines
2.5 KiB
Python
75 lines
2.5 KiB
Python
from igf_api.BaseEntity import BaseEntity
|
|
|
|
|
|
class PiapiTask(BaseEntity):
|
|
|
|
def get_list(self):
|
|
return self.connection.raw_request(method='get', endpoint='/piapi')
|
|
|
|
def get_one(self, data):
|
|
schema = {
|
|
'type': 'object',
|
|
'properties': {
|
|
'bot_id': {'type': 'integer'},
|
|
'dialog_id': {'type': 'integer'},
|
|
'task_id': {"type": ["string", "null"]},
|
|
'model': {'type': 'string'},
|
|
'task_type': {'type': 'string'},
|
|
'prompt': {'type': 'string'},
|
|
'status': {'type': 'integer'},
|
|
},
|
|
'required': []
|
|
}
|
|
|
|
if self.validate_dict(data, schema):
|
|
return self.connection.raw_request('get', '/piapi/get-one', params=data)
|
|
|
|
return False
|
|
|
|
def create(self, data):
|
|
schema = {
|
|
'type': 'object',
|
|
'properties': {
|
|
'bot_id': {'type': 'integer'},
|
|
'dialog_id': {'type': 'integer'},
|
|
'task_id': {"type": ["string", "null"]},
|
|
'model': {'type': 'string'},
|
|
'task_type': {'type': 'string'},
|
|
'prompt': {'type': 'string'},
|
|
'status': {'type': 'integer'},
|
|
},
|
|
'required': ['dialog_id', 'bot_id']
|
|
}
|
|
|
|
if self.validate_dict(data, schema):
|
|
return self.connection.raw_request('post', '/piapi', data=data)
|
|
|
|
return False
|
|
|
|
def update(self, id: int, data):
|
|
schema = {
|
|
'type': 'object',
|
|
'properties': {
|
|
'bot_id': {'type': 'integer'},
|
|
'dialog_id': {'type': 'integer'},
|
|
'task_id': {"type": ["string", "null"]},
|
|
'model': {'type': 'string'},
|
|
'task_type': {'type': 'string'},
|
|
'prompt': {'type': 'string'},
|
|
'status': {'type': 'integer'},
|
|
},
|
|
'required': []
|
|
}
|
|
|
|
if self.validate_dict(data, schema):
|
|
return self.connection.raw_request('post', '/piapi/update/{id}'.format(id=id), data=data)
|
|
|
|
return False
|
|
|
|
def to_archive_all_new(self, bot_id: int, dialog_id: int):
|
|
return self.connection.raw_request('get', '/piapi/to-archive-all-new/{bot_id}/{dialog_id}'.format(bot_id=bot_id,
|
|
dialog_id=dialog_id))
|
|
|
|
def get_new_tasks(self):
|
|
return self.connection.raw_request('get', '/piapi/find?status={status}'.format(status=1))
|