from _pydevd_bundle.pydevd_constants import ForkSafeLock, get_global_debugger, IS_PY2 import os import sys from contextlib import contextmanager class IORedirector: ''' This class works to wrap a stream (stdout/stderr) with an additional redirect. ''' def __init__(self, original, new_redirect, wrap_buffer=False): ''' :param stream original: The stream to be wrapped (usually stdout/stderr, but could be None). :param stream new_redirect: Usually IOBuf (below). :param bool wrap_buffer: Whether to create a buffer attribute (needed to mimick python 3 s tdout/stderr which has a buffer to write binary data). ''' self._lock = ForkSafeLock(rlock=True) self._writing = False self._redirect_to = (original, new_redirect) if wrap_buffer and hasattr(original, 'buffer'): self.buffer = IORedirector(original.buffer, new_redirect.buffer, False) def write(self, s): # Note that writing to the original stream may fail for some reasons # (such as trying to write something that's not a string or having it closed). with self._lock: if self._writing: return self._writing = True try: for r in self._redirect_to: if hasattr(r, 'write'): r.write(s) finally: self._writing = False def isatty(self): for r in self._redirect_to: if hasattr(r, 'isatty'): return r.isatty() return False def flush(self): for r in self._redirect_to: if hasattr(r, 'flush'): r.flush() def __getattr__(self, name): for r in self._redirect_to: if hasattr(r, name): return getattr(r, name) raise AttributeError(name) class RedirectToPyDBIoMessages(object): def __init__(self, out_ctx, wrap_stream, wrap_buffer, on_write=None): ''' :param out_ctx: 1=stdout and 2=stderr :param wrap_stream: Either sys.stdout or sys.stderr. :param bool wrap_buffer: If True the buffer attribute (which wraps writing bytes) should be wrapped. :param callable(str) on_write: May be a custom callable to be called when to write something. If not passed the default implementation will create an io message and send it through the debugger. ''' encoding = getattr(wrap_stream, 'encoding', None) if not encoding: encoding = os.environ.get('PYTHONIOENCODING', 'utf-8') self.encoding = encoding self._out_ctx = out_ctx if wrap_buffer: self.buffer = RedirectToPyDBIoMessages(out_ctx, wrap_stream, wrap_buffer=False, on_write=on_write) self._on_write = on_write def get_pydb(self): # Note: separate method for mocking on tests. return get_global_debugger() def flush(self): pass # no-op here def write(self, s): if self._on_write is not None: self._on_write(s) return if s: if IS_PY2: # Need s in utf-8 bytes if isinstance(s, unicode): # noqa # Note: python 2.6 does not accept the "errors" keyword. s = s.encode('utf-8', 'replace') else: s = s.decode(self.encoding, 'replace').encode('utf-8', 'replace') else: # Need s in str if isinstance(s, bytes): s = s.decode(self.encoding, errors='replace') py_db = self.get_pydb() if py_db is not None: # Note that the actual message contents will be a xml with utf-8, although # the entry is str on py3 and bytes on py2. cmd = py_db.cmd_factory.make_io_message(s, self._out_ctx) if py_db.writer is not None: py_db.writer.add_command(cmd) class IOBuf: '''This class works as a replacement for stdio and stderr. It is a buffer and when its contents are requested, it will erase what it has so far so that the next return will not return the same contents again. ''' def __init__(self): self.buflist = [] import os self.encoding = os.environ.get('PYTHONIOENCODING', 'utf-8') def getvalue(self): b = self.buflist self.buflist = [] # clear it return ''.join(b) # bytes on py2, str on py3. def write(self, s): if IS_PY2: if isinstance(s, unicode): # can't use 'errors' as kwargs in py 2.6 s = s.encode(self.encoding, 'replace') else: if isinstance(s, bytes): s = s.decode(self.encoding, errors='replace') self.buflist.append(s) def isatty(self): return False def flush(self): pass def empty(self): return len(self.buflist) == 0 class _RedirectInfo(object): def __init__(self, original, redirect_to): self.original = original self.redirect_to = redirect_to class _RedirectionsHolder: _lock = ForkSafeLock(rlock=True) _stack_stdout = [] _stack_stderr = [] _pydevd_stdout_redirect_ = None _pydevd_stderr_redirect_ = None def start_redirect(keep_original_redirection=False, std='stdout', redirect_to=None): ''' @param std: 'stdout', 'stderr', or 'both' ''' with _RedirectionsHolder._lock: if redirect_to is None: redirect_to = IOBuf() if std == 'both': config_stds = ['stdout', 'stderr'] else: config_stds = [std] for std in config_stds: original = getattr(sys, std) stack = getattr(_RedirectionsHolder, '_stack_%s' % std) if keep_original_redirection: wrap_buffer = True if not IS_PY2 and hasattr(redirect_to, 'buffer') else False new_std_instance = IORedirector(getattr(sys, std), redirect_to, wrap_buffer=wrap_buffer) setattr(sys, std, new_std_instance) else: new_std_instance = redirect_to setattr(sys, std, redirect_to) stack.append(_RedirectInfo(original, new_std_instance)) return redirect_to def end_redirect(std='stdout'): with _RedirectionsHolder._lock: if std == 'both': config_stds = ['stdout', 'stderr'] else: config_stds = [std] for std in config_stds: stack = getattr(_RedirectionsHolder, '_stack_%s' % std) redirect_info = stack.pop() setattr(sys, std, redirect_info.original) def redirect_stream_to_pydb_io_messages(std): ''' :param std: 'stdout' or 'stderr' ''' with _RedirectionsHolder._lock: redirect_to_name = '_pydevd_%s_redirect_' % (std,) if getattr(_RedirectionsHolder, redirect_to_name) is None: wrap_buffer = True if not IS_PY2 else False original = getattr(sys, std) redirect_to = RedirectToPyDBIoMessages(1 if std == 'stdout' else 2, original, wrap_buffer) start_redirect(keep_original_redirection=True, std=std, redirect_to=redirect_to) stack = getattr(_RedirectionsHolder, '_stack_%s' % std) setattr(_RedirectionsHolder, redirect_to_name, stack[-1]) return True return False def stop_redirect_stream_to_pydb_io_messages(std): ''' :param std: 'stdout' or 'stderr' ''' with _RedirectionsHolder._lock: redirect_to_name = '_pydevd_%s_redirect_' % (std,) redirect_info = getattr(_RedirectionsHolder, redirect_to_name) if redirect_info is not None: # :type redirect_info: _RedirectInfo setattr(_RedirectionsHolder, redirect_to_name, None) stack = getattr(_RedirectionsHolder, '_stack_%s' % std) prev_info = stack.pop() curr = getattr(sys, std) if curr is redirect_info.redirect_to: setattr(sys, std, redirect_info.original) @contextmanager def redirect_stream_to_pydb_io_messages_context(): with _RedirectionsHolder._lock: redirecting = [] for std in ('stdout', 'stderr'): if redirect_stream_to_pydb_io_messages(std): redirecting.append(std) try: yield finally: for std in redirecting: stop_redirect_stream_to_pydb_io_messages(std)