auxlib._vendor.exec_() - Code Metrics - Inspection of "add scrutinizer" - kalefranz/auxlib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — develop ( c97d07...11b882 )

by Kale

created 2016-01-20 02:01 UTC

auxlib._vendor.exec_() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	4
dl	0
loc	11
rs	9.2

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

    Compatibility implementations of features
    only available in newer Python versions.


"""
from __future__ import absolute_import

import io
import sys

try:
    from collections import Counter
except ImportError:  # pragma: no cover
    from collections import defaultdict

    def Counter():  # noqa
        return defaultdict(int)

try:
    buffer_t = buffer
except NameError:  # pragma: no cover
    # Py3 does not have buffer, only use this for isa checks.

    class buffer_t(object):  # noqa
        pass

bytes_t = bytes

__all__ = ['Counter', 'reload', 'UserList', 'UserDict',
           'Queue', 'Empty', 'Full', 'LifoQueue', 'builtins',
           'zip_longest', 'map', 'zip', 'string', 'string_t', 'bytes_t',
           'long_t', 'text_t', 'int_types', 'module_name_t',
           'range', 'items', 'keys', 'values', 'nextfun', 'reraise',
           'WhateverIO', 'with_metaclass', 'open_fqdn', 'StringIO',
           'THREAD_TIMEOUT_MAX', 'format_d', 'monotonic', 'buffer_t']


#  ############# py3k ########################################################
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

#  ############# 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>

        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 itertools import zip_longest

    map = map
    zip = zip
    string = str
    string_t = str
    long_t = int
    text_t = str
    range = range
    int_types = (int,)
    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')

    def reraise(tp, value, tb=None):
        if value.__traceback__ is not tb:
            raise value.with_traceback(tb)
        raise value

else:
    import __builtin__ as builtins  # noqa
    from itertools import (               # noqa
        imap as map,
        izip as zip,
        izip_longest as zip_longest,
    )

    string = unicode                # noqa
    string_t = basestring           # noqa
    text_t = unicode
    long_t = long                   # noqa
    range = xrange
    module_name_t = str
    int_types = (int, long)

    open_fqdn = '__builtin__.open'

    def items(d):                   # noqa
        return d.iteritems()

    def keys(d):                    # noqa
        return d.iterkeys()

    def values(d):                  # noqa
        return d.itervalues()

    def nextfun(it):                # noqa
        return it.next

    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""")


