Barrier.wait() - Code Metrics - Inspection of "harden tarfile stream implementation" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — file-actions ( 3503b9...9b9199 )

by Felipe A.

created 2018-04-12 07:42 UTC

Barrier.wait() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	17
rs	8
c	0
b	0
f	0
cc	6

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

import os
import os.path
import sys
import abc
import itertools

import warnings
import functools

import posixpath
import ntpath

FS_ENCODING = sys.getfilesystemencoding()
PY_LEGACY = sys.version_info < (3, )
TRUE_VALUES = frozenset(('true', 'yes', '1', 'enable', 'enabled', True, 1))

try:
    from os import scandir, walk
except ImportError:
    from scandir import scandir, walk  # noqa

try:
    from shutil import get_terminal_size
except ImportError:
    from backports.shutil_get_terminal_size import get_terminal_size  # noqa

try:
    from queue import Queue, Empty, Full
except ImportError:
    from Queue import Queue, Empty, Full  # noqa

try:
    from threading import Barrier, BrokenBarrierError
except ImportError:
    from threading import Lock  # noqa

    class BrokenBarrierError(RuntimeError):
        pass

    class Barrier(object):
        def __init__(self, parties, action=None, timeout=0):
            self.parties = parties
            self.n_waiting = 0
            self.broken = False
            self._queue = Queue(maxsize=1)
            self._lock = Lock()

        def _uncork(self, broken=False):
            # allow every exit
            while self.n_waiting > 1:
                self._queue.put(broken)
                self.n_waiting -= 1
            self.n_waiting = 0

        def reset(self):
            with self._lock:
                self._uncork(True)

        def abort(self):
            with self._lock:
                self.broken = True
                self._uncork(True)

        def wait(self):
            with self._lock:
                if self.broken:
                    raise BrokenBarrierError()

                self.n_waiting += 1

                if self.n_waiting == self.parties:
                    self._uncork()

                    if self.action:
                        self.action()

                    return

            if self._queue.get():
                raise BrokenBarrierError()


def isexec(path):
    '''
    Check if given path points to an executable file.

    :param path: file path
    :type path: str
    :return: True if executable, False otherwise
    :rtype: bool
    '''
    return os.path.isfile(path) and os.access(path, os.X_OK)


def fsdecode(path, os_name=os.name, fs_encoding=FS_ENCODING, errors=None):
    '''
    Decode given path.

    :param path: path will be decoded if using bytes
    :type path: bytes or str
    :param os_name: operative system name, defaults to os.name
    :type os_name: str
    :param fs_encoding: current filesystem encoding, defaults to autodetected
    :type fs_encoding: str
    :return: decoded path
    :rtype: str
    '''
    if not isinstance(path, bytes):
        return path
    if not errors:
        use_strict = PY_LEGACY or os_name == 'nt'
        errors = 'strict' if use_strict else 'surrogateescape'
    return path.decode(fs_encoding, errors=errors)


def fsencode(path, os_name=os.name, fs_encoding=FS_ENCODING, errors=None):
    '''
    Encode given path.

    :param path: path will be encoded if not using bytes
    :type path: bytes or str
    :param os_name: operative system name, defaults to os.name
    :type os_name: str
    :param fs_encoding: current filesystem encoding, defaults to autodetected
    :type fs_encoding: str
    :return: encoded path
    :rtype: bytes
    '''
    if isinstance(path, bytes):
        return path
    if not errors:
        use_strict = PY_LEGACY or os_name == 'nt'
        errors = 'strict' if use_strict else 'surrogateescape'
    return path.encode(fs_encoding, errors=errors)


def getcwd(fs_encoding=FS_ENCODING, cwd_fnc=os.getcwd):
    '''
    Get current work directory's absolute path.
    Like os.getcwd but garanteed to return an unicode-str object.

    :param fs_encoding: filesystem encoding, defaults to autodetected
    :type fs_encoding: str
    :param cwd_fnc: callable used to get the path, defaults to os.getcwd
    :type cwd_fnc: Callable
    :return: path
    :rtype: str
    '''
    path = fsdecode(cwd_fnc(), fs_encoding=fs_encoding)
    return os.path.abspath(path)


