output_stream: permit ingore or interrupt when serialization error

hapic/decorator.py

@@ -341,6 +341,22 @@ class AsyncOutputStreamControllerWrapper(OutputControllerWrapper):
     This controller wrapper produce a wrapper who caught the http view items
     to check and serialize them into a stream response.
     """
+    def __init__(
+        self,
+        context: typing.Union[ContextInterface, typing.Callable[[], ContextInterface]],  # nopep8
+        processor: ProcessorInterface,
+        error_http_code: HTTPStatus=HTTPStatus.INTERNAL_SERVER_ERROR,
+        default_http_code: HTTPStatus=HTTPStatus.OK,
+        ignore_on_error: bool = True,
+    ) -> None:
+        super().__init__(
+            context,
+            processor,
+            error_http_code,
+            default_http_code,
+        )
+        self.ignore_on_error = ignore_on_error
+
     def get_wrapper(
         self,
         func: 'typing.Callable[..., typing.Any]',
@@ -362,11 +378,17 @@ class AsyncOutputStreamControllerWrapper(OutputControllerWrapper):
                 args,
                 kwargs,
             ):
-                serialized_item = self._get_serialized_item(stream_item)
-                await self.context.feed_stream_response(
-                    stream_response,
-                    serialized_item,
-                )
+                try:
+                    serialized_item = self._get_serialized_item(stream_item)
+                    await self.context.feed_stream_response(
+                        stream_response,
+                        serialized_item,
+                    )
+                except OutputValidationException as exc:
+                    if not self.ignore_on_error:
+                        # TODO BS 2018-07-31: Something should inform about
+                        # error, a log ?
+                        return stream_response
 
             return stream_response
 
@@ -376,12 +398,7 @@ class AsyncOutputStreamControllerWrapper(OutputControllerWrapper):
         self,
         item_object: typing.Any,
     ) -> dict:
-        try:
-            return self.processor.process(item_object)
-        except ProcessException:
-            # TODO BS 2018-07-25: Must interrupt stream response: but how
-            # inform about error ?
-            raise NotImplementedError('todo')
+        return self.processor.process(item_object)
 
 
 class OutputHeadersControllerWrapper(OutputControllerWrapper):

hapic/hapic.py

@@ -179,6 +179,7 @@ class Hapic(object):
         context: ContextInterface = None,
         error_http_code: HTTPStatus = HTTPStatus.INTERNAL_SERVER_ERROR,
         default_http_code: HTTPStatus = HTTPStatus.OK,
+        ignore_on_error: bool = True,
     ) -> typing.Callable[[typing.Callable[..., typing.Any]], typing.Any]:
         """
         Decorate with a wrapper who check and serialize each items in output
@@ -190,6 +191,8 @@ class Hapic(object):
         :param context: Context to use here
         :param error_http_code: http code in case of error
         :param default_http_code: http code in case of success
+        :param ignore_on_error: if set, an error of serialization will be
+        ignored: stream will not send this failed object
         :return: decorator
         """
         processor = processor or MarshmallowOutputProcessor()
@@ -202,6 +205,7 @@ class Hapic(object):
                 processor=processor,
                 error_http_code=error_http_code,
                 default_http_code=default_http_code,
+                ignore_on_error=ignore_on_error,
             )
         else:
             # TODO BS 2018-07-25: To do

tests/ext/unit/test_aiohttp.py

@@ -228,7 +228,89 @@ class TestAiohttpExt(object):
         line = await resp.content.readline()
         assert b'{"name": "Hello, franck"}\n' == line
 
-        # TODO BS 2018-07-26: How to ensure we are at end of response ?
+    async def test_aiohttp_output_stream__error__ignore(
+        self,
+        aiohttp_client,
+        loop,
+    ):
+        hapic = Hapic(async_=True)
+
+        class AsyncGenerator:
+            def __init__(self):
+                self._iterator = iter([
+                    {'name': 'Hello, bob'},
+                    {'nameZ': 'Hello, Z'},  # This line is incorrect
+                    {'name': 'Hello, franck'},
+                ])
+
+            async def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                return next(self._iterator)
+
+        class OuputStreamItemSchema(marshmallow.Schema):
+            name = marshmallow.fields.String(required=True)
+
+        @hapic.output_stream(OuputStreamItemSchema(), ignore_on_error=True)
+        async def hello(request):
+            return AsyncGenerator()
+
+        app = web.Application(debug=True)
+        app.router.add_get('/', hello)
+        hapic.set_context(AiohttpContext(app))
+        client = await aiohttp_client(app)
+
+        resp = await client.get('/')
+        assert resp.status == 200
+
+        line = await resp.content.readline()
+        assert b'{"name": "Hello, bob"}\n' == line
+
+        line = await resp.content.readline()
+        assert b'{"name": "Hello, franck"}\n' == line
+
+    async def test_aiohttp_output_stream__error__interrupt(
+        self,
+        aiohttp_client,
+        loop,
+    ):
+        hapic = Hapic(async_=True)
+
+        class AsyncGenerator:
+            def __init__(self):
+                self._iterator = iter([
+                    {'name': 'Hello, bob'},
+                    {'nameZ': 'Hello, Z'},  # This line is incorrect
+                    {'name': 'Hello, franck'},  # This line must not be reached
+                ])
+
+            async def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                return next(self._iterator)
+
+        class OuputStreamItemSchema(marshmallow.Schema):
+            name = marshmallow.fields.String(required=True)
+
+        @hapic.output_stream(OuputStreamItemSchema(), ignore_on_error=False)
+        async def hello(request):
+            return AsyncGenerator()
+
+        app = web.Application(debug=True)
+        app.router.add_get('/', hello)
+        hapic.set_context(AiohttpContext(app))
+        client = await aiohttp_client(app)
+
+        resp = await client.get('/')
+        assert resp.status == 200
+
+        line = await resp.content.readline()
+        assert b'{"name": "Hello, bob"}\n' == line
+
+        line = await resp.content.readline()
+        assert b'' == line
 
     def test_unit__generate_doc__ok__nominal_case(
         self,