497 lines
23 KiB
Python
497 lines
23 KiB
Python
try:
|
|
from tiramisu3 import Config
|
|
from tiramisu3.error import ValueOptionError
|
|
except:
|
|
from tiramisu import Config
|
|
from tiramisu.error import ValueOptionError
|
|
from asyncio import get_event_loop, ensure_future
|
|
from traceback import print_exc
|
|
from copy import copy
|
|
from typing import Dict, Callable, List, Optional
|
|
from json import dumps, loads
|
|
|
|
from .utils import _
|
|
from .error import CallError, NotAllowedError
|
|
from .logger import log
|
|
from .config import get_config
|
|
from .context import Context
|
|
from . import register
|
|
|
|
|
|
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 = 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')
|
|
raise CallError(err)
|
|
else:
|
|
if not isinstance(returns, dict):
|
|
err = _(f'function {module_name}.{function_name} has to return a dict')
|
|
raise CallError(err)
|
|
returns = [returns]
|
|
if response is None:
|
|
raise Exception('hu?')
|
|
else:
|
|
for ret in returns:
|
|
async with await Config(response, display_name=lambda self, dyn_name, suffix: self.impl_getname()) as config:
|
|
await config.property.read_write()
|
|
key = None
|
|
try:
|
|
for key, value in ret.items():
|
|
await config.option(key).value.set(value)
|
|
except AttributeError as err:
|
|
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)
|
|
except ValueError as 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)
|
|
await config.property.read_only()
|
|
mandatories = await config.value.mandatory()
|
|
if mandatories:
|
|
mand = [mand.split('.')[-1] for mand in mandatories]
|
|
raise ValueError(_(f'missing parameters in response of the uri "{risotto_context.version}.{risotto_context.message}": {mand} in message'))
|
|
try:
|
|
await config.value.dict()
|
|
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}"')
|
|
raise CallError(err)
|
|
|
|
async def call(self,
|
|
version: str,
|
|
message: str,
|
|
old_risotto_context: Context,
|
|
check_role: bool=False,
|
|
internal: bool=True,
|
|
**kwargs,
|
|
):
|
|
""" execute the function associate with specified uri
|
|
arguments are validate before
|
|
"""
|
|
risotto_context = self.build_new_context(old_risotto_context.__dict__,
|
|
version,
|
|
message,
|
|
'rpc',
|
|
)
|
|
if version not in self.messages:
|
|
raise CallError(_(f'cannot find version of message "{version}"'))
|
|
if message not in self.messages[version]:
|
|
raise CallError(_(f'cannot find message "{version}.{message}"'))
|
|
function_obj = self.messages[version][message]
|
|
# 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'):
|
|
# do not start a new database connection
|
|
risotto_context.connection = old_risotto_context.connection
|
|
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,
|
|
)
|
|
try:
|
|
ret = await self.launch(risotto_context,
|
|
kwargs,
|
|
config_arguments,
|
|
function_obj,
|
|
)
|
|
await log.success(risotto_context,
|
|
ret,
|
|
)
|
|
except Exception as err:
|
|
await log.failed(risotto_context,
|
|
str(err),
|
|
)
|
|
raise CallError(err) from err
|
|
else:
|
|
error = None
|
|
try:
|
|
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:
|
|
async def register_remote(self) -> None:
|
|
print()
|
|
print(_('======== Registered remote event ========'))
|
|
self.listened_connection = await self.pool.acquire()
|
|
for version, messages in self.messages.items():
|
|
for message, message_infos in messages.items():
|
|
# event not emit locally
|
|
if message_infos['pattern'] == 'event' and 'functions' in message_infos and message_infos['functions']:
|
|
uri = f'{version}.{message}'
|
|
print(f' - {uri}')
|
|
await self.listened_connection.add_listener(uri,
|
|
self.to_async_publish,
|
|
)
|
|
|
|
async def publish(self,
|
|
version: str,
|
|
message: str,
|
|
risotto_context: Context,
|
|
**kwargs,
|
|
) -> None:
|
|
if version not in self.messages or message not in self.messages[version]:
|
|
raise ValueError(_(f'cannot find URI "{version}.{message}"'))
|
|
|
|
# publish to remote
|
|
remote_kw = dumps({'kwargs': kwargs,
|
|
'context': {'username': risotto_context.username,
|
|
'paths': risotto_context.paths,
|
|
'context_id': risotto_context.context_id,
|
|
}
|
|
})
|
|
# FIXME should be better :/
|
|
remote_kw = remote_kw.replace("'", "''")
|
|
await risotto_context.connection.execute(f'NOTIFY "{version}.{message}", \'{remote_kw}\'')
|
|
|
|
def to_async_publish(self,
|
|
con: 'asyncpg.connection.Connection',
|
|
pid: int,
|
|
uri: str,
|
|
payload: str,
|
|
) -> None:
|
|
version, message = uri.split('.', 1)
|
|
loop = get_event_loop()
|
|
remote_kw = loads(payload)
|
|
for function_obj in self.messages[version][message]['functions']:
|
|
risotto_context = self.build_new_context(remote_kw['context'],
|
|
version,
|
|
message,
|
|
'event',
|
|
)
|
|
callback = self.get_callback(version, message, function_obj, risotto_context, remote_kw['kwargs'],)
|
|
loop.call_soon(callback)
|
|
|
|
def get_callback(self,
|
|
version,
|
|
message,
|
|
function_obj,
|
|
risotto_context,
|
|
kwargs,
|
|
):
|
|
return lambda: ensure_future(self._publish(version,
|
|
message,
|
|
function_obj,
|
|
risotto_context,
|
|
**kwargs,
|
|
))
|
|
|
|
|
|
async def _publish(self,
|
|
version: str,
|
|
message: str,
|
|
function_obj,
|
|
risotto_context: Context,
|
|
**kwargs,
|
|
) -> None:
|
|
config_arguments = await self.load_kwargs_to_config(risotto_context,
|
|
f'{version}.{message}',
|
|
kwargs,
|
|
False,
|
|
False,
|
|
)
|
|
async with self.pool.acquire() as connection:
|
|
await connection.set_type_codec(
|
|
'json',
|
|
encoder=dumps,
|
|
decoder=loads,
|
|
schema='pg_catalog'
|
|
)
|
|
risotto_context.connection = connection
|
|
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,
|
|
config_arguments,
|
|
function_obj,
|
|
)
|
|
# log the success
|
|
await log.success(risotto_context)
|
|
except CallError as 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()
|
|
await log.failed(risotto_context,
|
|
str(err),
|
|
)
|
|
|
|
|
|
class Dispatcher(register.RegisterDispatcher,
|
|
CallDispatcher,
|
|
PublishDispatcher,
|
|
):
|
|
""" Manage message (call or publish)
|
|
so launch a function when a message is called
|
|
"""
|
|
def build_new_context(self,
|
|
context: dict,
|
|
version: str,
|
|
message: str,
|
|
type: str,
|
|
) -> Context:
|
|
""" This is a new call or a new publish, so create a new context
|
|
"""
|
|
uri = version + '.' + message
|
|
risotto_context = Context()
|
|
risotto_context.username = context['username']
|
|
risotto_context.paths = copy(context['paths'])
|
|
risotto_context.context_id = context['context_id']
|
|
risotto_context.paths.append(uri)
|
|
risotto_context.uri = uri
|
|
risotto_context.type = type
|
|
risotto_context.message = message
|
|
risotto_context.version = version
|
|
risotto_context.pool = self.pool
|
|
return risotto_context
|
|
|
|
async def check_message_type(self,
|
|
risotto_context: Context,
|
|
kwargs: Dict,
|
|
) -> None:
|
|
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')
|
|
raise CallError(msg)
|
|
|
|
async def load_kwargs_to_config(self,
|
|
risotto_context: Context,
|
|
uri: str,
|
|
kwargs: Dict,
|
|
check_role: bool,
|
|
internal: bool,
|
|
):
|
|
""" create a new Config et set values to it
|
|
"""
|
|
# create a new config
|
|
async with await Config(self.option) as config:
|
|
await config.property.read_write()
|
|
# set message's option
|
|
await config.option('message').value.set(uri)
|
|
# store values
|
|
subconfig = config.option(uri)
|
|
extra_parameters = {}
|
|
for key, value in kwargs.items():
|
|
if not internal or not key.startswith('_'):
|
|
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}'))
|
|
else:
|
|
extra_parameters[key] = value
|
|
# check mandatories options
|
|
if check_role and get_config().get('global').get('check_role'):
|
|
await self.check_role(subconfig,
|
|
risotto_context.username,
|
|
uri)
|
|
await config.property.read_only()
|
|
mandatories = await config.value.mandatory()
|
|
if mandatories:
|
|
mand = [mand.split('.')[-1] for mand in mandatories]
|
|
raise ValueError(_(f'missing parameters in "{uri}": {mand}'))
|
|
# return complete an validated kwargs
|
|
parameters = await subconfig.value.dict()
|
|
if extra_parameters:
|
|
parameters.update(extra_parameters)
|
|
return parameters
|
|
|
|
def get_service(self,
|
|
name: str):
|
|
return self.injected_self[name]
|
|
|
|
async def check_role(self,
|
|
config: Config,
|
|
user_login: str,
|
|
uri: str,
|
|
) -> None:
|
|
async with self.pool.acquire() as connection:
|
|
async with connection.transaction():
|
|
# Verify if user exists and get ID
|
|
sql = '''
|
|
SELECT UserId
|
|
FROM UserUser
|
|
WHERE Login = $1
|
|
'''
|
|
user_id = await connection.fetchval(sql,
|
|
user_login)
|
|
if user_id is None:
|
|
raise NotAllowedError(_(f"You ({user_login}) don't have any account in this application"))
|
|
|
|
# Get all references for this message
|
|
refs = {}
|
|
for option in await config.list('all'):
|
|
ref = await option.information.get('ref')
|
|
if ref:
|
|
refs[ref] = str(await option.value.get())
|
|
|
|
# Check role
|
|
select_role_uri = '''
|
|
SELECT RoleName
|
|
FROM UserURI, UserRoleURI
|
|
WHERE UserURI.URIName = $1 AND UserRoleURI.URIId = UserURI.URIId
|
|
'''
|
|
select_role_user = '''
|
|
SELECT RoleAttribute, RoleAttributeValue
|
|
FROM UserRole
|
|
WHERE RoleUserId = $1 AND RoleName = $2
|
|
'''
|
|
for uri_role in await connection.fetch(select_role_uri,
|
|
uri):
|
|
for user_role in await connection.fetch(select_role_user,
|
|
user_id,
|
|
uri_role['rolename']):
|
|
if not user_role['roleattribute']:
|
|
return
|
|
if user_role['roleattribute'] in refs and \
|
|
user_role['roleattributevalue'] == refs[user_role['roleattribute']]:
|
|
return
|
|
raise NotAllowedError(_(f'You ({user_login}) don\'t have any authorisation to access to "{uri}"'))
|
|
|
|
async def launch(self,
|
|
risotto_context: Context,
|
|
kwargs: Dict,
|
|
config_arguments: dict,
|
|
function_obj: Callable,
|
|
) -> Optional[Dict]:
|
|
# so send the message
|
|
function = function_obj['function']
|
|
risotto_context.module = function_obj['module'].split('.', 1)[0]
|
|
# 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
|
|
|
|
kw['risotto_context'] = risotto_context
|
|
# launch
|
|
returns = await function(self.get_service(function_obj['module']), **kw)
|
|
if risotto_context.type == 'rpc':
|
|
# valid returns
|
|
await self.valid_call_returns(risotto_context,
|
|
function,
|
|
returns,
|
|
kwargs,
|
|
)
|
|
# 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)
|
|
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
|