Created
January 9, 2019 10:45
-
-
Save jmetzz/074ca84298602a9b55dc3bb67b706e22 to your computer and use it in GitHub Desktop.
Load and register a set of classes from a python package
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import importlib | |
import inspect | |
import logging | |
import pkgutil | |
import six | |
from typing import Text, List, Dict, Any | |
class Action(object): | |
"""A generic action class.""" | |
def name(self): | |
# type: () -> Text | |
"""Unique identifier of this simple action.""" | |
raise NotImplementedError("An action must implement a name") | |
def run(self, **kwargs): | |
# type: (...) -> List[Dict[Text, Any]] | |
"""Execute the side effects of this action. | |
Args: | |
kwargs: the specific arguments for the action implementation | |
Returns: | |
List[Dict[Any, Any]]: A dictionary representing the results of the | |
action | |
""" | |
raise NotImplementedError("An action must implement its run method") | |
def __str__(self): | |
return f"Action('{self.name()}')" | |
class ActionExecutorRegister(object): | |
def __init__(self): | |
self.actions = {} | |
self.logger = logging.getLogger(self.__class__.__name__) | |
def register_package(self, package, cls, filter_package=None): | |
try: | |
self._import_submodules(package) | |
except ImportError: | |
logging.exception(f"Failed to register package '{package}'.") | |
actions = self._all_subclasses(cls) | |
for action in actions: | |
self.register_action(action, filter_package) | |
def register_action(self, action, filter_package=None): | |
if inspect.isclass(action): | |
if filter_package and action.__module__.startswith(filter_package): | |
self.logger.warning(f"Skipping filtered Action '{action}'.") | |
return | |
else: | |
action = action() | |
if isinstance(action, Action): | |
self.register_function(action.name(), action.run) | |
else: | |
raise Exception("You can only register instances or subclasses of " | |
"type Action. If you want to directly register " | |
"a function, use `register_function` instead.") | |
def register_function(self, name, f): | |
self.logger.info(f"Registered function for '{name}'.") | |
self.actions[name] = f | |
def _all_subclasses(self, cls): | |
"""Returns all known (imported) subclasses of a class.""" | |
return cls.__subclasses__() + [g for s in cls.__subclasses__() | |
for g in self._all_subclasses(s)] | |
def _arguments_of(func): | |
"""Return the parameters of the function `func` as a list of their names.""" | |
try: | |
# python 3.x is used | |
return inspect.signature(func).parameters.keys() | |
except AttributeError: | |
# python 2.x is used | |
return inspect.getargspec(func).args | |
def _import_submodules(self, package, recursive=True): | |
""" Import all submodules of a module, recursively, including | |
subpackages | |
:param package: package (name or actual module) | |
:type package: str | module | |
:rtype: dict[str, types.ModuleType] | |
""" | |
if isinstance(package, six.string_types): | |
package = importlib.import_module(package) | |
if not getattr(package, '__path__', None): | |
return | |
results = {} | |
for loader, name, is_pkg in pkgutil.walk_packages(package.__path__): | |
full_name = package.__name__ + '.' + name | |
results[full_name] = importlib.import_module(full_name) | |
if recursive and is_pkg: | |
self._import_submodules(full_name) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment