Compare commits

..

25 Commits

Author SHA1 Message Date
52209a5ebd Merge branch 'develop' into dist/risotto/risotto-2.8.0/develop 2021-04-24 14:36:02 +02:00
0442e772c2 support str value 2021-04-24 14:35:51 +02:00
b0edfb7b01 Merge branch 'develop' into dist/risotto/risotto-2.8.0/develop 2021-04-24 14:16:11 +02:00
27031dbf0e log_connexion 2021-04-24 14:15:54 +02:00
cb4dde1dc4 Merge branch 'develop' into dist/risotto/risotto-2.8.0/develop 2021-04-24 12:56:52 +02:00
9ebe79d533 special connexion for log (do not rollback if error) 2021-04-24 12:56:44 +02:00
c740ec3fe3 Merge branch 'develop' into dist/risotto/risotto-2.8.0/develop 2021-04-24 10:12:51 +02:00
4c83e6d89d better log support 2021-04-24 10:12:39 +02:00
19240489db add http static support 2021-04-24 10:12:32 +02:00
30a267bf4a add TiramisuController 2021-04-24 10:12:13 +02:00
56b1f12a4a Merge branch 'develop' into dist/risotto/risotto-2.8.0/develop 2021-04-16 09:34:21 +02:00
f88bcef5c0 do not stop daemon when on_join failed 2021-04-16 09:33:25 +02:00
4fc3e74bbd Merge branch 'develop' into dist/risotto/risotto-2.8.0/develop 2021-04-13 10:31:24 +02:00
5663b2768b if not Risotto module, do not failed 2021-04-13 10:31:14 +02:00
83d74c2b06 Merge branch 'develop' into dist/risotto/risotto-2.8.0/develop 2021-04-12 15:11:54 +02:00
01834c6ba7 add check_role to dispatcher 2021-04-12 15:11:46 +02:00
6a27b002ff Merge branch 'develop' into dist/risotto/risotto-2.8.0/develop 2021-03-27 10:59:19 +01:00
8fdc34c4d3 fix 2021-03-27 10:59:10 +01:00
e2d73932c0 add sdnotify dependency 2020-11-14 19:11:57 +01:00
980a119ef9 Merge branch 'develop' into dist/risotto/risotto-2.8.0/develop 2020-11-14 19:01:34 +01:00
f623feb8a8 add systemd notifier 2020-11-14 19:01:28 +01:00
b9da2ce686 Merge branch 'develop' into dist/risotto/risotto-2.8.0/develop 2020-11-14 08:12:50 +01:00
46f8a4323b add pki informations 2020-11-14 08:12:39 +01:00
941261c830 Merge branch 'develop' into dist/risotto/risotto-2.8.0/develop 2020-10-14 18:30:13 +02:00
6c4bbb3dca add password support 2020-10-14 18:30:05 +02:00
12 changed files with 801 additions and 202 deletions

6
debian/control vendored
View File

@ -9,7 +9,11 @@ Homepage: https://forge.cadoles.com/Infra/risotto
Package: python3-risotto Package: python3-risotto
Architecture: any Architecture: any
Pre-Depends: dpkg, python3, ${misc:Pre-Depends} Pre-Depends: dpkg, python3, ${misc:Pre-Depends}
Depends: ${python:Depends}, ${misc:Depends}, python3-asyncpg, python3-rougail, python3-aiohttp Depends: ${python:Depends}, ${misc:Depends},
python3-asyncpg,
python3-rougail,
python3-aiohttp,
python3-sdnotify
Description: configuration manager libraries Description: configuration manager libraries
Package: risotto Package: risotto

View File

@ -1,13 +1,16 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
from sdnotify import SystemdNotifier
from asyncio import get_event_loop from asyncio import get_event_loop
from risotto import get_app from risotto import get_app
if __name__ == '__main__': if __name__ == '__main__':
notifier = SystemdNotifier()
loop = get_event_loop() loop = get_event_loop()
loop.run_until_complete(get_app(loop)) loop.run_until_complete(get_app(loop))
print('HTTP server ready')
notifier.notify("READY=1")
try: try:
print('HTTP server ready')
loop.run_forever() loop.run_forever()
except KeyboardInterrupt: except KeyboardInterrupt:
pass pass

View File

@ -1,8 +1,16 @@
CREATE TABLE log( CREATE TABLE RisottoLog(
LogId SERIAL PRIMARY KEY,
Msg VARCHAR(255) NOT NULL, Msg VARCHAR(255) NOT NULL,
URI VARCHAR(255),
URIS VARCHAR(255),
UserLogin VARCHAR(100) NOT NULL,
Level VARCHAR(10) NOT NULL, Level VARCHAR(10) NOT NULL,
Path VARCHAR(255), ContextId INTEGER,
Username VARCHAR(100) NOT NULL,
Data JSON, Data JSON,
Date timestamp DEFAULT current_timestamp Returns JSON,
StartDate timestamp DEFAULT current_timestamp,
StopDate timestamp
); );
CREATE INDEX RisottoLog_ContextId_index ON RisottoLog(ContextId);
CREATE INDEX RisottoLog_Login_index ON RisottoLog(UserLogin);
CREATE INDEX RisottoLog_URI_index ON RisottoLog(URI);

View File

