# -*- coding: utf-8 -*-
"""
Package-wide useful routines
============================
"""
import logging
import os
from abc import ABCMeta
from collections import defaultdict
from glob import glob
from importlib import import_module
import pkg_resources
log = logging.getLogger(__name__)
[docs]def nesteddict():
"""
Define type of arbitrary nested defaultdicts
Extend defaultdict to arbitrary nested levels.
"""
return defaultdict(nesteddict)
[docs]def get_all_subclasses(parent):
"""Get set of subclasses recursively"""
subclasses = set()
for subclass in parent.__subclasses__():
subclasses.add(subclass)
subclasses |= get_all_subclasses(subclass)
return subclasses
[docs]def get_all_types(parent_cls, cls_name):
"""Get all subclasses and lowercase subclass names"""
types = list(get_all_subclasses(parent_cls))
types = [class_ for class_ in types if class_.__name__ != cls_name]
return {class_.__name__.lower(): class_ for class_ in types}
def _import_modules(cls):
cls.modules = []
base = import_module(cls.__base__.__module__)
try:
py_files = glob(os.path.abspath(os.path.join(base.__path__[0], "[A-Za-z]*.py")))
py_mods = map(
lambda x: "." + os.path.split(os.path.splitext(x)[0])[1], py_files
)
for py_mod in py_mods:
cls.modules.append(import_module(py_mod, package=cls.__base__.__module__))
except AttributeError:
# This means that base class and implementations reside in a module
# itself and not a subpackage.
pass
# Get types advertised through entry points!
for entry_point in pkg_resources.iter_entry_points(cls.__name__):
entry_point.load()
log.debug(
"Found a %s %s from distribution: %s=%s",
entry_point.name,
cls.__name__,
entry_point.dist.project_name,
entry_point.dist.version,
)
def _set_typenames(cls):
# Get types visible from base module or package, but internal
cls.types.update(get_all_types(cls.__base__, cls.__name__))
log.debug("Implementations found: %s", sorted(cls.types.keys()))
[docs]class Factory(ABCMeta):
"""Instantiate appropriate wrapper for the infrastructure based on input
argument, ``of_type``.
Attributes
----------
types : dict of subclasses of ``cls.__base__``
Updated to contain all possible implementations currently. Check out code.
"""
def __init__(cls, names, bases, dictionary):
"""Search in directory for attribute names subclassing `bases[0]`"""
super(Factory, cls).__init__(names, bases, dictionary)
cls.types = {}
try:
_import_modules(cls)
except ImportError:
pass
_set_typenames(cls)
[docs] def __call__(cls, of_type, *args, **kwargs):
"""Create an object, instance of ``cls.__base__``, on first call.
:param of_type: Name of class, subclass of ``cls.__base__``, wrapper
of a database framework that will be instantiated on the first call.
:param args: positional arguments to initialize ``cls.__base__``'s instance (if any)
:param kwargs: keyword arguments to initialize ``cls.__base__``'s instance (if any)
.. seealso::
`Factory.types` keys for values of argument `of_type`.
.. seealso::
Attributes of ``cls.__base__`` and ``cls.__base__.__init__`` for
values of `args` and `kwargs`.
.. note:: New object is saved as `Factory`'s internal state.
:return: The object which was created on the first call.
"""
_import_modules(cls)
_set_typenames(cls)
for name, inherited_class in cls.types.items():
if name == of_type.lower():
return inherited_class(*args, **kwargs)
error = "Could not find implementation of {0}, type = '{1}'".format(
cls.__base__.__name__, of_type
)
error += "\nCurrently, there is an implementation for types:\n"
error += str(sorted(cls.types.keys()))
raise NotImplementedError(error)