from datetime import datetime import json import os import requests import urllib class Replay: """ ## Replay --- Esta classe é utilizada para fazer manipulações com o Replay instalado na máquina. Ela concede acesso às diferentes funcionalidades que o mesmo apresenta na interface do Console, sem que o programador precise realizar WebScraping para acessá-lo. Permite por exemplo a extração de configurações definidas para certos repositórios, registro de Logs, manipulação de banco de dados (SQLite), entre várias outras coisas. """ ep: str = "" def __init__ (self): self.ep = "https://localhost:8443" def __request_get__ (self, data: str): """ ## HTTP GET --- Este método é responsável por realizar requisições HTTP do tipo GET. 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 + data print("Calling: " + url) apikey = os.environ.get('REPLAY_APIKEY') headers = {"X-API-KEY": apikey} res = requests.get(url, 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 __request_json_post__ (self, path: str, 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 + path print("Calling: " + url) apikey = os.environ.get('REPLAY_APIKEY') headers = {"X-API-KEY": apikey} 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 __request_raw_post__ (self, path: str, data: str): """ ## HTTP RAW POST --- Este método é responsável por realizar requisições HTTP do tipo POST para objetos RAW. 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 + path print("Calling: " + url) apikey = os.environ.get('REPLAY_APIKEY') headers = {"X-API-KEY": apikey} res = requests.post(url, data = data, 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 replay_env_queue_id (self): """ ## Replay Environment Queue ID --- Esta função retorna o valor atribuído à variável de ambiente "REPLAY_QUEUEID", caso exista. """ return os.environ.get ("REPLAY_QUEUEID") def replay_env_alias (self): """ ## Replay Environment Alias --- Esta função retorna o valor atribuído à variável de ambiente "REPLAY_ALIAS", caso exista. """ return os.environ.get ("REPLAY_ALIAS") def replay_env_feature_dir (self): """ ## Replay Environment Feature Directory --- Esta função retorna o valor atribuído à variável de ambiente "REPLAY_FEATUREDIR", caso exista. """ return os.environ.get ("REPLAY_FEATUREDIR") def replay_env_root (self): """ ## Replay Environment Root --- Esta função retorna o valor atribuído à variável de ambiente "REPLAY_ROOT", caso exista. """ return os.environ.get ("REPLAY_ROOT") def replay_env_addr (self): """ ## Replay Environment Address --- Esta função retorna o valor atribuído à variável de ambiente "REPLAY_ADDR", caso exista. """ return os.environ.get ("REPLAY_ADDR") def replay_env_home_dir (self): """ ## Replay Environment Home Directory --- Esta função retorna o valor atribuído à variável de ambiente "REPLAY_HOMEDIR", caso exista. """ return os.environ.get ("REPLAY_HOMEDIR") def replay_env_repo (self): """ ## Replay Environment Repository --- Esta função retorna o valor atribuído à variável de ambiente "REPLAY_REPO", caso exista. """ return os.environ.get ("REPLAY_REPO") def replay_env_ver (self): """ ## Replay Environment Version --- Esta função retorna o valor atribuído à variável de ambiente "REPLAY_VER", caso exista. """ return os.environ.get ("REPLAY_VER") def replay_env_repo_dir (self): """ ## Replay Environment Repository Directory --- Esta função retorna o valor atribuído à variável de ambiente "REPLAY_REPODIR", caso exista. """ return os.environ.get ("REPLAY_REPODIR") def replay_env_data_dir (self): """ ## Replay Environment Data Directory --- Esta função retorna o valor atribuído à variável de ambiente "REPLAY_DATADIR", caso exista. """ return os.environ.get ("REPLAY_DATADIR") def replay_env_instance_alias (self): """ ## Replay Environment Instance Alias --- Esta função retorna o valor atribuído à variável de ambiente "REPLAY_INSTANCE_ALIAS", caso exista. """ return os.environ.get ("REPLAY_INSTANCE_ALIAS") def replay_env_api_key (self): """ ## Replay Environment API Key --- Esta função retorna o valor atribuído à variável de ambiente "REPLAY_APIKEY", caso exista. """ return os.environ.get ("REPLAY_APIKEY") def sql (self, command:str): """ ## SQL Commands Executer Este método faz um requisição SQL no Replay na aba "Banco de Dados" --- #### Parâmetros: - command: Comando SQL que se deseja executar. --- #### Retorna: -> Qualquer que seja o retorno do comando executado. """ sql = { "Sql": command } return self.__request_json_post__("/api/v1/sql", sql) def new_log (self): """ ## New Log Este método faz a instanciação de um objeto do tipo "Log" para que ele possa ser registrado no Replay. """ log = { "Alias": str (self.replay_env_alias()), "Dtlog": datetime.now().isoformat() + "Z" } return log def log (self, log: dict): """ ## Log Esta função realiza a postagem de um Log na base de dados do replay, podendo ser visualizado pelo Console. --- #### Parâmetros: - log: Este parâmetro é uma instância da classe Logs que deve ser criada a partir do método "new_logs". --- #### Retorna: --- """ log["Dtlogend"] = datetime.now().isoformat() + "Z" return self.__request_json_post__ ("/api/v1/log/add", log) def config_get (self, config_name: str): """ ## Configuration Getter Este método busca a configuração associada ao nome recebido no repositório atual. --- #### Parâmetros: - config_name: Nome da configuração que se quer procurar o valor. --- #### Retorna: -> O valor associado à configuração procurada. """ config_name = urllib.parse.quote(config_name, "") return self.__request_get__(f"/api/v1/config/byrepo/{int(os.path.basename(os.getcwd()))}/{config_name}") def config_get_global (self, config_name: str): """ ## Global Configuration Getter Este método procura em todos os repositórios uma correspondência à configuração recebida. --- #### Parâmetros: - config_name: Nome da configuração que se quer buscar em todos os repositórios instalados. --- #### Retorna: -> O valor da configuração encontrada. """ config_name = urllib.parse.quote(config_name, "") return self.__request_get__("/api/v1/config/global/"+config_name) def config_get_all (self): """ ## All Configuration Getter Este método busca por todas as configurações contidas no repositório atual. --- #### Parâmetros: --- --- #### Retorna: -> Uma lista de dicionários contendo todas as configs e informações sobre elas. Por exemplo: [ { "id": número associado a config na base de dados do replay, "k": nome associado à config na base de dados do replay, "v" valor atribuído à config, "t": tipo da config em questão (que pode ser 't', 'p' ou 'json'), "repo": repositório ao qual a configuração está associada } ] """ return self.__request_get__(f"/api/v1/config/byrepo/{int(os.path.basename(os.getcwd()))}") def config_get_map (self) -> dict[str, dict]: """ ## Configuration Getter Mapper Este método executa o método "config_get_all" para conseguir todas as configurações alocadas para o reposiório do projeto atual. Após isso, cria um dicionário de strings para outros dicionários contendo as configurações recebidas. As strings do primeiro dicionário são o nome associado à configuração. --- #### Parâmetros: --- --- #### Retorna: -> Dicionário de strings (nome da configuração) para dicionários (todas as propriedades da configuração). """ dict_configs = {} configs = self.config_get_all() for config in configs: dict_configs[config["k"]] = config return dict_configs