63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
from typing import Type, Union, Tuple, Any, List
|
|
|
|
from biopro.text import *
|
|
|
|
|
|
class APIDispatcher:
|
|
__slots__ = ('__api_class', '__target', '__method_cache')
|
|
|
|
def __init__(self, api_class: Union[Type, Tuple[Type, ...]], target: Any):
|
|
self.__api_class = api_class
|
|
self.__target = target
|
|
self.__method_cache = set()
|
|
|
|
def _add_method(_cls):
|
|
for attr_name in dir(_cls):
|
|
if not attr_name.startswith('_') and callable(getattr(target, attr_name)):
|
|
self.__method_cache.add(attr_name)
|
|
|
|
if isinstance(api_class, tuple):
|
|
for cls in api_class:
|
|
if not isinstance(target, cls):
|
|
raise RuntimeError('target not subclass of api_class')
|
|
|
|
_add_method(cls)
|
|
|
|
else:
|
|
if not isinstance(target, api_class):
|
|
raise RuntimeError('target not subclass of api_class')
|
|
|
|
_add_method(api_class)
|
|
|
|
@property
|
|
def api_class(self) -> Union[Type, Tuple[Type, ...]]:
|
|
return self.__api_class
|
|
|
|
@property
|
|
def api_methods(self) -> List[str]:
|
|
return list(self.__method_cache)
|
|
|
|
def method_dispatch(self, method: str, argument, options) -> Any:
|
|
"""dispatch according the **method**.
|
|
|
|
:param method: method name
|
|
:param argument: method arguments
|
|
:param options: method keyword arguments
|
|
:return: method invoked return.
|
|
:raises RuntimeError: method's name startswith '_' which is protected.
|
|
or method not defined in api_class T.
|
|
or method not found in runtime.
|
|
"""
|
|
if method.startswith('_'):
|
|
raise RuntimeError(COMMAND_UNKNOWN, method)
|
|
|
|
if method not in self.__method_cache:
|
|
raise RuntimeError(COMMAND_UNKNOWN, method)
|
|
|
|
func = getattr(self.__target, method, None)
|
|
|
|
if func is not None:
|
|
return func(*argument, **options)
|
|
else:
|
|
raise RuntimeError(COMMAND_UNKNOWN, method)
|