[#23] reworking exception chaining (#28)

* Changed exception handling
  * now exceptions are chained (before they were added in `args`)
  * timeout errors are now chained (before they were not included at all)
  * in case of dogpiling, all callers are now notified about the error (see issue #23)

Author: zmumi

CHANGELOG.rst

@@ -1,3 +1,11 @@
+2.0.0
+-----
+
+* Changed exception handling
+  * now exceptions are chained (before they were added in `args`)
+  * timeout errors are now chained (before they were not included at all)
+  * in case of dogpiling, all callers are now notified about the error (see issue #23)
+
 1.2.2
 -----
 

memoize/statuses.py

@@ -4,7 +4,7 @@
 import datetime
 import logging
 from asyncio import Future
-from typing import Optional, Dict, Awaitable
+from typing import Optional, Dict, Awaitable, Union
 
 from memoize import coerced
 from memoize.entry import CacheKey, CacheEntry
@@ -14,8 +14,7 @@ class UpdateStatuses:
     def __init__(self, update_lock_timeout: datetime.timedelta = datetime.timedelta(minutes=5)) -> None:
         self.logger = logging.getLogger(__name__)
         self._update_lock_timeout = update_lock_timeout
-        # type declaration should not be in comment once we drop py35 support
-        self._updates_in_progress = {}  # type: Dict[CacheKey, Future]
+        self._updates_in_progress: Dict[CacheKey, Future] = {}
 
     def is_being_updated(self, key: CacheKey) -> bool:
         """Checks if update for given key is in progress. Obtained info is valid until control gets back to IO-loop."""
@@ -49,17 +48,19 @@ def mark_updated(self, key: CacheKey, entry: CacheEntry) -> None:
         update = self._updates_in_progress.pop(key)
         update.set_result(entry)
 
-    def mark_update_aborted(self, key: CacheKey) -> None:
+    def mark_update_aborted(self, key: CacheKey, exception: Exception) -> None:
         """Informs that update failed to complete.
-        Calls to 'is_being_updated' will return False until 'mark_being_updated' will be called."""
+        Calls to 'is_being_updated' will return False until 'mark_being_updated' will be called.
+        Accepts exception to propagate it across all clients awaiting an update."""
         if key not in self._updates_in_progress:
             raise ValueError('Key {} is not being updated'.format(key))
         update = self._updates_in_progress.pop(key)
-        update.set_result(None)
+        update.set_result(exception)
 
-    def await_updated(self, key: CacheKey) -> Awaitable[Optional[CacheEntry]]:
+    def await_updated(self, key: CacheKey) -> Awaitable[Union[CacheEntry, Exception]]:
         """Waits (asynchronously) until update in progress has benn finished.
-        Returns updated entry or None if update failed/timed-out.
+        Returns awaitable with the updated entry
+        (or awaitable with an exception if update failed/timed-out).
         Should be called only if 'is_being_updated' returned True (and since then IO-loop has not been lost)."""
         if not self.is_being_updated(key):
             raise ValueError('Key {} is not being updated'.format(key))

memoize/wrapper.py

@@ -77,8 +77,8 @@ async def refresh(actual_entry: Optional[CacheEntry], key: CacheKey,
         if actual_entry is None and update_statuses.is_being_updated(key):
             logger.debug('As entry expired, waiting for results of concurrent refresh %s', key)
             entry = await update_statuses.await_updated(key)
-            if entry is None:
-                raise CachedMethodFailedException('Concurrent refresh failed to complete')
+            if isinstance(entry, Exception):
+                raise CachedMethodFailedException('Concurrent refresh failed to complete') from entry
             return entry
         elif actual_entry is not None and update_statuses.is_being_updated(key):
             logger.debug('As update point reached but concurrent update already in progress, '
@@ -103,12 +103,12 @@ async def refresh(actual_entry: Optional[CacheEntry], key: CacheKey,
                 return offered_entry
             except (asyncio.TimeoutError, _timeout_error_type()) as e:
                 logger.debug('Timeout for %s: %s', key, e)
-                update_statuses.mark_update_aborted(key)
-                raise CachedMethodFailedException('Refresh timed out')
+                update_statuses.mark_update_aborted(key, e)
+                raise CachedMethodFailedException('Refresh timed out') from e
             except Exception as e:
                 logger.debug('Error while refreshing cache for %s: %s', key, e)
-                update_statuses.mark_update_aborted(key)
-                raise CachedMethodFailedException('Refresh failed to complete', e)
+                update_statuses.mark_update_aborted(key, e)
+                raise CachedMethodFailedException('Refresh failed to complete') from e
 
     @functools.wraps(method)
     async def wrapper(*args, **kwargs):

setup.py

@@ -11,7 +11,7 @@ def prepare_description():
 
 setup(
     name='py-memoize',
-    version='1.2.2',
+    version='2.0.0',
     author='Michal Zmuda',
     author_email='[email protected]',
     url='https://github.com/DreamLab/memoize',

tests/asynciotests/test_showcase.py

@@ -94,5 +94,5 @@ async def get_value_or_throw(arg, kwarg=None):
         self.assertEqual('ok #2', res4)  # value from cache - still relevant
         self.assertEqual('ok #2', res5)  # stale from cache - refresh in background
         self.assertEqual('ok #2', res6)  # stale from cache - should be updated but method throws
-        expected = CachedMethodFailedException('Refresh failed to complete', ValueError('throws #4', ))
-        self.assertEqual(str(expected), str(context.exception))  # ToDo: consider better comparision
+        self.assertEqual(str(context.exception), str(CachedMethodFailedException('Refresh failed to complete')))
+        self.assertEqual(str(context.exception.__cause__), str(ValueError("throws #4")))

tests/asynciotests/test_wrapper_manually_applied_on_asyncio.py

@@ -1,3 +1,4 @@
+from memoize.coerced import _timeout_error_type
 from tests.py310workaround import fix_python_3_10_compatibility
 
 fix_python_3_10_compatibility()
@@ -274,8 +275,8 @@ async def get_value(arg, kwarg=None):
             await get_value_cached('test1', kwarg='args1')
 
         # then
-        expected = CachedMethodFailedException('Refresh failed to complete', ValueError('Get lost', ))
-        self.assertEqual(str(expected), str(context.exception))  # ToDo: consider better comparision
+        self.assertEqual(str(context.exception), str(CachedMethodFailedException('Refresh failed to complete')))
+        self.assertEqual(str(context.exception.__cause__), str(ValueError("Get lost")))
 
     @gen_test
     async def test_should_throw_exception_on_refresh_timeout(self):
@@ -295,8 +296,8 @@ async def get_value(arg, kwarg=None):
             await get_value_cached('test1', kwarg='args1')
 
         # then
-        expected = CachedMethodFailedException('Refresh timed out')
-        self.assertEqual(str(expected), str(context.exception))  # ToDo: consider better comparision
+        self.assertEqual(context.exception.__class__, CachedMethodFailedException)
+        self.assertEqual(context.exception.__cause__.__class__, _timeout_error_type())
 
     @staticmethod
     async def _call_thrice(call):

tests/asynciotests/test_wrapper_on_asyncio.py

@@ -1,3 +1,4 @@
+from memoize.coerced import _timeout_error_type
 from tests.py310workaround import fix_python_3_10_compatibility
 
 fix_python_3_10_compatibility()
@@ -160,6 +161,69 @@ async def get_value(arg, kwarg=None):
         self.assertEqual(0, res1)
         self.assertEqual(1, res2)
 
+    @gen_test
+    async def test_should_return_exception_for_all_concurrent_callers(self):
+        # given
+        value = 0
+
+        @memoize()
+        async def get_value(arg, kwarg=None):
+            raise ValueError(f'stub{value}')
+
+        # when
+        res1 = get_value('test', kwarg='args1')
+        res2 = get_value('test', kwarg='args1')
+        res3 = get_value('test', kwarg='args1')
+
+        # then
+        with self.assertRaises(Exception) as context:
+            await res1
+        self.assertEqual(context.exception.__class__, CachedMethodFailedException)
+        self.assertEqual(str(context.exception.__cause__), str(ValueError('stub0')))
+
+        with self.assertRaises(Exception) as context:
+            await res2
+        self.assertEqual(context.exception.__class__, CachedMethodFailedException)
+        self.assertEqual(str(context.exception.__cause__), str(ValueError('stub0')))
+
+        with self.assertRaises(Exception) as context:
+            await res3
+        self.assertEqual(context.exception.__class__, CachedMethodFailedException)
+        self.assertEqual(str(context.exception.__cause__), str(ValueError('stub0')))
+
+    @gen_test
+    async def test_should_return_timeout_for_all_concurrent_callers(self):
+        # given
+        value = 0
+
+        @memoize(configuration=DefaultInMemoryCacheConfiguration(method_timeout=timedelta(milliseconds=1)))
+        async def get_value(arg, kwarg=None):
+            await _ensure_asyncio_background_tasks_finished()
+            time.sleep(.200)
+            await _ensure_asyncio_background_tasks_finished()
+            return value
+
+        # when
+        res1 = get_value('test', kwarg='args1')
+        res2 = get_value('test', kwarg='args1')
+        res3 = get_value('test', kwarg='args1')
+
+        # then
+        with self.assertRaises(Exception) as context:
+            await res1
+        self.assertEqual(context.exception.__class__, CachedMethodFailedException)
+        self.assertEqual(context.exception.__cause__.__class__, _timeout_error_type())
+
+        with self.assertRaises(Exception) as context:
+            await res2
+        self.assertEqual(context.exception.__class__, CachedMethodFailedException)
+        self.assertEqual(context.exception.__cause__.__class__, _timeout_error_type())
+
+        with self.assertRaises(Exception) as context:
+            await res3
+        self.assertEqual(context.exception.__class__, CachedMethodFailedException)
+        self.assertEqual(context.exception.__cause__.__class__, _timeout_error_type())
+
     @gen_test
     async def test_should_return_same_value_on_constant_key_function(self):
         # given
@@ -253,8 +317,8 @@ async def get_value(arg, kwarg=None):
             await get_value('test1', kwarg='args1')
 
         # then
-        expected = CachedMethodFailedException('Refresh failed to complete', ValueError('Get lost', ))
-        self.assertEqual(str(expected), str(context.exception))  # ToDo: consider better comparision
+        self.assertEqual(str(context.exception), str(CachedMethodFailedException('Refresh failed to complete')))
+        self.assertEqual(str(context.exception.__cause__), str(ValueError("Get lost")))
 
     @gen_test
     async def test_should_throw_exception_on_refresh_timeout(self):
@@ -272,8 +336,8 @@ async def get_value(arg, kwarg=None):
             await get_value('test1', kwarg='args1')
 
         # then
-        expected = CachedMethodFailedException('Refresh timed out')
-        self.assertEqual(str(expected), str(context.exception))  # ToDo: consider better comparision
+        self.assertEqual(context.exception.__class__, CachedMethodFailedException)
+        self.assertEqual(context.exception.__cause__.__class__, _timeout_error_type())
 
     @staticmethod
     async def _call_thrice(call):

tests/tornadotests/test_wrapper.py

@@ -1,3 +1,4 @@
+from memoize.coerced import _timeout_error_type
 from tests.py310workaround import fix_python_3_10_compatibility
 
 fix_python_3_10_compatibility()
@@ -265,8 +266,8 @@ def get_value(arg, kwarg=None):
             yield get_value('test1', kwarg='args1')
 
         # then
-        expected = CachedMethodFailedException('Refresh failed to complete', ValueError('Get lost', ))
-        self.assertEqual(str(expected), str(context.exception))  # ToDo: consider better comparision
+        self.assertEqual(str(context.exception), str(CachedMethodFailedException('Refresh failed to complete')))
+        self.assertEqual(str(context.exception.__cause__), str(ValueError("Get lost")))
 
     @gen_test
     def test_should_throw_exception_on_refresh_timeout(self):
@@ -285,5 +286,6 @@ def get_value(arg, kwarg=None):
             yield get_value('test1', kwarg='args1')
 
         # then
-        expected = CachedMethodFailedException('Refresh timed out')
-        self.assertEqual(str(expected), str(context.exception))  # ToDo: consider better comparision
+        self.assertEqual(context.exception.__class__, CachedMethodFailedException)
+        self.assertEqual(context.exception.__cause__.__class__, _timeout_error_type())
+

tests/unit/test_statuses.py

@@ -62,14 +62,14 @@ def test_should_be_mark_as_updated(self):
     def test_should_raise_exception_during_mark_update_as_aborted(self):
         # given/when/then
         with self.assertRaises(ValueError):
-            self.update_statuses.mark_update_aborted('key')
+            self.update_statuses.mark_update_aborted('key', Exception('stub'))
 
     def test_should_mark_update_as_aborted(self):
         # given
         self.update_statuses.mark_being_updated('key')
 
         # when
-        self.update_statuses.mark_update_aborted('key')
+        self.update_statuses.mark_update_aborted('key', Exception('stub'))
 
         # then
         self.assertFalse(self.update_statuses.is_being_updated('key'))
@@ -105,3 +105,23 @@ async def test_should_await_updated_return_entry(self):
         # then
         self.assertIsNone(result)
         self.assertFalse(self.update_statuses.is_being_updated('key'))
+
+    @gen_test
+    async def test_concurrent_callers_should_all_get_exception_on_aborted_update(self):
+        # given
+        self.update_statuses.mark_being_updated('key')
+
+        # when
+        result1 = self.update_statuses.await_updated('key')
+        result2 = self.update_statuses.await_updated('key')
+        result3 = self.update_statuses.await_updated('key')
+        self.update_statuses.mark_update_aborted('key', ValueError('stub'))
+        result1 = await result1
+        result2 = await result2
+        result3 = await result3
+
+        # then
+        self.assertFalse(self.update_statuses.is_being_updated('key'))
+        self.assertEqual(str(result1), str(ValueError('stub')))
+        self.assertEqual(str(result2), str(ValueError('stub')))
+        self.assertEqual(str(result3), str(ValueError('stub')))