replaycli-py/api/dataapi/cli.py

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"