python类format_exception()的实例源码

def handle_exception(self, exception, debug_mode):
    """Called if this handler throws an exception during execution.

    The default behavior is to call self.error(500) and print a stack trace
    if debug_mode is True.

    Args:
      exception: the exception that was thrown
      debug_mode: True if the web application is running in debug mode
    """
    self.error(500)
    logging.exception(exception)
    if debug_mode:
      lines = ''.join(traceback.format_exception(*sys.exc_info()))
      self.response.clear()
      self.response.out.write('<pre>%s</pre>' % (cgi.escape(lines, quote=True)))

def format_error(e):
    # errback automatically wraps everything in a Twisted Failure
    if isinstance(e, failure.Failure):
        e = e.value

    if isinstance(e, str):
        err_string = e
    elif six.PY2:
        err_string = traceback.format_exc(e).rstrip()
    else:
        err_string = ''.join(traceback.format_exception(type(e), e, e.__traceback__)).rstrip()

    if err_string == 'None':
        # Reasonable heuristic for exceptions that were created by hand
        last = traceback.format_stack()[-2]
        err_string = '{}\n  {}'.format(e, last)
    # Quick and dirty hack for now.
    err_string = err_string.replace('Connection to the other side was lost in a non-clean fashion', 'Connection to the other side was lost in a non-clean fashion (HINT: this generally actually means we got a connection refused error. Check that the remote is actually running.)')
    return error.Error(err_string)

def octario_excepthook(exc_type, exc_value, exc_traceback):
    """exception hook that sends OctarioException to log and other

    exceptions to stderr (default excepthook)
    """
    from octario.lib.exceptions import OctarioException

    # sends full exception with trace to log
    if not isinstance(exc_value, OctarioException):
        return sys.__excepthook__(exc_type, exc_value, exc_traceback)

    if LOG.getEffectiveLevel() <= logging.DEBUG:
        formated_exception = "".join(
            traceback.format_exception(exc_type, exc_value, exc_traceback))
        LOG.error(formated_exception + exc_value.message)
    else:
        LOG.error(exc_value.message)

def ssh_exec_live(ssh, cmd, timeout=None, check_retcode=True):
    LOG.debug(u"{}Calling SSH live: '{}'{}".format(Style.DIM, cmd,
                                                   Style.RESET_ALL))
    try:
        interact = SSHClientInteraction(ssh, timeout=timeout, display=True,
                                        logger=logging.getLogger())
        interact.expect('.*')
        interact.send(cmd + "; exit $?")  # needed to not depend on prompt type
        interact.tail()
        ret_code = interact.channel.recv_exit_status()
    except Exception:
        LOG.debug("Something went wrong in 'ssh_exec_live':\n{}".format(
            format_exception(sys.exc_info())
        ))
        raise

    _proceed_exec_result("", "", ret_code, check_retcode)
    return ret_code, "", ""

def test_logging(caplog):
    exc1 = get_exc(raiser1)
    exc2 = get_exc(raiser2)

    m = MultiError([exc1, exc2])

    message = "test test test"
    try:
        raise m
    except MultiError as exc:
        logging.getLogger().exception(message)
        # Join lines together
        formatted = "".join(
            format_exception(type(exc), exc, exc.__traceback__)
        )
        assert message in caplog.text
        assert formatted in caplog.text

def _worker_run_collect(all_args):
    try:
        collect_once, counter, lock, threshold, args = all_args
        collected = []
        while True:
            with lock:
                if counter.value >= threshold:
                    return collected
            result, inc = collect_once(singleton_pool.G, *args)
            collected.append(result)
            with lock:
                counter.value += inc
                if counter.value >= threshold:
                    return collected
    except Exception:
        raise Exception("".join(traceback.format_exception(*sys.exc_info())))

def user_exception(self, frame, info):
        """This function is called if an exception occurs,
        but only if we are to stop at or just below this level."""
        if self._wait_for_mainpyfile or self._wait_for_breakpoint:
            return
        extype, exvalue, trace = info
        # pre-process stack trace as it isn't pickeable (cannot be sent pure)
        msg = ''.join(traceback.format_exception(extype, exvalue, trace))
        trace = traceback.extract_tb(trace)
        title = traceback.format_exception_only(extype, exvalue)[0]
        # send an Exception notification
        msg = {'method': 'exception',
               'args': (title, extype.__name__, exvalue, trace, msg),
               'id': None}
        self.pipe.send(msg)
        self.interaction(frame, info)

def __init__(self, test_name, exc_info):
    """Constructs a LinkerExceptionTestResult object.

    Args:
      test_name: name of the test which raised an exception.
      exc_info: exception info, ostensibly from sys.exc_info().
    """
    exc_type, exc_value, exc_traceback = exc_info
    trace_info = ''.join(traceback.format_exception(exc_type, exc_value,
                                                    exc_traceback))
    log_msg = 'Exception:\n' + trace_info

    super(LinkerExceptionTestResult, self).__init__(
        test_name,
        base_test_result.ResultType.FAIL,
        log="%s %s" % (exc_type, log_msg))