def with_metaclass(Type, skip_attrs={'__dict__', '__weakref__'}):
    """Class decorator to set metaclass.

    Works with both Python 2 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 = {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

# ############# threading.TIMEOUT_MAX ########################################
try:
    from threading import TIMEOUT_MAX as THREAD_TIMEOUT_MAX
except ImportError:
    THREAD_TIMEOUT_MAX = 1e10  # noqa

# ############# format(int, ',d') ############################################

if sys.version_info >= (2, 7):  # pragma: no cover
    def format_d(i):
        return format(i, ',d')
else:  # pragma: no cover
    def format_d(i):  # noqa
        s = '%d' % i
        groups = []
        while s and s[-1].isdigit():
            groups.append(s[-3:])
            s = s[:-3]
        return s + ','.join(reversed(groups))

StringIO = io.StringIO
_SIO_write = StringIO.write
_SIO_init = StringIO.__init__


class WhateverIO(StringIO):

    def __init__(self, v=None, *a, **kw):
        _SIO_init(self, v.decode() if isinstance(v, bytes) else v, *a, **kw)

    def write(self, data):
        _SIO_write(self, data.decode() if isinstance(data, bytes) else data)

1			# -- coding: utf-8 --
2			"""
3			amqp.five
4			~~~~~~~~~~~
5
6			Compatibility implementations of features
7			only available in newer Python versions.
8
9
10			"""
11			from __future__ import absolute_import
12
13			import io
14			import sys
15
16			try:
17			from collections import Counter
18			except ImportError: # pragma: no cover
19			from collections import defaultdict
20
21			def Counter(): # noqa
22			return defaultdict(int)
23
24			try:
25			buffer_t = buffer
26			except NameError: # pragma: no cover
27			# Py3 does not have buffer, only use this for isa checks.
28
29			class buffer_t(object): # noqa
30			pass
31
32			bytes_t = bytes
33
34			__all__ = ['Counter', 'reload', 'UserList', 'UserDict',
35			'Queue', 'Empty', 'Full', 'LifoQueue', 'builtins',
36			'zip_longest', 'map', 'zip', 'string', 'string_t', 'bytes_t',
37			'long_t', 'text_t', 'int_types', 'module_name_t',
38			'range', 'items', 'keys', 'values', 'nextfun', 'reraise',
39			'WhateverIO', 'with_metaclass', 'open_fqdn', 'StringIO',
40			'THREAD_TIMEOUT_MAX', 'format_d', 'monotonic', 'buffer_t']
41
42
43			# ############# py3k ########################################################
44			PY3 = sys.version_info[0] == 3
45
46			try:
47			reload = reload # noqa
48			except NameError: # pragma: no cover
49			from imp import reload # noqa
50
51			try:
52			from collections import UserList # noqa
53			except ImportError: # pragma: no cover
54			from UserList import UserList # noqa
55
56			try:
57			from collections import UserDict # noqa
58			except ImportError: # pragma: no cover
59			from UserDict import UserDict # noqa
60
61			# ############# time.monotonic #############################################
62
63			if sys.version_info < (3, 3):
64
65			import platform
66			SYSTEM = platform.system()
67
68			try:
69			import ctypes
70			except ImportError: # pragma: no cover
71			ctypes = None # noqa
72
73			if SYSTEM == 'Darwin' and ctypes is not None:
74			from ctypes.util import find_library
75			libSystem = ctypes.CDLL(find_library('libSystem.dylib'))
76			CoreServices = ctypes.CDLL(find_library('CoreServices'),
77			use_errno=True)
78			mach_absolute_time = libSystem.mach_absolute_time
79			mach_absolute_time.restype = ctypes.c_uint64
80			absolute_to_nanoseconds = CoreServices.AbsoluteToNanoseconds
81			absolute_to_nanoseconds.restype = ctypes.c_uint64
82			absolute_to_nanoseconds.argtypes = [ctypes.c_uint64]
83
84			def _monotonic():
85			return absolute_to_nanoseconds(mach_absolute_time()) * 1e-9
86
87			elif SYSTEM == 'Linux' and ctypes is not None:
88			# from stackoverflow:
89			# questions/1205722/how-do-i-get-monotonic-time-durations-in-python
90			import os
91
92			CLOCK_MONOTONIC = 1 # see <linux/time.h>
93
94			class timespec(ctypes.Structure):
95			_fields_ = [
96			('tv_sec', ctypes.c_long),
97			('tv_nsec', ctypes.c_long),
98			]
99
100			librt = ctypes.CDLL('librt.so.1', use_errno=True)
101			clock_gettime = librt.clock_gettime
102			clock_gettime.argtypes = [
103			ctypes.c_int, ctypes.POINTER(timespec),
104			]
105
106			def _monotonic(): # noqa
107			t = timespec()
108			if clock_gettime(CLOCK_MONOTONIC, ctypes.pointer(t)) != 0:
109			errno_ = ctypes.get_errno()
110			raise OSError(errno_, os.strerror(errno_))
111			return t.tv_sec + t.tv_nsec * 1e-9
112			else:
113			from time import time as _monotonic
114			try:
115			from time import monotonic
116			except ImportError:
117			monotonic = _monotonic # noqa
118
119			# ############# Py3 <-> Py2 #################################################
120
121			if PY3: # pragma: no cover
122			import builtins
123
124			from itertools import zip_longest
125
126			map = map
127			zip = zip
128			string = str
129			string_t = str
130			long_t = int
131			text_t = str
132			range = range
133			int_types = (int,)
134			module_name_t = str
135
136			open_fqdn = 'builtins.open'
137
138			def items(d):
139			return d.items()
140
141			def keys(d):
142			return d.keys()
143
144			def values(d):
145			return d.values()
146
147			def nextfun(it):
148			return it.__next__
149
150			exec_ = getattr(builtins, 'exec')
151
152			def reraise(tp, value, tb=None):
153			if value.__traceback__ is not tb:
154			raise value.with_traceback(tb)
155			raise value
156
157			else:
158			import __builtin__ as builtins # noqa
159			from itertools import ( # noqa
160			imap as map,
161			izip as zip,
162			izip_longest as zip_longest,
163			)
164
165			string = unicode # noqa
166			string_t = basestring # noqa
167			text_t = unicode
168			long_t = long # noqa
169			range = xrange
170			module_name_t = str
171			int_types = (int, long)
172
173			open_fqdn = '__builtin__.open'
174
175			def items(d): # noqa
176			return d.iteritems()
177
178			def keys(d): # noqa
179			return d.iterkeys()
180
181			def values(d): # noqa
182			return d.itervalues()
183
184			def nextfun(it): # noqa
185			return it.next
186
187			def exec_(code, globs=None, locs=None): # pragma: no cover
188			"""Execute code in a namespace."""
189			if globs is None:
190			frame = sys._getframe(1)
191			globs = frame.f_globals
192			if locs is None:
193			locs = frame.f_locals
194			del frame
195			elif locs is None:
196			locs = globs
197			exec("""exec code in globs, locs""")
198
199			exec_("""def reraise(tp, value, tb=None): raise tp, value, tb""")
200
201
202			def with_metaclass(Type, skip_attrs={'__dict__', '__weakref__'}):
203			"""Class decorator to set metaclass.
204
205			Works with both Python 2 and Python 3 and it does not add
206			an extra class in the lookup order like ``six.with_metaclass`` does
207			(that is -- it copies the original class instead of using inheritance).
208
209			"""
210
211			def _clone_with_metaclass(Class):
212			attrs = {key: value for key, value in items(vars(Class))
213			if key not in skip_attrs}
214			return Type(Class.__name__, Class.__bases__, attrs)
215
216			return _clone_with_metaclass
217
218			# ############# threading.TIMEOUT_MAX ########################################
219			try:
220			from threading import TIMEOUT_MAX as THREAD_TIMEOUT_MAX
221			except ImportError:
222			THREAD_TIMEOUT_MAX = 1e10 # noqa
223
224			# ############# format(int, ',d') ############################################
225
226			if sys.version_info >= (2, 7): # pragma: no cover
227			def format_d(i):
228			return format(i, ',d')
229			else: # pragma: no cover
230			def format_d(i): # noqa
231			s = '%d' % i
232			groups = []
233			while s and s[-1].isdigit():
234			groups.append(s[-3:])
235			s = s[:-3]
236			return s + ','.join(reversed(groups))
237
238			StringIO = io.StringIO
239			_SIO_write = StringIO.write
240			_SIO_init = StringIO.__init__
241
242
243			class WhateverIO(StringIO):
244
245			def __init__(self, v=None, a, *kw):
246			_SIO_init(self, v.decode() if isinstance(v, bytes) else v, a, *kw)
247
248			def write(self, data):
249			_SIO_write(self, data.decode() if isinstance(data, bytes) else data)

kalefranz / auxlib

Push — develop ( c97d07...11b882 )

auxlib._vendor.exec_() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like