Source code for orion.core.utils

# -*- 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] typenames = list(map(lambda x: x.__name__.lower(), types)) return types, typenames
[docs]class Factory(ABCMeta): """Instantiate appropriate wrapper for the infrastructure based on input argument, ``of_type``. Attributes ---------- types : list of subclasses of ``cls.__base__`` Updated to contain all possible implementations currently. Check out code. typenames : list of str Names of implemented wrapper classes, correspond to possible ``of_type`` values. """ def __init__(cls, names, bases, dictionary): """Search in directory for attribute names subclassing `bases[0]`""" super(Factory, cls).__init__(names, bases, dictionary) 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, ) # Get types visible from base module or package, but internal cls.types, cls.typenames = get_all_types(cls.__base__, cls.__name__) log.debug("Implementations found: %s", cls.typenames) 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.typenames` 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. """ if of_type.lower() not in cls.typenames: cls.types, cls.typenames = get_all_types(cls.__base__, cls.__name__) for inherited_class in cls.types: if inherited_class.__name__.lower() == 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(cls.typenames) raise NotImplementedError(error)