121 lines
3.5 KiB
Python
121 lines
3.5 KiB
Python
from os import environ
|
|
from os.path import isfile
|
|
from requests import get, post
|
|
from json import dumps
|
|
from typing import Dict
|
|
|
|
from tiramisu_api import Config
|
|
|
|
from .config import config
|
|
from .i18n import _
|
|
|
|
|
|
if config.allow_insecure_https:
|
|
import warnings
|
|
from urllib3.exceptions import InsecureRequestWarning
|
|
warnings.simplefilter('ignore', InsecureRequestWarning)
|
|
|
|
|
|
class JsonError(Exception):
|
|
pass
|
|
|
|
|
|
class Common:
|
|
def __init__(self):
|
|
self.cucchiaiata_config = config
|
|
|
|
def get_error_from_http(self,
|
|
req):
|
|
try:
|
|
json = req.json()
|
|
err = json['error']['kwargs']['reason']
|
|
except:
|
|
err = req.text
|
|
return err
|
|
|
|
def remote_json_to_config(self,
|
|
url,
|
|
config_type=Config,
|
|
):
|
|
"retrieves the remote config from the distant api description"
|
|
headers = get_headers()
|
|
req = get(url,
|
|
headers=headers,
|
|
verify=config.allow_insecure_https,
|
|
)
|
|
code = req.status_code
|
|
if code != 200:
|
|
raise Exception(self.get_error_from_http(req))
|
|
json = req.json()
|
|
return config_type(json)
|
|
|
|
def get_payload(self,
|
|
message: str):
|
|
# remove symlinkoption and default value from payload
|
|
payload = {}
|
|
for option in self.remote_config.option(message).list():
|
|
if not option.owner.isdefault() and not option.option.issymlinkoption():
|
|
payload[option.option.name()] = option.value.get()
|
|
return payload
|
|
|
|
def send(self,
|
|
message: str,
|
|
**kwargs,
|
|
) -> None:
|
|
self.remote_config.option('message').value.set(message)
|
|
for key, value in kwargs.items():
|
|
self.remote_config.option(f'{message}.{key}').value.set(value)
|
|
payload = self.get_payload(message)
|
|
return send_data(message,
|
|
payload,
|
|
)
|
|
|
|
def send_configuration(self,
|
|
tiramisu: 'ConfigAPI',
|
|
session_id: str,
|
|
) -> None:
|
|
tiramisu.send()
|
|
try:
|
|
self.send('v1.setting.session.validate',
|
|
session_id=session_id,
|
|
)
|
|
except JsonError as err:
|
|
self.send('v1.setting.session.stop',
|
|
session_id=session_id,
|
|
)
|
|
raise err from err
|
|
|
|
|
|
def get_headers():
|
|
headers = {}
|
|
if isfile(config.token_file):
|
|
with open(config.token_file) as token_file:
|
|
token = token_file.read()
|
|
headers['Authorization'] = f'Bearer {token}'
|
|
if 'FORCE_RISOTTO_USER' in environ:
|
|
headers['username'] = environ['FORCE_RISOTTO_USER']
|
|
return headers
|
|
|
|
|
|
def send_data(uri: str,
|
|
payload: Dict,
|
|
):
|
|
version, message = uri.split('.', 1)
|
|
final_url = '{}/{}/{}'.format(config.remote_url,
|
|
version,
|
|
message,
|
|
)
|
|
ret = post(final_url,
|
|
data=dumps(payload),
|
|
headers=get_headers(),
|
|
verify=config.allow_insecure_https)
|
|
try:
|
|
response = ret.json()
|
|
except:
|
|
raise Exception(ret.text)
|
|
if response['type'] == 'error':
|
|
err = JsonError()
|
|
err.message = response['response']
|
|
raise err
|
|
return response['response']
|