@ -1,6 +1,7 @@
from os import environ from os import environ
from os.path import isfile from os.path import isfile
from configobj import ConfigObj from configobj import ConfigObj
from uuid import uuid4
CONFIG_FILE = environ.get('CONFIG_FILE', '/etc/risotto/risotto.conf') CONFIG_FILE = environ.get('CONFIG_FILE', '/etc/risotto/risotto.conf')
@ -16,6 +17,10 @@ if 'RISOTTO_PORT' in environ:
RISOTTO_PORT = environ['RISOTTO_PORT'] RISOTTO_PORT = environ['RISOTTO_PORT']
else: else:
RISOTTO_PORT = config.get('RISOTTO_PORT', 8080) RISOTTO_PORT = config.get('RISOTTO_PORT', 8080)
if 'RISOTTO_URL' in environ:
RISOTTO_URL = environ['RISOTTO_URL']
else:
RISOTTO_URL = config.get('RISOTTO_URL', 'http://localhost:8080/')
if 'CONFIGURATION_DIR' in environ: if 'CONFIGURATION_DIR' in environ:
CONFIGURATION_DIR = environ['CONFIGURATION_DIR'] CONFIGURATION_DIR = environ['CONFIGURATION_DIR']
else: else:
@ -88,6 +93,40 @@ if 'IMAGE_PATH' in environ:
IMAGE_PATH = environ['IMAGE_PATH'] IMAGE_PATH = environ['IMAGE_PATH']
else: else:
IMAGE_PATH = config.get('IMAGE_PATH', '/tmp') IMAGE_PATH = config.get('IMAGE_PATH', '/tmp')
if 'PASSWORD_ADMIN_USERNAME' in environ:
PASSWORD_ADMIN_USERNAME = environ['PASSWORD_ADMIN_USERNAME']
else:
PASSWORD_ADMIN_USERNAME = config.get('PASSWORD_ADMIN_USERNAME', 'risotto')
if 'PASSWORD_ADMIN_EMAIL' in environ:
PASSWORD_ADMIN_EMAIL = environ['PASSWORD_ADMIN_EMAIL']
else:
# this parameter is mandatory
PASSWORD_ADMIN_EMAIL = config['PASSWORD_ADMIN_EMAIL']
if 'PASSWORD_ADMIN_PASSWORD' in environ:
PASSWORD_ADMIN_PASSWORD = environ['PASSWORD_ADMIN_PASSWORD']
else:
# this parameter is mandatory
PASSWORD_ADMIN_PASSWORD = config['PASSWORD_ADMIN_PASSWORD']
if 'PASSWORD_DEVICE_IDENTIFIER' in environ:
PASSWORD_DEVICE_IDENTIFIER = environ['PASSWORD_DEVICE_IDENTIFIER']
else:
PASSWORD_DEVICE_IDENTIFIER = config.get('PASSWORD_DEVICE_IDENTIFIER', uuid4())
if 'PASSWORD_URL' in environ:
PASSWORD_URL = environ['PASSWORD_URL']
else:
PASSWORD_URL = config.get('PASSWORD_URL', 'https://localhost:8001/')
if 'PKI_ADMIN_PASSWORD' in environ:
PKI_ADMIN_PASSWORD = environ['PKI_ADMIN_PASSWORD']
else:
PKI_ADMIN_PASSWORD = config['PKI_ADMIN_PASSWORD']
if 'PKI_ADMIN_EMAIL' in environ:
PKI_ADMIN_EMAIL = environ['PKI_ADMIN_EMAIL']
else:
PKI_ADMIN_EMAIL = config['PKI_ADMIN_EMAIL']
if 'PKI_URL' in environ:
PKI_URL = environ['PKI_URL']
else:
PKI_URL = config.get('PKI_URL', 'http://localhost:8002')
def dsn_factory(database, user, password, address=DB_ADDRESS): def dsn_factory(database, user, password, address=DB_ADDRESS):
@ -100,7 +139,8 @@ _config = {'database': {'dsn': dsn_factory(RISOTTO_DB_NAME, RISOTTO_DB_USER, RIS
'celery_dsn': dsn_factory(CELERYRISOTTO_DB_NAME, CELERYRISOTTO_DB_USER, CELERYRISOTTO_DB_PASSWORD) 'celery_dsn': dsn_factory(CELERYRISOTTO_DB_NAME, CELERYRISOTTO_DB_USER, CELERYRISOTTO_DB_PASSWORD)
}, },
'http_server': {'port': RISOTTO_PORT, 'http_server': {'port': RISOTTO_PORT,
'default_user': DEFAULT_USER}, 'default_user': DEFAULT_USER,
'url': RISOTTO_URL},
'global': {'message_root_path': MESSAGE_PATH, 'global': {'message_root_path': MESSAGE_PATH,
'configurations_dir': CONFIGURATION_DIR, 'configurations_dir': CONFIGURATION_DIR,
'debug': True, 'debug': True,
@ -110,6 +150,16 @@ _config = {'database': {'dsn': dsn_factory(RISOTTO_DB_NAME, RISOTTO_DB_USER, RIS
'sql_dir': SQL_DIR, 'sql_dir': SQL_DIR,
'tmp_dir': TMP_DIR, 'tmp_dir': TMP_DIR,
}, },
'password': {'admin_username': PASSWORD_ADMIN_USERNAME,
'admin_email': PASSWORD_ADMIN_EMAIL,
'admin_password': PASSWORD_ADMIN_PASSWORD,
'device_identifier': PASSWORD_DEVICE_IDENTIFIER,
'service_url': PASSWORD_URL,
},
'pki': {'admin_password': PKI_ADMIN_PASSWORD,
'owner': PKI_ADMIN_EMAIL,
'url': PKI_URL,
},
'cache': {'root_path': CACHE_ROOT_PATH}, 'cache': {'root_path': CACHE_ROOT_PATH},
'servermodel': {'internal_source_path': SRV_SEED_PATH, 'servermodel': {'internal_source_path': SRV_SEED_PATH,
'internal_source': 'internal'}, 'internal_source': 'internal'},

View File

@ -1,3 +1,5 @@
class Context: class Context:
def __init__(self): def __init__(self):
self.paths = [] self.paths = []
self.context_id = None
self.start_id = None

View File

@ -1,6 +1,22 @@
from os import listdir, makedirs
from os.path import join, isdir, isfile
from shutil import rmtree
from traceback import print_exc
from typing import Dict
from rougail import RougailConvert, RougailConfig, RougailUpgrade
try:
from tiramisu3 import Storage, Config
except:
from tiramisu import Storage, Config
from .config import get_config
from .utils import _, tiramisu_display_name
from .logger import log
from .dispatcher import dispatcher from .dispatcher import dispatcher
from .context import Context from .context import Context
from .utils import _
RougailConfig['variable_namespace'] = 'configuration'
class Controller: class Controller:
@ -8,7 +24,7 @@ class Controller:
""" """
def __init__(self, def __init__(self,
test: bool, test: bool,
): ) -> None:
pass pass
async def call(self, async def call(self,
@ -48,7 +64,268 @@ class Controller:
**kwargs, **kwargs,
) )
@staticmethod
async def check_role(self,
uri: str,
username: str,
**kwargs: dict,
) -> None:
# create a new config
async with await Config(dispatcher.option) as config:
await config.property.read_write()
await config.option('message').value.set(uri)
subconfig = config.option(uri)
for key, value in kwargs.items():
try:
await subconfig.option(key).value.set(value)
except AttributeError:
if get_config()['global']['debug']:
print_exc()
raise ValueError(_(f'unknown parameter in "{uri}": "{key}"'))
except ValueOptionError as err:
raise ValueError(_(f'invalid parameter in "{uri}": {err}'))
await dispatcher.check_role(subconfig,
username,
uri,
)
async def on_join(self, async def on_join(self,
risotto_context, risotto_context,
): ):
pass pass
class TiramisuController(Controller):
def __init__(self,
test: bool,
) -> None:
if not 'dataset_name' in vars(self):
raise Exception(f'please specify "dataset_name" to "{self.__class__.__name__}"')
self.tiramisu_cache_root_path = join(get_config()['cache']['root_path'], self.dataset_name)
if not test:
db_conf = get_config()['database']['tiramisu_dsn']
self.save_storage = Storage(engine='postgres')
self.save_storage.setting(dsn=db_conf)
if self.dataset_name != 'servermodel':
self.optiondescription = None
dispatcher.set_function('v1.setting.dataset.updated',
None,
TiramisuController.dataset_updated,
self.__class__.__module__,
)
async def on_join(self,
risotto_context: Context,
) -> None:
if isdir(self.tiramisu_cache_root_path):
await self.load_datas(risotto_context)
async def dataset_updated(self,
risotto_context: Context,
) -> Dict:
await self.gen_dictionaries(risotto_context)
await self.load_datas(risotto_context)
async def gen_dictionaries(self,
risotto_context: Context,
) -> None:
sources = await self.get_sources(risotto_context)
self._aggregate_tiramisu_funcs(sources)
self._convert_dictionaries_to_tiramisu(sources)
async def get_sources(self,
risotto_context: Context,
) -> None:
return await self.call('v1.setting.source.list',
risotto_context,
)
def _aggregate_tiramisu_funcs(self,
sources: list,
) -> None:
dest_file = join(self.tiramisu_cache_root_path, 'funcs.py')
if not isdir(self.tiramisu_cache_root_path):
makedirs(self.tiramisu_cache_root_path)
with open(dest_file, 'wb') as funcs:
funcs.write(b"""try:
from tiramisu3 import valid_network_netmask, valid_ip_netmask, valid_broadcast, valid_in_network, valid_not_equal as valid_differ, valid_not_equal, calc_value
except:
from tiramisu import valid_network_netmask, valid_ip_netmask, valid_broadcast, valid_in_network, valid_not_equal as valid_differ, valid_not_equal, calc_value
""")
for source in sources:
root_path = join(source['source_directory'],
self.dataset_name,
)
if not isdir(root_path):
continue
for service in listdir(root_path):
path = join(root_path,
service,
'funcs',
)
if not isdir(path):
continue
for filename in listdir(path):
if not filename.endswith('.py'):
continue
filename_path = join(path, filename)
with open(filename_path, 'rb') as fh:
funcs.write(f'# {filename_path}\n'.encode())
funcs.write(fh.read())
funcs.write(b'\n')
def _convert_dictionaries_to_tiramisu(self, sources: list) -> None:
funcs_file = join(self.tiramisu_cache_root_path, 'funcs.py')
tiramisu_file = join(self.tiramisu_cache_root_path, 'tiramisu.py')
dictionaries_dir = join(self.tiramisu_cache_root_path, 'dictionaries')
extras_dictionaries_dir = join(self.tiramisu_cache_root_path, 'extra_dictionaries')
if isdir(dictionaries_dir):
rmtree(dictionaries_dir)
makedirs(dictionaries_dir)
if isdir(extras_dictionaries_dir):
rmtree(extras_dictionaries_dir)
makedirs(extras_dictionaries_dir)
extras = []
upgrade = RougailUpgrade()
for source in sources:
root_path = join(source['source_directory'],
self.dataset_name,
)
if not isdir(root_path):
continue
for service in listdir(root_path):
# upgrade dictionaries
path = join(root_path,
service,
'dictionaries',
)
if not isdir(path):
continue
upgrade.load_xml_from_folders(path,
dictionaries_dir,
RougailConfig['variable_namespace'],
)
for service in listdir(root_path):
# upgrade extra dictionaries
path = join(root_path,
service,
'extras',
)
if not isdir(path):
continue
for namespace in listdir(path):
extra_dir = join(path,
namespace,
)
if not isdir(extra_dir):
continue
extra_dictionaries_dir = join(extras_dictionaries_dir,
namespace,
)
if not isdir(extra_dictionaries_dir):
makedirs(extra_dictionaries_dir)
extras.append((namespace, [extra_dictionaries_dir]))
upgrade.load_xml_from_folders(extra_dir,
extra_dictionaries_dir,
namespace,
)
del upgrade
config = RougailConfig.copy()
config['functions_file'] = funcs_file
config['dictionaries_dir'] = [dictionaries_dir]
config['extra_dictionaries'] = {}
for extra in extras:
config['extra_dictionaries'][extra[0]] = extra[1]
eolobj = RougailConvert(rougailconfig=config)
eolobj.save(tiramisu_file)
async def load(self,
risotto_context: Context,
name: str,
to_deploy: bool=False,
) -> Config:
if self.optiondescription is None:
# use file in cache
tiramisu_file = join(self.tiramisu_cache_root_path, 'tiramisu.py')
if not isfile(tiramisu_file):
raise Exception(_(f'unable to load the "{self.dataset_name}" configuration, is dataset loaded?'))
with open(tiramisu_file) as fileio:
tiramisu_locals = {}
try:
exec(fileio.read(), None, tiramisu_locals)
except Exception as err:
raise Exception(_(f'unable to load tiramisu file {tiramisu_file}: {err}'))
self.optiondescription = tiramisu_locals['option_0']
del tiramisu_locals
try:
letter = self.dataset_name[0]
if not to_deploy:
session_id = f'{letter}_{name}'
else:
session_id = f'{letter}td_{name}'
config = await Config(self.optiondescription,
session_id=session_id,
storage=self.save_storage,
display_name=tiramisu_display_name,
)
# change default rights
await config.property.read_only()
await config.permissive.add('basic')
await config.permissive.add('normal')
await config.permissive.add('expert')
# set information and owner
await config.owner.set(session_id)
await config.information.set(f'{self.dataset_name}_name', name)
except Exception as err:
if get_config()['global']['debug']:
print_exc()
msg = _(f'unable to load config for {self.dataset_name} "{name}": {err}')
await log.error_msg(risotto_context,
None,
msg,
)
return config
async def _deploy_configuration(self,
dico: dict,
) -> None:
config_std = dico['config_to_deploy']
config = dico['config']
# when deploy, calculate force_store_value
ro = await config_std.property.getdefault('read_only', 'append')
if 'force_store_value' not in ro:
await config_std.property.read_write()
if self.dataset_name == 'servermodel':
# server_deployed should be hidden
await config_std.forcepermissive.option('configuration.general.server_deployed').value.set(True)
ro = frozenset(list(ro) + ['force_store_value'])
rw = await config_std.property.getdefault('read_write', 'append')
rw = frozenset(list(rw) + ['force_store_value'])
await config_std.property.setdefault(ro, 'read_only', 'append')
await config_std.property.setdefault(rw, 'read_write', 'append')
await config_std.property.read_only()
# copy informations from 'to deploy' configuration to configuration
await config.value.importation(await config_std.value.exportation())
await config.permissive.importation(await config_std.permissive.exportation())
await config.property.importation(await config_std.property.exportation())
async def build_configuration(self,
config: Config,
) -> dict:
configuration = {}
for option in await config.option.list('optiondescription'):
name = await option.option.name()
if name == 'services':
continue
if name == RougailConfig['variable_namespace']:
fullpath = False
flatten = True
else:
fullpath = True
flatten = False
configuration.update(await option.value.dict(leader_to_list=True, fullpath=fullpath, flatten=flatten))
return configuration

View File

@ -30,14 +30,11 @@ class CallDispatcher:
if response.impl_get_information('multi'): if response.impl_get_information('multi'):
if not isinstance(returns, list): if not isinstance(returns, list):
err = _(f'function {module_name}.{function_name} has to return a list') err = _(f'function {module_name}.{function_name} has to return a list')
await log.error_msg(risotto_context, kwargs, err) raise CallError(err)
raise CallError(str(err))
else: else:
if not isinstance(returns, dict): if not isinstance(returns, dict):
await log.error_msg(risotto_context, kwargs, returns)
err = _(f'function {module_name}.{function_name} has to return a dict') err = _(f'function {module_name}.{function_name} has to return a dict')
await log.error_msg(risotto_context, kwargs, err) raise CallError(err)
raise CallError(str(err))
returns = [returns] returns = [returns]
if response is None: if response is None:
raise Exception('hu?') raise Exception('hu?')
@ -50,12 +47,10 @@ class CallDispatcher:
await config.option(key).value.set(value) await config.option(key).value.set(value)
except AttributeError: except AttributeError:
err = _(f'function {module_name}.{function_name} return the unknown parameter "{key}" for the uri "{risotto_context.version}.{risotto_context.message}"') err = _(f'function {module_name}.{function_name} return the unknown parameter "{key}" for the uri "{risotto_context.version}.{risotto_context.message}"')
await log.error_msg(risotto_context, kwargs, err) raise CallError(err)
raise CallError(str(err)) except ValueError as err:
except ValueError: err = _(f'function {module_name}.{function_name} return the invalid parameter "{key}" for the uri "{risotto_context.version}.{risotto_context.message}": {err}')
err = _(f'function {module_name}.{function_name} return the parameter "{key}" with an unvalid value "{value}" for the uri "{risotto_context.version}.{risotto_context.message}"') raise CallError(err)
await log.error_msg(risotto_context, kwargs, err)
raise CallError(str(err))
await config.property.read_only() await config.property.read_only()
mandatories = await config.value.mandatory() mandatories = await config.value.mandatory()
if mandatories: if mandatories:
@ -65,8 +60,7 @@ class CallDispatcher:
await config.value.dict() await config.value.dict()
except Exception as err: except Exception as err:
err = _(f'function {module_name}.{function_name} return an invalid response {err} for the uri "{risotto_context.version}.{risotto_context.message}"') err = _(f'function {module_name}.{function_name} return an invalid response {err} for the uri "{risotto_context.version}.{risotto_context.message}"')
await log.error_msg(risotto_context, kwargs, err) raise CallError(err)
raise CallError(str(err))
async def call(self, async def call(self,
version: str, version: str,
@ -89,9 +83,17 @@ class CallDispatcher:
if message not in self.messages[version]: if message not in self.messages[version]:
raise CallError(_(f'cannot find message "{version}.{message}"')) raise CallError(_(f'cannot find message "{version}.{message}"'))
function_obj = self.messages[version][message] function_obj = self.messages[version][message]
# do not start a new database connection # log
function_name = function_obj['function'].__name__
info_msg = _(f"call function {function_obj['full_module_name']}.{function_name}")
if hasattr(old_risotto_context, 'connection'): if hasattr(old_risotto_context, 'connection'):
# do not start a new database connection
risotto_context.connection = old_risotto_context.connection risotto_context.connection = old_risotto_context.connection
risotto_context.log_connection = old_risotto_context.log_connection
await log.start(risotto_context,
kwargs,
info_msg,
)
await self.check_message_type(risotto_context, await self.check_message_type(risotto_context,
kwargs, kwargs,
) )
@ -101,53 +103,91 @@ class CallDispatcher:
check_role, check_role,
internal, internal,
) )
return await self.launch(risotto_context,
kwargs,
config_arguments,
function_obj,
)
else:
try: try:
await self.check_message_type(risotto_context, ret = await self.launch(risotto_context,
kwargs, kwargs,
) config_arguments,
config_arguments = await self.load_kwargs_to_config(risotto_context, function_obj,
f'{version}.{message}', )
kwargs, await log.success(risotto_context,
check_role, ret,
internal, )
)
async with self.pool.acquire() as connection:
await connection.set_type_codec(
'json',
encoder=dumps,
decoder=loads,
schema='pg_catalog'
)
risotto_context.connection = connection
async with connection.transaction():
return await self.launch(risotto_context,
kwargs,
config_arguments,
function_obj,
)
except CallError as err:
raise err
except Exception as err: except Exception as err:
# if there is a problem with arguments, just send an error and do nothing await log.failed(risotto_context,
if get_config()['global']['debug']: str(err),
print_exc() )
async with self.pool.acquire() as connection: raise CallError(err) from err
await connection.set_type_codec( else:
'json', error = None
encoder=dumps, async with self.pool.acquire() as log_connection:
decoder=loads, await log_connection.set_type_codec(
schema='pg_catalog' 'json',
) encoder=dumps,
risotto_context.connection = connection decoder=loads,
async with connection.transaction(): schema='pg_catalog'
await log.error_msg(risotto_context, kwargs, err) )
raise err async with log_connection.transaction():
try:
risotto_context.log_connection = log_connection
async with self.pool.acquire() as connection:
await connection.set_type_codec(
'json',
encoder=dumps,
decoder=loads,
schema='pg_catalog'
)
risotto_context.connection = connection
async with connection.transaction():
try:
await log.start(risotto_context,
kwargs,
info_msg,
)
await self.check_message_type(risotto_context,
kwargs,
)
config_arguments = await self.load_kwargs_to_config(risotto_context,
f'{version}.{message}',
kwargs,
check_role,
internal,
)
ret = await self.launch(risotto_context,
kwargs,
config_arguments,
function_obj,
)
# log the success
await log.success(risotto_context,
ret,
)
if not internal and isinstance(ret, dict):
ret['context_id'] = risotto_context.context_id
except CallError as err:
if get_config()['global']['debug']:
print_exc()
await log.failed(risotto_context,
str(err),
)
raise err from err
except CallError as err:
error = err
except Exception as err:
# if there is a problem with arguments, just send an error and do nothing
if get_config()['global']['debug']:
print_exc()
await log.failed(risotto_context,
str(err),
)
error = err
if error:
if not internal:
err = CallError(str(error))
err.context_id = risotto_context.context_id
else:
err = error
raise err from error
return ret
class PublishDispatcher: class PublishDispatcher:
@ -180,6 +220,7 @@ class PublishDispatcher:
remote_kw = dumps({'kwargs': kwargs, remote_kw = dumps({'kwargs': kwargs,
'context': {'username': risotto_context.username, 'context': {'username': risotto_context.username,
'paths': risotto_context.paths, 'paths': risotto_context.paths,
'context_id': risotto_context.context_id,
} }
}) })
# FIXME should be better :/ # FIXME should be better :/
@ -219,12 +260,16 @@ class PublishDispatcher:
False, False,
False, False,
) )
for function_obj in self.messages[version][message]['functions']: async with self.pool.acquire() as log_connection:
async with self.pool.acquire() as connection: await log_connection.set_type_codec(
try: 'json',
await self.check_message_type(risotto_context, encoder=dumps,
kwargs, decoder=loads,
) schema='pg_catalog'
)
async with log_connection.transaction():
risotto_context.log_connection = log_connection
async with self.pool.acquire() as connection:
await connection.set_type_codec( await connection.set_type_codec(
'json', 'json',
encoder=dumps, encoder=dumps,
@ -232,33 +277,56 @@ class PublishDispatcher:
schema='pg_catalog' schema='pg_catalog'
) )
risotto_context.connection = connection risotto_context.connection = connection
async with connection.transaction(): for function_obj in self.messages[version][message]['functions']:
await self.launch(risotto_context, function_name = function_obj['function'].__name__
kwargs, info_msg = _(f"call function {function_obj['full_module_name']}.{function_name}")
config_arguments, try:
function_obj, async with connection.transaction():
) try:
except CallError as err: await log.start(risotto_context,
pass kwargs,
except Exception as err: info_msg,
# if there is a problem with arguments, log and do nothing )
if get_config()['global']['debug']: await self.check_message_type(risotto_context,
print_exc() kwargs,
async with self.pool.acquire() as connection: )
await connection.set_type_codec( await self.launch(risotto_context,
'json', kwargs,
encoder=dumps, config_arguments,
decoder=loads, function_obj,
schema='pg_catalog' )
) # log the success
risotto_context.connection = connection await log.success(risotto_context)
async with connection.transaction(): except CallError as err:
await log.error_msg(risotto_context, kwargs, err) if get_config()['global']['debug']:
print_exc()
await log.failed(risotto_context,
str(err),
)
except CallError:
pass
except Exception as err:
# if there is a problem with arguments, log and do nothing
if get_config()['global']['debug']:
print_exc()
async with self.pool.acquire() as connection:
await connection.set_type_codec(
'json',
encoder=dumps,
decoder=loads,
schema='pg_catalog'
)
risotto_context.connection = connection
async with connection.transaction():
await log.failed(risotto_context,
str(err),
)
class Dispatcher(register.RegisterDispatcher, class Dispatcher(register.RegisterDispatcher,
CallDispatcher, CallDispatcher,
PublishDispatcher): PublishDispatcher,
):
""" Manage message (call or publish) """ Manage message (call or publish)
so launch a function when a message is called so launch a function when a message is called
""" """
@ -274,6 +342,7 @@ class Dispatcher(register.RegisterDispatcher,
risotto_context = Context() risotto_context = Context()
risotto_context.username = context['username'] risotto_context.username = context['username']
risotto_context.paths = copy(context['paths']) risotto_context.paths = copy(context['paths'])
risotto_context.context_id = context['context_id']
risotto_context.paths.append(uri) risotto_context.paths.append(uri)
risotto_context.uri = uri risotto_context.uri = uri
risotto_context.type = type risotto_context.type = type
@ -287,7 +356,6 @@ class Dispatcher(register.RegisterDispatcher,
) -> None: ) -> None:
if self.messages[risotto_context.version][risotto_context.message]['pattern'] != risotto_context.type: if self.messages[risotto_context.version][risotto_context.message]['pattern'] != risotto_context.type:
msg = _(f'{risotto_context.uri} is not a {risotto_context.type} message') msg = _(f'{risotto_context.uri} is not a {risotto_context.type} message')
await log.error_msg(risotto_context, kwargs, msg)
raise CallError(msg) raise CallError(msg)
async def load_kwargs_to_config(self, async def load_kwargs_to_config(self,
@ -333,7 +401,7 @@ class Dispatcher(register.RegisterDispatcher,
parameters = await subconfig.value.dict() parameters = await subconfig.value.dict()
if extra_parameters: if extra_parameters:
parameters.update(extra_parameters) parameters.update(extra_parameters)
return parameters return parameters
def get_service(self, def get_service(self,
name: str): name: str):
@ -342,14 +410,15 @@ class Dispatcher(register.RegisterDispatcher,
async def check_role(self, async def check_role(self,
config: Config, config: Config,
user_login: str, user_login: str,
uri: str) -> None: uri: str,
) -> None:
async with self.pool.acquire() as connection: async with self.pool.acquire() as connection:
async with connection.transaction(): async with connection.transaction():
# Verify if user exists and get ID # Verify if user exists and get ID
sql = ''' sql = '''
SELECT UserId SELECT UserId
FROM UserUser FROM UserUser
WHERE UserLogin = $1 WHERE Login = $1
''' '''
user_id = await connection.fetchval(sql, user_id = await connection.fetchval(sql,
user_login) user_login)
@ -395,8 +464,6 @@ class Dispatcher(register.RegisterDispatcher,
# so send the message # so send the message
function = function_obj['function'] function = function_obj['function']
risotto_context.module = function_obj['module'].split('.', 1)[0] risotto_context.module = function_obj['module'].split('.', 1)[0]
function_name = function.__name__
info_msg = _(f"in function {function_obj['full_module_name']}.{function_name}")
# build argument for this function # build argument for this function
if risotto_context.type == 'rpc': if risotto_context.type == 'rpc':
kw = config_arguments kw = config_arguments
@ -407,7 +474,8 @@ class Dispatcher(register.RegisterDispatcher,
kw[key] = value kw[key] = value
kw['risotto_context'] = risotto_context kw['risotto_context'] = risotto_context
returns = await function(self.injected_self[function_obj['module']], **kw) # launch
returns = await function(self.get_service(function_obj['module']), **kw)
if risotto_context.type == 'rpc': if risotto_context.type == 'rpc':
# valid returns # valid returns
await self.valid_call_returns(risotto_context, await self.valid_call_returns(risotto_context,
@ -415,12 +483,6 @@ class Dispatcher(register.RegisterDispatcher,
returns, returns,
kwargs, kwargs,
) )
# log the success
await log.info_msg(risotto_context,
{'arguments': kwargs,
'returns': returns},
info_msg,
)
# notification # notification
if function_obj.get('notification'): if function_obj.get('notification'):
notif_version, notif_message = function_obj['notification'].split('.', 1) notif_version, notif_message = function_obj['notification'].split('.', 1)

View File

@ -1,4 +1,4 @@
from aiohttp.web import Application, Response, get, post, HTTPBadRequest, HTTPInternalServerError, HTTPNotFound from aiohttp.web import Application, Response, get, post, HTTPBadRequest, HTTPInternalServerError, HTTPNotFound, static
from json import dumps from json import dumps
from traceback import print_exc from traceback import print_exc
try: try:
@ -12,12 +12,13 @@ from .utils import _
from .context import Context from .context import Context
from .error import CallError, NotAllowedError, RegistrationError from .error import CallError, NotAllowedError, RegistrationError
from .message import get_messages from .message import get_messages
from .logger import log #from .logger import log
from .config import get_config from .config import get_config
from . import services from . import services
extra_routes = {} extra_routes = {}
extra_statics = {}
def create_context(request): def create_context(request):
@ -29,19 +30,31 @@ def create_context(request):
def register(version: str, def register(version: str,
path: str): path: str,
):
""" Decorator to register function to the http route """ Decorator to register function to the http route
""" """
def decorator(function): def decorator(function):
if path in extra_routes: if path in extra_routes:
raise RegistrationError(f'the route {path} is already registered') raise RegistrationError(f'the route "{path}" is already registered')
extra_routes[path] = {'function': function, extra_routes[path] = {'function': function,
'version': version} 'version': version,
}
return decorator return decorator
def register_static(path: str,
directory: str,
) -> None:
if path in extra_statics:
raise RegistrationError(f'the static path "{path}" is already registered')
extra_statics[path] = directory
class extra_route_handler: class extra_route_handler:
async def __new__(cls, request): async def __new__(cls,
request,
):
kwargs = dict(request.match_info) kwargs = dict(request.match_info)
kwargs['request'] = request kwargs['request'] = request
kwargs['risotto_context'] = create_context(request) kwargs['risotto_context'] = create_context(request)
@ -67,7 +80,8 @@ class extra_route_handler:
# await log.info_msg(kwargs['risotto_context'], # await log.info_msg(kwargs['risotto_context'],
# dict(request.match_info)) # dict(request.match_info))
return Response(text=dumps(returns), return Response(text=dumps(returns),
content_type='application/json') content_type='application/json',
)
async def handle(request): async def handle(request):
@ -87,20 +101,38 @@ async def handle(request):
internal=False, internal=False,
**kwargs, **kwargs,
) )
except NotAllowedError as err:
raise HTTPNotFound(reason=str(err))
except CallError as err:
raise HTTPBadRequest(reason=str(err).replace('\n', ' '))
except Exception as err: except Exception as err:
if get_config()['global']['debug']: context_id = None
print_exc() if isinstance(err, NotAllowedError):
raise HTTPInternalServerError(reason=str(err)) error_type = HTTPNotFound
return Response(text=dumps({'response': text}), elif isinstance(err, CallError):
content_type='application/json') error_type = HTTPBadRequest
context_id = err.context_id
else:
if get_config()['global']['debug']:
print_exc()
error_type = HTTPInternalServerError
response = {'type': 'error',
'reason': str(err).replace('\n', ' '),
}
if context_id is not None:
response['context_id'] = context_id
err = dumps({'response': response,
'type': 'error',
})
raise error_type(text=err,
content_type='application/json',
)
return Response(text=dumps({'response': text,
'type': 'success',
}),
content_type='application/json',
)
async def api(request, async def api(request,
risotto_context): risotto_context,
):
global TIRAMISU global TIRAMISU
if not TIRAMISU: if not TIRAMISU:
# check all URI that have an associated role # check all URI that have an associated role
@ -130,7 +162,7 @@ async def api(request,
async def get_app(loop): async def get_app(loop):
""" build all routes """ build all routes
""" """
global extra_routes global extra_routes, extra_statics
services.link_to_dispatcher(dispatcher) services.link_to_dispatcher(dispatcher)
app = Application(loop=loop) app = Application(loop=loop)
routes = [] routes = []
@ -152,7 +184,8 @@ async def get_app(loop):
for version in versions: for version in versions:
api_route = {'function': api, api_route = {'function': api,
'version': version, 'version': version,
'path': f'/api/{version}'} 'path': f'/api/{version}',
}
extra_handler = type(api_route['path'], (extra_route_handler,), api_route) extra_handler = type(api_route['path'], (extra_route_handler,), api_route)
routes.append(get(api_route['path'], extra_handler)) routes.append(get(api_route['path'], extra_handler))
print(f' - {api_route["path"]} (http_get)') print(f' - {api_route["path"]} (http_get)')
@ -169,12 +202,22 @@ async def get_app(loop):
extra_handler = type(path, (extra_route_handler,), extra) extra_handler = type(path, (extra_route_handler,), extra)
routes.append(get(path, extra_handler)) routes.append(get(path, extra_handler))
print(f' - {path} (http_get)') print(f' - {path} (http_get)')
if extra_statics:
if not extra_routes:
print(_('======== Registered static routes ========'))
for path, directory in extra_statics.items():
routes.append(static(path, directory))
print(f' - {path} (static)')
del extra_routes del extra_routes
del extra_statics
app.router.add_routes(routes) app.router.add_routes(routes)
await dispatcher.register_remote() await dispatcher.register_remote()
print() print()
await dispatcher.on_join() await dispatcher.on_join()
return await loop.create_server(app.make_handler(), '*', get_config()['http_server']['port']) return await loop.create_server(app.make_handler(),
'*',
get_config()['http_server']['port'],
)
TIRAMISU = None TIRAMISU = None

View File

@ -1,7 +1,7 @@
from typing import Dict, Any from typing import Dict, Any, Optional
from json import dumps from json import dumps, loads
from asyncpg.exceptions import UndefinedTableError from asyncpg.exceptions import UndefinedTableError
from datetime import datetime
from .context import Context from .context import Context
from .utils import _ from .utils import _
@ -13,26 +13,81 @@ class Logger:
""" """
async def insert(self, async def insert(self,
msg: str, msg: str,
path: str, uri: str,
risotto_context: str, uris: str,
risotto_context: Context,
level: str, level: str,
data: Any= None) -> None: data: Any=None,
insert = 'INSERT INTO log(Msg, Path, Username, Level' start: bool=False,
values = 'VALUES($1,$2,$3,$4' ) -> None:
args = [msg, path, risotto_context.username, level] insert = 'INSERT INTO RisottoLog(Msg, URI, URIS, UserLogin, Level'
values = 'VALUES($1,$2,$3,$4,$5'
args = [msg, uri, uris, risotto_context.username, level]
if data: if data:
insert += ', Data' insert += ', Data'
values += ',$5' values += ',$6'
args.append(dumps(data)) args.append(dumps(data))
context_id = risotto_context.context_id
if context_id is not None:
insert += ', ContextId'
if data:
values += ',$7'
else:
values += ',$6'
args.append(context_id)
sql = insert + ') ' + values + ')' sql = insert + ') ' + values + ') RETURNING LogId'
try: try:
await risotto_context.connection.fetch(sql, *args) log_id = await risotto_context.log_connection.fetchval(sql, *args)
if context_id is None and start:
risotto_context.context_id = log_id
if start:
risotto_context.start_id = log_id
except UndefinedTableError as err: except UndefinedTableError as err:
raise Exception(_(f'cannot access to database ({err}), was the database really created?')) raise Exception(_(f'cannot access to database ({err}), was the database really created?'))
async def query(self,
risotto_context: Context,
context_id: int,
uri: Optional[str],
) -> list:
sql = '''SELECT Msg as msg, URI as uri_name, URIS as uris, UserLogin as user_login, Level as level, Data as data, StartDate as start_date, StopDate as stop_date
FROM RisottoLog
WHERE UserLogin = $1 AND (LogId = $2 OR ContextId = $2)
'''
args = [sql, risotto_context.username, context_id]
if uri is not None:
sql += ' AND URI = $3'
args.append(uri)
ret = []
for row in await risotto_context.log_connection.fetch(*args):
d = {}
for key, value in row.items():
if key == 'data':
if isinstance(value, dict):
pass
elif not value:
value = {}
else:
value = loads(value)
elif key in ['start_date', 'stop_date']:
value = str(value)
d[key] = value
ret.append(d)
return ret
def _get_last_uri(self,
risotto_context: Context,
) -> str:
if risotto_context.paths:
return risotto_context.paths[-1]
return ''
def _get_message_paths(self, def _get_message_paths(self,
risotto_context: Context): risotto_context: Context,
) -> str:
if not risotto_context.paths:
return ''
paths = risotto_context.paths paths = risotto_context.paths
if risotto_context.type: if risotto_context.type:
paths_msg = f' {risotto_context.type} ' paths_msg = f' {risotto_context.type} '
@ -49,44 +104,117 @@ class Logger:
risotto_context: Context, risotto_context: Context,
arguments, arguments,
error: str, error: str,
msg: str=''): msg: str='',
):
""" send message when an error append """ send message when an error append
""" """
paths_msg = self._get_message_paths(risotto_context) paths_msg = self._get_message_paths(risotto_context)
print(_(f'{risotto_context.username}: ERROR: {error} ({paths_msg} with arguments "{arguments}": {msg})')) print(_(f'{risotto_context.username}: ERROR: {error} ({paths_msg} with arguments "{arguments}": {msg})'))
await self.insert(msg, await self.insert(msg,
self._get_last_uri(risotto_context),
paths_msg, paths_msg,
risotto_context, risotto_context,
'Error', 'Error',
arguments) arguments,
)
async def info_msg(self, async def info_msg(self,
risotto_context: Context, risotto_context: Context,
arguments: Dict, arguments: Dict,
msg: str=''): msg: str='',
) -> None:
""" send message with common information """ send message with common information
""" """
if risotto_context.paths: paths_msg = self._get_message_paths(risotto_context)
paths_msg = self._get_message_paths(risotto_context)
else:
paths_msg = ''
if get_config()['global']['debug']: if get_config()['global']['debug']:
print(_(f'{risotto_context.username}: INFO:{paths_msg}: {msg}')) print(_(f'{risotto_context.username}: INFO:{paths_msg}: {msg}'))
await self.insert(msg, await self.insert(msg,
self._get_last_uri(risotto_context),
paths_msg, paths_msg,
risotto_context, risotto_context,
'Info', 'Info',
arguments) arguments,
)
async def start(self,
risotto_context: Context,
arguments: dict,
msg: str,
) -> None:
paths_msg = self._get_message_paths(risotto_context)
if get_config()['global']['debug']:
if risotto_context.context_id != None:
context = f'({risotto_context.context_id})'
else:
context = ''
print(_(f'{risotto_context.username}: START{context}:{paths_msg}: {msg}'))
await self.insert(msg,
self._get_last_uri(risotto_context),
paths_msg,
risotto_context,
'Start',
arguments,
start=True,
)
async def success(self,
risotto_context: Context,
returns: Optional[dict]=None,
) -> None:
if get_config()['global']['debug']:
paths_msg = self._get_message_paths(risotto_context)
print(_(f'{risotto_context.username}: SUCCESS({risotto_context.context_id}):{paths_msg}'))
sql = """UPDATE RisottoLog
SET StopDate = $2,
Level = 'SUCCESS'
"""
args = [datetime.now()]
if returns:
sql += """, Returns = $3
"""
args.append(dumps(returns))
sql += """WHERE LogId = $1
"""
await risotto_context.log_connection.execute(sql,
risotto_context.start_id,
*args,
)
async def failed(self,
risotto_context: Context,
err: str,
) -> None:
if get_config()['global']['debug']:
paths_msg = self._get_message_paths(risotto_context)
if risotto_context.context_id != None:
context = f'({risotto_context.context_id})'
else:
context = ''
print(_(f'{risotto_context.username}: FAILED({risotto_context.context_id}):{paths_msg}: err'))
sql = """UPDATE RisottoLog
SET StopDate = $2,
Level = 'FAILED',
Msg = $3
WHERE LogId = $1
"""
await risotto_context.log_connection.execute(sql,
risotto_context.start_id,
datetime.now(),
err,
)
async def info(self, async def info(self,
risotto_context, risotto_context,
msg): msg,
):
if get_config()['global']['debug']: if get_config()['global']['debug']:
print(msg) print(msg)
await self.insert(msg, await self.insert(msg,
'',
None, None,
risotto_context, risotto_context,
'Info') 'Info',
)
log = Logger() log = Logger()

View File

@ -7,6 +7,7 @@ from typing import Callable, Optional, List
from asyncpg import create_pool from asyncpg import create_pool
from json import dumps, loads from json import dumps, loads
from pkg_resources import iter_entry_points from pkg_resources import iter_entry_points
from traceback import print_exc
import risotto import risotto
from .utils import _ from .utils import _
from .error import RegistrationError from .error import RegistrationError
@ -23,7 +24,7 @@ class Services():
def load_services(self): def load_services(self):
for entry_point in iter_entry_points(group='risotto_services'): for entry_point in iter_entry_points(group='risotto_services'):
self.services.setdefault(entry_point.name, []) self.services.setdefault(entry_point.name, {})
self.services_loaded = True self.services_loaded = True
def load_modules(self, def load_modules(self,
@ -32,21 +33,20 @@ class Services():
for entry_point in iter_entry_points(group='risotto_modules'): for entry_point in iter_entry_points(group='risotto_modules'):
service_name, module_name = entry_point.name.split('.') service_name, module_name = entry_point.name.split('.')
if limit_services is None or service_name in limit_services: if limit_services is None or service_name in limit_services:
setattr(self, module_name, entry_point.load()) self.services[service_name][module_name] = entry_point.load()
self.services[service_name].append(module_name)
self.modules_loaded = True self.modules_loaded = True
#
def get_services(self): # def get_services(self):
if not self.services_loaded: # if not self.services_loaded:
self.load_services() # self.load_services()
return [(s, getattr(self, s)) for s in self.services] # return [(service, getattr(self, service)) for service in self.services]
def get_modules(self, def get_modules(self,
limit_services: Optional[List[str]]=None, limit_services: Optional[List[str]]=None,
) -> List[str]: ) -> List[str]:
if not self.modules_loaded: if not self.modules_loaded:
self.load_modules(limit_services=limit_services) self.load_modules(limit_services=limit_services)
return [(module + '.' + submodule, getattr(self, submodule)) for module, submodules in self.services.items() for submodule in submodules] return [(module + '.' + submodule, entry_point) for module, submodules in self.services.items() for submodule, entry_point in submodules.items()]
def get_services_list(self): def get_services_list(self):
return self.services.keys() return self.services.keys()
@ -84,11 +84,11 @@ def register(uris: str,
def decorator(function): def decorator(function):
for uri in uris: for uri in uris:
version, message = uri.split('.', 1) dispatcher.set_function(uri,
dispatcher.set_function(version,
message,
notification, notification,
function) function,
function.__module__
)
return decorator return decorator
@ -185,21 +185,20 @@ class RegisterDispatcher:
raise RegistrationError(_(f'error with {module_name}.{function_name} arguments: {msg}')) raise RegistrationError(_(f'error with {module_name}.{function_name} arguments: {msg}'))
def set_function(self, def set_function(self,
version: str, uri: str,
message: str,
notification: str, notification: str,
function: Callable, function: Callable,
full_module_name: str,
): ):
""" register a function to an URI """ register a function to an URI
URI is a message URI is a message
""" """
version, message = uri.split('.', 1)
# check if message exists # check if message exists
if message not in self.messages[version]: if message not in self.messages[version]:
raise RegistrationError(_(f'the message {message} not exists')) raise RegistrationError(_(f'the message {message} not exists'))
# xxx submodule can only be register with v1.yyy.xxx..... message # xxx submodule can only be register with v1.yyy.xxx..... message
full_module_name = function.__module__
risotto_module_name, submodule_name = full_module_name.split('.')[-3:-1] risotto_module_name, submodule_name = full_module_name.split('.')[-3:-1]
module_name = risotto_module_name.split('_')[-1] module_name = risotto_module_name.split('_')[-1]
message_module, message_submodule, message_name = message.split('.', 2) message_module, message_submodule, message_name = message.split('.', 2)
@ -215,7 +214,7 @@ class RegisterDispatcher:
# check if already register # check if already register
if 'function' in self.messages[version][message]: if 'function' in self.messages[version][message]:
raise RegistrationError(_(f'uri {version}.{message} already registered')) raise RegistrationError(_(f'uri {uri} already registered'))
# register # register
if self.messages[version][message]['pattern'] == 'rpc': if self.messages[version][message]['pattern'] == 'rpc':
@ -278,7 +277,7 @@ class RegisterDispatcher:
try: try:
self.injected_self[submodule_name] = module.Risotto(test) self.injected_self[submodule_name] = module.Risotto(test)
except AttributeError as err: except AttributeError as err:
raise RegistrationError(_(f'unable to register the module {submodule_name}, this module must have Risotto class')) print(_(f'unable to register the module {submodule_name}, this module must have Risotto class'))
def validate(self): def validate(self):
""" check if all messages have a function """ check if all messages have a function
@ -288,9 +287,9 @@ class RegisterDispatcher:
for message, message_obj in messages.items(): for message, message_obj in messages.items():
if not 'functions' in message_obj and not 'function' in message_obj: if not 'functions' in message_obj and not 'function' in message_obj:
if message_obj['pattern'] == 'event': if message_obj['pattern'] == 'event':
print(f'{message} prêche dans le désert') print(f'{version}.{message} prêche dans le désert')
else: else:
missing_messages.append(message) missing_messages.append(f'{version}.{message}')
if missing_messages: if missing_messages:
raise RegistrationError(_(f'no matching function for uri {missing_messages}')) raise RegistrationError(_(f'no matching function for uri {missing_messages}'))
@ -298,29 +297,38 @@ class RegisterDispatcher:
truncate: bool=False, truncate: bool=False,
) -> None: ) -> None:
internal_user = get_config()['global']['internal_user'] internal_user = get_config()['global']['internal_user']
async with self.pool.acquire() as connection: async with self.pool.acquire() as log_connection:
await connection.set_type_codec( async with log_connection.transaction():
'json', async with self.pool.acquire() as connection:
encoder=dumps, await connection.set_type_codec(
decoder=loads, 'json',
schema='pg_catalog' encoder=dumps,
) decoder=loads,
if truncate: schema='pg_catalog'
async with connection.transaction(): )
await connection.execute('TRUNCATE InfraServer, InfraSite, InfraZone, Log, ProviderDeployment, ProviderFactoryCluster, ProviderFactoryClusterNode, SettingApplicationservice, SettingApplicationServiceDependency, SettingRelease, SettingServer, SettingServermodel, SettingSource, UserRole, UserRoleURI, UserURI, UserUser, InfraServermodel, ProviderZone, ProviderServer, ProviderSource, ProviderApplicationservice, ProviderServermodel') if truncate:
async with connection.transaction(): async with connection.transaction():
for submodule_name, module in self.injected_self.items(): await connection.execute('TRUNCATE InfraServer, InfraSite, InfraZone, Log, ProviderDeployment, ProviderFactoryCluster, ProviderFactoryClusterNode, SettingApplicationservice, SettingApplicationServiceDependency, SettingRelease, SettingServer, SettingServermodel, SettingSource, UserRole, UserRoleURI, UserURI, UserUser, InfraServermodel, ProviderZone, ProviderServer, ProviderSource, ProviderApplicationservice, ProviderServermodel')
risotto_context = Context() async with connection.transaction():
risotto_context.username = internal_user for submodule_name, module in self.injected_self.items():
risotto_context.paths.append(f'internal.{submodule_name}.on_join') risotto_context = Context()
risotto_context.type = None risotto_context.username = internal_user
risotto_context.connection = connection risotto_context.paths.append(f'internal.{submodule_name}.on_join')
risotto_context.module = submodule_name.split('.', 1)[0] risotto_context.type = None
info_msg = _(f'in function risotto_{submodule_name}.on_join') risotto_context.log_connection = log_connection
await log.info_msg(risotto_context, risotto_context.connection = connection
None, risotto_context.module = submodule_name.split('.', 1)[0]
info_msg) info_msg = _(f'in function risotto_{submodule_name}.on_join')
await module.on_join(risotto_context) await log.info_msg(risotto_context,
None,
info_msg)
try:
await module.on_join(risotto_context)
except Exception as err:
if get_config()['global']['debug']:
print_exc()
msg = _(f'on_join returns an error in module {submodule_name}: {err}')
await log.error_msg(risotto_context, {}, msg)
async def load(self): async def load(self):
# valid function's arguments # valid function's arguments

View File

@ -1,9 +1,27 @@
class Undefined: class Undefined:
pass pass
undefined = Undefined()
def _(s): def _(s):
return s return s
undefined = Undefined() def tiramisu_display_name(kls,
dyn_name: 'Base'=None,
suffix: str=None,
) -> str:
if dyn_name is not None:
name = dyn_name
else:
name = kls.impl_getname()
doc = kls.impl_get_information('doc', None)
if doc:
doc = str(doc)
if doc.endswith('.'):
doc = doc[:-1]
if suffix:
doc += suffix
if name != doc:
name += f' ({doc})'
return name

View File

@ -392,7 +392,6 @@ async def test_server_created_base():
release_distribution='last', release_distribution='last',
site_name='site_1', site_name='site_1',
zones_name=['zones'], zones_name=['zones'],
zones_ip=['1.1.1.1'],
) )
assert list(config_module.server) == [server_name] assert list(config_module.server) == [server_name]
assert set(config_module.server[server_name]) == {'server', 'server_to_deploy', 'funcs_file'} assert set(config_module.server[server_name]) == {'server', 'server_to_deploy', 'funcs_file'}
@ -420,7 +419,6 @@ async def test_server_created_own_sm():
release_distribution='last', release_distribution='last',
site_name='site_1', site_name='site_1',
zones_name=['zones'], zones_name=['zones'],
zones_ip=['1.1.1.1'],
) )
assert list(config_module.server) == [server_name] assert list(config_module.server) == [server_name]
assert set(config_module.server[server_name]) == {'server', 'server_to_deploy', 'funcs_file'} assert set(config_module.server[server_name]) == {'server', 'server_to_deploy', 'funcs_file'}
@ -469,7 +467,6 @@ async def test_server_configuration_get():
release_distribution='last', release_distribution='last',
site_name='site_1', site_name='site_1',
zones_name=['zones'], zones_name=['zones'],
zones_ip=['1.1.1.1'],
) )
# #
await config_module.server[server_name]['server'].property.read_write() await config_module.server[server_name]['server'].property.read_write()
@ -515,7 +512,6 @@ async def test_server_configuration_deployed():
release_distribution='last', release_distribution='last',
site_name='site_1', site_name='site_1',
zones_name=['zones'], zones_name=['zones'],
zones_ip=['1.1.1.1'],
) )
# #
await config_module.server[server_name]['server'].property.read_write() await config_module.server[server_name]['server'].property.read_write()