Source code for kombu.five

# -*- coding: utf-8 -*-
"""
    celery.five
    ~~~~~~~~~~~

    Compatibility implementations of features
    only available in newer Python versions.


"""
from __future__ import absolute_import

# ############# py3k #########################################################
import sys
PY3 = sys.version_info[0] == 3

try:
    reload = reload                         # noqa
except NameError:                           # pragma: no cover
    from imp import reload                  # noqa

try:
    from collections import UserList        # noqa
except ImportError:                         # pragma: no cover
    from UserList import UserList           # noqa

try:
    from collections import UserDict        # noqa
except ImportError:                         # pragma: no cover
    from UserDict import UserDict           # noqa

try:
    bytes_t = bytes
except NameError:  # pragma: no cover
    bytes_t = str  # noqa

# ############# time.monotonic ###############################################

if sys.version_info < (3, 3):

    import platform
    SYSTEM = platform.system()

    try:
        import ctypes
    except ImportError:  # pragma: no cover
        ctypes = None  # noqa

    if SYSTEM == 'Darwin' and ctypes is not None:
        from ctypes.util import find_library
        libSystem = ctypes.CDLL(find_library('libSystem.dylib'))
        CoreServices = ctypes.CDLL(find_library('CoreServices'),
                                   use_errno=True)
        mach_absolute_time = libSystem.mach_absolute_time
        mach_absolute_time.restype = ctypes.c_uint64
        absolute_to_nanoseconds = CoreServices.AbsoluteToNanoseconds
        absolute_to_nanoseconds.restype = ctypes.c_uint64
        absolute_to_nanoseconds.argtypes = [ctypes.c_uint64]

        def _monotonic():
            return absolute_to_nanoseconds(mach_absolute_time()) * 1e-9

    elif SYSTEM == 'Linux' and ctypes is not None:
        # from stackoverflow:
        # questions/1205722/how-do-i-get-monotonic-time-durations-in-python
        import os

        CLOCK_MONOTONIC = 1  # see <linux/time.h>

[docs] class timespec(ctypes.Structure): _fields_ = [ ('tv_sec', ctypes.c_long), ('tv_nsec', ctypes.c_long), ]
librt = ctypes.CDLL('librt.so.1', use_errno=True) clock_gettime = librt.clock_gettime clock_gettime.argtypes = [ ctypes.c_int, ctypes.POINTER(timespec), ] def _monotonic(): # noqa t = timespec() if clock_gettime(CLOCK_MONOTONIC, ctypes.pointer(t)) != 0: errno_ = ctypes.get_errno() raise OSError(errno_, os.strerror(errno_)) return t.tv_sec + t.tv_nsec * 1e-9 else: from time import time as _monotonic try: from time import monotonic except ImportError: monotonic = _monotonic # noqa # ############# Py3 <-> Py2 ################################################## if PY3: # pragma: no cover import builtins from queue import Queue, Empty, Full, LifoQueue from itertools import zip_longest from io import StringIO, BytesIO map = map zip = zip string = str string_t = str long_t = int text_t = str range = range module_name_t = str open_fqdn = 'builtins.open' def items(d): return d.items() def keys(d): return d.keys() def values(d): return d.values() def nextfun(it): return it.__next__ exec_ = getattr(builtins, 'exec')
[docs] def reraise(tp, value, tb=None): if value.__traceback__ is not tb: raise value.with_traceback(tb) raise value
class WhateverIO(StringIO): def write(self, data): if isinstance(data, bytes): data = data.encode() StringIO.write(self, data) else: import __builtin__ as builtins # noqa from Queue import Queue, Empty, Full, LifoQueue # noqa from itertools import ( # noqa imap as map, izip as zip, izip_longest as zip_longest, ) try: from cStringIO import StringIO # noqa except ImportError: # pragma: no cover from StringIO import StringIO # noqa string = unicode # noqa string_t = basestring # noqa text_t = unicode long_t = long # noqa range = xrange module_name_t = str open_fqdn = '__builtin__.open'
[docs] def items(d): # noqa return d.iteritems()
[docs] def keys(d): # noqa return d.iterkeys()
[docs] def values(d): # noqa return d.itervalues()
[docs] def nextfun(it): # noqa return it.next
[docs] def exec_(code, globs=None, locs=None): # pragma: no cover """Execute code in a namespace.""" if globs is None: frame = sys._getframe(1) globs = frame.f_globals if locs is None: locs = frame.f_locals del frame elif locs is None: locs = globs exec("""exec code in globs, locs""")
exec_("""def reraise(tp, value, tb=None): raise tp, value, tb""") BytesIO = WhateverIO = StringIO # noqa
[docs]def with_metaclass(Type, skip_attrs=set(['__dict__', '__weakref__'])): """Class decorator to set metaclass. Works with both Python 3 and Python 3 and it does not add an extra class in the lookup order like ``six.with_metaclass`` does (that is -- it copies the original class instead of using inheritance). """ def _clone_with_metaclass(Class): attrs = dict((key, value) for key, value in items(vars(Class)) if key not in skip_attrs) return Type(Class.__name__, Class.__bases__, attrs) return _clone_with_metaclass