Add excutils.exception_filter
This is a useful decorator for writing a simple function to filter or translate exceptions based on criteria more complex than simply the (super)class (i.e. in situations that aren't amenable to being handled by simply matching subclasses with Python's exception handling syntax). For example, this can be used to filter on the HTTP status code from a generic client exception. Unlike a naive implementation at the point of need or using contextlib to create a reusable context manager, using this library class also makes it easy to ensure that any exceptions that are re-raised (not filtered) retain the same traceback. Change-Id: Ib986ccaf95a19ef14fef1cebe39fc87c17c2769f
This commit is contained in:
Zane Bitter
parent
7e45bcc47d
commit
081669be91
@ -17,6 +17,7 @@
|
||||
Exception related utilities.
|
||||
"""
|
||||
|
||||
import functools
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
@ -278,3 +279,69 @@ def forever_retry_uncaught_exceptions(*args, **kwargs):
|
||||
return decorator(args[0])
|
||||
else:
|
||||
return decorator
|
||||
|
||||
|
||||
class exception_filter(object):
|
||||
"""A context manager that prevents some exceptions from being raised.
|
||||
|
||||
Use this class as a decorator for a function that returns whether a given
|
||||
exception should be ignored, in cases where complex logic beyond subclass
|
||||
matching is required. e.g.
|
||||
|
||||
>>> @exception_filter
|
||||
>>> def ignore_test_assertions(ex):
|
||||
... return isinstance(ex, AssertionError) and 'test' in str(ex)
|
||||
|
||||
The filter matching function can then be used as a context manager:
|
||||
|
||||
>>> with ignore_test_assertions:
|
||||
... assert False, 'This is a test'
|
||||
|
||||
or called directly:
|
||||
|
||||
>>> try:
|
||||
... assert False, 'This is a test'
|
||||
... except Exception as ex:
|
||||
... ignore_test_assertions(ex)
|
||||
|
||||
Any non-matching exception will be re-raised. When the filter is used as a
|
||||
context manager, the traceback for re-raised exceptions is always
|
||||
preserved. When the filter is called as a function, the traceback is
|
||||
preserved provided that no other exceptions have been raised in the
|
||||
intervening time. The context manager method is preferred for this reason
|
||||
except in cases where the ignored exception affects control flow.
|
||||
"""
|
||||
|
||||
def __init__(self, should_ignore_ex):
|
||||
self._should_ignore_ex = should_ignore_ex
|
||||
|
||||
if all(hasattr(should_ignore_ex, a)
|
||||
for a in functools.WRAPPER_ASSIGNMENTS):
|
||||
functools.update_wrapper(self, should_ignore_ex)
|
||||
|
||||
def __get__(self, obj, owner):
|
||||
return type(self)(self._should_ignore_ex.__get__(obj, owner))
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
if exc_val is not None:
|
||||
return self._should_ignore_ex(exc_val)
|
||||
|
||||
def __call__(self, ex):
|
||||
"""Re-raise any exception value not being filtered out.
|
||||
|
||||
If the exception was the last to be raised, it will be re-raised with
|
||||
its original traceback.
|
||||
"""
|
||||
exc_type, exc_val, traceback = sys.exc_info()
|
||||
|
||||
try:
|
||||
if not self._should_ignore_ex(ex):
|
||||
if exc_val is ex:
|
||||
six.reraise(exc_type, exc_val, traceback)
|
||||
else:
|
||||
raise ex
|
||||
finally:
|
||||
del exc_type, exc_val, traceback
|
||||
|
||||
@ -320,3 +320,250 @@ class ForeverRetryUncaughtExceptionsTest(test_base.BaseTestCase):
|
||||
self.exc_retrier_sequence(exc_id=2, exc_count=1,
|
||||
after_timestamp_calls=[110, 111])
|
||||
self.exc_retrier_common_end()
|
||||
|
||||
|
||||
class ExceptionFilterTest(test_base.BaseTestCase):
|
||||
|
||||
def _make_filter_func(self, ignore_classes=AssertionError):
|
||||
@excutils.exception_filter
|
||||
def ignore_exceptions(ex):
|
||||
'''Ignore some exceptions F'''
|
||||
return isinstance(ex, ignore_classes)
|
||||
|
||||
return ignore_exceptions
|
||||
|
||||
def _make_filter_method(self, ignore_classes=AssertionError):
|
||||
class ExceptionIgnorer(object):
|
||||
def __init__(self, ignore):
|
||||
self.ignore = ignore
|
||||
|
||||
@excutils.exception_filter
|
||||
def ignore_exceptions(self, ex):
|
||||
'''Ignore some exceptions M'''
|
||||
return isinstance(ex, self.ignore)
|
||||
|
||||
return ExceptionIgnorer(ignore_classes).ignore_exceptions
|
||||
|
||||
def _make_filter_classmethod(self, ignore_classes=AssertionError):
|
||||
class ExceptionIgnorer(object):
|
||||
ignore = ignore_classes
|
||||
|
||||
@excutils.exception_filter
|
||||
@classmethod
|
||||
def ignore_exceptions(cls, ex):
|
||||
'''Ignore some exceptions C'''
|
||||
return isinstance(ex, cls.ignore)
|
||||
|
||||
return ExceptionIgnorer.ignore_exceptions
|
||||
|
||||
def _make_filter_staticmethod(self, ignore_classes=AssertionError):
|
||||
class ExceptionIgnorer(object):
|
||||
@excutils.exception_filter
|
||||
@staticmethod
|
||||
def ignore_exceptions(ex):
|
||||
'''Ignore some exceptions S'''
|
||||
return isinstance(ex, ignore_classes)
|
||||
|
||||
return ExceptionIgnorer.ignore_exceptions
|
||||
|
||||
def test_filter_func_call(self):
|
||||
ignore_assertion_error = self._make_filter_func()
|
||||
|
||||
try:
|
||||
assert False, "This is a test"
|
||||
except Exception as exc:
|
||||
ignore_assertion_error(exc)
|
||||
|
||||
def test_raise_func_call(self):
|
||||
ignore_assertion_error = self._make_filter_func()
|
||||
|
||||
try:
|
||||
raise RuntimeError
|
||||
except Exception as exc:
|
||||
self.assertRaises(RuntimeError, ignore_assertion_error, exc)
|
||||
|
||||
def test_raise_previous_func_call(self):
|
||||
ignore_assertion_error = self._make_filter_func()
|
||||
|
||||
try:
|
||||
raise RuntimeError
|
||||
except Exception as exc1:
|
||||
try:
|
||||
raise RuntimeError
|
||||
except Exception as exc2:
|
||||
self.assertIsNot(exc1, exc2)
|
||||
raised = self.assertRaises(RuntimeError,
|
||||
ignore_assertion_error,
|
||||
exc1)
|
||||
self.assertIs(exc1, raised)
|
||||
|
||||
def test_raise_previous_after_filtered_func_call(self):
|
||||
ignore_assertion_error = self._make_filter_func()
|
||||
|
||||
try:
|
||||
raise RuntimeError
|
||||
except Exception as exc1:
|
||||
try:
|
||||
assert False, "This is a test"
|
||||
except Exception:
|
||||
pass
|
||||
self.assertRaises(RuntimeError, ignore_assertion_error, exc1)
|
||||
|
||||
def test_raise_other_func_call(self):
|
||||
@excutils.exception_filter
|
||||
def translate_exceptions(ex):
|
||||
raise RuntimeError
|
||||
|
||||
try:
|
||||
assert False, "This is a test"
|
||||
except Exception as exc:
|
||||
self.assertRaises(RuntimeError, translate_exceptions, exc)
|
||||
|
||||
def test_filter_func_context_manager(self):
|
||||
ignore_assertion_error = self._make_filter_func()
|
||||
|
||||
with ignore_assertion_error:
|
||||
assert False, "This is a test"
|
||||
|
||||
def test_raise_func_context_manager(self):
|
||||
ignore_assertion_error = self._make_filter_func()
|
||||
|
||||
def try_runtime_err():
|
||||
with ignore_assertion_error:
|
||||
raise RuntimeError
|
||||
|
||||
self.assertRaises(RuntimeError, try_runtime_err)
|
||||
|
||||
def test_raise_other_func_context_manager(self):
|
||||
@excutils.exception_filter
|
||||
def translate_exceptions(ex):
|
||||
raise RuntimeError
|
||||
|
||||
def try_assertion():
|
||||
with translate_exceptions:
|
||||
assert False, "This is a test"
|
||||
|
||||
self.assertRaises(RuntimeError, try_assertion)
|
||||
|
||||
def test_noexc_func_context_manager(self):
|
||||
ignore_assertion_error = self._make_filter_func()
|
||||
|
||||
with ignore_assertion_error:
|
||||
pass
|
||||
|
||||
def test_noexc_nocall_func_context_manager(self):
|
||||
@excutils.exception_filter
|
||||
def translate_exceptions(ex):
|
||||
raise RuntimeError
|
||||
|
||||
with translate_exceptions:
|
||||
pass
|
||||
|
||||
def test_func_docstring(self):
|
||||
ignore_func = self._make_filter_func()
|
||||
self.assertEqual('Ignore some exceptions F', ignore_func.__doc__)
|
||||
|
||||
def test_filter_method_call(self):
|
||||
ignore_assertion_error = self._make_filter_method()
|
||||
|
||||
try:
|
||||
assert False, "This is a test"
|
||||
except Exception as exc:
|
||||
ignore_assertion_error(exc)
|
||||
|
||||
def test_raise_method_call(self):
|
||||
ignore_assertion_error = self._make_filter_method()
|
||||
|
||||
try:
|
||||
raise RuntimeError
|
||||
except Exception as exc:
|
||||
self.assertRaises(RuntimeError, ignore_assertion_error, exc)
|
||||
|
||||
def test_filter_method_context_manager(self):
|
||||
ignore_assertion_error = self._make_filter_method()
|
||||
|
||||
with ignore_assertion_error:
|
||||
assert False, "This is a test"
|
||||
|
||||
def test_raise_method_context_manager(self):
|
||||
ignore_assertion_error = self._make_filter_method()
|
||||
|
||||
def try_runtime_err():
|
||||
with ignore_assertion_error:
|
||||
raise RuntimeError
|
||||
|
||||
self.assertRaises(RuntimeError, try_runtime_err)
|
||||
|
||||
def test_method_docstring(self):
|
||||
ignore_func = self._make_filter_method()
|
||||
self.assertEqual('Ignore some exceptions M', ignore_func.__doc__)
|
||||
|
||||
def test_filter_classmethod_call(self):
|
||||
ignore_assertion_error = self._make_filter_classmethod()
|
||||
|
||||
try:
|
||||
assert False, "This is a test"
|
||||
except Exception as exc:
|
||||
ignore_assertion_error(exc)
|
||||
|
||||
def test_raise_classmethod_call(self):
|
||||
ignore_assertion_error = self._make_filter_classmethod()
|
||||
|
||||
try:
|
||||
raise RuntimeError
|
||||
except Exception as exc:
|
||||
self.assertRaises(RuntimeError, ignore_assertion_error, exc)
|
||||
|
||||
def test_filter_classmethod_context_manager(self):
|
||||
ignore_assertion_error = self._make_filter_classmethod()
|
||||
|
||||
with ignore_assertion_error:
|
||||
assert False, "This is a test"
|
||||
|
||||
def test_raise_classmethod_context_manager(self):
|
||||
ignore_assertion_error = self._make_filter_classmethod()
|
||||
|
||||
def try_runtime_err():
|
||||
with ignore_assertion_error:
|
||||
raise RuntimeError
|
||||
|
||||
self.assertRaises(RuntimeError, try_runtime_err)
|
||||
|
||||
def test_classmethod_docstring(self):
|
||||
ignore_func = self._make_filter_classmethod()
|
||||
self.assertEqual('Ignore some exceptions C', ignore_func.__doc__)
|
||||
|
||||
def test_filter_staticmethod_call(self):
|
||||
ignore_assertion_error = self._make_filter_staticmethod()
|
||||
|
||||
try:
|
||||
assert False, "This is a test"
|
||||
except Exception as exc:
|
||||
ignore_assertion_error(exc)
|
||||
|
||||
def test_raise_staticmethod_call(self):
|
||||
ignore_assertion_error = self._make_filter_staticmethod()
|
||||
|
||||
try:
|
||||
raise RuntimeError
|
||||
except Exception as exc:
|
||||
self.assertRaises(RuntimeError, ignore_assertion_error, exc)
|
||||
|
||||
def test_filter_staticmethod_context_manager(self):
|
||||
ignore_assertion_error = self._make_filter_staticmethod()
|
||||
|
||||
with ignore_assertion_error:
|
||||
assert False, "This is a test"
|
||||
|
||||
def test_raise_staticmethod_context_manager(self):
|
||||
ignore_assertion_error = self._make_filter_staticmethod()
|
||||
|
||||
def try_runtime_err():
|
||||
with ignore_assertion_error:
|
||||
raise RuntimeError
|
||||
|
||||
self.assertRaises(RuntimeError, try_runtime_err)
|
||||
|
||||
def test_staticmethod_docstring(self):
|
||||
ignore_func = self._make_filter_staticmethod()
|
||||
self.assertEqual('Ignore some exceptions S', ignore_func.__doc__)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user