"""API class which serves as a base for IP block emulator API implementations."""
from functools import wraps
from typing import Any, Callable, Self
from ska_mid_cbf_emulators.common import InternalRestResponse, LoggingBase
[docs]
class RouterImpl(LoggingBase):
"""API class which serves as a base for IP block emulator API implementations.
This class is designed to internally process requests
forwarded from the respective Client via RPC.
"""
def __init__(self: Self, *args, **kwargs) -> None:
super().__init__()
self._setup(*args, **kwargs)
[docs]
def call(self: Self, method_name: str, **kwargs) -> InternalRestResponse:
"""Call a method of this instance by name and return its result.
This method allows for a method name to be provided in an RPC request
so that the client request may be mapped to the correct implementation.
Args:
method_name (:obj:`str`): The method name to call.
**kwargs: Arbitrary keyword arguments.
Returns:
:obj:`InternalRestResponse` The response from the called method.
"""
try:
fn: Callable[[Any], Any] = getattr(self, method_name)
except (AttributeError, TypeError):
self.log_error(f'Invalid method name: {method_name}')
return
return fn(**kwargs)
[docs]
def error_catcher(fn: Callable) -> Callable: # pylint: disable=no-self-argument
"""Decorator which converts Python errors into API responses.
This decorator wraps implementation methods to catch all unhandled exceptions
and wraps the exception messages in a 500 (Internal Server Error) API response
to allow for graceful failures in the event of an error
instead of hanging the entire application.
Args:
fn (:obj:`Callable`): The method to wrap.
Returns:
:obj:`Callable` The wrapped method.
"""
@wraps(fn)
def inner(self: Self, *args, **kwargs):
try:
return fn(self, *args, **kwargs) # pylint: disable=not-callable
except Exception as e:
self.log_error(e)
return InternalRestResponse.internal_server_error(str(e))
return inner
def _setup(self: Self, *args, **kwargs):
return