Python part of the warnings subsystem.
This commit is contained in:
Guido van Rossum
parent
cfd42b556b
commit
2a862c614d
|
|
@ -0,0 +1,227 @@
|
|||
"""Python part of the warnings subsystem."""
|
||||
|
||||
import sys, re, types
|
||||
|
||||
defaultaction = "default"
|
||||
filters = []
|
||||
onceregistry = {}
|
||||
|
||||
def warn(message, category=None, stacklevel=1):
|
||||
"""Issue a warning, or maybe ignore it or raise an exception."""
|
||||
# Check category argument
|
||||
if category is None:
|
||||
category = UserWarning
|
||||
assert issubclass(category, Warning)
|
||||
# Get context information
|
||||
try:
|
||||
caller = sys._getframe(stacklevel)
|
||||
except ValueError:
|
||||
globals = sys.__dict__
|
||||
lineno = 1
|
||||
else:
|
||||
globals = caller.f_globals
|
||||
lineno = caller.f_lineno
|
||||
module = globals['__name__']
|
||||
filename = globals.get('__file__')
|
||||
if filename:
|
||||
fnl = filename.lower()
|
||||
if fnl.endswith(".pyc") or fnl.endswith(".pyo"):
|
||||
filename = filename[:-1]
|
||||
else:
|
||||
if module == "__main__":
|
||||
filename = sys.argv[0]
|
||||
if not filename:
|
||||
filename = module
|
||||
# Quick test for common case
|
||||
registry = globals.setdefault("__warningregistry__", {})
|
||||
key = (message, category, lineno)
|
||||
if registry.get(key):
|
||||
return
|
||||
# Search the filters
|
||||
for item in filters:
|
||||
action, msg, cat, mod, ln = item
|
||||
if (msg.match(message) and
|
||||
issubclass(category, cat) and
|
||||
mod.match(module) and
|
||||
(ln == 0 or lineno == ln)):
|
||||
break
|
||||
else:
|
||||
action = defaultaction
|
||||
# Early exit actions
|
||||
if action == "ignore":
|
||||
registry[key] = 1
|
||||
return
|
||||
if action == "error":
|
||||
raise category(message)
|
||||
# Other actions
|
||||
if action == "once":
|
||||
registry[key] = 1
|
||||
oncekey = (message, category)
|
||||
if onceregistry.get(oncekey):
|
||||
return
|
||||
onceregistry[oncekey] = 1
|
||||
elif action == "always":
|
||||
pass
|
||||
elif action == "module":
|
||||
registry[key] = 1
|
||||
altkey = (message, category, 0)
|
||||
if registry.get(altkey):
|
||||
return
|
||||
registry[altkey] = 1
|
||||
elif action == "default":
|
||||
registry[key] = 1
|
||||
else:
|
||||
# Unrecognized actions are errors
|
||||
raise RuntimeError(
|
||||
"Unrecognized action (%s) in warnings.filters:\n %s" %
|
||||
(`action`, str(item)))
|
||||
# Print message and context
|
||||
showwarning(message, category, filename, lineno)
|
||||
|
||||
def showwarning(message, category, filename, lineno, file=None):
|
||||
"""Hook to write a warning to a file; replace if you like."""
|
||||
if file is None:
|
||||
file = sys.stderr
|
||||
file.write(formatwarning(message, category, filename, lineno))
|
||||
|
||||
def formatwarning(message, category, filename, lineno):
|
||||
"""Hook to format a warning the standard way."""
|
||||
import linecache
|
||||
s = "%s:%s: %s: %s\n" % (filename, lineno, category.__name__, message)
|
||||
line = linecache.getline(filename, lineno).strip()
|
||||
if line:
|
||||
s = s + " " + line + "\n"
|
||||
return s
|
||||
|
||||
def filterwarnings(action, message="", category=Warning, module="", lineno=0):
|
||||
"""Insert an entry into the list of warnings filters (at the front).
|
||||
|
||||
Use assertions to check that all arguments have the right type."""
|
||||
assert action in ("error", "ignore", "always", "default", "module",
|
||||
"once"), "invalid action: %s" % `action`
|
||||
assert isinstance(message, types.StringType), "message must be a string"
|
||||
assert isinstance(category, types.ClassType), "category must be a class"
|
||||
assert issubclass(category, Warning), "category must be a Warning subclass"
|
||||
assert type(module) is types.StringType, "module must be a string"
|
||||
assert type(lineno) is types.IntType and lineno >= 0, \
|
||||
"lineno must be an int >= 0"
|
||||
filters.insert(0, (action, re.compile(message, re.I), category,
|
||||
re.compile(module), lineno))
|
||||
|
||||
def resetwarnings():
|
||||
"""Reset the list of warnings filters to its default state."""
|
||||
filters[:] = []
|
||||
|
||||
class _OptionError(Exception):
|
||||
"""Exception used by option processing helpers."""
|
||||
pass
|
||||
|
||||
# Helper to process -W options passed via sys.warnoptions
|
||||
def _processoptions(args):
|
||||
for arg in args:
|
||||
try:
|
||||
_setoption(arg)
|
||||
except _OptionError, msg:
|
||||
print >>sys.stderr, "Invalid -W option ignored:", msg
|
||||
|
||||
# Helper for _processoptions()
|
||||
def _setoption(arg):
|
||||
parts = arg.split(':')
|
||||
if len(parts) > 5:
|
||||
raise _OptionError("unparsable -W option %s" % `arg`)
|
||||
while len(parts) < 5:
|
||||
parts.append('')
|
||||
action, message, category, module, lineno = [s.strip()
|
||||
for s in parts]
|
||||
action = _getaction(action)
|
||||
message = re.escape(message)
|
||||
category = _getcategory(category)
|
||||
module = re.escape(module)
|
||||
if module:
|
||||
module = module + '$'
|
||||
if lineno:
|
||||
try:
|
||||
lineno = int(lineno)
|
||||
if lineno < 0:
|
||||
raise ValueError
|
||||
except (ValueError, OverflowError):
|
||||
raise _OptionError("invalid lineno %s" % `lineno`)
|
||||
else:
|
||||
lineno = 0
|
||||
filterwarnings(action, message, category, module, lineno)
|
||||
|
||||
# Helper for _setoption()
|
||||
def _getaction(action):
|
||||
if not action:
|
||||
return "default"
|
||||
if action == "all": return "always" # Alias
|
||||
for a in ['default', 'always', 'ignore', 'module', 'once', 'error']:
|
||||
if a.startswith(action):
|
||||
return a
|
||||
raise _OptionError("invalid action: %s" % `action`)
|
||||
|
||||
# Helper for _setoption()
|
||||
def _getcategory(category):
|
||||
if not category:
|
||||
return Warning
|
||||
if re.match("^[a-zA-Z0-9_]+$", category):
|
||||
try:
|
||||
cat = eval(category)
|
||||
except KeyError:
|
||||
raise _OptionError("invalid warning category: %s" % `category`)
|
||||
else:
|
||||
i = category.rfind(".")
|
||||
module = category[:i]
|
||||
klass = category[i+1:]
|
||||
m = __import__(module, None, None, [klass])
|
||||
cat = getattr(m, klass)
|
||||
if (not isinstance(cat, types.ClassType) or
|
||||
not issubclass(cat, Warning)):
|
||||
raise _OptionError("invalid warning category: %s" % `category`)
|
||||
return cat
|
||||
|
||||
# Self-test
|
||||
def _test():
|
||||
import getopt
|
||||
testoptions = []
|
||||
try:
|
||||
opts, args = getopt.getopt(sys.argv[1:], "W:")
|
||||
except getopt.error, msg:
|
||||
print >>sys.stderr, msg
|
||||
return
|
||||
for o, a in opts:
|
||||
testoptions.append(a)
|
||||
try:
|
||||
_processoptions(testoptions)
|
||||
except _OptionError, msg:
|
||||
print >>sys.stderr, msg
|
||||
return
|
||||
for item in filters: print item
|
||||
hello = "hello world"
|
||||
warn(hello); warn(hello); warn(hello); warn(hello)
|
||||
warn(hello, UserWarning)
|
||||
warn(hello, DeprecationWarning)
|
||||
for i in range(3):
|
||||
warn(hello)
|
||||
filterwarnings("error", "", Warning, "", 0)
|
||||
try:
|
||||
warn(hello)
|
||||
except Exception, msg:
|
||||
print "Caught", msg.__class__.__name__ + ":", msg
|
||||
else:
|
||||
print "No exception"
|
||||
resetwarnings()
|
||||
try:
|
||||
filterwarnings("booh", "", Warning, "", 0)
|
||||
except Exception, msg:
|
||||
print "Caught", msg.__class__.__name__ + ":", msg
|
||||
else:
|
||||
print "No exception"
|
||||
|
||||
# Module initialization
|
||||
if __name__ == "__main__":
|
||||
import __main__
|
||||
sys.modules['warnings'] = __main__
|
||||
_test()
|
||||
else:
|
||||
_processoptions(sys.warnoptions)
|
||||
Loading…
Reference in New Issue