fix: use body for streaming instead of params

jina/clients/base/helper.py

@@ -203,7 +203,7 @@
             'json': doc.dict() if docarray_v2 else doc.to_dict(),
         }
 
-        async with self.session.post(**request_kwargs) as response:
+        async with self.session.get(**request_kwargs) as response:
             async for chunk in response.content.iter_any():
                 events = chunk.split(b'event: ')[1:]
                 for event in events:

jina/serve/runtimes/gateway/http_fastapi_app_docarrayv2.py

@@ -252,35 +252,14 @@
             methods=['GET'],
             summary=f'Streaming Endpoint {endpoint_path}',
         )
-        async def streaming_get(request: Request):
-            query_params = dict(request.query_params)
+        async def streaming_get(request: Request, body: input_doc_model = None):
+            if not body:
+                query_params = dict(request.query_params)
+                body = input_doc_model.parse_obj(query_params)
 
             async def event_generator():
                 async for doc, error in streamer.stream_doc(
-                    doc=input_doc_model(**query_params), exec_endpoint=endpoint_path
-                ):
-                    if error:
-                        raise HTTPException(status_code=499, detail=str(error))
-                    yield {
-                        'event': 'update',
-                        'data': doc.dict()
-                    }
-                yield {
-                    'event': 'end'
-                }
-
-            return EventSourceResponse(event_generator())
-
-        @app.api_route(
-            path=f'/{endpoint_path.strip("/")}',
-            methods=['POST'],
-            summary=f'Streaming Endpoint {endpoint_path}',
-        )
-        async def streaming_post(body: dict):
-
-            async def event_generator():
-                async for doc, error in streamer.stream_doc(
-                    doc=input_doc_model.parse_obj(body), exec_endpoint=endpoint_path
+                    doc=body, exec_endpoint=endpoint_path
                 ):
                     if error:
                         raise HTTPException(status_code=499, detail=str(error))

jina/serve/runtimes/worker/http_fastapi_app.py

@@ -1,7 +1,7 @@
 import inspect
 from typing import TYPE_CHECKING, Callable, Dict, List, Optional, Union
 
-from jina import DocumentArray, Document
+from jina import Document, DocumentArray
 from jina._docarray import docarray_v2
 from jina.importer import ImportExtensions
 from jina.serve.networking.sse import EventSourceResponse
@@ -11,15 +11,15 @@
     from jina.logging.logger import JinaLogger
 
 if docarray_v2:
