igf_api_py_client/entities/PiapiTask.py

75 lines
2.5 KiB
Python
Raw Permalink Normal View History

2024-12-22 14:10:39 +03:00
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))