def LOGEXCEPTION(message):
if LOGGER:
LOGGER.exception(message)
else:
print(
'%sZ [EXC!]: %s\nexception was: %s' % (
datetime.utcnow().isoformat(),
message, format_exc()
)
)
###################
## LOCAL IMPORTS ##
###################
# LC reading functions
python类format_exc()的实例源码
def LOGEXCEPTION(message):
if LOGGER:
LOGGER.exception(message)
else:
print(
'%sZ [EXC!]: %s\nexception was: %s' % (
datetime.utcnow().isoformat(),
message, format_exc()
)
)
#######################
## UTILITY FUNCTIONS ##
#######################
# this is from Tornado's source (MIT License):
# http://www.tornadoweb.org/en/stable/_modules/tornado/escape.html#squeeze
def LOGEXCEPTION(message):
if LOGGER:
LOGGER.exception(message)
else:
print(
'%sZ [EXC!]: %s\nexception was: %s' % (
datetime.utcnow().isoformat(),
message, format_exc()
)
)
###################
## FORM SETTINGS ##
###################
def retry_wrapper(func):
def try_and_retry(*args, **kwargs):
retry = False
try:
return func(*args, **kwargs)
except utils.SOSError as e:
# if we got an http error and
# the string contains 401 or if the string contains the word cookie
if (e.err_code == utils.SOSError.HTTP_ERR and
(e.err_text.find('401') != -1 or
e.err_text.lower().find('cookie') != -1)):
retry = True
CoprHDCLIDriver.AUTHENTICATED = False
else:
exception_message = "\nViPR Exception: %s\nStack Trace:\n%s" \
% (e.err_text, traceback.format_exc())
raise utils.SOSError(utils.SOSError.SOS_FAILURE_ERR,"Exception is : "+exception_message)
except Exception:
exception_message = "\nGeneral Exception: %s\nStack Trace:\n%s" \
% (sys.exc_info()[0], traceback.format_exc())
raise utils.SOSError(utils.SOSError.SOS_FAILURE_ERR,"Exception is : "+exception_message)
if retry:
return func(*args, **kwargs)
return try_and_retry
def __init__(self, *args):
# build the message from the user
user_msg = args[0] if args else "An error has occurred"
# get the traceback from the last error if it exists
try: tb = traceback.format_exc()
# otherwise, get the tb prior to this frame and pretend its us
except:
tb = "Traceback (most recent call last):\n"
tb += ''.join(traceback.format_stack()[:-1])
tb += self.__class__.__name__ + ": " + user_msg
tb += '\n'
# build the complete log message
log_msg = user_msg + "\n" + tb
# and log it
self.log.error(log_msg)
# store the args
self.args = args
def update():
try:
ident = screen.after(1000 / FPS, update)
for mutate in container, governor:
for ball in balls:
mutate(ball)
for index, ball_1 in enumerate(balls[:-1]):
for ball_2 in balls[index+1:]:
ball_1.crash(ball_2)
for ball in balls:
ball.move(FPS)
screen.delete(ALL)
for ball in balls:
x1 = ball.pos.x - BALL_RADIUS
y1 = ball.pos.y - BALL_RADIUS
x2 = ball.pos.x + BALL_RADIUS
y2 = ball.pos.y + BALL_RADIUS
screen.create_oval(x1, y1, x2, y2, fill=BALL_COLOR)
except:
screen.after_cancel(ident)
screen.delete(ALL)
screen.create_text(SCREEN_WIDTH / 2, SCREEN_HEIGHT / 2, text=format_exc(), font='Courier 10', fill='red')
def update():
'Run physics and update screen.'
global frame
try:
for mutate in wall, floor, gravity, friction, governor:
for ball in balls:
mutate(ball)
for index, ball_1 in enumerate(balls[:-1]):
for ball_2 in balls[index+1:]:
ball_1.crash(ball_2)
for ball in balls:
ball.move(FPS)
screen.delete('animate')
for ball in balls:
x1 = ball.pos.x - ball.rad
y1 = ball.pos.y - ball.rad
x2 = ball.pos.x + ball.rad
y2 = ball.pos.y + ball.rad
screen.create_oval(x1, y1, x2, y2, fill=BALL_COLOR, tag='animate')
frame += 1
screen.after(int((start + frame / FPS - time.clock()) * 1000), update)
except:
screen.delete(Tkinter.ALL)
screen.create_text(SCREEN_WIDTH / 2, SCREEN_HEIGHT / 2, text=traceback.format_exc(), font='Courier 10', fill='red', tag='animate')
def main(args=None):
if args is None:
args = sys.argv[1:]
if not len(args):
sys.stderr.write(USAGE)
sys.exit(1)
elif args[0] in ("-h", "--help"):
sys.stderr.write(__doc__ + "\n")
sys.exit(1)
elif args[0][:1] == "-":
sys.stderr.write(USAGE)
sys.exit(1)
module_name = args.pop(0)
try:
module = import_module(".utilities." + module_name,
package="mjolnir")
except ImportError:
sys.stderr.write(traceback.format_exc())
sys.stderr.write("Could not find utility %s" % (module_name))
sys.exit(1)
module.main(args)
def git_reinstall():
"""Reinstall from source and restart services.
If the openstack-origin-git config option was used to install openstack
from source git repositories, then this action can be used to reinstall
from updated git repositories, followed by a restart of services."""
if not git_install_requested():
action_fail('openstack-origin-git is not configured')
return
try:
git_install(config('openstack-origin-git'))
config_changed()
except:
action_set({'traceback': traceback.format_exc()})
action_fail('git-reinstall resulted in an unexpected error')
def catch_error(func):
def __tryexcept__(*args, **kwargs):
try:
func(*args, **kwargs)
except Exception as expected:
traceback.print_exc(file=sys.stdout)
stacktrace = traceback.format_exc()
exception = {
'exception': expected,
'stacktrace': stacktrace,
}
stacktrace_handle(exception)
return __tryexcept__
def process(self):
""" Main entry point. All archives are processed """
success = 0
self.archives_processed = 0
for nodename, nodeidx, archive in self.job.nodearchives():
try:
self.processarchive(nodename, nodeidx, archive)
self.archives_processed += 1
except pmapi.pmErr as exc:
success -= 1
#pylint: disable=not-callable
self.adderror("archive", "{0}: pmapi.pmErr: {1}".format(archive, exc.message()))
except Exception as exc:
success -= 1
self.adderror("archive", "{0}: Exception: {1}. {2}".format(archive, str(exc), traceback.format_exc()))
return success == 0
def setup_py(self):
assert self.source_dir, "No source dir for %s" % self
try:
import setuptools # noqa
except ImportError:
if get_installed_version('setuptools') is None:
add_msg = "Please install setuptools."
else:
add_msg = traceback.format_exc()
# Setuptools is not available
raise InstallationError(
"Could not import setuptools which is required to "
"install from a source distribution.\n%s" % add_msg
)
setup_py = os.path.join(self.setup_py_dir, 'setup.py')
# Python2 __file__ should not be unicode
if six.PY2 and isinstance(setup_py, six.text_type):
setup_py = setup_py.encode(sys.getfilesystemencoding())
return setup_py
def git_reinstall():
"""Reinstall from source and restart services.
If the openstack-origin-git config option was used to install openstack
from source git repositories, then this action can be used to reinstall
from updated git repositories, followed by a restart of services."""
if not git_install_requested():
action_fail('openstack-origin-git is not configured')
return
try:
git_install(config('openstack-origin-git'))
config_changed()
except:
action_set({'traceback': traceback.format_exc()})
action_fail('git-reinstall resulted in an unexpected error')
def setup_py(self):
assert self.source_dir, "No source dir for %s" % self
try:
import setuptools # noqa
except ImportError:
if get_installed_version('setuptools') is None:
add_msg = "Please install setuptools."
else:
add_msg = traceback.format_exc()
# Setuptools is not available
raise InstallationError(
"Could not import setuptools which is required to "
"install from a source distribution.\n%s" % add_msg
)
setup_py = os.path.join(self.setup_py_dir, 'setup.py')
# Python2 __file__ should not be unicode
if six.PY2 and isinstance(setup_py, six.text_type):
setup_py = setup_py.encode(sys.getfilesystemencoding())
return setup_py
def test_timeout(self):
# Set a short timeout and exceed it.
@gen_test(timeout=0.1)
def test(self):
yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)
# This can't use assertRaises because we need to inspect the
# exc_info triple (and not just the exception object)
try:
test(self)
self.fail("did not get expected exception")
except ioloop.TimeoutError:
# The stack trace should blame the add_timeout line, not just
# unrelated IOLoop/testing internals.
self.assertIn(
"gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
traceback.format_exc())
self.finished = True
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 setup_py(self):
assert self.source_dir, "No source dir for %s" % self
try:
import setuptools # noqa
except ImportError:
if get_installed_version('setuptools') is None:
add_msg = "Please install setuptools."
else:
add_msg = traceback.format_exc()
# Setuptools is not available
raise InstallationError(
"Could not import setuptools which is required to "
"install from a source distribution.\n%s" % add_msg
)
setup_py = os.path.join(self.setup_py_dir, 'setup.py')
# Python2 __file__ should not be unicode
if six.PY2 and isinstance(setup_py, six.text_type):
setup_py = setup_py.encode(sys.getfilesystemencoding())
return setup_py
def test_timeout(self):
# Set a short timeout and exceed it.
@gen_test(timeout=0.1)
def test(self):
yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)
# This can't use assertRaises because we need to inspect the
# exc_info triple (and not just the exception object)
try:
test(self)
self.fail("did not get expected exception")
except ioloop.TimeoutError:
# The stack trace should blame the add_timeout line, not just
# unrelated IOLoop/testing internals.
self.assertIn(
"gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
traceback.format_exc())
self.finished = True
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,
})
