Async mode fixes (#399)

* Fix double printing of exception when initializing resource causes error

* Improve async mode exceptions handling to prevent infinite hanging when exception occurs

* Improve async mode exceptions handling

* Update changelog

* Update tests

docs/main/changelog.rst

@@ -7,6 +7,11 @@ that were made in every particular version.
 From version 0.7.6 *Dependency Injector* framework strictly 
 follows `Semantic versioning`_
 
+Development version
+-------------------
+- Improve async mode exceptions handling.
+- Fix double printing of exception when async resource initialization causes an error.
+
 4.23.1
 ------
 - Hotfix a bug with importing FastAPI ``Request``.

src/dependency_injector/providers.pxd

@@ -455,10 +455,15 @@ cdef inline void __async_prepare_args_kwargs_callback(
         object awaitables,
         object future,
 ):
-    awaited = future.result()
+    try:
-    for value, (key, _) in zip(awaited, awaitables):
+        awaited = future.result()
-        args[key] = value
+
-    future_result.set_result(args)
+        for value, (key, _) in zip(awaited, awaitables):
+            args[key] = value
+    except Exception as exception:
+        future_result.set_exception(exception)
+    else:
+        future_result.set_result(args)
 
 
 @cython.boundscheck(False)
@@ -498,9 +503,15 @@ cdef inline object __async_inject_attributes(future_instance, future_attributes)
 
 
 cdef inline void __async_inject_attributes_callback(object future_result, object future):
-    instance, attributes = future.result()
+    try:
-    __inject_attributes(instance, attributes)
+        instance, attributes = future.result()
-    future_result.set_result(instance)
+
+        for name, value in attributes.items():
+            setattr(instance, name, value)
+    except Exception as exception:
+        future_result.set_exception(exception)
+    else:
+        future_result.set_result(instance)
 
 
 cdef inline void __inject_attributes(object instance, dict attributes):
@@ -560,19 +571,26 @@ cdef inline object __call(
 
 
 cdef inline void __async_call_callback(object future_result, object call, object future):
-    args, kwargs = future.result()
+    try:
-    result = call(*args, **kwargs)
+        args, kwargs = future.result()
-
+        result = call(*args, **kwargs)
-    if __isawaitable(result):
+    except Exception as exception:
-        result = asyncio.ensure_future(result)
+        future_result.set_exception(exception)
-        result.add_done_callback(functools.partial(__async_result_callback, future_result))
+    else:
-        return
+        if __isawaitable(result):
-
+            result = asyncio.ensure_future(result)
-    future_result.set_result(result)
+            result.add_done_callback(functools.partial(__async_result_callback, future_result))
+            return
+        future_result.set_result(result)
 
 
 cdef inline object __async_result_callback(object future_result, object future):
-    future_result.set_result(future.result())
+    try:
+        result = future.result()
+    except Exception as exception:
+        future_result.set_exception(exception)
+    else:
+        future_result.set_result(result)
 
 
 cdef inline object __callable_call(Callable self, tuple args, dict kwargs):

src/dependency_injector/providers.pyx

@@ -3486,7 +3486,6 @@ cdef class Resource(Provider):
             resource = initializer.result()
         except Exception:
             self.__initialized = False
-            raise
         else:
             self.__resource = resource
             self.__shutdowner = shutdowner

tests/unit/providers/test_async_py36.py

@@ -190,6 +190,64 @@ class FactoryTests(AsyncTestCase):
 
         self.assertIsNot(service1.client, service2.client)
 
+    def test_injection_error(self):
+        async def init_resource():
+            raise Exception('Something went wrong')
+
+        class Container(containers.DeclarativeContainer):
+            resource_with_error = providers.Resource(init_resource)
+
+            client = providers.Factory(
+                Client,
+                resource1=resource_with_error,
+                resource2=None,
+            )
+
+        container = Container()
+
+        with self.assertRaises(Exception) as context:
+            self._run(container.client())
+        self.assertEqual(str(context.exception), 'Something went wrong')
+
+    def test_injection_runtime_error_async_provides(self):
+        async def create_client(*args,  **kwargs):
+            raise Exception('Something went wrong')
+
+        class Container(containers.DeclarativeContainer):
+            resource = providers.Resource(init_resource, providers.Object(RESOURCE1))
+
+            client = providers.Factory(
+                create_client,
+                resource1=resource,
+                resource2=None,
+            )
+
+        container = Container()
+
+        with self.assertRaises(Exception) as context:
+            self._run(container.client())
+        self.assertEqual(str(context.exception), 'Something went wrong')
+
+    def test_injection_call_error_async_provides(self):
+        async def create_client():  # <-- no args defined
+            ...
+
+        class Container(containers.DeclarativeContainer):
+            resource = providers.Resource(init_resource, providers.Object(RESOURCE1))
+
+            client = providers.Factory(
+                create_client,
+                resource1=resource,
+                resource2=None,
+            )
+
+        container = Container()
+
+        with self.assertRaises(TypeError) as context:
+            self._run(container.client())
+        self.assertIn("create_client() got", str(context.exception))
+        self.assertIn("unexpected keyword argument", str(context.exception))
+
     def test_attributes_injection(self):
         class ContainerWithAttributes(containers.DeclarativeContainer):
             resource1 = providers.Resource(init_resource, providers.Object(RESOURCE1))
@@ -236,6 +294,53 @@ class FactoryTests(AsyncTestCase):
 
         self.assertIsNot(service1.client, service2.client)
 
+    def test_attributes_injection_attribute_error(self):
+        class ClientWithException(Client):
+            @property
+            def attribute_set_error(self):
+                return None
+
+            @attribute_set_error.setter
+            def attribute_set_error(self, value):
+                raise Exception('Something went wrong')
+
+        class Container(containers.DeclarativeContainer):
+            resource = providers.Resource(init_resource, providers.Object(RESOURCE1))
+
+            client = providers.Factory(
+                ClientWithException,
+                resource1=resource,
+                resource2=resource,
+            )
+            client.add_attributes(attribute_set_error=123)
+
+        container = Container()
+
+        with self.assertRaises(Exception) as context:
+            self._run(container.client())
+        self.assertEqual(str(context.exception), 'Something went wrong')
+
+    def test_attributes_injection_runtime_error(self):
+        async def init_resource():
+            raise Exception('Something went wrong')
+
+        class Container(containers.DeclarativeContainer):
+            resource = providers.Resource(init_resource)
+
+            client = providers.Factory(
+                Client,
+                resource1=None,
+                resource2=None,
+            )
+            client.add_attributes(resource1=resource)
+            client.add_attributes(resource2=resource)
+
+        container = Container()
+
+        with self.assertRaises(Exception) as context:
+            self._run(container.client())
+        self.assertEqual(str(context.exception), 'Something went wrong')
+
     def test_async_instance_and_sync_attributes_injection(self):
         class ContainerWithAttributes(containers.DeclarativeContainer):
             resource1 = providers.Resource(init_resource, providers.Object(RESOURCE1))

tests/unit/providers/test_resource_py35.py

@@ -449,14 +449,45 @@ class AsyncResourceTest(AsyncTestCase):
         self.assertTrue(provider.initialized)
         self.assertTrue(provider.is_async_mode_enabled())
 
-        # Disable default exception handling to prevent output
+        with self.assertRaises(RuntimeError):
-        asyncio.get_event_loop().set_exception_handler(lambda loop, context: ...)
+            self._run(future)
+
+        self.assertFalse(provider.initialized)
+        self.assertTrue(provider.is_async_mode_enabled())
+
+    def test_init_async_gen_with_error(self):
+        async def _init():
+            raise RuntimeError()
+            yield
+
+        provider = providers.Resource(_init)
+
+        future = provider()
+        self.assertTrue(provider.initialized)
+        self.assertTrue(provider.is_async_mode_enabled())
 
         with self.assertRaises(RuntimeError):
             self._run(future)
 
-        # Restore default exception handling
+        self.assertFalse(provider.initialized)
-        asyncio.get_event_loop().set_exception_handler(None)
+        self.assertTrue(provider.is_async_mode_enabled())
+
+    def test_init_async_subclass_with_error(self):
+        class _Resource(resources.AsyncResource):
+            async def init(self):
+                raise RuntimeError()
+
+            async def shutdown(self, resource):
+                pass
+
+        provider = providers.Resource(_Resource)
+
+        future = provider()
+        self.assertTrue(provider.initialized)
+        self.assertTrue(provider.is_async_mode_enabled())
+
+        with self.assertRaises(RuntimeError):
+            self._run(future)
 
         self.assertFalse(provider.initialized)
         self.assertTrue(provider.is_async_mode_enabled())