-    from docarray import DocList, BaseDoc
+    from docarray import BaseDoc, DocList
 
 
 def get_fastapi_app(
-        request_models_map: Dict,
-        caller: Callable,
-        logger: 'JinaLogger',
-        cors: bool = False,
-        **kwargs,
+    request_models_map: Dict,
+    caller: Callable,
+    logger: 'JinaLogger',
+    cors: bool = False,
+    **kwargs,
 ):
     """
     Get the app from FastAPI as the REST interface.
@@ -35,15 +35,18 @@
         from fastapi import FastAPI, Response, HTTPException
         import pydantic
         from fastapi.middleware.cors import CORSMiddleware
+    import os
+
     from pydantic import BaseModel, Field
     from pydantic.config import BaseConfig, inherit_config
 
     from jina.proto import jina_pb2
     from jina.serve.runtimes.gateway.models import _to_camel_case
-    import os
 
     class Header(BaseModel):
-        request_id: Optional[str] = Field(description='Request ID', example=os.urandom(16).hex())
+        request_id: Optional[str] = Field(
+            description='Request ID', example=os.urandom(16).hex()
+        )
 
         class Config(BaseConfig):
             alias_generator = _to_camel_case
@@ -66,11 +69,11 @@
         logger.warning('CORS is enabled. This service is accessible from any website!')
 
     def add_post_route(
-            endpoint_path,
-            input_model,
-            output_model,
-            input_doc_list_model=None,
-            output_doc_list_model=None,
+        endpoint_path,
+        input_model,
+        output_model,
+        input_doc_list_model=None,
+        output_doc_list_model=None,
     ):
         app_kwargs = dict(
             path=f'/{endpoint_path.strip("/")}',
@@ -123,8 +126,8 @@
                 return ret
 
     def add_streaming_routes(
-            endpoint_path,
-            input_doc_model=None,
+        endpoint_path,
+        input_doc_model=None,
     ):
         from fastapi import Request
 
@@ -133,26 +136,14 @@
             methods=['GET'],
             summary=f'Streaming Endpoint {endpoint_path}',
         )
-        async def streaming_get(request: Request):
-            query_params = dict(request.query_params)
-            req = DataRequest()
-            req.header.exec_endpoint = endpoint_path
-            if not docarray_v2:
-                req.data.docs = DocumentArray([Document.from_dict(query_params)])
-            else:
-                req.document_array_cls = DocList[input_doc_model]
-                req.data.docs = DocList[input_doc_model](
-                    [input_doc_model(**query_params)]
+        async def streaming_get(request: Request, body: input_doc_model = None):
+            if not body:
+                query_params = dict(request.query_params)
+                body = (
+                    input_doc_model.parse_obj(query_params)
+                    if docarray_v2
+                    else Document.from_dict(query_params)
                 )
-            event_generator = _gen_dict_documents(await caller(req))
-            return EventSourceResponse(event_generator)
-
-        @app.api_route(
-            path=f'/{endpoint_path.strip("/")}',
-            methods=['POST'],
-            summary=f'Streaming Endpoint {endpoint_path}',
-        )
-        async def streaming_post(body: input_doc_model, request: Request):
             req = DataRequest()
             req.header.exec_endpoint = endpoint_path
             if not docarray_v2:
@@ -169,7 +160,9 @@
             output_doc_model = input_output_map['output']['model']
             is_generator = input_output_map['is_generator']
             parameters_model = input_output_map['parameters']['model'] or Optional[Dict]
-            default_parameters = ... if input_output_map['parameters']['model'] else None
+            default_parameters = (
+                ... if input_output_map['parameters']['model'] else None
+            )
 
             if docarray_v2:
                 _config = inherit_config(InnerConfig, BaseDoc.__config__)

tests/integration/docarray_v2/test_issues.py

@@ -1,10 +1,10 @@
-from typing import List, Optional, Dict
+from typing import Dict, List, Optional
 
 import pytest
 from docarray import BaseDoc, DocList
 from pydantic import Field
 
-from jina import Executor, Flow, requests, Deployment, Client
+from jina import Client, Deployment, Executor, Flow, requests
 
 
 class Nested2Doc(BaseDoc):
@@ -78,6 +78,7 @@ def test_issue_6019_with_nested_list():
         assert res[0].text == 'hello world'
         assert res[0].nested[0].nested.value == 'test'
 
+
 def test_issue_6084():
     class EnvInfo(BaseDoc):
         history: str = ''
@@ -86,7 +87,6 @@ class A(BaseDoc):
         b: EnvInfo
 
     class MyIssue6084Exec(Executor):
-
         @requests
         def foo(self, docs: DocList[A], **kwargs) -> DocList[A]:
             pass
@@ -115,7 +115,10 @@ class InputWithComplexFields(BaseDoc):
     class MyExecutor(Executor):
         @requests(on="/stream")
         async def stream(
-            self, doc: InputWithComplexFields, parameters: Optional[Dict] = None, **kwargs
+            self,
+            doc: InputWithComplexFields,
+            parameters: Optional[Dict] = None,
+            **kwargs,
         ) -> InputWithComplexFields:
             for i in range(4):
                 yield InputWithComplexFields(text=f"hello world {doc.text} {i}")
@@ -134,10 +137,9 @@ async def stream(
             docs.append(doc)
 
     assert [d.text for d in docs] == [
-        "hello world my input text 0",
-        "hello world my input text 1",
-        "hello world my input text 2",
-        "hello world my input text 3",
+        'hello world test 0',
+        'hello world test 1',
+        'hello world test 2',
+        'hello world test 3',
     ]
     assert docs[0].nested_field.name == "test_name"
-

tests/integration/docarray_v2/test_streaming.py

@@ -143,19 +143,17 @@ async def test_streaming_delay(protocol, include_gateway):
     ):
         client = Client(port=port, protocol=protocol, asyncio=True)
         i = 0
-        stream = client.stream_doc(
+        start_time = time.time()
+        async for doc in client.stream_doc(
             on='/hello',
             inputs=MyDocument(text='hello world', number=i),
             return_type=MyDocument,
-        )
-        start_time = None
-        async for doc in stream:
-            start_time = start_time or time.time()
+        ):
             assert doc.text == f'hello world {i}'
             i += 1
-            delay = time.time() - start_time
+
             # 0.5 seconds between each request + 0.5 seconds tolerance interval
-            assert delay < (0.5 * i), f'Expected delay to be less than {0.5 * i}, got {delay} on iteration {i}'
+            assert time.time() - start_time < (0.5 * i) + 0.5
 
 
 @pytest.mark.asyncio