Skip to content

Instantly share code, notes, and snippets.

@jmetzz
Created January 9, 2019 10:45
Show Gist options
  • Save jmetzz/074ca84298602a9b55dc3bb67b706e22 to your computer and use it in GitHub Desktop.
Save jmetzz/074ca84298602a9b55dc3bb67b706e22 to your computer and use it in GitHub Desktop.
Load and register a set of classes from a python package
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