open
/
replaycli-py
2
0
Fork 0
Code Issues 7 Pull Requests Projects Releases Wiki Activity
replaycli-py/api/replay/cli.py

440 lines
14 KiB
Python
Raw Normal View History

Adding new methods to the client.
2022-01-17 15:25:55 +00:00
from datetime import datetime
Creating the requests methods.
2022-01-14 11:16:57 +00:00
import json
import os
import requests
Adding configuration getters.
2022-01-18 11:29:02 +00:00
import urllib
Creating the requests methods.
2022-01-14 11:16:57 +00:00
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"
Implementing the environment variables getters.
2022-01-14 11:46:23 +00:00
def __request_get__ (self, data: str):
Creating the requests methods.
2022-01-14 11:16:57 +00:00
"""
## 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
Implementing the environment variables getters.
2022-01-14 11:46:23 +00:00
def __request_json_post__ (self, path: str, object: dict):
Creating the requests methods.
2022-01-14 11:16:57 +00:00
"""
## 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
Implementing the environment variables getters.
2022-01-14 11:46:23 +00:00
def __request_raw_post__ (self, path: str, data: str):
Creating the requests methods.
2022-01-14 11:16:57 +00:00
"""
## 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
Implementing the environment variables getters.
2022-01-14 11:46:23 +00:00
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")
Adding new methods to the client.
2022-01-17 15:25:55 +00:00
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)
Adding configuration getters.
2022-01-18 11:29:02 +00:00
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
Adding new methods.
2022-01-18 13:57:37 +00:00
Este método procura entre as configurações globais, uma correspondência ao nome de configuração recebido.
---
### Obs:
Uma configuração global é definida diretamente na aba de Configurações do Console do Replay.
Deve-se clicar em "Adicionar" e digitar o nome da configuração, depois editar o nome da mesma.
Se uma configuração não está associada a nenhum repositório em específico, isso quer dizer que ela é global.
Adding configuration getters.
2022-01-18 11:29:02 +00:00
---
#### Parâmetros:
Adding new methods.
2022-01-18 13:57:37 +00:00
- config_name: Nome da configuração global que se quer saber o valor.
Adding configuration getters.
2022-01-18 11:29:02 +00:00
---
#### Retorna:
-> O valor da configuração encontrada.
Adding new methods.
2022-01-18 13:57:37 +00:00
-> "" caso a configuração não exista.
Adding configuration getters.
2022-01-18 11:29:02 +00:00
"""
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
Adding new methods.
2022-01-18 13:57:37 +00:00
def open_app (self, id: int):
"""
## Open Apps
Este método realiza a execução de um aplicativo cujo ID foi recebido.
---
### Obs:
É importante ressaltar que a execução do aplicativo lançado é síncrona, ou seja, o código não terá sua execução completada antes da finalização do aplicativo.
Para alguns casos de uso, a execução síncrona é bastante útil, mas para outros, pode dificultar bastante o processo de desenvolvimento.
Sendo assim, é possível utilizar o recurso de multiThreading do python para que o aplicativo seja executado assíncronamente. Isso é feito através da biblioteca "threading", que não é nativa. Deve-se instalá-la e utilizar de acordo com a necessidade.
Para maiores informações sobre a biblioteca threading, visite o site:
http://pymotw.com/2/threading/
---
#### Parâmetros:
- id: Identificador do aplicativo que se deseja executar.
---
#### Retorna:
---
"""
return self.__request_get__(f"/api/v1/app/run/{id}")
def cron_reload (self):
"""
## Cronometer Reload
Este método está ligado à página de Agendamentos no Console do Replay. Ele recarrega esta funcionalidade para atualizar os registros de agendamento.
---
#### Parâmetros:
---
---
#### Retorna:
-> ???
"""
return self.__request_get__("/ipc/cron/reload")
def menu_get_all_enabled (self):
"""
Get All Enabled Menus
Este método está relacionado à aba de "Menus" do Console do Replay. Todos os menus cadastrados naquela aba e que tenham a checkbox do lado esquerdo marcada deverão ser buscados por este método.
---
#### Parâmetros:
---
---
#### Retorna:
-> Lista de dicionários, onde os dicionários são os dados/informações dos menus encontrados.
"""
return self.__request_get__("/api/v1/menu/enabled")
def service_stop_all (self):
"""
## Stop All Services
Este método realiza a parada forçada de todos os serviços que estiverem ativos. É possível visualizar quais serviços estão ativos pela aba "Serviços" do Console do Replay. Todos cujo PID tenha um valor diferente de -1 e 0 são serviços que estão, atualmente, ativos.
---
#### Parâmetros:
---
---
#### Retorna:
-> "ok" caso o método tenha sido executado corretamente
"""
return self.__request_get__("/api/v1/service/op/stopall")
def exit (self):
"""
## Exit
Finaliza a sessão do Replay. Equivalente a clicar no ícone e depois em "Quit".
---
#### Parâmetros:
---
---
#### Retorna:
---
"""
try:
self.__request_get__("/api/v1/exit")
except:
return "ok"