def getdebug(environ=os.environ, true_values=TRUE_VALUES):
    '''
    Get if app is expected to be ran in debug mode looking at environment
    variables.

    :param environ: environment dict-like object
    :type environ: collections.abc.Mapping
    :returns: True if debug contains a true-like string, False otherwise
    :rtype: bool
    '''
    return environ.get('DEBUG', '').lower() in true_values


def deprecated(func_or_text, environ=os.environ):
    '''
    Decorator used to mark functions as deprecated. It will result in a
    warning being emmitted hen the function is called.

    Usage:

    >>> @deprecated
    ... def fnc():
    ...     pass

    Usage (custom message):

    >>> @deprecated('This is deprecated')
    ... def fnc():
    ...     pass

    :param func_or_text: message or callable to decorate
    :type func_or_text: callable
    :param environ: optional environment mapping
    :type environ: collections.abc.Mapping
    :returns: nested decorator or new decorated function (depending on params)
    :rtype: callable
    '''
    def inner(func):
        message = (
            'Deprecated function {}.'.format(func.__name__)
            if callable(func_or_text) else
            func_or_text
            )

        @functools.wraps(func)
        def new_func(*args, **kwargs):
            with warnings.catch_warnings():
                if getdebug(environ):
                    warnings.simplefilter('always', DeprecationWarning)
                warnings.warn(message, category=DeprecationWarning,
                              stacklevel=3)
            return func(*args, **kwargs)
        return new_func
    return inner(func_or_text) if callable(func_or_text) else inner


def usedoc(other):
    '''
    Decorator which copies __doc__ of given object into decorated one.

    Usage:

    >>> def fnc1():
    ...     """docstring"""
    ...     pass
    >>> @usedoc(fnc1)
    ... def fnc2():
    ...     pass
    >>> fnc2.__doc__
    'docstring'collections.abc.D

    :param other: anything with a __doc__ attribute
    :type other: any
    :returns: decorator function
    :rtype: callable
    '''
    def inner(fnc):
        fnc.__doc__ = fnc.__doc__ or getattr(other, '__doc__')
        return fnc
    return inner


def pathsplit(value, sep=os.pathsep):
    '''
    Get enviroment PATH elements as list.

    This function only cares about spliting across OSes.

    :param value: path string, as given by os.environ['PATH']
    :type value: str
    :param sep: PATH separator, defaults to os.pathsep
    :type sep: str
    :yields: every path
    :ytype: str
    '''
    for part in value.split(sep):
        if part[:1] == part[-1:] == '"' or part[:1] == part[-1:] == '\'':
            part = part[1:-1]
        yield part


def pathparse(value, sep=os.pathsep, os_sep=os.sep):
    '''
    Get enviroment PATH directories as list.

    This function cares about spliting, escapes and normalization of paths
    across OSes.

    :param value: path string, as given by os.environ['PATH']
    :type value: str
    :param sep: PATH separator, defaults to os.pathsep
    :type sep: str
    :param os_sep: OS filesystem path separator, defaults to os.sep
    :type os_sep: str
    :yields: every path
    :ytype: str
    '''
    escapes = []
    normpath = ntpath.normpath if os_sep == '\\' else posixpath.normpath
    if '\\' not in (os_sep, sep):
        escapes.extend((
            ('\\\\', '<ESCAPE-ESCAPE>', '\\'),
            ('\\"', '<ESCAPE-DQUOTE>', '"'),
            ('\\\'', '<ESCAPE-SQUOTE>', '\''),
            ('\\%s' % sep, '<ESCAPE-PATHSEP>', sep),
            ))
    for original, escape, unescape in escapes:
        value = value.replace(original, escape)
    for part in pathsplit(value, sep=sep):
        if part[-1:] == os_sep and part != os_sep:
            part = part[:-1]
        for original, escape, unescape in escapes:
            part = part.replace(escape, unescape)
        yield normpath(fsdecode(part))