def findvideos(data):
    logger.info("[servertools.py] findvideos")
    encontrados = set()
    devuelve = []

    # Ejecuta el findvideos en cada servidor
    for serverid in ALL_SERVERS:
        try:
            exec "from servers import "+serverid
            exec "devuelve.extend("+serverid+".find_videos(data))"
        except ImportError:
            logger.info("No existe conector para "+serverid)
        except:
            logger.info("Error en el conector "+serverid)
            import traceback,sys
            from pprint import pprint
            exc_type, exc_value, exc_tb = sys.exc_info()
            lines = traceback.format_exception(exc_type, exc_value, exc_tb)
            for line in lines:
                line_splits = line.split("\n")
                for line_split in line_splits:
                    logger.error(line_split)

    return devuelve

def user_exception(self, frame, info):
        """This function is called if an exception occurs,
        but only if we are to stop at or just below this level."""
        if self._wait_for_mainpyfile or self._wait_for_breakpoint:
            return
        extype, exvalue, trace = info
        # pre-process stack trace as it isn't pickeable (cannot be sent pure)
        msg = ''.join(traceback.format_exception(extype, exvalue, trace))
        trace = traceback.extract_tb(trace)
        title = traceback.format_exception_only(extype, exvalue)[0]
        # send an Exception notification
        msg = {'method': 'exception',
               'args': (title, extype.__name__, exvalue, trace, msg),
               'id': None}
        self.pipe.send(msg)
        self.interaction(frame, info)

def format_message(self, message):
        now = datetime.datetime.utcnow()
        timestamp = now.strftime("%Y-%m-%dT%H:%M:%S") + ".%03d" % (now.microsecond / 1000) + "Z"

        return_json = {
            "logger": message.name,
            "line_number": message.lineno,
            "path_name": message.pathname,
            "log_level": message.levelname,
            "type": self.logzio_type,
            "message": message.getMessage(),
            "@timestamp": timestamp
        }

        if message.exc_info:
            return_json["exception"] = self.format_exception(message.exc_info)
        else:
            formatted_message = self.format(message)
            if isinstance(formatted_message, dict):
                return_json.update(formatted_message)
            else:
                return_json["message"] = formatted_message

        return return_json

def render_POST(self, request):
        """
        Handle a request from the client.
        """
        script_env = {
            method: api_method(request, method)
            for method in request.sdata.api.fns
        }

        # Make get do auto-formatting for convenience, even though this
        # breaks if you try to use literal '{}' named arguments
        # @@@ reconsider whether this is at all a good idea
        def get_with_formatting(path, *args):
            return api_method(request, 'get')(path.format(*args))
        script_env['get'] = get_with_formatting

        script_env['re'] = re
        script_env['dumps'] = dumps
        script_env['defaultdict'] = defaultdict
        script_env['OrderedDict'] = OrderedDict

        buf = []
        def dummy_print(*args):
            if len(args) == 1 and (isinstance(args[0], list) or isinstance(args[0], dict)):
                buf.append(dumps(args[0], indent=4))
            else:
                buf.append(' '.join(map(str, args)))
        script_env['print'] = dummy_print

        def run_script(script):
            try:
                exec script in script_env
            except:
                exception_info = sys.exc_info()
                buf.extend(traceback.format_exception(*exception_info))
            request.sdata.log('got reply {}'.format(buf))
            request.sdata.add_to_push_queue('script', text=dumps(buf))

        script = request.args['script'][0]
        reactor.callInThread(run_script, script)

def _exc_info_to_string(self, err, test):
        """Converts a sys.exc_info()-style tuple of values into a string."""
        exctype, value, tb = err
        # Skip test runner traceback levels
        while tb and self._is_relevant_tb_level(tb):
            tb = tb.tb_next

        if exctype is test.failureException:
            # Skip assert*() traceback levels
            length = self._count_relevant_tb_levels(tb)
            msgLines = traceback.format_exception(exctype, value, tb, length)
        else:
            msgLines = traceback.format_exception(exctype, value, tb)

        if self.buffer:
            output = sys.stdout.getvalue()
            error = sys.stderr.getvalue()
            if output:
                if not output.endswith('\n'):
                    output += '\n'
                msgLines.append(STDOUT_LINE % output)
            if error:
                if not error.endswith('\n'):
                    error += '\n'
                msgLines.append(STDERR_LINE % error)
        return ''.join(msgLines)

