from datetime import datetime import json import requests class DataAPI: """ ## Data API --- Esta classe permite a execução de comandos básicos SQL para o banco de dados remoto do Replay. Para que a classe funcione corretamente, é necessário que o usuário tenha acesso à chave da API do BD remoto. Devendo passá-la como parâmetro ao instanciar a classe DataAPI. Por exemplo: d_api = DataAPI ("ahsgcn21390-dsaf_dain2eq81288sad9") """ ep: str = "" def __init__ (self, api_key: str): self.ep = "https://dataapi.digitalcircle.com.br" self.api_key = api_key def __request_json_post__ (self, object: dict): """ ## HTTP JSON POST --- Este método é responsável por realizar requisições HTTP do tipo POST para objetos JSON. Ele retorna o corpo de resposta da requisição, ou uma mensagem de erro, que indica qual foi a irregularidade ocorrida ao chamar a API. """ url = self.ep + "/" print("Calling: " + url) headers = {"X-API-KEY": self.api_key} res = requests.post(url, json = object, headers = headers, verify = False) if res.status_code >= 400: raise Exception(f"HTTP ERROR: {str(res.status_code)} - {res.text}") if res.headers.get("Content-Type") != None and res.headers.get("Content-Type").find("json") != -1: return json.loads(res.text) else: return res.text def do (self, dataAPI_request: dict): """ ## DataAPI Do Este método é responsável pela execução dos comandos SQL no Banco de Dados remoto. --- #### Parâmetros de dataAPI_request: - Col: Nome da tabela SQL em que se quer realizar a operação (OBRIGATÓRIO) - Op: Operação que se quer desenvolver. São elas: "R", "C", "D", "U" (OBRIGATÓRIO) - R: Retrieve (Busca informações da coluna da tabela) - C: Create (Cria uma nova tabela) - D: Delete (Deleta uma tabela ou as colunas de uma tabela) - U: Update (Atualiza as informações de uma tabela) - Q: Nome do atributo (coluna) que se quer consultar ou alterar na tabela SQL - Id: Identificador da instância (linha) da tabela que se quer realizar a operação - Data: Dados que se quer inserir na tabela --- #### Retorna: -> O retorno do comando feito, qualquer que seja ele. """ return self.__request_json_post__(dataAPI_request) def registrar_exec (self, register_table: str, check: bool): """ ## Registrar Execução Esta função é utilizada para realizar um registro de execuções falhas e bem sucedidas de um robô. Tudo é feito diretamente na Base de Dados remota, eliminando a necessidade de planilhas Excel locais. --- #### Parâmetros: - register_table: Tabela utilizada para os registros de execução. Para começos de projeto, procure criar uma nova tabela, cheque se ela já existe utilizando a operação de Retrieve pelo método Do. - check: Assume dois valores dependendo do sucesso da execução do robô: True ou False. - True: Execução obteve sucesso. - False: Execução falhou em algum ponto. --- #### Retorna: --- """ today_date = datetime.now().isoformat()[:10] table_check = self.do({ "Col": register_table, "Op": "R", "Q": f"@[?date=='{today_date}']" }) if len(table_check["data"]) > 0: data = table_check["data"][0] id = str(int(data["ID"])) if check: success = int(data["exec"]) + 1 try: self.do({ "Col": register_table, "Op": "U", "Id": id, "Data": { "exec": success } }) except: return "Algo falhou ao registrar a execução" else: fail = int(data["err"]) + 1 try: self.do({ "Col": register_table, "Op": "U", "Id": id, "Data": { "err": fail } }) except: return "Algo falhou ao registrar a execução" else: if check: success = 1 fail = 0 else: success = 0 fail = 1 try: self.do({ "Col": register_table, "Op": "C", "Data": { "date": today_date, "exec": success, "err": fail, } }) except: return "Algo falhou ao registrar a execução"