def pathconf(path,
             os_name=os.name,
             isdir_fnc=os.path.isdir,
             pathconf_fnc=getattr(os, 'pathconf', None),
             pathconf_names=getattr(os, 'pathconf_names', ())):
    '''
    Get all pathconf variables for given path.

    :param path: absolute fs path
    :type path: str
    :returns: dictionary containing pathconf keys and their values (both str)
    :rtype: dict
    '''

    if pathconf_fnc and pathconf_names:
        return {key: pathconf_fnc(path, key) for key in pathconf_names}
    if os_name == 'nt':
        maxpath = 246 if isdir_fnc(path) else 259  # 260 minus <END>
    else:
        maxpath = 255  # conservative sane default
    return {
        'PC_PATH_MAX': maxpath,
        'PC_NAME_MAX': maxpath - len(path),
        }


ENV_PATH = tuple(pathparse(os.getenv('PATH', '')))
ENV_PATHEXT = tuple(pathsplit(os.getenv('PATHEXT', '')))


def which(name,
          env_path=ENV_PATH,
          env_path_ext=ENV_PATHEXT,
          is_executable_fnc=isexec,
          path_join_fnc=os.path.join,
          os_name=os.name):
    '''
    Get command absolute path.

    :param name: name of executable command
    :type name: str
    :param env_path: OS environment executable paths, defaults to autodetected
    :type env_path: list of str
    :param is_executable_fnc: callable will be used to detect if path is
                              executable, defaults to `isexec`
    :type is_executable_fnc: Callable
    :param path_join_fnc: callable will be used to join path components
    :type path_join_fnc: Callable
    :param os_name: os name, defaults to os.name
    :type os_name: str
    :return: absolute path
    :rtype: str or None
    '''
    for path in env_path:
        for suffix in env_path_ext:
            exe_file = path_join_fnc(path, name) + suffix
            if is_executable_fnc(exe_file):
                return exe_file
    return None


def re_escape(pattern, chars=frozenset("()[]{}?*+|^$\\.-#")):
    '''
    Escape all special regex characters in pattern and converts non-ascii
    characters into unicode escape sequences.

    Logic taken from regex module.

    :param pattern: regex pattern to escape
    :type patterm: str
    :returns: escaped pattern
    :rtype: str
    '''
    chr_escape = '\\{}'.format
    uni_escape = '\\u{:04d}'.format
    return ''.join(
        chr_escape(c) if c in chars or c.isspace() else
        c if '\x19' < c < '\x80' else
        uni_escape(ord(c))
        for c in pattern
        )


if PY_LEGACY:

    class FileNotFoundError(BaseException):
        __metaclass__ = abc.ABCMeta

    FileNotFoundError.register(OSError)
    FileNotFoundError.register(IOError)

    range = xrange  # noqa
    filter = itertools.ifilter
    map = itertools.imap
    basestring = basestring  # noqa
    unicode = unicode  # noqa
    chr = unichr  # noqa
    bytes = str  # noqa
else:
    FileNotFoundError = FileNotFoundError
    range = range
    filter = filter
    map = map
    basestring = str
    unicode = str
    chr = chr
    bytes = bytes


