Compare commits
14 Commits
52209a5ebd
...
pkg/dev/ri
Author | SHA1 | Date | |
---|---|---|---|
9423cdef8d | |||
ef43b197a1 | |||
fc5a13152b | |||
94b6563d8f | |||
617ef55dd3 | |||
88c2c168ac | |||
8c5a2bb702 | |||
09cd0a4e4c | |||
92846052b1 | |||
3085bf67d6 | |||
c63170be1d | |||
1063d2e735 | |||
2a98575790 | |||
ed51bc483d |
@ -1,12 +1,12 @@
|
|||||||
CREATE TABLE RisottoLog(
|
CREATE TABLE RisottoLog(
|
||||||
LogId SERIAL PRIMARY KEY,
|
LogId SERIAL PRIMARY KEY,
|
||||||
|
ContextId INTEGER,
|
||||||
Msg VARCHAR(255) NOT NULL,
|
Msg VARCHAR(255) NOT NULL,
|
||||||
URI VARCHAR(255),
|
URI VARCHAR(255),
|
||||||
URIS VARCHAR(255),
|
URIS VARCHAR(255),
|
||||||
UserLogin VARCHAR(100) NOT NULL,
|
UserLogin VARCHAR(100) NOT NULL,
|
||||||
Level VARCHAR(10) NOT NULL,
|
Status INTEGER NOT NULL,
|
||||||
ContextId INTEGER,
|
Kwargs JSON,
|
||||||
Data JSON,
|
|
||||||
Returns JSON,
|
Returns JSON,
|
||||||
StartDate timestamp DEFAULT current_timestamp,
|
StartDate timestamp DEFAULT current_timestamp,
|
||||||
StopDate timestamp
|
StopDate timestamp
|
||||||
|
@ -65,6 +65,18 @@ if 'CELERYRISOTTO_DB_USER' in environ:
|
|||||||
CELERYRISOTTO_DB_USER = environ['CELERYRISOTTO_DB_USER']
|
CELERYRISOTTO_DB_USER = environ['CELERYRISOTTO_DB_USER']
|
||||||
else:
|
else:
|
||||||
CELERYRISOTTO_DB_USER = config.get('CELERYRISOTTO_DB_USER', None)
|
CELERYRISOTTO_DB_USER = config.get('CELERYRISOTTO_DB_USER', None)
|
||||||
|
if 'LEMUR_DB_NAME' in environ:
|
||||||
|
LEMUR_DB_NAME = environ['LEMUR_DB_NAME']
|
||||||
|
else:
|
||||||
|
LEMUR_DB_NAME = config.get('LEMUR_DB_NAME', None)
|
||||||
|
if 'LEMUR_DB_PASSWORD' in environ:
|
||||||
|
LEMUR_DB_PASSWORD = environ['LEMUR_DB_PASSWORD']
|
||||||
|
else:
|
||||||
|
LEMUR_DB_PASSWORD = config.get('LEMUR_DB_PASSWORD', None)
|
||||||
|
if 'LEMUR_DB_USER' in environ:
|
||||||
|
LEMUR_DB_USER = environ['LEMUR_DB_USER']
|
||||||
|
else:
|
||||||
|
LEMUR_DB_USER = config.get('LEMUR_DB_USER', None)
|
||||||
if 'DB_ADDRESS' in environ:
|
if 'DB_ADDRESS' in environ:
|
||||||
DB_ADDRESS = environ['DB_ADDRESS']
|
DB_ADDRESS = environ['DB_ADDRESS']
|
||||||
else:
|
else:
|
||||||
@ -115,6 +127,11 @@ if 'PASSWORD_URL' in environ:
|
|||||||
PASSWORD_URL = environ['PASSWORD_URL']
|
PASSWORD_URL = environ['PASSWORD_URL']
|
||||||
else:
|
else:
|
||||||
PASSWORD_URL = config.get('PASSWORD_URL', 'https://localhost:8001/')
|
PASSWORD_URL = config.get('PASSWORD_URL', 'https://localhost:8001/')
|
||||||
|
|
||||||
|
if 'PASSWORD_LENGTH' in environ:
|
||||||
|
PASSWORD_LENGTH = int(environ['PASSWORD_LENGTH'])
|
||||||
|
else:
|
||||||
|
PASSWORD_LENGTH = int(config.get('PASSWORD_LENGTH', 20))
|
||||||
if 'PKI_ADMIN_PASSWORD' in environ:
|
if 'PKI_ADMIN_PASSWORD' in environ:
|
||||||
PKI_ADMIN_PASSWORD = environ['PKI_ADMIN_PASSWORD']
|
PKI_ADMIN_PASSWORD = environ['PKI_ADMIN_PASSWORD']
|
||||||
else:
|
else:
|
||||||
@ -136,7 +153,8 @@ def dsn_factory(database, user, password, address=DB_ADDRESS):
|
|||||||
|
|
||||||
_config = {'database': {'dsn': dsn_factory(RISOTTO_DB_NAME, RISOTTO_DB_USER, RISOTTO_DB_PASSWORD),
|
_config = {'database': {'dsn': dsn_factory(RISOTTO_DB_NAME, RISOTTO_DB_USER, RISOTTO_DB_PASSWORD),
|
||||||
'tiramisu_dsn': dsn_factory(TIRAMISU_DB_NAME, TIRAMISU_DB_USER, TIRAMISU_DB_PASSWORD),
|
'tiramisu_dsn': dsn_factory(TIRAMISU_DB_NAME, TIRAMISU_DB_USER, TIRAMISU_DB_PASSWORD),
|
||||||
'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),
|
||||||
|
'lemur_dns': dsn_factory(LEMUR_DB_NAME, LEMUR_DB_USER, LEMUR_DB_PASSWORD),
|
||||||
},
|
},
|
||||||
'http_server': {'port': RISOTTO_PORT,
|
'http_server': {'port': RISOTTO_PORT,
|
||||||
'default_user': DEFAULT_USER,
|
'default_user': DEFAULT_USER,
|
||||||
@ -155,6 +173,7 @@ _config = {'database': {'dsn': dsn_factory(RISOTTO_DB_NAME, RISOTTO_DB_USER, RIS
|
|||||||
'admin_password': PASSWORD_ADMIN_PASSWORD,
|
'admin_password': PASSWORD_ADMIN_PASSWORD,
|
||||||
'device_identifier': PASSWORD_DEVICE_IDENTIFIER,
|
'device_identifier': PASSWORD_DEVICE_IDENTIFIER,
|
||||||
'service_url': PASSWORD_URL,
|
'service_url': PASSWORD_URL,
|
||||||
|
'length': PASSWORD_LENGTH,
|
||||||
},
|
},
|
||||||
'pki': {'admin_password': PKI_ADMIN_PASSWORD,
|
'pki': {'admin_password': PKI_ADMIN_PASSWORD,
|
||||||
'owner': PKI_ADMIN_EMAIL,
|
'owner': PKI_ADMIN_EMAIL,
|
||||||
|
@ -3,3 +3,11 @@ class Context:
|
|||||||
self.paths = []
|
self.paths = []
|
||||||
self.context_id = None
|
self.context_id = None
|
||||||
self.start_id = None
|
self.start_id = None
|
||||||
|
|
||||||
|
def copy(self):
|
||||||
|
context = Context()
|
||||||
|
for key, value in self.__dict__.items():
|
||||||
|
if key.startswith('__'):
|
||||||
|
continue
|
||||||
|
setattr(context, key, value)
|
||||||
|
return context
|
||||||
|
@ -309,6 +309,7 @@ except:
|
|||||||
await config_std.property.read_only()
|
await config_std.property.read_only()
|
||||||
|
|
||||||
# copy informations from 'to deploy' configuration to configuration
|
# copy informations from 'to deploy' configuration to configuration
|
||||||
|
await config.information.importation(await config_std.information.exportation())
|
||||||
await config.value.importation(await config_std.value.exportation())
|
await config.value.importation(await config_std.value.exportation())
|
||||||
await config.permissive.importation(await config_std.permissive.exportation())
|
await config.permissive.importation(await config_std.permissive.exportation())
|
||||||
await config.property.importation(await config_std.property.exportation())
|
await config.property.importation(await config_std.property.exportation())
|
||||||
|
@ -42,14 +42,21 @@ class CallDispatcher:
|
|||||||
for ret in returns:
|
for ret in returns:
|
||||||
async with await Config(response, display_name=lambda self, dyn_name, suffix: self.impl_getname()) as config:
|
async with await Config(response, display_name=lambda self, dyn_name, suffix: self.impl_getname()) as config:
|
||||||
await config.property.read_write()
|
await config.property.read_write()
|
||||||
|
key = None
|
||||||
try:
|
try:
|
||||||
for key, value in ret.items():
|
for key, value in ret.items():
|
||||||
await config.option(key).value.set(value)
|
await config.option(key).value.set(value)
|
||||||
except AttributeError:
|
except AttributeError as err:
|
||||||
err = _(f'function {module_name}.{function_name} return the unknown parameter "{key}" for the uri "{risotto_context.version}.{risotto_context.message}"')
|
if key is not None:
|
||||||
|
err = _(f'function {module_name}.{function_name} return the unknown parameter "{key}" for the uri "{risotto_context.version}.{risotto_context.message}"')
|
||||||
|
else:
|
||||||
|
err = _(f'function {module_name}.{function_name} return unconsistency data "{err}" for the uri "{risotto_context.version}.{risotto_context.message}"')
|
||||||
raise CallError(err)
|
raise CallError(err)
|
||||||
except ValueError as err:
|
except ValueError as err:
|
||||||
err = _(f'function {module_name}.{function_name} return the invalid parameter "{key}" for the uri "{risotto_context.version}.{risotto_context.message}": {err}')
|
if key is not None:
|
||||||
|
err = _(f'function {module_name}.{function_name} return the invalid parameter "{key}" for the uri "{risotto_context.version}.{risotto_context.message}": {err}')
|
||||||
|
else:
|
||||||
|
err = _(f'function {module_name}.{function_name} return unconsistency error for the uri "{risotto_context.version}.{risotto_context.message}": {err}')
|
||||||
raise CallError(err)
|
raise CallError(err)
|
||||||
await config.property.read_only()
|
await config.property.read_only()
|
||||||
mandatories = await config.value.mandatory()
|
mandatories = await config.value.mandatory()
|
||||||
@ -89,7 +96,6 @@ class CallDispatcher:
|
|||||||
if hasattr(old_risotto_context, 'connection'):
|
if hasattr(old_risotto_context, 'connection'):
|
||||||
# do not start a new database 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,
|
await log.start(risotto_context,
|
||||||
kwargs,
|
kwargs,
|
||||||
info_msg,
|
info_msg,
|
||||||
@ -119,67 +125,58 @@ class CallDispatcher:
|
|||||||
raise CallError(err) from err
|
raise CallError(err) from err
|
||||||
else:
|
else:
|
||||||
error = None
|
error = None
|
||||||
async with self.pool.acquire() as log_connection:
|
try:
|
||||||
await log_connection.set_type_codec(
|
async with self.pool.acquire() as connection:
|
||||||
'json',
|
await connection.set_type_codec(
|
||||||
encoder=dumps,
|
'json',
|
||||||
decoder=loads,
|
encoder=dumps,
|
||||||
schema='pg_catalog'
|
decoder=loads,
|
||||||
)
|
schema='pg_catalog'
|
||||||
async with log_connection.transaction():
|
)
|
||||||
try:
|
risotto_context.connection = connection
|
||||||
risotto_context.log_connection = log_connection
|
async with connection.transaction():
|
||||||
async with self.pool.acquire() as connection:
|
try:
|
||||||
await connection.set_type_codec(
|
await log.start(risotto_context,
|
||||||
'json',
|
kwargs,
|
||||||
encoder=dumps,
|
info_msg,
|
||||||
decoder=loads,
|
)
|
||||||
schema='pg_catalog'
|
await self.check_message_type(risotto_context,
|
||||||
)
|
kwargs,
|
||||||
risotto_context.connection = connection
|
)
|
||||||
async with connection.transaction():
|
config_arguments = await self.load_kwargs_to_config(risotto_context,
|
||||||
try:
|
f'{version}.{message}',
|
||||||
await log.start(risotto_context,
|
kwargs,
|
||||||
|
check_role,
|
||||||
|
internal,
|
||||||
|
)
|
||||||
|
ret = await self.launch(risotto_context,
|
||||||
kwargs,
|
kwargs,
|
||||||
info_msg,
|
config_arguments,
|
||||||
|
function_obj,
|
||||||
)
|
)
|
||||||
await self.check_message_type(risotto_context,
|
# log the success
|
||||||
kwargs,
|
await log.success(risotto_context,
|
||||||
)
|
ret,
|
||||||
config_arguments = await self.load_kwargs_to_config(risotto_context,
|
)
|
||||||
f'{version}.{message}',
|
if not internal and isinstance(ret, dict):
|
||||||
kwargs,
|
ret['context_id'] = risotto_context.context_id
|
||||||
check_role,
|
except CallError as err:
|
||||||
internal,
|
if get_config()['global']['debug']:
|
||||||
)
|
print_exc()
|
||||||
ret = await self.launch(risotto_context,
|
await log.failed(risotto_context,
|
||||||
kwargs,
|
str(err),
|
||||||
config_arguments,
|
)
|
||||||
function_obj,
|
raise err from err
|
||||||
)
|
except CallError as err:
|
||||||
# log the success
|
error = err
|
||||||
await log.success(risotto_context,
|
except Exception as err:
|
||||||
ret,
|
# if there is a problem with arguments, just send an error and do nothing
|
||||||
)
|
if get_config()['global']['debug']:
|
||||||
if not internal and isinstance(ret, dict):
|
print_exc()
|
||||||
ret['context_id'] = risotto_context.context_id
|
await log.failed(risotto_context,
|
||||||
except CallError as err:
|
str(err),
|
||||||
if get_config()['global']['debug']:
|
)
|
||||||
print_exc()
|
error = err
|
||||||
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 error:
|
||||||
if not internal:
|
if not internal:
|
||||||
err = CallError(str(error))
|
err = CallError(str(error))
|
||||||
@ -199,8 +196,6 @@ class PublishDispatcher:
|
|||||||
for message, message_infos in messages.items():
|
for message, message_infos in messages.items():
|
||||||
# event not emit locally
|
# event not emit locally
|
||||||
if message_infos['pattern'] == 'event' and 'functions' in message_infos and message_infos['functions']:
|
if message_infos['pattern'] == 'event' and 'functions' in message_infos and message_infos['functions']:
|
||||||
# module, submodule, submessage = message.split('.', 2)
|
|
||||||
# if f'{module}.{submodule}' not in self.injected_self:
|
|
||||||
uri = f'{version}.{message}'
|
uri = f'{version}.{message}'
|
||||||
print(f' - {uri}')
|
print(f' - {uri}')
|
||||||
await self.listened_connection.add_listener(uri,
|
await self.listened_connection.add_listener(uri,
|
||||||
@ -236,21 +231,34 @@ class PublishDispatcher:
|
|||||||
version, message = uri.split('.', 1)
|
version, message = uri.split('.', 1)
|
||||||
loop = get_event_loop()
|
loop = get_event_loop()
|
||||||
remote_kw = loads(payload)
|
remote_kw = loads(payload)
|
||||||
risotto_context = self.build_new_context(remote_kw['context'],
|
for function_obj in self.messages[version][message]['functions']:
|
||||||
version,
|
risotto_context = self.build_new_context(remote_kw['context'],
|
||||||
message,
|
version,
|
||||||
'event',
|
message,
|
||||||
)
|
'event',
|
||||||
callback = lambda: ensure_future(self._publish(version,
|
)
|
||||||
message,
|
callback = self.get_callback(version, message, function_obj, risotto_context, remote_kw['kwargs'],)
|
||||||
risotto_context,
|
loop.call_soon(callback)
|
||||||
**remote_kw['kwargs'],
|
|
||||||
))
|
def get_callback(self,
|
||||||
loop.call_soon(callback)
|
version,
|
||||||
|
message,
|
||||||
|
function_obj,
|
||||||
|
risotto_context,
|
||||||
|
kwargs,
|
||||||
|
):
|
||||||
|
return lambda: ensure_future(self._publish(version,
|
||||||
|
message,
|
||||||
|
function_obj,
|
||||||
|
risotto_context,
|
||||||
|
**kwargs,
|
||||||
|
))
|
||||||
|
|
||||||
|
|
||||||
async def _publish(self,
|
async def _publish(self,
|
||||||
version: str,
|
version: str,
|
||||||
message: str,
|
message: str,
|
||||||
|
function_obj,
|
||||||
risotto_context: Context,
|
risotto_context: Context,
|
||||||
**kwargs,
|
**kwargs,
|
||||||
) -> None:
|
) -> None:
|
||||||
@ -260,67 +268,48 @@ class PublishDispatcher:
|
|||||||
False,
|
False,
|
||||||
False,
|
False,
|
||||||
)
|
)
|
||||||
async with self.pool.acquire() as log_connection:
|
async with self.pool.acquire() as connection:
|
||||||
await log_connection.set_type_codec(
|
await connection.set_type_codec(
|
||||||
'json',
|
'json',
|
||||||
encoder=dumps,
|
encoder=dumps,
|
||||||
decoder=loads,
|
decoder=loads,
|
||||||
schema='pg_catalog'
|
schema='pg_catalog'
|
||||||
)
|
)
|
||||||
async with log_connection.transaction():
|
risotto_context.connection = connection
|
||||||
risotto_context.log_connection = log_connection
|
function_name = function_obj['function'].__name__
|
||||||
async with self.pool.acquire() as connection:
|
info_msg = _(f"call function {function_obj['full_module_name']}.{function_name}")
|
||||||
await connection.set_type_codec(
|
try:
|
||||||
'json',
|
async with connection.transaction():
|
||||||
encoder=dumps,
|
try:
|
||||||
decoder=loads,
|
await log.start(risotto_context,
|
||||||
schema='pg_catalog'
|
kwargs,
|
||||||
)
|
info_msg,
|
||||||
risotto_context.connection = connection
|
)
|
||||||
for function_obj in self.messages[version][message]['functions']:
|
await self.check_message_type(risotto_context,
|
||||||
function_name = function_obj['function'].__name__
|
|
||||||
info_msg = _(f"call function {function_obj['full_module_name']}.{function_name}")
|
|
||||||
try:
|
|
||||||
async with connection.transaction():
|
|
||||||
try:
|
|
||||||
await log.start(risotto_context,
|
|
||||||
kwargs,
|
|
||||||
info_msg,
|
|
||||||
)
|
|
||||||
await self.check_message_type(risotto_context,
|
|
||||||
kwargs,
|
|
||||||
)
|
|
||||||
await self.launch(risotto_context,
|
|
||||||
kwargs,
|
kwargs,
|
||||||
config_arguments,
|
|
||||||
function_obj,
|
|
||||||
)
|
)
|
||||||
# log the success
|
await self.launch(risotto_context,
|
||||||
await log.success(risotto_context)
|
kwargs,
|
||||||
except CallError as err:
|
config_arguments,
|
||||||
if get_config()['global']['debug']:
|
function_obj,
|
||||||
print_exc()
|
)
|
||||||
await log.failed(risotto_context,
|
# log the success
|
||||||
str(err),
|
await log.success(risotto_context)
|
||||||
)
|
except CallError as err:
|
||||||
except CallError:
|
if get_config()['global']['debug']:
|
||||||
pass
|
print_exc()
|
||||||
except Exception as err:
|
await log.failed(risotto_context,
|
||||||
# if there is a problem with arguments, log and do nothing
|
str(err),
|
||||||
if get_config()['global']['debug']:
|
)
|
||||||
print_exc()
|
except CallError:
|
||||||
async with self.pool.acquire() as connection:
|
pass
|
||||||
await connection.set_type_codec(
|
except Exception as err:
|
||||||
'json',
|
# if there is a problem with arguments, log and do nothing
|
||||||
encoder=dumps,
|
if get_config()['global']['debug']:
|
||||||
decoder=loads,
|
print_exc()
|
||||||
schema='pg_catalog'
|
await log.failed(risotto_context,
|
||||||
)
|
str(err),
|
||||||
risotto_context.connection = connection
|
)
|
||||||
async with connection.transaction():
|
|
||||||
await log.failed(risotto_context,
|
|
||||||
str(err),
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class Dispatcher(register.RegisterDispatcher,
|
class Dispatcher(register.RegisterDispatcher,
|
||||||
@ -348,6 +337,7 @@ class Dispatcher(register.RegisterDispatcher,
|
|||||||
risotto_context.type = type
|
risotto_context.type = type
|
||||||
risotto_context.message = message
|
risotto_context.message = message
|
||||||
risotto_context.version = version
|
risotto_context.version = version
|
||||||
|
risotto_context.pool = self.pool
|
||||||
return risotto_context
|
return risotto_context
|
||||||
|
|
||||||
async def check_message_type(self,
|
async def check_message_type(self,
|
||||||
@ -485,6 +475,8 @@ class Dispatcher(register.RegisterDispatcher,
|
|||||||
)
|
)
|
||||||
# notification
|
# notification
|
||||||
if function_obj.get('notification'):
|
if function_obj.get('notification'):
|
||||||
|
if returns is None:
|
||||||
|
raise Exception(_(f'function "{function_obj["full_module_name"]}.{function_obj["function"].__name__}" must returns something for {function_obj["notification"]}!'))
|
||||||
notif_version, notif_message = function_obj['notification'].split('.', 1)
|
notif_version, notif_message = function_obj['notification'].split('.', 1)
|
||||||
if not isinstance(returns, list):
|
if not isinstance(returns, list):
|
||||||
send_returns = [returns]
|
send_returns = [returns]
|
||||||
|
@ -23,9 +23,13 @@ extra_statics = {}
|
|||||||
|
|
||||||
def create_context(request):
|
def create_context(request):
|
||||||
risotto_context = Context()
|
risotto_context = Context()
|
||||||
risotto_context.username = request.match_info.get('username',
|
if 'username' in dict(request.match_info):
|
||||||
get_config()['http_server']['default_user'],
|
username = request.match_info['username']
|
||||||
)
|
elif 'username' in request.headers:
|
||||||
|
username = request.headers['username']
|
||||||
|
else:
|
||||||
|
username = get_config()['http_server']['default_user']
|
||||||
|
risotto_context.username = username
|
||||||
return risotto_context
|
return risotto_context
|
||||||
|
|
||||||
|
|
||||||
|
@ -2,35 +2,57 @@ from typing import Dict, Any, Optional
|
|||||||
from json import dumps, loads
|
from json import dumps, loads
|
||||||
from asyncpg.exceptions import UndefinedTableError
|
from asyncpg.exceptions import UndefinedTableError
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
from asyncio import Lock
|
||||||
|
|
||||||
from .context import Context
|
from .context import Context
|
||||||
from .utils import _
|
from .utils import _
|
||||||
from .config import get_config
|
from .config import get_config
|
||||||
|
|
||||||
|
database_lock = Lock()
|
||||||
|
|
||||||
|
|
||||||
|
LEVELS = ['Error', 'Info', 'Success', 'Started', 'Failure']
|
||||||
|
|
||||||
|
|
||||||
class Logger:
|
class Logger:
|
||||||
""" An object to manager log
|
""" An object to manager log
|
||||||
"""
|
"""
|
||||||
|
def __init__(self) -> None:
|
||||||
|
self.log_connection = None
|
||||||
|
|
||||||
|
async def get_connection(self,
|
||||||
|
risotto_context: Context,
|
||||||
|
):
|
||||||
|
if not self.log_connection:
|
||||||
|
self.log_connection = await risotto_context.pool.acquire()
|
||||||
|
await self.log_connection.set_type_codec(
|
||||||
|
'json',
|
||||||
|
encoder=dumps,
|
||||||
|
decoder=loads,
|
||||||
|
schema='pg_catalog'
|
||||||
|
)
|
||||||
|
return self.log_connection
|
||||||
|
|
||||||
async def insert(self,
|
async def insert(self,
|
||||||
msg: str,
|
msg: str,
|
||||||
uri: str,
|
|
||||||
uris: str,
|
|
||||||
risotto_context: Context,
|
risotto_context: Context,
|
||||||
level: str,
|
level: str,
|
||||||
data: Any=None,
|
kwargs: Any=None,
|
||||||
start: bool=False,
|
start: bool=False,
|
||||||
) -> None:
|
) -> None:
|
||||||
insert = 'INSERT INTO RisottoLog(Msg, URI, URIS, UserLogin, Level'
|
uri = self._get_last_uri(risotto_context)
|
||||||
|
uris = " ".join(risotto_context.paths)
|
||||||
|
insert = 'INSERT INTO RisottoLog(Msg, URI, URIS, UserLogin, Status'
|
||||||
values = 'VALUES($1,$2,$3,$4,$5'
|
values = 'VALUES($1,$2,$3,$4,$5'
|
||||||
args = [msg, uri, uris, risotto_context.username, level]
|
args = [msg, uri, uris, risotto_context.username, LEVELS.index(level)]
|
||||||
if data:
|
if kwargs:
|
||||||
insert += ', Data'
|
insert += ', Kwargs'
|
||||||
values += ',$6'
|
values += ',$6'
|
||||||
args.append(dumps(data))
|
args.append(dumps(kwargs))
|
||||||
context_id = risotto_context.context_id
|
context_id = risotto_context.context_id
|
||||||
if context_id is not None:
|
if context_id is not None:
|
||||||
insert += ', ContextId'
|
insert += ', ContextId'
|
||||||
if data:
|
if kwargs:
|
||||||
values += ',$7'
|
values += ',$7'
|
||||||
else:
|
else:
|
||||||
values += ',$6'
|
values += ',$6'
|
||||||
@ -38,7 +60,9 @@ class Logger:
|
|||||||
|
|
||||||
sql = insert + ') ' + values + ') RETURNING LogId'
|
sql = insert + ') ' + values + ') RETURNING LogId'
|
||||||
try:
|
try:
|
||||||
log_id = await risotto_context.log_connection.fetchval(sql, *args)
|
async with database_lock:
|
||||||
|
connection = await self.get_connection(risotto_context)
|
||||||
|
log_id = await connection.fetchval(sql, *args)
|
||||||
if context_id is None and start:
|
if context_id is None and start:
|
||||||
risotto_context.context_id = log_id
|
risotto_context.context_id = log_id
|
||||||
if start:
|
if start:
|
||||||
@ -46,36 +70,6 @@ class Logger:
|
|||||||
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,
|
def _get_last_uri(self,
|
||||||
risotto_context: Context,
|
risotto_context: Context,
|
||||||
) -> str:
|
) -> str:
|
||||||
@ -111,8 +105,6 @@ class Logger:
|
|||||||
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,
|
|
||||||
risotto_context,
|
risotto_context,
|
||||||
'Error',
|
'Error',
|
||||||
arguments,
|
arguments,
|
||||||
@ -129,8 +121,6 @@ class Logger:
|
|||||||
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,
|
|
||||||
risotto_context,
|
risotto_context,
|
||||||
'Info',
|
'Info',
|
||||||
arguments,
|
arguments,
|
||||||
@ -149,10 +139,8 @@ class Logger:
|
|||||||
context = ''
|
context = ''
|
||||||
print(_(f'{risotto_context.username}: START{context}:{paths_msg}: {msg}'))
|
print(_(f'{risotto_context.username}: START{context}:{paths_msg}: {msg}'))
|
||||||
await self.insert(msg,
|
await self.insert(msg,
|
||||||
self._get_last_uri(risotto_context),
|
|
||||||
paths_msg,
|
|
||||||
risotto_context,
|
risotto_context,
|
||||||
'Start',
|
'Started',
|
||||||
arguments,
|
arguments,
|
||||||
start=True,
|
start=True,
|
||||||
)
|
)
|
||||||
@ -166,19 +154,21 @@ class Logger:
|
|||||||
print(_(f'{risotto_context.username}: SUCCESS({risotto_context.context_id}):{paths_msg}'))
|
print(_(f'{risotto_context.username}: SUCCESS({risotto_context.context_id}):{paths_msg}'))
|
||||||
sql = """UPDATE RisottoLog
|
sql = """UPDATE RisottoLog
|
||||||
SET StopDate = $2,
|
SET StopDate = $2,
|
||||||
Level = 'SUCCESS'
|
Status = $3
|
||||||
"""
|
"""
|
||||||
args = [datetime.now()]
|
args = [datetime.now(), LEVELS.index('Success')]
|
||||||
if returns:
|
if returns:
|
||||||
sql += """, Returns = $3
|
sql += """, Returns = $4
|
||||||
"""
|
"""
|
||||||
args.append(dumps(returns))
|
args.append(dumps(returns))
|
||||||
sql += """WHERE LogId = $1
|
sql += """WHERE LogId = $1
|
||||||
"""
|
"""
|
||||||
await risotto_context.log_connection.execute(sql,
|
async with database_lock:
|
||||||
risotto_context.start_id,
|
connection = await self.get_connection(risotto_context)
|
||||||
*args,
|
await connection.execute(sql,
|
||||||
)
|
risotto_context.start_id,
|
||||||
|
*args,
|
||||||
|
)
|
||||||
|
|
||||||
async def failed(self,
|
async def failed(self,
|
||||||
risotto_context: Context,
|
risotto_context: Context,
|
||||||
@ -190,18 +180,21 @@ class Logger:
|
|||||||
context = f'({risotto_context.context_id})'
|
context = f'({risotto_context.context_id})'
|
||||||
else:
|
else:
|
||||||
context = ''
|
context = ''
|
||||||
print(_(f'{risotto_context.username}: FAILED({risotto_context.context_id}):{paths_msg}: err'))
|
print(_(f'{risotto_context.username}: FAILED({risotto_context.context_id}):{paths_msg}: {err}'))
|
||||||
sql = """UPDATE RisottoLog
|
sql = """UPDATE RisottoLog
|
||||||
SET StopDate = $2,
|
SET StopDate = $2,
|
||||||
Level = 'FAILED',
|
Status = $4,
|
||||||
Msg = $3
|
Msg = $3
|
||||||
WHERE LogId = $1
|
WHERE LogId = $1
|
||||||
"""
|
"""
|
||||||
await risotto_context.log_connection.execute(sql,
|
async with database_lock:
|
||||||
risotto_context.start_id,
|
connection = await self.get_connection(risotto_context)
|
||||||
datetime.now(),
|
await connection.execute(sql,
|
||||||
err,
|
risotto_context.start_id,
|
||||||
)
|
datetime.now(),
|
||||||
|
err[:254],
|
||||||
|
LEVELS.index('Failure'),
|
||||||
|
)
|
||||||
|
|
||||||
async def info(self,
|
async def info(self,
|
||||||
risotto_context,
|
risotto_context,
|
||||||
@ -210,8 +203,6 @@ class Logger:
|
|||||||
if get_config()['global']['debug']:
|
if get_config()['global']['debug']:
|
||||||
print(msg)
|
print(msg)
|
||||||
await self.insert(msg,
|
await self.insert(msg,
|
||||||
'',
|
|
||||||
None,
|
|
||||||
risotto_context,
|
risotto_context,
|
||||||
'Info',
|
'Info',
|
||||||
)
|
)
|
||||||
|
@ -313,6 +313,7 @@ class CustomParam:
|
|||||||
'string': 'String',
|
'string': 'String',
|
||||||
'number': 'Number',
|
'number': 'Number',
|
||||||
'object': 'Dict',
|
'object': 'Dict',
|
||||||
|
'any': 'Any',
|
||||||
'array': 'Array',
|
'array': 'Array',
|
||||||
'file': 'File',
|
'file': 'File',
|
||||||
'float': 'Float'}
|
'float': 'Float'}
|
||||||
@ -448,6 +449,7 @@ def _get_option(name,
|
|||||||
'reverse_condition': ParamValue(True)}),
|
'reverse_condition': ParamValue(True)}),
|
||||||
calc_value_property_help))
|
calc_value_property_help))
|
||||||
|
|
||||||
|
props.append('notunique')
|
||||||
description = arg.description.strip().rstrip()
|
description = arg.description.strip().rstrip()
|
||||||
kwargs = {'name': name,
|
kwargs = {'name': name,
|
||||||
'doc': _get_description(description, name),
|
'doc': _get_description(description, name),
|
||||||
@ -523,6 +525,7 @@ def _parse_responses(message_def,
|
|||||||
'Number': IntOption,
|
'Number': IntOption,
|
||||||
'Boolean': BoolOption,
|
'Boolean': BoolOption,
|
||||||
'Dict': DictOption,
|
'Dict': DictOption,
|
||||||
|
'Any': AnyOption,
|
||||||
'Float': FloatOption,
|
'Float': FloatOption,
|
||||||
# FIXME
|
# FIXME
|
||||||
'File': StrOption}.get(type_)
|
'File': StrOption}.get(type_)
|
||||||
@ -530,8 +533,9 @@ def _parse_responses(message_def,
|
|||||||
raise Exception(f'unknown param type {obj.type} in responses of message {message_def.message}')
|
raise Exception(f'unknown param type {obj.type} in responses of message {message_def.message}')
|
||||||
if hasattr(obj, 'default'):
|
if hasattr(obj, 'default'):
|
||||||
kwargs['default'] = obj.default
|
kwargs['default'] = obj.default
|
||||||
|
kwargs['properties'] = ('notunique',)
|
||||||
else:
|
else:
|
||||||
kwargs['properties'] = ('mandatory',)
|
kwargs['properties'] = ('mandatory', 'notunique')
|
||||||
options.append(option(**kwargs))
|
options.append(option(**kwargs))
|
||||||
od = OptionDescription(uri,
|
od = OptionDescription(uri,
|
||||||
message_def.response.description,
|
message_def.response.description,
|
||||||
@ -600,7 +604,7 @@ def get_messages(current_module_names,
|
|||||||
select_option = ChoiceOption('message',
|
select_option = ChoiceOption('message',
|
||||||
'Nom du message.',
|
'Nom du message.',
|
||||||
tuple(messages),
|
tuple(messages),
|
||||||
properties=frozenset(['mandatory', 'positional']))
|
properties=frozenset(['mandatory', 'positional', 'notunique']))
|
||||||
for uri in messages:
|
for uri in messages:
|
||||||
message_def = get_message(uri,
|
message_def = get_message(uri,
|
||||||
current_module_names,
|
current_module_names,
|
||||||
|
@ -107,6 +107,7 @@ class RegisterDispatcher:
|
|||||||
version = obj['version']
|
version = obj['version']
|
||||||
if version not in self.messages:
|
if version not in self.messages:
|
||||||
self.messages[version] = {}
|
self.messages[version] = {}
|
||||||
|
obj['message'] = tiramisu_message
|
||||||
self.messages[version][tiramisu_message] = obj
|
self.messages[version][tiramisu_message] = obj
|
||||||
|
|
||||||
def get_function_args(self,
|
def get_function_args(self,
|
||||||
@ -297,38 +298,36 @@ 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 log_connection:
|
async with self.pool.acquire() as connection:
|
||||||
async with log_connection.transaction():
|
await connection.set_type_codec(
|
||||||
async with self.pool.acquire() as connection:
|
'json',
|
||||||
await connection.set_type_codec(
|
encoder=dumps,
|
||||||
'json',
|
decoder=loads,
|
||||||
encoder=dumps,
|
schema='pg_catalog'
|
||||||
decoder=loads,
|
)
|
||||||
schema='pg_catalog'
|
if truncate:
|
||||||
)
|
async with connection.transaction():
|
||||||
if truncate:
|
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')
|
||||||
async with connection.transaction():
|
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')
|
for submodule_name, module in self.injected_self.items():
|
||||||
async with connection.transaction():
|
risotto_context = Context()
|
||||||
for submodule_name, module in self.injected_self.items():
|
risotto_context.username = internal_user
|
||||||
risotto_context = Context()
|
risotto_context.paths.append(f'internal.{submodule_name}.on_join')
|
||||||
risotto_context.username = internal_user
|
risotto_context.type = None
|
||||||
risotto_context.paths.append(f'internal.{submodule_name}.on_join')
|
risotto_context.pool = self.pool
|
||||||
risotto_context.type = None
|
risotto_context.connection = connection
|
||||||
risotto_context.log_connection = log_connection
|
risotto_context.module = submodule_name.split('.', 1)[0]
|
||||||
risotto_context.connection = connection
|
info_msg = _(f'in function risotto_{submodule_name}.on_join')
|
||||||
risotto_context.module = submodule_name.split('.', 1)[0]
|
await log.info_msg(risotto_context,
|
||||||
info_msg = _(f'in function risotto_{submodule_name}.on_join')
|
None,
|
||||||
await log.info_msg(risotto_context,
|
info_msg)
|
||||||
None,
|
try:
|
||||||
info_msg)
|
await module.on_join(risotto_context)
|
||||||
try:
|
except Exception as err:
|
||||||
await module.on_join(risotto_context)
|
if get_config()['global']['debug']:
|
||||||
except Exception as err:
|
print_exc()
|
||||||
if get_config()['global']['debug']:
|
msg = _(f'on_join returns an error in module {submodule_name}: {err}')
|
||||||
print_exc()
|
await log.error_msg(risotto_context, {}, msg)
|
||||||
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
|
||||||
|
@ -23,5 +23,5 @@ def tiramisu_display_name(kls,
|
|||||||
if suffix:
|
if suffix:
|
||||||
doc += suffix
|
doc += suffix
|
||||||
if name != doc:
|
if name != doc:
|
||||||
name += f' ({doc})'
|
name += f'" "{doc}'
|
||||||
return name
|
return name
|
||||||
|
Reference in New Issue
Block a user