def addpackage(sitedir, name, known_paths):
    """Process a .pth file within the site-packages directory:
       For each line in the file, either combine it with sitedir to a path
       and add that to known_paths, or execute it if it starts with 'import '.
    """
    if known_paths is None:
        _init_pathinfo()
        reset = 1
    else:
        reset = 0
    fullname = os.path.join(sitedir, name)
    try:
        f = open(fullname, "rU")
    except IOError:
        return
    with f:
        for n, line in enumerate(f):
            if line.startswith("#"):
                continue
            try:
                if line.startswith(("import ", "import\t")):
                    exec line
                    continue
                line = line.rstrip()
                dir, dircase = makepath(sitedir, line)
                if not dircase in known_paths and os.path.exists(dir):
                    sys.path.append(dir)
                    known_paths.add(dircase)
            except Exception as err:
                print >>sys.stderr, "Error processing line {:d} of {}:\n".format(
                    n+1, fullname)
                for record in traceback.format_exception(*sys.exc_info()):
                    for line in record.splitlines():
                        print >>sys.stderr, '  '+line
                print >>sys.stderr, "\nRemainder of file ignored"
                break
    if reset:
        known_paths = None
    return known_paths

def handle(self, info=None):
        info = info or sys.exc_info()
        if self.format == "html":
            self.file.write(reset())

        formatter = (self.format=="html") and html or text
        plain = False
        try:
            doc = formatter(info, self.context)
        except:                         # just in case something goes wrong
            doc = ''.join(traceback.format_exception(*info))
            plain = True

        if self.display:
            if plain:
                doc = doc.replace('&', '&amp;').replace('<', '&lt;')
                self.file.write('<pre>' + doc + '</pre>\n')
            else:
                self.file.write(doc + '\n')
        else:
            self.file.write('<p>A problem occurred in a Python script.\n')

        if self.logdir is not None:
            suffix = ['.txt', '.html'][self.format=="html"]
            (fd, path) = tempfile.mkstemp(suffix=suffix, dir=self.logdir)
            try:
                file = os.fdopen(fd, 'w')
                file.write(doc)
                file.close()
                msg = '<p> %s contains the description of this error.' % path
            except:
                msg = '<p> Tried to save traceback to %s, but failed.' % path
            self.file.write(msg + '\n')
        try:
            self.file.flush()
        except: pass

def setup(level=logging.WARNING, outputs=[output.STDERR], program_name=None,
          capture_warnings=True):
    """Setup Python logging.

    This will setup basic handlers for Python logging.

    :param level: Root log level.
    :param outputs: Iterable of outputs to log to.
    :param program_name: The name of the program. Auto-detected if not set.
    :param capture_warnings: Capture warnings from the `warnings' module.
    """
    root_logger = logging.getLogger(None)

    # Remove all handlers
    for handler in list(root_logger.handlers):
        root_logger.removeHandler(handler)

    # Add configured handlers
    for out in outputs:
        if isinstance(out, str):
            out = output.preconfigured.get(out)
            if out is None:
                raise RuntimeError("Output {} is not available".format(out))
        out.add_to_logger(root_logger)

    root_logger.setLevel(level)

    program_logger = logging.getLogger(program_name)

    def logging_excepthook(exc_type, value, tb):
        program_logger.critical(
            "".join(traceback.format_exception(exc_type, value, tb)))

    sys.excepthook = logging_excepthook

    if capture_warnings:
        logging.captureWarnings(True)

def exception(self, message=None):
        "Log an exception as an error and debug if we're doing that."
        extype, ex, tb = sys.exc_info()
        message = "Exception: %s%s%s" % (
            message + ': ' if message is not None else '',
            ''.join(traceback.format_exception_only(extype, ex)),
            ''.join(traceback.format_exception(extype, ex, tb)).strip()
        )
        if self.forced_debug:
            self.debug(message)
        else:
            self.error(message)

    # Forced setting of debug level

def formatException(exctype, value, tb, skip=0):
    """Return a list of formatted exception strings.

    Similar to traceback.format_exception, but displays the entire stack trace
    rather than just the portion downstream of the point where the exception is
    caught. In particular, unhandled exceptions that occur during Qt signal
    handling do not usually show the portion of the stack that emitted the
    signal.
    """
    lines = traceback.format_exception(exctype, value, tb)
    lines = [lines[0]] + traceback.format_stack()[:-(skip+1)] + ['  --- exception caught here ---\n'] + lines[1:]
    return lines

def replyError(self, reqId, *exc):
        print("error: %s %s %s" % (self.name, str(reqId), str(exc[1])))
        excStr = traceback.format_exception(*exc)
        try:
            self.send(request='error', reqId=reqId, callSync='off', opts=dict(exception=exc[1], excString=excStr))
        except:
            self.send(request='error', reqId=reqId, callSync='off', opts=dict(exception=None, excString=excStr))

def formatException(exctype, value, tb, skip=0):
    """Return a list of formatted exception strings.

    Similar to traceback.format_exception, but displays the entire stack trace
    rather than just the portion downstream of the point where the exception is
    caught. In particular, unhandled exceptions that occur during Qt signal
    handling do not usually show the portion of the stack that emitted the
    signal.
    """
    lines = traceback.format_exception(exctype, value, tb)
    lines = [lines[0]] + traceback.format_stack()[:-(skip+1)] + ['  --- exception caught here ---\n'] + lines[1:]
    return lines