output stream async

aiopoc.py

@@ -1,25 +1,61 @@
+import asyncio
+import json
+
 from sh import tail
 from asyncio import sleep
 from aiohttp import web
+import marshmallow
+
+from hapic import async as hapic
+from hapic.ext.aiohttp.context import AiohttpContext
+
+
+class OutputStreamItemSchema(marshmallow.Schema):
+    i = marshmallow.fields.Integer(required=True)
+
+
+# Python 3.6 async generator: http://rickyhan.com/jekyll/update/2018/01/27/python36.html
+# Python 3.5 solution: https://stackoverflow.com/questions/37549846/how-to-use-yield-inside-async-function
+
 
+class LinesAsyncGenerator:
+    def __init__(self):
+        self.iterable = tail("-f", "aiopocdata.txt", _iter=True)
 
-async def handle(request):
-    response = web.StreamResponse(
-        status=200,
-        reason='OK',
-        headers={
-            'Content-Type': 'text/plain; charset=utf-8',
-        },
-    )
+    async def __aiter__(self):
+        return self
 
-    await response.prepare(request)
-    response.enable_chunked_encoding()
+    async def __anext__(self):
+        line = next(self.iterable)
 
-    for line in tail("-f", "aiopocdata.txt", _iter=True):
-        await response.write(line.encode('utf-8'))
-        await sleep(0.1)
+        if 'STOP' in line:
+            raise StopAsyncIteration
 
-    return response
+        await asyncio.sleep(0.025)
+        return json.loads(line)
+
+
+@hapic.with_api_doc()
+@hapic.output_stream(item_schema=OutputStreamItemSchema())
+def handle(request):
+    # response = web.StreamResponse(
+    #     status=200,
+    #     reason='OK',
+    #     headers={
+    #         'Content-Type': 'text/plain; charset=utf-8',
+    #     },
+    # )
+    #
+    # await response.prepare(request)
+    # response.enable_chunked_encoding()
+
+    # for line in tail("-f", "aiopocdata.txt", _iter=True):
+        # await response.write(line.encode('utf-8'))
+        # await sleep(0.1)
+
+    # return response
+
+    return LinesAsyncGenerator()
 
 
 app = web.Application()
@@ -27,5 +63,5 @@ app.add_routes([
     web.get('/', handle)
 ])
 
-
+hapic.set_context(AiohttpContext(app))
 web.run_app(app)

hapic/__init__.py

@@ -20,3 +20,4 @@ set_context = _hapic_default.set_context
 reset_context = _hapic_default.reset_context
 add_documentation_view = _hapic_default.add_documentation_view
 handle_exception = _hapic_default.handle_exception
+output_stream = _hapic_default.output_stream

hapic/async.py

@@ -18,3 +18,4 @@ set_context = _hapic_default.set_context
 reset_context = _hapic_default.reset_context
 add_documentation_view = _hapic_default.add_documentation_view
 handle_exception = _hapic_default.handle_exception
+output_stream = _hapic_default.output_stream

hapic/decorator.py

@@ -313,6 +313,10 @@ class OutputBodyControllerWrapper(OutputControllerWrapper):
     pass
 
 
