Merge branch 'develop' into dist/risotto/risotto-2.8.0/develop
This commit is contained in:
commit
be97d757d9
|
@ -39,9 +39,9 @@ class Controller:
|
||||||
**kwargs,
|
**kwargs,
|
||||||
):
|
):
|
||||||
""" a wrapper to dispatcher's publish"""
|
""" a wrapper to dispatcher's publish"""
|
||||||
version, message = uri.split('.', 1)
|
|
||||||
if args:
|
if args:
|
||||||
raise ValueError(_(f'the URI "{uri}" can only be published with keyword arguments'))
|
raise ValueError(_(f'the URI "{uri}" can only be published with keyword arguments'))
|
||||||
|
version, message = uri.split('.', 1)
|
||||||
await dispatcher.publish(version,
|
await dispatcher.publish(version,
|
||||||
message,
|
message,
|
||||||
risotto_context,
|
risotto_context,
|
||||||
|
|
|
@ -4,6 +4,7 @@ try:
|
||||||
except:
|
except:
|
||||||
from tiramisu import Config
|
from tiramisu import Config
|
||||||
from tiramisu.error import ValueOptionError
|
from tiramisu.error import ValueOptionError
|
||||||
|
from asyncio import get_event_loop, ensure_future
|
||||||
from traceback import print_exc
|
from traceback import print_exc
|
||||||
from copy import copy
|
from copy import copy
|
||||||
from typing import Dict, Callable, List, Optional
|
from typing import Dict, Callable, List, Optional
|
||||||
|
@ -15,7 +16,6 @@ from .logger import log
|
||||||
from .config import get_config
|
from .config import get_config
|
||||||
from .context import Context
|
from .context import Context
|
||||||
from . import register
|
from . import register
|
||||||
from .remote import Remote
|
|
||||||
|
|
||||||
|
|
||||||
class CallDispatcher:
|
class CallDispatcher:
|
||||||
|
@ -79,7 +79,7 @@ class CallDispatcher:
|
||||||
""" execute the function associate with specified uri
|
""" execute the function associate with specified uri
|
||||||
arguments are validate before
|
arguments are validate before
|
||||||
"""
|
"""
|
||||||
risotto_context = self.build_new_context(old_risotto_context,
|
risotto_context = self.build_new_context(old_risotto_context.__dict__,
|
||||||
version,
|
version,
|
||||||
message,
|
message,
|
||||||
'rpc',
|
'rpc',
|
||||||
|
@ -88,20 +88,35 @@ class CallDispatcher:
|
||||||
raise CallError(_(f'cannot find version of message "{version}"'))
|
raise CallError(_(f'cannot find version of message "{version}"'))
|
||||||
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_objs = [self.messages[version][message]]
|
function_obj = self.messages[version][message]
|
||||||
# do not start a new database connection
|
# do not start a new database connection
|
||||||
if hasattr(old_risotto_context, 'connection'):
|
if hasattr(old_risotto_context, 'connection'):
|
||||||
risotto_context.connection = old_risotto_context.connection
|
risotto_context.connection = old_risotto_context.connection
|
||||||
return await self.launch(version,
|
await self.check_message_type(risotto_context,
|
||||||
message,
|
|
||||||
risotto_context,
|
|
||||||
check_role,
|
|
||||||
kwargs,
|
kwargs,
|
||||||
function_objs,
|
)
|
||||||
|
config_arguments = await self.load_kwargs_to_config(risotto_context,
|
||||||
|
f'{version}.{message}',
|
||||||
|
kwargs,
|
||||||
|
check_role,
|
||||||
internal,
|
internal,
|
||||||
)
|
)
|
||||||
|
return await self.launch(risotto_context,
|
||||||
|
kwargs,
|
||||||
|
config_arguments,
|
||||||
|
function_obj,
|
||||||
|
)
|
||||||
else:
|
else:
|
||||||
try:
|
try:
|
||||||
|
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,
|
||||||
|
)
|
||||||
async with self.pool.acquire() as connection:
|
async with self.pool.acquire() as connection:
|
||||||
await connection.set_type_codec(
|
await connection.set_type_codec(
|
||||||
'json',
|
'json',
|
||||||
|
@ -111,13 +126,10 @@ class CallDispatcher:
|
||||||
)
|
)
|
||||||
risotto_context.connection = connection
|
risotto_context.connection = connection
|
||||||
async with connection.transaction():
|
async with connection.transaction():
|
||||||
return await self.launch(version,
|
return await self.launch(risotto_context,
|
||||||
message,
|
|
||||||
risotto_context,
|
|
||||||
check_role,
|
|
||||||
kwargs,
|
kwargs,
|
||||||
function_objs,
|
config_arguments,
|
||||||
internal,
|
function_obj,
|
||||||
)
|
)
|
||||||
except CallError as err:
|
except CallError as err:
|
||||||
raise err
|
raise err
|
||||||
|
@ -139,43 +151,81 @@ class CallDispatcher:
|
||||||
|
|
||||||
|
|
||||||
class PublishDispatcher:
|
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']:
|
||||||
|
# module, submodule, submessage = message.split('.', 2)
|
||||||
|
# if f'{module}.{submodule}' not in self.injected_self:
|
||||||
|
uri = f'{version}.{message}'
|
||||||
|
print(f' - {uri}')
|
||||||
|
await self.listened_connection.add_listener(uri,
|
||||||
|
self.to_async_publish,
|
||||||
|
)
|
||||||
|
|
||||||
async def publish(self,
|
async def publish(self,
|
||||||
version: str,
|
version: str,
|
||||||
message: str,
|
message: str,
|
||||||
old_risotto_context: Context,
|
risotto_context: Context,
|
||||||
check_role: bool=False,
|
|
||||||
internal: bool=True,
|
|
||||||
**kwargs,
|
**kwargs,
|
||||||
) -> None:
|
) -> None:
|
||||||
risotto_context = self.build_new_context(old_risotto_context,
|
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,
|
||||||
|
}
|
||||||
|
})
|
||||||
|
# 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)
|
||||||
|
risotto_context = self.build_new_context(remote_kw['context'],
|
||||||
version,
|
version,
|
||||||
message,
|
message,
|
||||||
'event',
|
'event',
|
||||||
)
|
)
|
||||||
try:
|
callback = lambda: ensure_future(self._publish(version,
|
||||||
function_objs = self.messages[version][message].get('functions', [])
|
|
||||||
except KeyError:
|
|
||||||
raise ValueError(_(f'cannot find message {version}.{message}'))
|
|
||||||
# do not start a new database connection
|
|
||||||
if hasattr(old_risotto_context, 'connection'):
|
|
||||||
# publish to remove
|
|
||||||
remote_kw = dumps({'kwargs': kwargs,
|
|
||||||
'context': risotto_context.__dict__,
|
|
||||||
})
|
|
||||||
risotto_context.connection = old_risotto_context.connection
|
|
||||||
# FIXME should be better :/
|
|
||||||
remote_kw = remote_kw.replace("'", "''")
|
|
||||||
await risotto_context.connection.execute(f'NOTIFY "{version}.{message}", \'{remote_kw}\'')
|
|
||||||
return await self.launch(version,
|
|
||||||
message,
|
message,
|
||||||
risotto_context,
|
risotto_context,
|
||||||
check_role,
|
**remote_kw['kwargs'],
|
||||||
|
))
|
||||||
|
loop.call_soon(callback)
|
||||||
|
|
||||||
|
async def _publish(self,
|
||||||
|
version: str,
|
||||||
|
message: str,
|
||||||
|
risotto_context: Context,
|
||||||
|
**kwargs,
|
||||||
|
) -> None:
|
||||||
|
config_arguments = await self.load_kwargs_to_config(risotto_context,
|
||||||
|
f'{version}.{message}',
|
||||||
kwargs,
|
kwargs,
|
||||||
function_objs,
|
False,
|
||||||
internal,
|
False,
|
||||||
)
|
)
|
||||||
|
for function_obj in self.messages[version][message]['functions']:
|
||||||
|
print('======', function_obj['function'].__name__)
|
||||||
async with self.pool.acquire() as connection:
|
async with self.pool.acquire() as connection:
|
||||||
try:
|
try:
|
||||||
|
await self.check_message_type(risotto_context,
|
||||||
|
kwargs,
|
||||||
|
)
|
||||||
await connection.set_type_codec(
|
await connection.set_type_codec(
|
||||||
'json',
|
'json',
|
||||||
encoder=dumps,
|
encoder=dumps,
|
||||||
|
@ -184,13 +234,10 @@ class PublishDispatcher:
|
||||||
)
|
)
|
||||||
risotto_context.connection = connection
|
risotto_context.connection = connection
|
||||||
async with connection.transaction():
|
async with connection.transaction():
|
||||||
return await self.launch(version,
|
await self.launch(risotto_context,
|
||||||
message,
|
|
||||||
risotto_context,
|
|
||||||
check_role,
|
|
||||||
kwargs,
|
kwargs,
|
||||||
function_objs,
|
config_arguments,
|
||||||
internal,
|
function_obj,
|
||||||
)
|
)
|
||||||
except CallError as err:
|
except CallError as err:
|
||||||
pass
|
pass
|
||||||
|
@ -211,14 +258,13 @@ class PublishDispatcher:
|
||||||
|
|
||||||
|
|
||||||
class Dispatcher(register.RegisterDispatcher,
|
class Dispatcher(register.RegisterDispatcher,
|
||||||
Remote,
|
|
||||||
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
|
||||||
"""
|
"""
|
||||||
def build_new_context(self,
|
def build_new_context(self,
|
||||||
old_risotto_context: Context,
|
context: dict,
|
||||||
version: str,
|
version: str,
|
||||||
message: str,
|
message: str,
|
||||||
type: str,
|
type: str,
|
||||||
|
@ -227,8 +273,8 @@ class Dispatcher(register.RegisterDispatcher,
|
||||||
"""
|
"""
|
||||||
uri = version + '.' + message
|
uri = version + '.' + message
|
||||||
risotto_context = Context()
|
risotto_context = Context()
|
||||||
risotto_context.username = old_risotto_context.username
|
risotto_context.username = context['username']
|
||||||
risotto_context.paths = copy(old_risotto_context.paths)
|
risotto_context.paths = copy(context['paths'])
|
||||||
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
|
||||||
|
@ -342,24 +388,12 @@ class Dispatcher(register.RegisterDispatcher,
|
||||||
raise NotAllowedError(_(f'You ({user_login}) don\'t have any authorisation to access to "{uri}"'))
|
raise NotAllowedError(_(f'You ({user_login}) don\'t have any authorisation to access to "{uri}"'))
|
||||||
|
|
||||||
async def launch(self,
|
async def launch(self,
|
||||||
version: str,
|
|
||||||
message: str,
|
|
||||||
risotto_context: Context,
|
risotto_context: Context,
|
||||||
check_role: bool,
|
|
||||||
kwargs: Dict,
|
kwargs: Dict,
|
||||||
function_objs: List,
|
config_arguments: dict,
|
||||||
internal: bool,
|
function_obj: Callable,
|
||||||
) -> Optional[Dict]:
|
) -> Optional[Dict]:
|
||||||
await self.check_message_type(risotto_context,
|
# so send the message
|
||||||
kwargs)
|
|
||||||
config_arguments = await self.load_kwargs_to_config(risotto_context,
|
|
||||||
f'{version}.{message}',
|
|
||||||
kwargs,
|
|
||||||
check_role,
|
|
||||||
internal,
|
|
||||||
)
|
|
||||||
# config is ok, so send the message
|
|
||||||
for function_obj in function_objs:
|
|
||||||
function = function_obj['function']
|
function = function_obj['function']
|
||||||
submodule_name = function_obj['module']
|
submodule_name = function_obj['module']
|
||||||
function_name = function.__name__
|
function_name = function.__name__
|
||||||
|
@ -381,12 +415,14 @@ class Dispatcher(register.RegisterDispatcher,
|
||||||
await self.valid_call_returns(risotto_context,
|
await self.valid_call_returns(risotto_context,
|
||||||
function,
|
function,
|
||||||
returns,
|
returns,
|
||||||
kwargs)
|
kwargs,
|
||||||
|
)
|
||||||
# log the success
|
# log the success
|
||||||
await log.info_msg(risotto_context,
|
await log.info_msg(risotto_context,
|
||||||
{'arguments': kwargs,
|
{'arguments': kwargs,
|
||||||
'returns': returns},
|
'returns': returns},
|
||||||
info_msg)
|
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)
|
||||||
|
@ -398,7 +434,8 @@ class Dispatcher(register.RegisterDispatcher,
|
||||||
await self.publish(notif_version,
|
await self.publish(notif_version,
|
||||||
notif_message,
|
notif_message,
|
||||||
risotto_context,
|
risotto_context,
|
||||||
**ret)
|
**ret,
|
||||||
|
)
|
||||||
if risotto_context.type == 'rpc':
|
if risotto_context.type == 'rpc':
|
||||||
return returns
|
return returns
|
||||||
|
|
||||||
|
|
|
@ -300,7 +300,7 @@ class RegisterDispatcher:
|
||||||
)
|
)
|
||||||
if truncate:
|
if truncate:
|
||||||
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')
|
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():
|
||||||
for submodule_name, module in self.injected_self.items():
|
for submodule_name, module in self.injected_self.items():
|
||||||
risotto_context = Context()
|
risotto_context = Context()
|
||||||
|
|
|
@ -1,42 +0,0 @@
|
||||||
from asyncio import get_event_loop, ensure_future
|
|
||||||
from json import loads
|
|
||||||
|
|
||||||
|
|
||||||
from .context import Context
|
|
||||||
from .config import get_config
|
|
||||||
from .utils import _
|
|
||||||
|
|
||||||
|
|
||||||
class Remote:
|
|
||||||
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':
|
|
||||||
module, submodule, submessage = message.split('.', 2)
|
|
||||||
if f'{module}.{submodule}' not in self.injected_self:
|
|
||||||
uri = f'{version}.{message}'
|
|
||||||
print(f' - {uri}')
|
|
||||||
await self.listened_connection.add_listener(uri, self.to_async_publish)
|
|
||||||
|
|
||||||
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)
|
|
||||||
context = Context()
|
|
||||||
for key, value in remote_kw['context'].items():
|
|
||||||
setattr(context, key, value)
|
|
||||||
callback = lambda: ensure_future(self.publish(version,
|
|
||||||
message,
|
|
||||||
context,
|
|
||||||
**remote_kw['kwargs'],
|
|
||||||
))
|
|
||||||
loop.call_soon(callback)
|
|
Loading…
Reference in New Issue