Merge branch 'ionelmc/master'

kombu/exceptions.py

@@ -22,7 +22,15 @@ TimeoutError = socket.timeout
 
 class KombuError(Exception):
     """Common subclass for all Kombu exceptions."""
+    pass
 
+class SerializationError(KombuError):
+    """Failed to encode a message."""
+    pass
+
+class DeserializationError(KombuError):
+    """Failed to decode a message."""
+    pass
 
 class NotBoundError(KombuError):
     """Trying to call channel dependent method on unbound entity."""

kombu/serialization.py

@@ -19,9 +19,9 @@ except ImportError:  # pragma: no cover
 
 from collections import namedtuple
 
-from .exceptions import SerializerNotInstalled, ContentDisallowed
+from .exceptions import SerializerNotInstalled, ContentDisallowed, SerializationError, DeserializationError
 from .five import BytesIO, text_t
-from .utils import entrypoints
+from .utils import entrypoints, wrap_exceptions
 from .utils.encoding import str_to_bytes, bytes_t
 
 __all__ = ['pickle', 'loads', 'dumps', 'register', 'unregister']
@@ -116,6 +116,7 @@ class SerializerRegistry(object):
             raise SerializerNotInstalled(
                 'No encoder installed for {0}'.format(name))
 
+    @wrap_exceptions(SerializationError)
     def dumps(self, data, serializer=None):
         if serializer == 'raw':
             return raw_encode(data)
@@ -148,6 +149,7 @@ class SerializerRegistry(object):
         return content_type, content_encoding, payload
     encode = dumps  # XXX compat
 
+    @wrap_exceptions(DeserializationError)
     def loads(self, data, content_type, content_encoding,
               accept=None, force=False):
         if accept is not None:

kombu/utils/__init__.py

@@ -13,6 +13,7 @@ import sys
 
 from contextlib import contextmanager
 from itertools import count, repeat
+from functools import wraps
 from time import sleep
 from uuid import UUID, uuid4 as _uuid4, _uuid_generate_random
 
@@ -428,3 +429,20 @@ def maybe_fileno(f):
         return fileno(f)
     except FILENO_ERRORS:
         pass
+
+def wrap_exceptions(exception, catch=Exception):
+    """
+    Catch the exception specified by ``catch`` and raise ``exception`` instead with
+    the old exception as the value.
+    """
+
+    def decorator(func):
+        @wraps(func)
+        def wrap_exceptions_wrapper(*args, **kwargs):
+            try:
+                return func(*args, **kwargs)
+            except catch as exc:
+                raise exception(exc)
+        return wrap_exceptions_wrapper
+
+    return decorator