def on_command_error(self, error, ctx):
ignored = (commands.NoPrivateMessage, commands.DisabledCommand, commands.CheckFailure,
commands.CommandNotFound, commands.UserInputError, discord.HTTPException)
error = getattr(error, 'original', error)
if isinstance(error, ignored):
return
if ctx.message.server:
fmt = 'Channel: {0} (ID: {0.id})\nGuild: {1} (ID: {1.id})'
else:
fmt = 'Channel: {0} (ID: {0.id})'
exc = traceback.format_exception(type(error), error, error.__traceback__, chain=False)
description = '```py\n%s\n```' % ''.join(exc)
time = datetime.datetime.utcnow()
name = ctx.command.qualified_name
author = '{0} (ID: {0.id})'.format(ctx.message.author)
location = fmt.format(ctx.message.channel, ctx.message.server)
message = '{0} at {1}: Called by: {2} in {3}. More info: {4}'.format(name, time, author, location, description)
self.bot.logs['discord'].critical(message)
python类format_exception()的实例源码
def upload_output_xml(request):
if request.method == 'POST':
form = UploadOutputXmlForm(request.POST, request.FILES)
print "HELLP"
if form.is_valid():
print "YES"
try:
handle_uploaded_file(request)
except:
tt, value, tb = sys.exc_info()
print {'exception_value': value,
'value': tt,
'tb': traceback.format_exception(tt, value, tb)}
return handler500(request)
return HttpResponseRedirect(reverse('home'))
else:
return handler500(request)
else:
print "No"
form = UploadOutputXmlForm()
return render(request, 'report/upload_xml_file.html', {'form': form})
def handler500(request, template_name='500.html'):
t = get_template(template_name)
tt, value, tb = sys.exc_info()
ctx = Context({'exception_value': value,
'value': tt,
'tb': traceback.format_exception(tt, value, tb)})
return HttpResponseServerError(t.render(ctx))
def export_as_string(self):
"""
Returns a CQL query string that can be used to recreate the entire keyspace,
including user-defined types and tables.
"""
cql = "\n\n".join([self.as_cql_query() + ';']
+ self.user_type_strings()
+ [f.export_as_string() for f in self.functions.values()]
+ [a.export_as_string() for a in self.aggregates.values()]
+ [t.export_as_string() for t in self.tables.values()])
if self._exc_info:
import traceback
ret = "/*\nWarning: Keyspace %s is incomplete because of an error processing metadata.\n" % \
(self.name)
for line in traceback.format_exception(*self._exc_info):
ret += line
ret += "\nApproximate structure, for reference:\n(this should not be used to reproduce this schema)\n\n%s\n*/" % cql
return ret
return cql
def export_as_string(self):
"""
Returns a string of CQL queries that can be used to recreate this table
along with all indexes on it. The returned string is formatted to
be human readable.
"""
if self._exc_info:
import traceback
ret = "/*\nWarning: Table %s.%s is incomplete because of an error processing metadata.\n" % \
(self.keyspace_name, self.name)
for line in traceback.format_exception(*self._exc_info):
ret += line
ret += "\nApproximate structure, for reference:\n(this should not be used to reproduce this schema)\n\n%s\n*/" % self._all_as_cql()
elif not self.is_cql_compatible:
# If we can't produce this table with CQL, comment inline
ret = "/*\nWarning: Table %s.%s omitted because it has constructs not compatible with CQL (was created via legacy API).\n" % \
(self.keyspace_name, self.name)
ret += "\nApproximate structure, for reference:\n(this should not be used to reproduce this schema)\n\n%s\n*/" % self._all_as_cql()
else:
ret = self._all_as_cql()
return ret
def error_response(func):
@wraps(func)
def wrapper(self, *args, **kwargs):
try:
result = func(self, *args, **kwargs)
except Exception as ex:
if not isinstance(ex, (web.HTTPError, ExecutionError, GraphQLError)):
tb = ''.join(traceback.format_exception(*sys.exc_info()))
app_log.error('Error: {0} {1}'.format(ex, tb))
self.set_status(error_status(ex))
error_json = json_encode({'errors': error_format(ex)})
app_log.debug('error_json: %s', error_json)
self.write(error_json)
else:
return result
return wrapper
def write_error(self, status_code, **kwargs):
"""Override to implement custom error pages.
``write_error`` may call `write`, `render`, `set_header`, etc
to produce output as usual.
If this error was caused by an uncaught exception (including
HTTPError), an ``exc_info`` triple will be available as
``kwargs["exc_info"]``. Note that this exception may not be
the "current" exception for purposes of methods like
``sys.exc_info()`` or ``traceback.format_exc``.
"""
if self.settings.get("serve_traceback") and "exc_info" in kwargs:
# in debug mode, try to send a traceback
self.set_header('Content-Type', 'text/plain')
for line in traceback.format_exception(*kwargs["exc_info"]):
self.write(line)
self.finish()
else:
self.finish("<html><title>%(code)d: %(message)s</title>"
"<body>%(code)d: %(message)s</body></html>" % {
"code": status_code,
"message": self._reason,
})
def write_error(self, status_code, **kwargs):
"""Override to implement custom error pages.
``write_error`` may call `write`, `render`, `set_header`, etc
to produce output as usual.
If this error was caused by an uncaught exception (including
HTTPError), an ``exc_info`` triple will be available as
``kwargs["exc_info"]``. Note that this exception may not be
the "current" exception for purposes of methods like
``sys.exc_info()`` or ``traceback.format_exc``.
"""
if self.settings.get("serve_traceback") and "exc_info" in kwargs:
# in debug mode, try to send a traceback
self.set_header('Content-Type', 'text/plain')
for line in traceback.format_exception(*kwargs["exc_info"]):
self.write(line)
self.finish()
else:
self.finish("<html><title>%(code)d: %(message)s</title>"
"<body>%(code)d: %(message)s</body></html>" % {
"code": status_code,
"message": self._reason,
})
def write_error(self, status_code, **kwargs):
self.set_header('Content-Type', 'text/json')
if self.settings.get("serve_traceback") and "exc_info" in kwargs:
# in debug mode, try to send a traceback
lines = []
for line in traceback.format_exception(*kwargs["exc_info"]):
lines.append(line)
self.finish(json.dumps({
'code': status_code,
'status': self._reason,
'traceback': lines,
'results': {},
}))
else:
self.finish(json.dumps({
'code': status_code,
'status': self._reason,
'results': {},
}))
def write_error(self, status_code, **kwargs):
self.set_header('Content-Type', 'text/json')
if self.settings.get("serve_traceback") and "exc_info" in kwargs:
# in debug mode, try to send a traceback
lines = []
for line in traceback.format_exception(*kwargs["exc_info"]):
lines.append(line)
self.finish(json.dumps({
'code': status_code,
'status': self._reason,
'traceback': lines,
'results': {},
}))
else:
self.finish(json.dumps({
'code': status_code,
'status': self._reason,
'results': {},
}))
def write_error(self, status_code, **kwargs):
self.set_header('Content-Type', 'text/json')
if self.settings.get("serve_traceback") and "exc_info" in kwargs:
# in debug mode, try to send a traceback
lines = []
for line in traceback.format_exception(*kwargs["exc_info"]):
lines.append(line)
self.finish(json.dumps({
'code': status_code,
'status': self._reason,
'traceback': lines,
'results': {},
}))
else:
self.finish(json.dumps({
'code': status_code,
'status': self._reason,
'results': {},
}))
def build_fault(cls, e, tb, include_traceback=False):
"""
Builds a L{ErrorFault<pyamf.remoting.ErrorFault>} object based on the last
exception raised.
If include_traceback is C{False} then the traceback will not be added to
the L{remoting.ErrorFault}.
"""
if hasattr(cls, '_amf_code'):
code = cls._amf_code
else:
code = cls.__name__
details = None
if include_traceback:
details = traceback.format_exception(cls, e, tb)
return remoting.ErrorFault(
code=code,
description=unicode(e),
details=details
)
def safe_str(o):
"""safe_str(anything) -> string
Returns a string representation of an object, or a string containing a
traceback, if that object's __str__ raised an exception.
"""
try:
return str(o)
except:
strExc = '\n'.join(traceback.format_exception(*sys.exc_info()))
clsName = _determineClassName(o)
obId = id(o)
return '<%s instance at %s with str error %s>' % (
clsName, obId, strExc)
##the following were factored out of usage
def format_exception(exc_info, encoding='UTF-8'):
ec, ev, tb = exc_info
# Our exception object may have been turned into a string, and Python 3's
# traceback.format_exception() doesn't take kindly to that (it expects an
# actual exception object). So we work around it, by doing the work
# ourselves if ev is not an exception object.
if not is_base_exception(ev):
tb_data = force_unicode(
''.join(traceback.format_tb(tb)),
encoding)
ev = exc_to_unicode(ev)
return tb_data + ev
else:
return force_unicode(
''.join(traceback.format_exception(*exc_info)),
encoding)
def write_error(self, status_code, **kwargs):
"""
Method gets traceback, writes it into response, finishes response.
:param status_code: tornado parameter to format html, we don't use it.
:type status_code: int
:param kwargs: in debug mode must contain exc_info.
:type kwargs: dict
"""
exc_info = kwargs.get('exc_info')
if self.settings.get(
"serve_traceback") and exc_info: # pragma: no cover
# in debug mode, try to send a traceback
self.set_header('Content-Type', 'text/plain')
for line in traceback.format_exception(*exc_info):
self.write(line)
# exc_info[1] - HTTPError instance
# Finish request with exception body or exception reason
err_text = getattr(exc_info[1], 'body', self._reason)
self.write(err_text)
self.finish()
visualstudio_py_testlauncher.py 文件源码
项目：pythonVSCode
作者: DonJayamanne
项目源码
文件源码
阅读 125
收藏 0
点赞 0
评论 0
def sendResult(self, test, outcome, trace = None):
if _channel is not None:
tb = None
message = None
if trace is not None:
traceback.print_exc()
formatted = traceback.format_exception(*trace)
# Remove the 'Traceback (most recent call last)'
formatted = formatted[1:]
tb = ''.join(formatted)
message = str(trace[1])
_channel.send_event(
name='result',
outcome=outcome,
traceback = tb,
message = message,
test = test.id()
)
def excepthook( self, type_, value, tb ):
# emergency write
self.__session_log.write( 'excepthook called\n' )
self.__session_log.flush()
all_exception_lines = traceback.format_exception( type_, value, tb )
for line in all_exception_lines:
self.__session_log.write( line )
self.__session_log.flush()
# cannot use the GUI window now app is not sane
self.app.log.removeHandler( self.widget_log_handler )
self.app.log.error( 'excepthook called' )
for line in all_exception_lines:
self.app.log.error( line.replace( '\n', '' ) )
self.app.runInForeground( self.app.log.addHandler, (self.widget_log_handler,) )
self.app.runInForeground( self.app.log.error, ('Check log for traceback details',) )
def _process_wrapper(function, upwards, profile, *args, **kwargs):
"""Wrap a process with additional features."""
try:
if profile:
_run_profiler(function, *args, **kwargs)
else:
function(*args, **kwargs)
except Exception:
process = multiprocessing.current_process()
info = sys.exc_info()
exception = traceback.format_exception(
info[0], info[1], info[2].tb_next)
_send_message(upwards, _MESSAGE_ERROR,
process_id=process.pid,
process_name=process.name,
message=''.join(exception).rstrip())
finally:
upwards.close()
def addFailure(self, test, err, systemout = '', systemerr = ''):
"""Add failure output to Xunit report.
"""
taken = self._timeTaken()
tb = format_exception(err)
self.stats['failures'] += 1
id = test.id()
self.errorlist.append(
'<testcase classname=%(cls)s name=%(name)s time="%(taken).3f">'
'<failure type=%(errtype)s message=%(message)s><![CDATA[%(tb)s]]>'
'</failure><system-out>%(systemout)s</system-out><system-err>%(systemerr)s</system-err></testcase>' %
{'cls': self._quoteattr(id_split(id)[0]),
'name': self._quoteattr(id_split(id)[-1]),
'taken': taken,
'errtype': self._quoteattr(nice_classname(err[0])),
'message': self._quoteattr(exc_message(err)),
'tb': escape_cdata(tb),
'systemout': systemout,
'systemerr': systemerr,
})
def lambda_handler(data, context):
"""
Execution wrapper on AWS Lambda
:param data: Inputs to the botengine
:param context: Ignored
:return: JSON structure with errors and debug information
"""
if data is None:
return 0
logger = LambdaLogger()
try:
bot = importlib.import_module('bot')
botengine._run(bot, data, logger)
except:
import traceback
import sys
(t, v, tb) = sys.exc_info()
logger.tracebacks = traceback.format_exception(t, v, tb)
return logger.get_lambda_return()
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 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))
# in python3.5, convert FrameSummary to tuples (py2.7+ compatibility)
tb = [tuple(fs) for fs in traceback.extract_tb(trace)]
title = traceback.format_exception_only(extype, exvalue)[0]
# send an Exception notification
msg = {'method': 'exception',
'args': (title, extype.__name__, repr(exvalue), tb, msg),
'id': None}
self.pipe.send(msg)
self.interaction(frame)
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 generic(context, request):
settings = request.registry.settings
with JSONAPIResponse(request.response) as resp:
_in = u'Failed'
code, status = JSONAPIResponse.INTERNAL_SERVER_ERROR
request.response.status_int = code
try:
message = {'message': context.args[0]}
except IndexError:
message = {'message': 'Unknown error'}
if settings.get('baka.debug', True):
message['traceback'] = ''.join(
traceback.format_exception(*request.exc_info))
return resp.to_json(
_in, code=code,
status=status, message=message)
def __exit__(self, exec_type, exec_value, trace):
self.body_executed = True
retval = False
# SystemExit is ignored to properly handle forks that don't exec
if exec_type in (None, SystemExit):
return retval
elif exec_type is errors.SignalExit:
logger.debug("Encountered signals: %s", self.received_signals)
retval = True
else:
logger.debug("Encountered exception:\n%s", "".join(
traceback.format_exception(exec_type, exec_value, trace)))
self._call_registered()
self._reset_signal_handlers()
self._call_signals()
return retval
def checkListenTask(self):
self.listener.unschedule_all()
if self.listenDirEnableFlag.get():
print("enable")
listen_dir = os.path.abspath(self.listenDirPath.get())
try:
self.listener.schedule(self._listen_handler,
listen_dir, recursive = True)
except Exception as e:
messagebox.showinfo("listen dir {dirname} failed"
.format(dirname = listen_dir),
"{reason}".format(
reason = traceback.format_exception(*sys.exc_info())
)
)
else:
print("disable")
def update_with_translator(self, canonical_ref, dataset_iterator):
translator = dataset_iterator.translator
translator_info = {
'id': unicode(translator.versioned_id()),
}
result = next(dataset_iterator)
translated = result['translated']
raw = result['raw_content']
local_assets = result.get('local_assets', {})
try:
refs = self.submit_translator_output(
translator_info,
translated,
raw,
local_assets,
existing_canonical_ref=multihash_ref(canonical_ref))
return refs
except AbortionError:
for line in traceback.format_exception(*sys.exc_info()):
print_err(line.rstrip('\n'))
def write_dataset(self, dataset_iterator):
translator = dataset_iterator.translator
translator_info = {
'id': translator.versioned_id(),
}
try:
translator_info['link'] = multihash_ref(
translator.__version__).to_map()
except (ValueError, TypeError):
pass
for result in dataset_iterator:
translated = result['translated']
raw = result['raw_content']
local_assets = result.get('local_assets', {})
try:
refs = self.submit_translator_output(translator_info, translated,
raw, local_assets)
yield refs
except AbortionError:
for line in traceback.format_exception(*sys.exc_info()):
print_err(line.rstrip('\n'))
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 on_command_error(ctx, error):
if isinstance(error, commands.errors.CommandNotFound):
pass
if isinstance(error, commands.errors.CheckFailure):
await ctx.send("{} You don't have permission to use this command.".format(ctx.message.author.mention))
elif isinstance(error, commands.errors.MissingRequiredArgument):
formatter = commands.formatter.HelpFormatter()
msg = await formatter.format_help_for(ctx, ctx.command)
await ctx.send("{} You are missing required arguments.\n{}".format(ctx.message.author.mention, msg[0]))
elif isinstance(error, commands.errors.CommandOnCooldown):
try:
await ctx.message.delete()
except discord.errors.NotFound:
pass
message = await ctx.message.channel.send("{} This command was used {:.2f}s ago and is on cooldown. Try again in {:.2f}s.".format(ctx.message.author.mention, error.cooldown.per - error.retry_after, error.retry_after))
await asyncio.sleep(10)
await message.delete()
else:
await ctx.send("An error occured while processing the `{}` command.".format(ctx.command.name))
print('Ignoring exception in command {0.command} in {0.message.channel}'.format(ctx))
botdev_msg = "Exception occured in `{0.command}` in {0.message.channel.mention}".format(ctx)
tb = traceback.format_exception(type(error), error, error.__traceback__)
print(''.join(tb))
await bot.botdev_channel.send(botdev_msg + '\n```' + ''.join(tb) + '\n```')
