argo_navis: venv/lib/python2.7/site-packages/click/

comparison venv/lib/python2.7/site-packages/click/_compat.py @ 0:d67268158946 draft

planemo upload commit a3f181f5f126803c654b3a66dd4e83a48f7e203b

author	bcclaywell
date	Mon, 12 Oct 2015 17:43:33 -0400
parents
children

comparison

equal deleted inserted replaced

--1:000000000000
+:d67268158946
+import re
+import io
+import os
+import sys
+import codecs
+from weakref import WeakKeyDictionary
+import click
+PY2 = sys.version_info[0] == 2
+WIN = sys.platform.startswith('win')
+DEFAULT_COLUMNS = 80
+_ansi_re = re.compile('\033\[((?:\d|;)*)([a-zA-Z])')
+def _find_unicode_literals_frame():
+import __future__
+frm = sys._getframe(1)
+idx = 1
+while frm is not None:
+if frm.f_globals.get('__name__', '').startswith('click.'):
+frm = frm.f_back
+idx += 1
+elif frm.f_code.co_flags & __future__.unicode_literals.compiler_flag:
+return idx
+else:
+break
+return 0
+def _check_for_unicode_literals():
+if not __debug__:
+return
+if not PY2 or click.disable_unicode_literals_warning:
+return
+bad_frame = _find_unicode_literals_frame()
+if bad_frame <= 0:
+return
+from warnings import warn
+warn(Warning('Click detected the use of the unicode_literals '
+'__future__ import.  This is heavily discouraged '
+'because it can introduce subtle bugs in your '
+'code.  You should instead use explicit u"" literals '
+'for your unicode strings.  For more information see '
+'http://click.pocoo.org/python3/'),
+stacklevel=bad_frame)
+def get_filesystem_encoding():
+return sys.getfilesystemencoding() or sys.getdefaultencoding()
+def _make_text_stream(stream, encoding, errors):
+if encoding is None:
+encoding = get_best_encoding(stream)
+if errors is None:
+errors = 'replace'
+return _NonClosingTextIOWrapper(stream, encoding, errors,
+line_buffering=True)
+def is_ascii_encoding(encoding):
+"""Checks if a given encoding is ascii."""
+try:
+return codecs.lookup(encoding).name == 'ascii'
+except LookupError:
+return False
+def get_best_encoding(stream):
+"""Returns the default stream encoding if not found."""
+rv = getattr(stream, 'encoding', None) or sys.getdefaultencoding()
+if is_ascii_encoding(rv):
+return 'utf-8'
+return rv
+class _NonClosingTextIOWrapper(io.TextIOWrapper):
+def __init__(self, stream, encoding, errors, **extra):
+self._stream = stream = _FixupStream(stream)
+io.TextIOWrapper.__init__(self, stream, encoding, errors, **extra)
+# The io module is a place where the Python 3 text behavior
+# was forced upon Python 2, so we need to unbreak
+# it to look like Python 2.
+if PY2:
+def write(self, x):
+if isinstance(x, str) or is_bytes(x):
+try:
+self.flush()
+except Exception:
+pass
+return self.buffer.write(str(x))
+return io.TextIOWrapper.write(self, x)
+def writelines(self, lines):
+for line in lines:
+self.write(line)
+def __del__(self):
+try:
+self.detach()
+except Exception:
+pass
+def isatty(self):
+# https://bitbucket.org/pypy/pypy/issue/1803
+return self._stream.isatty()
+class _FixupStream(object):
+"""The new io interface needs more from streams than streams
+traditionally implement.  As such, this fix-up code is necessary in
+some circumstances.
+"""
+def __init__(self, stream):
+self._stream = stream
+def __getattr__(self, name):
+return getattr(self._stream, name)
+def read1(self, size):
+f = getattr(self._stream, 'read1', None)
+if f is not None:
+return f(size)
+# We only dispatch to readline instead of read in Python 2 as we
+# do not want cause problems with the different implementation
+# of line buffering.
+if PY2:
+return self._stream.readline(size)
+return self._stream.read(size)
+def readable(self):
+x = getattr(self._stream, 'readable', None)
+if x is not None:
+return x()
+try:
+self._stream.read(0)
+except Exception:
+return False
+return True
+def writable(self):
+x = getattr(self._stream, 'writable', None)
+if x is not None:
+return x()
+try:
+self._stream.write('')
+except Exception:
+try:
+self._stream.write(b'')
+except Exception:
+return False
+return True
+def seekable(self):
+x = getattr(self._stream, 'seekable', None)
+if x is not None:
+return x()
+try:
+self._stream.seek(self._stream.tell())
+except Exception:
+return False
+return True
+if PY2:
+text_type = unicode
+bytes = str
+raw_input = raw_input
+string_types = (str, unicode)
+iteritems = lambda x: x.iteritems()
+range_type = xrange
+def is_bytes(x):
+return isinstance(x, (buffer, bytearray))
+_identifier_re = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$')
+# For Windows, we need to force stdout/stdin/stderr to binary if it's
+# fetched for that.  This obviously is not the most correct way to do
+# it as it changes global state.  Unfortunately, there does not seem to
+# be a clear better way to do it as just reopening the file in binary
+# mode does not change anything.
+#
+# An option would be to do what Python 3 does and to open the file as
+# binary only, patch it back to the system, and then use a wrapper
+# stream that converts newlines.  It's not quite clear what's the
+# correct option here.
+if WIN:
+import msvcrt
+def set_binary_mode(f):
+try:
+fileno = f.fileno()
+except Exception:
+pass
+else:
+msvcrt.setmode(fileno, os.O_BINARY)
+return f
+else:
+set_binary_mode = lambda x: x
+def isidentifier(x):
+return _identifier_re.search(x) is not None
+def get_binary_stdin():
+return set_binary_mode(sys.stdin)
+def get_binary_stdout():
+return set_binary_mode(sys.stdout)
+def get_binary_stderr():
+return set_binary_mode(sys.stderr)
+def get_text_stdin(encoding=None, errors=None):
+return _make_text_stream(sys.stdin, encoding, errors)
+def get_text_stdout(encoding=None, errors=None):
+return _make_text_stream(sys.stdout, encoding, errors)
+def get_text_stderr(encoding=None, errors=None):
+return _make_text_stream(sys.stderr, encoding, errors)
+def filename_to_ui(value):
+if isinstance(value, bytes):
+value = value.decode(get_filesystem_encoding(), 'replace')
+return value
+else:
+import io
+text_type = str
+raw_input = input
+string_types = (str,)
+range_type = range
+isidentifier = lambda x: x.isidentifier()
+iteritems = lambda x: iter(x.items())
+def is_bytes(x):
+return isinstance(x, (bytes, memoryview, bytearray))
+def _is_binary_reader(stream, default=False):
+try:
+return isinstance(stream.read(0), bytes)
+except Exception:
+return default
+# This happens in some cases where the stream was already
+# closed.  In this case, we assume the default.
+def _is_binary_writer(stream, default=False):
+try:
+stream.write(b'')
+except Exception:
+try:
+stream.write('')
+return False
+except Exception:
+pass
+return default
+return True
+def _find_binary_reader(stream):
+# We need to figure out if the given stream is already binary.
+# This can happen because the official docs recommend detaching
+# the streams to get binary streams.  Some code might do this, so
+# we need to deal with this case explicitly.
+if _is_binary_reader(stream, False):
+return stream
+buf = getattr(stream, 'buffer', None)
+# Same situation here; this time we assume that the buffer is
+# actually binary in case it's closed.
+if buf is not None and _is_binary_reader(buf, True):
+return buf
+def _find_binary_writer(stream):
+# We need to figure out if the given stream is already binary.
+# This can happen because the official docs recommend detatching
+# the streams to get binary streams.  Some code might do this, so
+# we need to deal with this case explicitly.
+if _is_binary_writer(stream, False):
+return stream
+buf = getattr(stream, 'buffer', None)
+# Same situation here; this time we assume that the buffer is
+# actually binary in case it's closed.
+if buf is not None and _is_binary_writer(buf, True):
+return buf
+def _stream_is_misconfigured(stream):
+"""A stream is misconfigured if its encoding is ASCII."""
+# If the stream does not have an encoding set, we assume it's set
+# to ASCII.  This appears to happen in certain unittest
+# environments.  It's not quite clear what the correct behavior is
+# but this at least will force Click to recover somehow.
+return is_ascii_encoding(getattr(stream, 'encoding', None) or 'ascii')
+def _is_compatible_text_stream(stream, encoding, errors):
+stream_encoding = getattr(stream, 'encoding', None)
+stream_errors = getattr(stream, 'errors', None)
+# Perfect match.
+if stream_encoding == encoding and stream_errors == errors:
+return True
+# Otherwise, it's only a compatible stream if we did not ask for
+# an encoding.
+if encoding is None:
+return stream_encoding is not None
+return False
+def _force_correct_text_reader(text_reader, encoding, errors):
+if _is_binary_reader(text_reader, False):
+binary_reader = text_reader
+else:
+# If there is no target encoding set, we need to verify that the
+# reader is not actually misconfigured.
+if encoding is None and not _stream_is_misconfigured(text_reader):
+return text_reader
+if _is_compatible_text_stream(text_reader, encoding, errors):
+return text_reader
+# If the reader has no encoding, we try to find the underlying
+# binary reader for it.  If that fails because the environment is
+# misconfigured, we silently go with the same reader because this
+# is too common to happen.  In that case, mojibake is better than
+# exceptions.
+binary_reader = _find_binary_reader(text_reader)
+if binary_reader is None:
+return text_reader
+# At this point, we default the errors to replace instead of strict
+# because nobody handles those errors anyways and at this point
+# we're so fundamentally fucked that nothing can repair it.
+if errors is None:
+errors = 'replace'
+return _make_text_stream(binary_reader, encoding, errors)
+def _force_correct_text_writer(text_writer, encoding, errors):
+if _is_binary_writer(text_writer, False):
+binary_writer = text_writer
+else:
+# If there is no target encoding set, we need to verify that the
+# writer is not actually misconfigured.
+if encoding is None and not _stream_is_misconfigured(text_writer):
+return text_writer
+if _is_compatible_text_stream(text_writer, encoding, errors):
+return text_writer
+# If the writer has no encoding, we try to find the underlying
+# binary writer for it.  If that fails because the environment is
+# misconfigured, we silently go with the same writer because this
+# is too common to happen.  In that case, mojibake is better than
+# exceptions.
+binary_writer = _find_binary_writer(text_writer)
+if binary_writer is None:
+return text_writer
+# At this point, we default the errors to replace instead of strict
+# because nobody handles those errors anyways and at this point
+# we're so fundamentally fucked that nothing can repair it.
+if errors is None:
+errors = 'replace'
+return _make_text_stream(binary_writer, encoding, errors)
+def get_binary_stdin():
+reader = _find_binary_reader(sys.stdin)
+if reader is None:
+raise RuntimeError('Was not able to determine binary '
+'stream for sys.stdin.')
+return reader
+def get_binary_stdout():
+writer = _find_binary_writer(sys.stdout)
+if writer is None:
+raise RuntimeError('Was not able to determine binary '
+'stream for sys.stdout.')
+return writer
+def get_binary_stderr():
+writer = _find_binary_writer(sys.stderr)
+if writer is None:
+raise RuntimeError('Was not able to determine binary '
+'stream for sys.stderr.')
+return writer
+def get_text_stdin(encoding=None, errors=None):
+return _force_correct_text_reader(sys.stdin, encoding, errors)
+def get_text_stdout(encoding=None, errors=None):
+return _force_correct_text_writer(sys.stdout, encoding, errors)
+def get_text_stderr(encoding=None, errors=None):
+return _force_correct_text_writer(sys.stderr, encoding, errors)
+def filename_to_ui(value):
+if isinstance(value, bytes):
+value = value.decode(get_filesystem_encoding(), 'replace')
+else:
+value = value.encode('utf-8', 'surrogateescape') \
+.decode('utf-8', 'replace')
+return value
+def get_streerror(e, default=None):
+if hasattr(e, 'strerror'):
+msg = e.strerror
+else:
+if default is not None:
+msg = default
+else:
+msg = str(e)
+if isinstance(msg, bytes):
+msg = msg.decode('utf-8', 'replace')
+return msg
+def open_stream(filename, mode='r', encoding=None, errors='strict',
+atomic=False):
+# Standard streams first.  These are simple because they don't need
+# special handling for the atomic flag.  It's entirely ignored.
+if filename == '-':
+if 'w' in mode:
+if 'b' in mode:
+return get_binary_stdout(), False
+return get_text_stdout(encoding=encoding, errors=errors), False
+if 'b' in mode:
+return get_binary_stdin(), False
+return get_text_stdin(encoding=encoding, errors=errors), False
+# Non-atomic writes directly go out through the regular open functions.
+if not atomic:
+if encoding is None:
+return open(filename, mode), True
+return io.open(filename, mode, encoding=encoding, errors=errors), True
+# Some usability stuff for atomic writes
+if 'a' in mode:
+raise ValueError(
+'Appending to an existing file is not supported, because that '
+'would involve an expensive `copy`-operation to a temporary '
+'file. Open the file in normal `w`-mode and copy explicitly '
+'if that\'s what you\'re after.'
+)
+if 'x' in mode:
+raise ValueError('Use the `overwrite`-parameter instead.')
+if 'w' not in mode:
+raise ValueError('Atomic writes only make sense with `w`-mode.')
+# Atomic writes are more complicated.  They work by opening a file
+# as a proxy in the same folder and then using the fdopen
+# functionality to wrap it in a Python file.  Then we wrap it in an
+# atomic file that moves the file over on close.
+import tempfile
+fd, tmp_filename = tempfile.mkstemp(dir=os.path.dirname(filename),
+prefix='.__atomic-write')
+if encoding is not None:
+f = io.open(fd, mode, encoding=encoding, errors=errors)
+else:
+f = os.fdopen(fd, mode)
+return _AtomicFile(f, tmp_filename, filename), True
+# Used in a destructor call, needs extra protection from interpreter cleanup.
+_rename = os.rename
+class _AtomicFile(object):
+def __init__(self, f, tmp_filename, real_filename):
+self._f = f
+self._tmp_filename = tmp_filename
+self._real_filename = real_filename
+self.closed = False
+@property
+def name(self):
+return self._real_filename
+def close(self, delete=False):
+if self.closed:
+return
+self._f.close()
+_rename(self._tmp_filename, self._real_filename)
+self.closed = True
+def __getattr__(self, name):
+return getattr(self._f, name)
+def __enter__(self):
+return self
+def __exit__(self, exc_type, exc_value, tb):
+self.close(delete=exc_type is not None)
+def __repr__(self):
+return repr(self._f)
+auto_wrap_for_ansi = None
+colorama = None
+get_winterm_size = None
+def strip_ansi(value):
+return _ansi_re.sub('', value)
+def should_strip_ansi(stream=None, color=None):
+if color is None:
+if stream is None:
+stream = sys.stdin
+return not isatty(stream)
+return not color
+# If we're on Windows, we provide transparent integration through
+# colorama.  This will make ANSI colors through the echo function
+# work automatically.
+if WIN:
+# Windows has a smaller terminal
+DEFAULT_COLUMNS = 79
+try:
+import colorama
+except ImportError:
+pass
+else:
+_ansi_stream_wrappers = WeakKeyDictionary()
+def auto_wrap_for_ansi(stream, color=None):
+"""This function wraps a stream so that calls through colorama
+are issued to the win32 console API to recolor on demand.  It
+also ensures to reset the colors if a write call is interrupted
+to not destroy the console afterwards.
+"""
+try:
+cached = _ansi_stream_wrappers.get(stream)
+except Exception:
+cached = None
+if cached is not None:
+return cached
+strip = should_strip_ansi(stream, color)
+ansi_wrapper = colorama.AnsiToWin32(stream, strip=strip)
+rv = ansi_wrapper.stream
+_write = rv.write
+def _safe_write(s):
+try:
+return _write(s)
+except:
+ansi_wrapper.reset_all()
+raise
+rv.write = _safe_write
+try:
+_ansi_stream_wrappers[stream] = rv
+except Exception:
+pass
+return rv
+def get_winterm_size():
+win = colorama.win32.GetConsoleScreenBufferInfo(
+colorama.win32.STDOUT).srWindow
+return win.Right - win.Left, win.Bottom - win.Top
+def term_len(x):
+return len(strip_ansi(x))
+def isatty(stream):
+try:
+return stream.isatty()
+except Exception:
+return False
+def _make_cached_stream_func(src_func, wrapper_func):
+cache = WeakKeyDictionary()
+def func():
+stream = src_func()
+try:
+rv = cache.get(stream)
+except Exception:
+rv = None
+if rv is not None:
+return rv
+rv = wrapper_func()
+try:
+cache[stream] = rv
+except Exception:
+pass
+return rv
+return func
+_default_text_stdout = _make_cached_stream_func(
+lambda: sys.stdout, get_text_stdout)
+_default_text_stderr = _make_cached_stream_func(
+lambda: sys.stderr, get_text_stderr)
+binary_streams = {
+'stdin': get_binary_stdin,
+'stdout': get_binary_stdout,
+'stderr': get_binary_stderr,
+}
+text_streams = {
+'stdin': get_text_stdin,
+'stdout': get_text_stdout,
+'stderr': get_text_stderr,
+}

Mercurial > repos > bcclaywell > argo_navis

comparison venv/lib/python2.7/site-packages/click/_compat.py @ 0:d67268158946 draft