celery/celery/five.py

# -*- coding: utf-8 -*-
"""Python 2/3 compatibility utilities."""
import operator
import sys

from importlib import import_module
from types import ModuleType

# extends vine.five
from vine import five
from vine.five import *  # noqa
from vine.five import __all__ as _all_five

# bloody flake8
items = five.items
string_t = five.string_t

try:
    from functools import reduce
except ImportError:
    pass

__all__ = [
    'class_property', 'reclassmethod', 'create_module', 'recreate_module',
]
__all__ += _all_five

#  ############# Module Generation ##########################

# Utilities to dynamically
# recreate modules, either for lazy loading or
# to create old modules at runtime instead of
# having them litter the source tree.

# import fails in python 2.5. fallback to reduce in stdlib

DEFAULT_ATTRS = {'__file__', '__path__', '__doc__', '__all__'}


def getappattr(path):
    """Gets attribute from the current_app recursively,
    e.g. getappattr('amqp.get_task_consumer')``."""
    from celery import current_app
    return current_app._rgetattr(path)


def _compat_periodic_task_decorator(*args, **kwargs):
    from celery.task import periodic_task
    return periodic_task(*args, **kwargs)


class class_property:

    def __init__(self, getter=None, setter=None):
        if getter is not None and not isinstance(getter, classmethod):
            getter = classmethod(getter)
        if setter is not None and not isinstance(setter, classmethod):
            setter = classmethod(setter)
        self.__get = getter
        self.__set = setter

        info = getter.__get__(object)  # just need the info attrs.
        self.__doc__ = info.__doc__
        self.__name__ = info.__name__
        self.__module__ = info.__module__

    def __get__(self, obj, type=None):
        if obj and type is None:
            type = obj.__class__
        return self.__get.__get__(obj, type)()

    def __set__(self, obj, value):
        if obj is None:
            return self
        return self.__set.__get__(obj)(value)

    def setter(self, setter):
        return self.__class__(self.__get, setter)


def reclassmethod(method):
    return classmethod(method)


class LazyModule(ModuleType):
    _compat_modules = ()
    _all_by_module = {}
    _direct = {}
    _object_origins = {}

    def __getattr__(self, name):
        if name in self._object_origins:
            module = __import__(self._object_origins[name], None, None, [name])
            for item in self._all_by_module[module.__name__]:
                setattr(self, item, getattr(module, item))
            return getattr(module, name)
        elif name in self._direct:  # pragma: no cover
            module = __import__(self._direct[name], None, None, [name])
            setattr(self, name, module)
            return module
        return ModuleType.__getattribute__(self, name)

    def __dir__(self):
        return list(set(self.__all__) | DEFAULT_ATTRS)

    def __reduce__(self):
        return import_module, (self.__name__,)


def create_module(name, attrs, cls_attrs=None, pkg=None,
                  base=LazyModule, prepare_attr=None):
    fqdn = '.'.join([pkg.__name__, name]) if pkg else name
    cls_attrs = {} if cls_attrs is None else cls_attrs
    pkg, _, modname = name.rpartition('.')
    cls_attrs['__module__'] = pkg

    attrs = {
        attr_name: (prepare_attr(attr) if prepare_attr else attr)
        for attr_name, attr in items(attrs)
    }
    module = sys.modules[fqdn] = type(modname, (base,), cls_attrs)(name)
    module.__dict__.update(attrs)
    return module


def recreate_module(name, compat_modules=(), by_module={}, direct={},
                    base=LazyModule, **attrs):
    old_module = sys.modules[name]
    origins = get_origins(by_module)

    _all = tuple(set(reduce(
        operator.add,
        [tuple(v) for v in [compat_modules, origins, direct, attrs]],
    )))
    cattrs = dict(
        _compat_modules=compat_modules,
        _all_by_module=by_module, _direct=direct,
        _object_origins=origins,
        __all__=_all,
    )
    new_module = create_module(name, attrs, cls_attrs=cattrs, base=base)
    return old_module, new_module


def get_origins(defs):
    origins = {}
    for module, attrs in items(defs):
        origins.update({attr: module for attr in attrs})
    return origins