def _exitfunc(cls):
# At shutdown invoke finalizers for which atexit is true.
# This is called once all other non-daemonic threads have been
# joined.
reenable_gc = False
try:
if cls._registry:
import gc
if gc.isenabled():
reenable_gc = True
gc.disable()
pending = None
while True:
if pending is None or finalize._dirty:
pending = cls._select_for_exit()
finalize._dirty = False
if not pending:
break
f = pending.pop()
try:
# gc is disabled, so (assuming no daemonic
# threads) the following is the only line in
# this function which might trigger creation
# of a new finalizer
f()
except Exception:
sys.excepthook(*sys.exc_info())
assert f not in cls._registry
finally:
# prevent any more finalizers from executing during shutdown
finalize._shutdown = True
if reenable_gc:
gc.enable()
python类excepthook()的实例源码
def execsitecustomize():
"""Run custom site specific code, if available."""
try:
import sitecustomize
except ImportError:
pass
except Exception:
if sys.flags.verbose:
sys.excepthook(*sys.exc_info())
else:
print >>sys.stderr, \
"'import sitecustomize' failed; use -v for traceback"
def execusercustomize():
"""Run custom user specific code, if available."""
try:
import usercustomize
except ImportError:
pass
except Exception:
if sys.flags.verbose:
sys.excepthook(*sys.exc_info())
else:
print>>sys.stderr, \
"'import usercustomize' failed; use -v for traceback"
def enable(display=1, logdir=None, context=5, format="html"):
"""Install an exception handler that formats tracebacks as HTML.
The optional argument 'display' can be set to 0 to suppress sending the
traceback to the browser, and 'logdir' can be set to a directory to cause
tracebacks to be written to files there."""
sys.excepthook = Hook(display=display, logdir=logdir,
context=context, format=format)
def excepthook(*exception_info):
logger.error('unexpected error',exc_info=exception_info)
message = QtGui.QMessageBox(None)
message.setWindowTitle(QtGui.qApp.applicationName())
message.setIcon(QtGui.QMessageBox.Critical)
message.setText('An unexpected error occurred.')
message.setDetailedText(logging_stream.getvalue())
message.setStandardButtons(QtGui.QMessageBox.Ok)
message.setDefaultButton(QtGui.QMessageBox.Ok)
message.exec_()
QtGui.qApp.exit()
def init_logger():
logging.basicConfig(filename="pjf_{0}.log".format(time.strftime("%d_%m_%Y")), level=PYJFUZZ_LOGLEVEL)
logger = logging.getLogger(__name__)
sys.tracebacklimit = 10
def handle_exception(exc_type, exc_value, exc_traceback):
if issubclass(exc_type, KeyboardInterrupt):
sys.__excepthook__(exc_type, exc_value, exc_traceback)
return
logger.error("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback))
sys.__excepthook__(exc_type, exc_value, None)
return
sys.excepthook = handle_exception
return logger
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 crashtest():
global allWidgets
try:
gc.disable()
actions = [
createWidget,
#setParent,
forgetWidget,
showWidget,
processEvents,
#raiseException,
#addReference,
]
thread = WorkThread()
thread.start()
while True:
try:
action = randItem(actions)
action()
print('[%d widgets alive, %d zombie]' % (len(allWidgets), len(allWidgets) - len(widgets)))
except KeyboardInterrupt:
print("Caught interrupt; send another to exit.")
try:
for i in range(100):
QtTest.QTest.qWait(100)
except KeyboardInterrupt:
thread.terminate()
break
except:
sys.excepthook(*sys.exc_info())
finally:
gc.enable()
def printException(exctype, value, traceback):
"""Print an exception with its full traceback.
Set `sys.excepthook = printException` to ensure that exceptions caught
inside Qt signal handlers are printed with their full stack trace.
"""
print(''.join(formatException(exctype, value, traceback, skip=1)))
def implements(self, interface=None):
## this just makes it easy for us to detect whether an ExceptionHook is already installed.
if interface is None:
return ['ExceptionHandler']
else:
return interface == 'ExceptionHandler'
## replace built-in excepthook only if this has not already been done
def eventLoop(self):
while True:
try:
self.processRequests() # exception raised when the loop should exit
time.sleep(0.01)
except ClosedError:
break
except:
print("Error occurred in forked event loop:")
sys.excepthook(*sys.exc_info())
sys.exit(0)
def __exit__(self, *exc_info):
if self.proc is not None: ## worker
exceptOccurred = exc_info[0] is not None ## hit an exception during processing.
try:
if exceptOccurred:
sys.excepthook(*exc_info)
finally:
#print os.getpid(), 'exit'
os._exit(1 if exceptOccurred else 0)
else: ## parent
if self.showProgress:
self.progressDlg.__exit__(None, None, None)
def run(self):
global path, version, initVersion, forcedVersion, installVersion
name = self.config_vars['dist_name']
print(name)
path = os.path.join(self.install_libbase, 'NeoAnalysis')
if os.path.exists(path):
raise Exception("It appears another version of %s is already "
"installed at %s; remove this before installing."
% (name, path))
print("Installing to %s" % path)
rval = install.install.run(self)
# If the version in __init__ is different from the automatically-generated
# version string, then we will update __init__ in the install directory
if initVersion == version:
return rval
try:
initfile = os.path.join(path, '__init__.py')
data = open(initfile, 'r').read()
open(initfile, 'w').write(re.sub(r"__version__ = .*", "__version__ = '%s'" % version, data))
installVersion = version
except:
sys.stderr.write("Warning: Error occurred while setting version string in build path. "
"Installation will use the original version string "
"%s instead.\n" % (initVersion)
)
if forcedVersion:
raise
installVersion = initVersion
sys.excepthook(*sys.exc_info())
return rval
def crashtest():
global allWidgets
try:
gc.disable()
actions = [
createWidget,
#setParent,
forgetWidget,
showWidget,
processEvents,
#raiseException,
#addReference,
]
thread = WorkThread()
thread.start()
while True:
try:
action = randItem(actions)
action()
print('[%d widgets alive, %d zombie]' % (len(allWidgets), len(allWidgets) - len(widgets)))
except KeyboardInterrupt:
print("Caught interrupt; send another to exit.")
try:
for i in range(100):
QtTest.QTest.qWait(100)
except KeyboardInterrupt:
thread.terminate()
break
except:
sys.excepthook(*sys.exc_info())
finally:
gc.enable()
def printException(exctype, value, traceback):
"""Print an exception with its full traceback.
Set `sys.excepthook = printException` to ensure that exceptions caught
inside Qt signal handlers are printed with their full stack trace.
"""
print(''.join(formatException(exctype, value, traceback, skip=1)))
def implements(self, interface=None):
## this just makes it easy for us to detect whether an ExceptionHook is already installed.
if interface is None:
return ['ExceptionHandler']
else:
return interface == 'ExceptionHandler'
## replace built-in excepthook only if this has not already been done
def __exit__(self, *exc_info):
if self.proc is not None: ## worker
exceptOccurred = exc_info[0] is not None ## hit an exception during processing.
try:
if exceptOccurred:
sys.excepthook(*exc_info)
finally:
#print os.getpid(), 'exit'
os._exit(1 if exceptOccurred else 0)
else: ## parent
if self.showProgress:
self.progressDlg.__exit__(None, None, None)
def run(self):
global path, version, initVersion, forcedVersion, installVersion
name = self.config_vars['dist_name']
print(name)
path = os.path.join(self.install_libbase, 'NeoAnalysis')
if os.path.exists(path):
raise Exception("It appears another version of %s is already "
"installed at %s; remove this before installing."
% (name, path))
print("Installing to %s" % path)
rval = install.install.run(self)
# If the version in __init__ is different from the automatically-generated
# version string, then we will update __init__ in the install directory
if initVersion == version:
return rval
try:
initfile = os.path.join(path, '__init__.py')
data = open(initfile, 'r').read()
open(initfile, 'w').write(re.sub(r"__version__ = .*", "__version__ = '%s'" % version, data))
installVersion = version
except:
sys.stderr.write("Warning: Error occurred while setting version string in build path. "
"Installation will use the original version string "
"%s instead.\n" % (initVersion)
)
if forcedVersion:
raise
installVersion = initVersion
sys.excepthook(*sys.exc_info())
return rval
def exceptionhook(exception_type, exception, traceback, default_hook = sys.excepthook):
if 'Toehold' in exception_type.__name__ :
print("{}: {}".format('RuntimeError', exception.message))
else:
default_hook(exception_type, exception, traceback)