1			#!/usr/bin/env python
2			# -- coding: UTF-8 --
3
4			import os
5			import os.path
6			import sys
7			import abc
8			import itertools
9
10			import warnings
11			import functools
12
13			import posixpath
14			import ntpath
15
16			FS_ENCODING = sys.getfilesystemencoding()
17			PY_LEGACY = sys.version_info < (3, )
18			TRUE_VALUES = frozenset(('true', 'yes', '1', 'enable', 'enabled', True, 1))
19
20			try:
21			from os import scandir, walk
22			except ImportError:
23			from scandir import scandir, walk # noqa
24
25			try:
26			from shutil import get_terminal_size
27			except ImportError:
28			from backports.shutil_get_terminal_size import get_terminal_size # noqa
29
30			try:
31			from queue import Queue, Empty, Full
32			except ImportError:
33			from Queue import Queue, Empty, Full # noqa
34
35			try:
36			from threading import Barrier, BrokenBarrierError
37			except ImportError:
38			from threading import Lock # noqa
39
40			class BrokenBarrierError(RuntimeError):
41			pass
42
43			class Barrier(object):
44			def __init__(self, parties, action=None, timeout=0):
45			self.parties = parties
46			self.n_waiting = 0
47			self.broken = False
48			self._queue = Queue(maxsize=1)
49			self._lock = Lock()
50
51			def _uncork(self, broken=False):
52			# allow every exit
53			while self.n_waiting > 1:
54			self._queue.put(broken)
55			self.n_waiting -= 1
56			self.n_waiting = 0
57
58			def reset(self):
59			with self._lock:
60			self._uncork(True)
61
62			def abort(self):
63			with self._lock:
64			self.broken = True
65			self._uncork(True)
66
67			def wait(self):
68			with self._lock:
69			if self.broken:
70			raise BrokenBarrierError()
71
72			self.n_waiting += 1
73
74			if self.n_waiting == self.parties:
75			self._uncork()
76
77			if self.action:
78			self.action()
79
80			return
81
82			if self._queue.get():
83			raise BrokenBarrierError()
84
85
86			def isexec(path):
87			'''
88			Check if given path points to an executable file.
89
90			:param path: file path
91			:type path: str
92			:return: True if executable, False otherwise
93			:rtype: bool
94			'''
95			return os.path.isfile(path) and os.access(path, os.X_OK)
96
97
98			def fsdecode(path, os_name=os.name, fs_encoding=FS_ENCODING, errors=None):
99			'''
100			Decode given path.
101
102			:param path: path will be decoded if using bytes
103			:type path: bytes or str
104			:param os_name: operative system name, defaults to os.name
105			:type os_name: str
106			:param fs_encoding: current filesystem encoding, defaults to autodetected
107			:type fs_encoding: str
108			:return: decoded path
109			:rtype: str
110			'''
111			if not isinstance(path, bytes):
112			return path
113			if not errors:
114			use_strict = PY_LEGACY or os_name == 'nt'
115			errors = 'strict' if use_strict else 'surrogateescape'
116			return path.decode(fs_encoding, errors=errors)
117
118
119			def fsencode(path, os_name=os.name, fs_encoding=FS_ENCODING, errors=None):
120			'''
121			Encode given path.
122
123			:param path: path will be encoded if not using bytes
124			:type path: bytes or str
125			:param os_name: operative system name, defaults to os.name
126			:type os_name: str
127			:param fs_encoding: current filesystem encoding, defaults to autodetected
128			:type fs_encoding: str
129			:return: encoded path
130			:rtype: bytes
131			'''
132			if isinstance(path, bytes):
133			return path
134			if not errors:
135			use_strict = PY_LEGACY or os_name == 'nt'
136			errors = 'strict' if use_strict else 'surrogateescape'
137			return path.encode(fs_encoding, errors=errors)
138
139
140			def getcwd(fs_encoding=FS_ENCODING, cwd_fnc=os.getcwd):
141			'''
142			Get current work directory's absolute path.
143			Like os.getcwd but garanteed to return an unicode-str object.
144
145			:param fs_encoding: filesystem encoding, defaults to autodetected
146			:type fs_encoding: str
147			:param cwd_fnc: callable used to get the path, defaults to os.getcwd
148			:type cwd_fnc: Callable
149			:return: path
150			:rtype: str
151			'''
152			path = fsdecode(cwd_fnc(), fs_encoding=fs_encoding)
153			return os.path.abspath(path)
154
155
156			def getdebug(environ=os.environ, true_values=TRUE_VALUES):
157			'''
158			Get if app is expected to be ran in debug mode looking at environment
159			variables.
160
161			:param environ: environment dict-like object
162			:type environ: collections.abc.Mapping
163			:returns: True if debug contains a true-like string, False otherwise
164			:rtype: bool
165			'''
166			return environ.get('DEBUG', '').lower() in true_values
167
168
169			def deprecated(func_or_text, environ=os.environ):
170			'''
171			Decorator used to mark functions as deprecated. It will result in a
172			warning being emmitted hen the function is called.
173
174			Usage:
175
176			>>> @deprecated
177			... def fnc():
178			... pass
179
180			Usage (custom message):
181
182			>>> @deprecated('This is deprecated')
183			... def fnc():
184			... pass
185
186			:param func_or_text: message or callable to decorate
187			:type func_or_text: callable
188			:param environ: optional environment mapping
189			:type environ: collections.abc.Mapping
190			:returns: nested decorator or new decorated function (depending on params)
191			:rtype: callable
192			'''
193			def inner(func):
194			message = (
195			'Deprecated function {}.'.format(func.__name__)
196			if callable(func_or_text) else
197			func_or_text
198			)
199
200			@functools.wraps(func)
201			def new_func(args, *kwargs):
202			with warnings.catch_warnings():
203			if getdebug(environ):
204			warnings.simplefilter('always', DeprecationWarning)
205			warnings.warn(message, category=DeprecationWarning,
206			stacklevel=3)
207			return func(args, *kwargs)
208			return new_func
209			return inner(func_or_text) if callable(func_or_text) else inner
210
211
212			def usedoc(other):
213			'''
214			Decorator which copies __doc__ of given object into decorated one.
215
216			Usage:
217
218			>>> def fnc1():
219			... """docstring"""
220			... pass
221			>>> @usedoc(fnc1)
222			... def fnc2():
223			... pass
224			>>> fnc2.__doc__
225			'docstring'collections.abc.D
226
227			:param other: anything with a __doc__ attribute
228			:type other: any
229			:returns: decorator function
230			:rtype: callable
231			'''
232			def inner(fnc):
233			fnc.__doc__ = fnc.__doc__ or getattr(other, '__doc__')
234			return fnc
235			return inner
236
237
238			def pathsplit(value, sep=os.pathsep):
239			'''
240			Get enviroment PATH elements as list.
241
242			This function only cares about spliting across OSes.
243
244			:param value: path string, as given by os.environ['PATH']
245			:type value: str
246			:param sep: PATH separator, defaults to os.pathsep
247			:type sep: str
248			:yields: every path
249			:ytype: str
250			'''
251			for part in value.split(sep):
252			if part[:1] == part[-1:] == '"' or part[:1] == part[-1:] == '\'':
253			part = part[1:-1]
254			yield part
255
256
257			def pathparse(value, sep=os.pathsep, os_sep=os.sep):
258			'''
259			Get enviroment PATH directories as list.
260
261			This function cares about spliting, escapes and normalization of paths
262			across OSes.
263
264			:param value: path string, as given by os.environ['PATH']
265			:type value: str
266			:param sep: PATH separator, defaults to os.pathsep
267			:type sep: str
268			:param os_sep: OS filesystem path separator, defaults to os.sep
269			:type os_sep: str
270			:yields: every path
271			:ytype: str
272			'''
273			escapes = []
274			normpath = ntpath.normpath if os_sep == '\\' else posixpath.normpath
275			if '\\' not in (os_sep, sep):
276			escapes.extend((
277			('\\\\', '<ESCAPE-ESCAPE>', '\\'),
278			('\\"', '<ESCAPE-DQUOTE>', '"'),
279			('\\\'', '<ESCAPE-SQUOTE>', '\''),
280			('\\%s' % sep, '<ESCAPE-PATHSEP>', sep),
281			))
282			for original, escape, unescape in escapes:
283			value = value.replace(original, escape)
284			for part in pathsplit(value, sep=sep):
285			if part[-1:] == os_sep and part != os_sep:
286			part = part[:-1]
287			for original, escape, unescape in escapes:
288			part = part.replace(escape, unescape)
289			yield normpath(fsdecode(part))
290
291
292			def pathconf(path,
293			os_name=os.name,
294			isdir_fnc=os.path.isdir,
295			pathconf_fnc=getattr(os, 'pathconf', None),
296			pathconf_names=getattr(os, 'pathconf_names', ())):
297			'''
298			Get all pathconf variables for given path.
299
300			:param path: absolute fs path
301			:type path: str
302			:returns: dictionary containing pathconf keys and their values (both str)
303			:rtype: dict
304			'''
305
306			if pathconf_fnc and pathconf_names:
307			return {key: pathconf_fnc(path, key) for key in pathconf_names}
308			if os_name == 'nt':
309			maxpath = 246 if isdir_fnc(path) else 259 # 260 minus <END>
310			else:
311			maxpath = 255 # conservative sane default
312			return {
313			'PC_PATH_MAX': maxpath,
314			'PC_NAME_MAX': maxpath - len(path),
315			}
316
317
318			ENV_PATH = tuple(pathparse(os.getenv('PATH', '')))
319			ENV_PATHEXT = tuple(pathsplit(os.getenv('PATHEXT', '')))
320
321
322			def which(name,
323			env_path=ENV_PATH,
324			env_path_ext=ENV_PATHEXT,
325			is_executable_fnc=isexec,
326			path_join_fnc=os.path.join,
327			os_name=os.name):
328			'''
329			Get command absolute path.
330
331			:param name: name of executable command
332			:type name: str
333			:param env_path: OS environment executable paths, defaults to autodetected
334			:type env_path: list of str
335			:param is_executable_fnc: callable will be used to detect if path is
336			executable, defaults to `isexec`
337			:type is_executable_fnc: Callable
338			:param path_join_fnc: callable will be used to join path components
339			:type path_join_fnc: Callable
340			:param os_name: os name, defaults to os.name
341			:type os_name: str
342			:return: absolute path
343			:rtype: str or None
344			'''
345			for path in env_path:
346			for suffix in env_path_ext:
347			exe_file = path_join_fnc(path, name) + suffix
348			if is_executable_fnc(exe_file):
349			return exe_file
350			return None
351
352
353			def re_escape(pattern, chars=frozenset("()[]{}?*+\|^$\\.-#")):
354			'''
355			Escape all special regex characters in pattern and converts non-ascii
356			characters into unicode escape sequences.
357
358			Logic taken from regex module.
359
360			:param pattern: regex pattern to escape
361			:type patterm: str
362			:returns: escaped pattern
363			:rtype: str
364			'''
365			chr_escape = '\\{}'.format
366			uni_escape = '\\u{:04d}'.format
367			return ''.join(
368			chr_escape(c) if c in chars or c.isspace() else
369			c if '\x19' < c < '\x80' else
370			uni_escape(ord(c))
371			for c in pattern
372			)
373
374
375			if PY_LEGACY:
376
377			class FileNotFoundError(BaseException):
378			__metaclass__ = abc.ABCMeta
379
380			FileNotFoundError.register(OSError)
381			FileNotFoundError.register(IOError)
382
383			range = xrange # noqa
384			filter = itertools.ifilter
385			map = itertools.imap
386			basestring = basestring # noqa
387			unicode = unicode # noqa
388			chr = unichr # noqa
389			bytes = str # noqa
390			else:
391			FileNotFoundError = FileNotFoundError
392			range = range
393			filter = filter
394			map = map
395			basestring = str
396			unicode = str
397			chr = chr
398			bytes = bytes
399

ergoithz / browsepy

Push — file-actions ( 3503b9...9b9199 )

Barrier.wait() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like