+# TODO BS 2018-07-23: This class is an async version of
+# OutputBodyControllerWrapper (ControllerWrapper.get_wrapper rewrite)
+# to permit async compatibility.
+# Please re-think about code refact. TAG: REFACT_ASYNC
 class AsyncOutputBodyControllerWrapper(OutputControllerWrapper):
     def get_wrapper(
         self,
@@ -332,6 +336,50 @@ class AsyncOutputBodyControllerWrapper(OutputControllerWrapper):
         return functools.update_wrapper(wrapper, func)
 
 
+class AsyncOutputStreamControllerWrapper(OutputControllerWrapper):
+    def get_wrapper(
+        self,
+        func: 'typing.Callable[..., typing.Any]',
+    ) -> 'typing.Callable[..., typing.Any]':
+        # async def wrapper(*args, **kwargs) -> typing.Any:
+        async def wrapper(*args, **kwargs) -> typing.Any:
+            # Note: Design of before_wrapped_func can be to update kwargs
+            # by reference here
+            replacement_response = self.before_wrapped_func(args, kwargs)
+            if replacement_response:
+                return replacement_response
+
+            stream_response = await self.context.get_stream_response_object(
+                args,
+                kwargs,
+            )
+            async for stream_item in self._execute_wrapped_function(
+                func,
+                args,
+                kwargs,
+            ):
+                serialized_item = self._get_serialized_item(stream_item)
+                await self.context.feed_stream_response(
+                    stream_response,
+                    serialized_item,
+                )
+
+            return stream_response
+
+        return functools.update_wrapper(wrapper, func)
+
+    def _get_serialized_item(
+        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')
+
+
 class OutputHeadersControllerWrapper(OutputControllerWrapper):
     pass
 

hapic/description.py

@@ -38,6 +38,10 @@ class OutputBodyDescription(Description):
     pass
 
 
+class OutputStreamDescription(Description):
+    pass
+
+
 class OutputFileDescription(Description):
     pass
 
@@ -60,6 +64,7 @@ class ControllerDescription(object):
         input_forms: InputFormsDescription=None,
         input_files: InputFilesDescription=None,
         output_body: OutputBodyDescription=None,
+        output_stream: OutputStreamDescription=None,
         output_file: OutputFileDescription=None,
         output_headers: OutputHeadersDescription=None,
         errors: typing.List[ErrorDescription]=None,
@@ -72,6 +77,7 @@ class ControllerDescription(object):
         self.input_forms = input_forms
         self.input_files = input_files
         self.output_body = output_body
+        self.output_stream = output_stream
         self.output_file = output_file
         self.output_headers = output_headers
         self.errors = errors or []

hapic/ext/aiohttp/context.py

@@ -1,5 +1,6 @@
 # coding: utf-8
 import asyncio
+import json
 import typing
 from http import HTTPStatus
 from json import JSONDecodeError
@@ -65,8 +66,10 @@ class AiohttpContext(BaseContext):
     def __init__(
         self,
         app: web.Application,
+        debug: bool = False,
     ) -> None:
         self._app = app
+        self._debug = debug
 
     @property
     def app(self) -> web.Application:
@@ -110,19 +113,22 @@ class AiohttpContext(BaseContext):
         self,
         decorated_controller: DecoratedController,
     ) -> RouteRepresentation:
-        pass
+        # TODO BS 2018-07-15: to do
+        raise NotImplementedError('todo')
 
     def get_swagger_path(
         self,
         contextualised_rule: str,
     ) -> str:
-        pass
+        # TODO BS 2018-07-15: to do
+        raise NotImplementedError('todo')
 
     def by_pass_output_wrapping(
         self,
         response: typing.Any,
     ) -> bool:
-        pass
+        # TODO BS 2018-07-15: to do
+        raise NotImplementedError('todo')
 
     def add_view(
         self,
@@ -130,30 +136,72 @@ class AiohttpContext(BaseContext):
         http_method: str,
         view_func: typing.Callable[..., typing.Any],
     ) -> None:
-        pass
+        # TODO BS 2018-07-15: to do
+        raise NotImplementedError('todo')
 
     def serve_directory(
         self,
         route_prefix: str,
         directory_path: str,
     ) -> None:
-        pass
+        # TODO BS 2018-07-15: to do
+        raise NotImplementedError('todo')
 
     def is_debug(
         self,
     ) -> bool:
-        pass
+        return self._debug
 
     def handle_exception(
         self,
         exception_class: typing.Type[Exception],
         http_code: int,
     ) -> None:
-        pass
+        # TODO BS 2018-07-15: to do
+        raise NotImplementedError('todo')
 
     def handle_exceptions(
         self,
         exception_classes: typing.List[typing.Type[Exception]],
         http_code: int,
     ) -> None:
-        pass
+        # TODO BS 2018-07-15: to do
+        raise NotImplementedError('todo')
+
+    async def get_stream_response_object(
+        self,
+        func_args,
+        func_kwargs,
+        http_code: HTTPStatus = HTTPStatus.OK,
+        headers: dict = None,
+    ) -> web.StreamResponse:
+        headers = headers or {
+            'Content-Type': 'text/plain; charset=utf-8',
+        }
+
+        response = web.StreamResponse(
+            status=http_code,
+            headers=headers,
+        )
+
+        try:
+            request = func_args[0]
+        except IndexError:
+            raise WorkflowException(
+                'Unable to get aiohttp request object',
+            )
+        request = typing.cast(Request, request)
+
+        await response.prepare(request)
+
+        return response
+
+    async def feed_stream_response(
+        self,
+        stream_response: web.StreamResponse,
+        serialized_item: dict,
+    ) -> None:
+        await stream_response.write(
+            # FIXME BS 2018-07-25: need \n :/
+            json.dumps(serialized_item).encode('utf-8') + b'\n',
+        )

hapic/hapic.py

@@ -22,6 +22,7 @@ from hapic.decorator import InputQueryControllerWrapper
 from hapic.decorator import InputFilesControllerWrapper
 from hapic.decorator import OutputBodyControllerWrapper
 from hapic.decorator import AsyncOutputBodyControllerWrapper
+from hapic.decorator import AsyncOutputStreamControllerWrapper
 from hapic.decorator import OutputHeadersControllerWrapper
 from hapic.decorator import OutputFileControllerWrapper
 from hapic.description import InputBodyDescription
@@ -32,6 +33,7 @@ from hapic.description import InputPathDescription
 from hapic.description import InputQueryDescription
 from hapic.description import InputFilesDescription
 from hapic.description import OutputBodyDescription
+from hapic.description import OutputStreamDescription
 from hapic.description import OutputHeadersDescription
 from hapic.description import OutputFileDescription
 from hapic.doc import DocGenerator
@@ -170,6 +172,34 @@ class Hapic(object):
             return decoration.get_wrapper(func)
         return decorator
 
+    def output_stream(
+        self,
+        item_schema: typing.Any,
+        processor: ProcessorInterface = None,
+        context: ContextInterface = None,
+        error_http_code: HTTPStatus = HTTPStatus.INTERNAL_SERVER_ERROR,
+        default_http_code: HTTPStatus = HTTPStatus.OK,
+    ) -> typing.Callable[[typing.Callable[..., typing.Any]], typing.Any]:
+        processor = processor or MarshmallowOutputProcessor()
+        processor.schema = item_schema
+        context = context or self._context_getter
+
+        if self._async:
+            decoration = AsyncOutputStreamControllerWrapper(
+                context=context,
+                processor=processor,
+                error_http_code=error_http_code,
+                default_http_code=default_http_code,
+            )
+        else:
+            # TODO BS 2018-07-25: To do
+            raise NotImplementedError('todo')
+
+        def decorator(func):
+            self._buffer.output_stream = OutputStreamDescription(decoration)
+            return decoration.get_wrapper(func)
+        return decorator
+
     def output_headers(
         self,
         schema: typing.Any,