80 lines
2.9 KiB
Python
80 lines
2.9 KiB
Python
from .dispatcher import dispatcher
|
|
from .context import Context
|
|
from .utils import _
|
|
|
|
|
|
class Controller:
|
|
"""Common controller used to add a service in Risotto
|
|
"""
|
|
def __init__(self,
|
|
test: bool,
|
|
):
|
|
pass
|
|
|
|
async def call(self,
|
|
uri: str,
|
|
risotto_context: Context,
|
|
*args,
|
|
**kwargs,
|
|
):
|
|
""" a wrapper to dispatcher's call"""
|
|
if args:
|
|
raise ValueError(_(f'the URI "{uri}" can only be called with keyword arguments'))
|
|
current_uri = risotto_context.paths[-1]
|
|
current_module = risotto_context.module
|
|
version, message = uri.split('.', 1)
|
|
module = message.split('.', 1)[0]
|
|
if current_module != module:
|
|
raise ValueError(_(f'cannot call to external module ("{module}") to the URI "{uri}" from "{current_module}"'))
|
|
return await dispatcher.call(version,
|
|
message,
|
|
risotto_context,
|
|
**kwargs,
|
|
)
|
|
|
|
async def publish(self,
|
|
uri: str,
|
|
risotto_context: Context,
|
|
*args,
|
|
**kwargs,
|
|
):
|
|
""" a wrapper to dispatcher's publish"""
|
|
if args:
|
|
raise ValueError(_(f'the URI "{uri}" can only be published with keyword arguments'))
|
|
version, message = uri.split('.', 1)
|
|
await dispatcher.publish(version,
|
|
message,
|
|
risotto_context,
|
|
**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,
|
|
risotto_context,
|
|
):
|
|
pass
|