Infra/risotto
9
0
Fork 1
Code Issues 1 Releases Wiki Activity

insert log in database

Browse Source
This commit is contained in:
Emmanuel Garette
2019-12-28 12:29:11 +01:00
parent a6383f0c2c
commit 4020f97db0
6 changed files with 292 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files

277
src/risotto/dispatcher.py
View File

@ -1,7 +1,7 @@
from tiramisu import Config
from traceback import print_exc
from copy import copy
from typing import Dict, Callable
from typing import Dict, Callable, List, Optional
from json import dumps, loads
from .utils import _
@ -17,21 +17,22 @@ import asyncpg
class CallDispatcher:
async def valid_call_returns(self,
risotto_context: Context,
function,
returns: Dict,
kwargs: Dict):
response = self.messages[risotto_context.version][risotto_context.message]['response']
module_name = risotto_context.function.__module__.split('.')[-2]
function_name = risotto_context.function.__name__
module_name = function.__module__.split('.')[-2]
function_name = function.__name__
if response.impl_get_information('multi'):
if not isinstance(returns, list):
err = _(f'function {module_name}.{function_name} has to return a list')
log.error_msg(risotto_context, kwargs, err)
await log.error_msg(risotto_context, kwargs, err)
raise CallError(str(err))
else:
if not isinstance(returns, dict):
log.error_msg(risotto_context, kwargs, returns)
await log.error_msg(risotto_context, kwargs, returns)
err = _(f'function {module_name}.{function_name} has to return a dict')
log.error_msg(risotto_context, kwargs, err)
await log.error_msg(risotto_context, kwargs, err)
raise CallError(str(err))
returns = [returns]
if response is None:
@ -46,11 +47,11 @@ class CallDispatcher:
await config.option(key).value.set(value)
except AttributeError:
err = _(f'function {module_name}.{function_name} return the unknown parameter "{key}"')
log.error_msg(risotto_context, kwargs, err)
await log.error_msg(risotto_context, kwargs, err)
raise CallError(str(err))
except ValueError:
err = _(f'function {module_name}.{function_name} return the parameter "{key}" with an unvalid value "{value}"')
log.error_msg(risotto_context, kwargs, err)
await log.error_msg(risotto_context, kwargs, err)
raise CallError(str(err))
await config.property.read_only()
mandatories = await config.value.mandatory()
@ -61,7 +62,7 @@ class CallDispatcher:
await config.value.dict()
except Exception as err:
err = _(f'function {module_name}.{function_name} return an invalid response {err}')
log.error_msg(risotto_context, kwargs, err)
await log.error_msg(risotto_context, kwargs, err)
raise CallError(str(err))
async def call(self,
@ -77,63 +78,32 @@ class CallDispatcher:
version,
message,
'rpc')
self.check_message_type(risotto_context,
kwargs)
try:
kw = await self.load_kwargs_to_config(risotto_context,
f'{version}.{message}',
kwargs,
check_role)
function_obj = self.messages[version][message]
risotto_context.function = function_obj['function']
if function_obj['risotto_context']:
kw['risotto_context'] = risotto_context
# do not start a new database connection
if function_obj['database'] and hasattr(old_risotto_context, 'connection'):
risotto_context.connection = old_risotto_context.connection
if function_obj['database'] and not hasattr(risotto_context, '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():
returns = await risotto_context.function(self.injected_self[function_obj['module']], **kw)
else:
returns = await risotto_context.function(self.injected_self[function_obj['module']], **kw)
except CallError as err:
raise err
except Exception as err:
if get_config().get('global').get('debug'):
print_exc()
log.error_msg(risotto_context,
kwargs,
err)
raise CallError(str(err))
# valid returns
await self.valid_call_returns(risotto_context,
returns,
kwargs)
# log the success
log.info_msg(risotto_context,
kwargs,
_(f'returns {returns}'))
# notification
if function_obj.get('notification'):
notif_version, notif_message = function_obj['notification'].split('.', 1)
if not isinstance(returns, list):
send_returns = [returns]
else:
send_returns = returns
for ret in send_returns:
await self.publish(notif_version,
notif_message,
risotto_context,
**ret)
return returns
function_objs = [self.messages[version][message]]
# do not start a new database connection
if hasattr(old_risotto_context, 'connection'):
risotto_context.connection = old_risotto_context.connection
return await self.launch(version,
message,
risotto_context,
check_role,
kwargs,
function_objs)
else:
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(version,
message,
risotto_context,
check_role,
kwargs,
function_objs)
class PublishDispatcher:
@ -147,66 +117,32 @@ class PublishDispatcher:
version,
message,
'event')
self.check_message_type(risotto_context,
kwargs)
try:
config_arguments = await self.load_kwargs_to_config(risotto_context,
f'{version}.{message}',
kwargs,
check_role)
except CallError as err:
return
except Exception as err:
# if there is a problem with arguments, just send an error et do nothing
if DEBUG:
print_exc()
log.error_msg(risotto_context, kwargs, err)
return
# config is ok, so publish the message
for function_obj in self.messages[version][message].get('functions', []):
function = function_obj['function']
module_name = function.__module__.split('.')[-2]
function_name = function.__name__
info_msg = _(f'in module {module_name}.{function_name}')
try:
# build argument for this function
kw = {}
for key, value in config_arguments.items():
if key in function_obj['arguments']:
kw[key] = value
if function_obj['risotto_context']:
kw['risotto_context'] = risotto_context
# send event
if function_obj['database'] and hasattr(old_risotto_context, 'connection'):
risotto_context.connection = old_risotto_context.connection
if function_obj['database'] and not hasattr(risotto_context, '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():
returns = await function(self.injected_self[function_obj['module']], **kw)
else:
returns = await function(self.injected_self[function_obj['module']], **kw)
except Exception as err:
if DEBUG:
print_exc()
log.error_msg(risotto_context, kwargs, err, info_msg)
continue
else:
log.info_msg(risotto_context, kwargs, info_msg)
# notification
if function_obj.get('notification'):
notif_version, notif_message = function_obj['notification'].split('.', 1)
await self.publish(notif_version,
notif_message,
risotto_context,
**returns)
function_objs = self.messages[version][message].get('functions', [])
# do not start a new database connection
if hasattr(old_risotto_context, 'connection'):
risotto_context.connection = old_risotto_context.connection
return await self.launch(version,
message,
risotto_context,
check_role,
kwargs,
function_objs)
else:
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(version,
message,
risotto_context,
check_role,
kwargs,
function_objs)
class Dispatcher(register.RegisterDispatcher, CallDispatcher, PublishDispatcher):
@ -231,12 +167,12 @@ class Dispatcher(register.RegisterDispatcher, CallDispatcher, PublishDispatcher)
risotto_context.version = version
return risotto_context
def check_message_type(self,
risotto_context: Context,
kwargs: Dict):
async def check_message_type(self,
risotto_context: Context,
kwargs: Dict):
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')
log.error_msg(risotto_context, kwargs, msg)
await log.error_msg(risotto_context, kwargs, msg)
raise CallError(msg)
async def load_kwargs_to_config(self,
@ -324,6 +260,87 @@ class Dispatcher(register.RegisterDispatcher, CallDispatcher, PublishDispatcher)
return
raise NotAllowedError(_(f'You ({user_login}) don\'t have any authorisation to access to "{uri}"'))
async def launch(self,
version: str,
message: str,
risotto_context: Context,
check_role: bool,
kwargs: Dict,
function_objs: List) -> Optional[Dict]:
await self.check_message_type(risotto_context,
kwargs)
try:
config_arguments = await self.load_kwargs_to_config(risotto_context,
f'{version}.{message}',
kwargs,
check_role)
except Exception as err:
# if there is a problem with arguments, just send an error et do nothing
if DEBUG:
print_exc()
await log.error_msg(risotto_context, kwargs, err)
return
# config is ok, so send the message
for function_obj in function_objs:
function = function_obj['function']
module_name = function.__module__.split('.')[-2]
function_name = function.__name__
info_msg = _(f'in module {module_name}.{function_name}')
try:
# build argument for this function
if risotto_context.type == 'rpc':
kw = config_arguments
else:
kw = {}
for key, value in config_arguments.items():
if key in function_obj['arguments']:
kw[key] = value
if function_obj['risotto_context']:
kw['risotto_context'] = risotto_context
returns = await function(self.injected_self[function_obj['module']], **kw)
except CallError as err:
if risotto_context.type == 'rpc':
raise err
continue
except Exception as err:
if get_config().get('global').get('debug'):
print_exc()
await log.error_msg(risotto_context,
kwargs,
err)
if risotto_context.type == 'rpc':
raise CallError(str(err))
continue
else:
if risotto_context.type == 'rpc':
# valid returns
await self.valid_call_returns(risotto_context,
function,
returns,
kwargs)
# log the success
await log.info_msg(risotto_context,
{'arguments': kwargs,
'returns': returns},
info_msg)
# notification
if function_obj.get('notification'):
notif_version, notif_message = function_obj['notification'].split('.', 1)
if not isinstance(returns, list):
send_returns = [returns]
else:
send_returns = returns
for ret in send_returns:
await self.publish(notif_version,
notif_message,
risotto_context,
**ret)
if risotto_context.type == 'rpc':
return returns
dispatcher = Dispatcher()
register.dispatcher = dispatcher