From 8fe667c94b29fdd276d5581cd87f1e0c70491336 Mon Sep 17 00:00:00 2001 From: Libba Lawrence Date: Wed, 15 Jul 2026 15:47:20 -0700 Subject: [PATCH] [azure-core/corehttp] Add JSONL streaming support Adds a stream-agnostic Stream/AsyncStream iterator plus JSONLDecoder/AsyncJSONLDecoder to both azure-core (azure.core.streaming) and corehttp (corehttp.streaming), kept in sync. The deserialization_callback receives the response and each decoded event so generated code can support the cls custom-deserializer pattern. Each package includes sync/async tests, test-server JSONL routes, a sample, CHANGELOG entry, and a README section (azure-core also bumps the version to 1.42.0). Based on the design in https://gist.github.com/kristapratico/d330af39962ea05b10384b865e37b36f and the original corehttp prototype in Azure/azure-sdk-for-python#39478. Part of Azure/azure-sdk-for-python#38806. Co-authored-by: Krista Pratico Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- sdk/core/azure-core/CHANGELOG.md | 6 + sdk/core/azure-core/README.md | 21 ++ sdk/core/azure-core/azure/core/_version.py | 2 +- .../azure/core/streaming/__init__.py | 38 ++++ .../azure/core/streaming/_decoders.py | 154 +++++++++++++ .../azure/core/streaming/_stream.py | 139 ++++++++++++ sdk/core/azure-core/samples/sample_stream.py | 75 +++++++ .../tests/async_tests/test_stream_async.py | 202 ++++++++++++++++++ sdk/core/azure-core/tests/test_stream.py | 187 ++++++++++++++++ .../coretestserver/test_routes/streams.py | 133 ++++++++++++ sdk/core/corehttp/CHANGELOG.md | 1 + sdk/core/corehttp/README.md | 21 ++ .../corehttp/corehttp/streaming/__init__.py | 38 ++++ .../corehttp/corehttp/streaming/_decoders.py | 154 +++++++++++++ .../corehttp/corehttp/streaming/_stream.py | 139 ++++++++++++ sdk/core/corehttp/samples/sample_stream.py | 75 +++++++ .../tests/async_tests/test_stream_async.py | 202 ++++++++++++++++++ sdk/core/corehttp/tests/test_stream.py | 187 ++++++++++++++++ .../coretestserver/test_routes/streams.py | 133 ++++++++++++ 19 files changed, 1906 insertions(+), 1 deletion(-) create mode 100644 sdk/core/azure-core/azure/core/streaming/__init__.py create mode 100644 sdk/core/azure-core/azure/core/streaming/_decoders.py create mode 100644 sdk/core/azure-core/azure/core/streaming/_stream.py create mode 100644 sdk/core/azure-core/samples/sample_stream.py create mode 100644 sdk/core/azure-core/tests/async_tests/test_stream_async.py create mode 100644 sdk/core/azure-core/tests/test_stream.py create mode 100644 sdk/core/corehttp/corehttp/streaming/__init__.py create mode 100644 sdk/core/corehttp/corehttp/streaming/_decoders.py create mode 100644 sdk/core/corehttp/corehttp/streaming/_stream.py create mode 100644 sdk/core/corehttp/samples/sample_stream.py create mode 100644 sdk/core/corehttp/tests/async_tests/test_stream_async.py create mode 100644 sdk/core/corehttp/tests/test_stream.py diff --git a/sdk/core/azure-core/CHANGELOG.md b/sdk/core/azure-core/CHANGELOG.md index f77fe1020e82..b863295a8243 100644 --- a/sdk/core/azure-core/CHANGELOG.md +++ b/sdk/core/azure-core/CHANGELOG.md @@ -1,5 +1,11 @@ # Release History +## 1.42.0 (Unreleased) + +### Features Added + +- Added JSONL streaming support via the `azure.core.streaming` module, including the `Stream`/`AsyncStream` iterators and `JSONLDecoder`/`AsyncJSONLDecoder`. #38806 + ## 1.41.0 (2026-05-07) ### Features Added diff --git a/sdk/core/azure-core/README.md b/sdk/core/azure-core/README.md index b58719c7a48f..7efe651c8e36 100644 --- a/sdk/core/azure-core/README.md +++ b/sdk/core/azure-core/README.md @@ -229,6 +229,27 @@ foo = Foo( ) ``` +#### Streaming + +`azure.core` provides a stream-agnostic `Stream` iterator for consuming streaming responses. Currently, JSON Lines (JSONL) streaming is supported via the `JSONLDecoder`. Pass the streamed response together with a decoder and a `deserialization_callback` that receives the response and each decoded event: + +```python +from azure.core.rest import HttpRequest, HttpResponse +from azure.core.streaming import Stream, JSONLDecoder + +request = HttpRequest("GET", "https://example.com/stream") +response = client.send_request(request, stream=True) + +def deserialize(response, event): + return event # or deserialize into a model + +with Stream(response=response, decoder=JSONLDecoder(), deserialization_callback=deserialize) as stream: + for item in stream: + print(item) +``` + +An asynchronous equivalent is available using `AsyncStream` and `AsyncJSONLDecoder`. + ## Logging Azure libraries follow the guidance of Python's standard [logging](https://docs.python.org/3/library/logging.html) module. By following the Python documentation on logging, you should be able to configure logging for Azure libraries effectively. diff --git a/sdk/core/azure-core/azure/core/_version.py b/sdk/core/azure-core/azure/core/_version.py index b61909f350dd..7e497df87224 100644 --- a/sdk/core/azure-core/azure/core/_version.py +++ b/sdk/core/azure-core/azure/core/_version.py @@ -9,4 +9,4 @@ # regenerated. # -------------------------------------------------------------------------- -VERSION = "1.41.0" +VERSION = "1.42.0" diff --git a/sdk/core/azure-core/azure/core/streaming/__init__.py b/sdk/core/azure-core/azure/core/streaming/__init__.py new file mode 100644 index 000000000000..89be35e00564 --- /dev/null +++ b/sdk/core/azure-core/azure/core/streaming/__init__.py @@ -0,0 +1,38 @@ +# -------------------------------------------------------------------------- +# +# Copyright (c) Microsoft Corporation. All rights reserved. +# +# The MIT License (MIT) +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the ""Software""), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS +# IN THE SOFTWARE. +# +# -------------------------------------------------------------------------- + +from ._stream import Stream, AsyncStream +from ._decoders import StreamDecoder, AsyncStreamDecoder, JSONLDecoder, AsyncJSONLDecoder + + +__all__ = [ + "Stream", + "AsyncStream", + "StreamDecoder", + "AsyncStreamDecoder", + "JSONLDecoder", + "AsyncJSONLDecoder", +] diff --git a/sdk/core/azure-core/azure/core/streaming/_decoders.py b/sdk/core/azure-core/azure/core/streaming/_decoders.py new file mode 100644 index 000000000000..29df2c8d8b83 --- /dev/null +++ b/sdk/core/azure-core/azure/core/streaming/_decoders.py @@ -0,0 +1,154 @@ +# -------------------------------------------------------------------------- +# +# Copyright (c) Microsoft Corporation. All rights reserved. +# +# The MIT License (MIT) +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the ""Software""), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS +# IN THE SOFTWARE. +# +# -------------------------------------------------------------------------- + +import codecs +import json +from typing import Iterator, AsyncIterator, Protocol, Any, MutableMapping, Generic + +from typing_extensions import runtime_checkable, TypeVar + +T_co = TypeVar("T_co", covariant=True) +T = TypeVar("T", default=MutableMapping[str, Any]) + + +@runtime_checkable +class StreamDecoder(Protocol[T_co]): + """Protocol for stream decoders.""" + + def iter_events(self, iter_bytes: Iterator[bytes]) -> Iterator[T_co]: + """Iterate over events from a byte iterator. + + :param iter_bytes: An iterator of byte chunks. + :type iter_bytes: Iterator[bytes] + :return: An iterator of decoded data. + :rtype: Iterator[DecodedType_co] + """ + ... + + +@runtime_checkable +class AsyncStreamDecoder(Protocol[T_co]): + """Protocol for async stream decoders.""" + + # Why this isn't async def: https://mypy.readthedocs.io/en/stable/more_types.html#asynchronous-iterators + def aiter_events(self, iter_bytes: AsyncIterator[bytes]) -> AsyncIterator[T_co]: + """Asynchronously iterate over events from a byte iterator. + + :param iter_bytes: An asynchronous iterator of byte chunks. + :type iter_bytes: AsyncIterator[bytes] + :return: An asynchronous iterator of decoded data. + :rtype: AsyncIterator[DecodedType_co] + """ + ... + + +def iter_lines(iter_bytes: Iterator[bytes]) -> Iterator[str]: + """Iterate over lines from a byte iterator. + + :param iter_bytes: An iterator of byte chunks. + :type iter_bytes: Iterator[bytes] + :rtype: Iterator[str] + :return: An iterator of lines. + """ + decoder = codecs.getincrementaldecoder("utf-8")() + + decoded = "" + for chunk in iter_bytes: + decoded += decoder.decode(chunk) + if decoded: + decoded_lines = decoded.splitlines() + if decoded.endswith(("\n", "\r\n")): + yield from decoded_lines + decoded = "" + else: + yield from decoded_lines[:-1] + decoded = decoded_lines[-1] + + decoded += decoder.decode(b"", final=True) + if decoded: + yield from decoded.splitlines() + + +async def aiter_lines(iter_bytes: AsyncIterator[bytes]) -> AsyncIterator[str]: + """Iterate over lines from a byte iterator. + + :param iter_bytes: An iterator of byte chunks. + :type iter_bytes: Iterator[bytes] + :rtype: Iterator[str] + :return: An iterator of lines. + """ + decoder = codecs.getincrementaldecoder("utf-8")() + + decoded = "" + async for chunk in iter_bytes: + decoded += decoder.decode(chunk) + if decoded: + decoded_lines = decoded.splitlines() + if decoded.endswith(("\n", "\r\n")): + for line in decoded_lines: + yield line + decoded = "" + else: + for line in decoded_lines[:-1]: + yield line + decoded = decoded_lines[-1] + + decoded += decoder.decode(b"", final=True) + if decoded: + for line in decoded.splitlines(): + yield line + + +class JSONLDecoder(Generic[T]): + """Decoder for JSON Lines (JSONL) format. https://jsonlines.org/""" + + def iter_events(self, iter_bytes: Iterator[bytes]) -> Iterator[T]: + """Iterate over JSONL events from a byte iterator. + + :param iter_bytes: An iterator of byte chunks. + :type iter_bytes: Iterator[bytes] + :rtype: Iterator[T] + :return: An iterator of objects. + """ + + yield from (json.loads(line) for line in iter_lines(iter_bytes)) + + +class AsyncJSONLDecoder(Generic[T]): + """Asynchronous decoder for JSON Lines (JSONL) format. https://jsonlines.org/""" + + # pylint: disable=invalid-overridden-method + async def aiter_events(self, iter_bytes: AsyncIterator[bytes]) -> AsyncIterator[T]: + """Asynchronously iterate over JSONL events from a byte iterator. + + :param iter_bytes: An asynchronous iterator of byte chunks. + :type iter_bytes: AsyncIterator[bytes] + :rtype: AsyncIterator[T] + :return: An asynchronous iterator of objects. + """ + + async for line in aiter_lines(iter_bytes): + yield json.loads(line) diff --git a/sdk/core/azure-core/azure/core/streaming/_stream.py b/sdk/core/azure-core/azure/core/streaming/_stream.py new file mode 100644 index 000000000000..54d64c6b0f2a --- /dev/null +++ b/sdk/core/azure-core/azure/core/streaming/_stream.py @@ -0,0 +1,139 @@ +# -------------------------------------------------------------------------- +# +# Copyright (c) Microsoft Corporation. All rights reserved. +# +# The MIT License (MIT) +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the ""Software""), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS +# IN THE SOFTWARE. +# +# -------------------------------------------------------------------------- + +from types import TracebackType +from typing import Iterator, AsyncIterator, TypeVar, Callable, Optional, Type + +from typing_extensions import Self + +from ..rest import HttpResponse, AsyncHttpResponse +from ._decoders import StreamDecoder, AsyncStreamDecoder + +DecodedType = TypeVar("DecodedType") +ReturnType_co = TypeVar("ReturnType_co", covariant=True) + + +class Stream(Iterator[ReturnType_co]): + """Stream class for streaming JSONL. + + :keyword response: The response object. + :paramtype response: ~azure.core.rest.HttpResponse + :keyword decoder: A decoder to use for the stream. + :paramtype decoder: ~azure.core.streaming.StreamDecoder + :keyword deserialization_callback: A callback that takes the response and the decoded event and + returns a deserialized object. + :paramtype deserialization_callback: Callable[[~azure.core.rest.HttpResponse, Any], ReturnType] + """ + + def __init__( + self, + *, + response: HttpResponse, + decoder: StreamDecoder[DecodedType], + deserialization_callback: Callable[[HttpResponse, DecodedType], ReturnType_co], + ) -> None: + self._response = response + self._decoder = decoder + self._deserialization_callback = deserialization_callback + self._iterator = self._iter_results() + + def __next__(self) -> ReturnType_co: + return self._iterator.__next__() + + def __iter__(self) -> Iterator[ReturnType_co]: + yield from self._iterator + + def _iter_results(self) -> Iterator[ReturnType_co]: + for event in self._decoder.iter_events(self._response.iter_bytes()): + + result = self._deserialization_callback(self._response, event) + yield result + + def __exit__( + self, + exc_type: Optional[Type[BaseException]] = None, + exc_value: Optional[BaseException] = None, + traceback: Optional[TracebackType] = None, + ) -> None: + self.close() + + def __enter__(self) -> Self: + return self + + def close(self) -> None: + self._response.close() + + +class AsyncStream(AsyncIterator[ReturnType_co]): + """AsyncStream class for asynchronously streaming JSONL. + + :keyword response: The response object. + :paramtype response: ~azure.core.rest.AsyncHttpResponse + :keyword decoder: A decoder to use for the stream. + :paramtype decoder: ~azure.core.streaming.AsyncStreamDecoder + :keyword deserialization_callback: A callback that takes the response and the decoded event and + returns a deserialized object. + :paramtype deserialization_callback: Callable[[~azure.core.rest.AsyncHttpResponse, Any], ReturnType] + """ + + def __init__( + self, + *, + response: AsyncHttpResponse, + decoder: AsyncStreamDecoder[DecodedType], + deserialization_callback: Callable[[AsyncHttpResponse, DecodedType], ReturnType_co], + ) -> None: + self._response = response + self._decoder = decoder + self._deserialization_callback = deserialization_callback + self._iterator = self._iter_results() + + async def __anext__(self) -> ReturnType_co: + return await self._iterator.__anext__() + + async def __aiter__(self) -> AsyncIterator[ReturnType_co]: # pylint: disable=invalid-overridden-method + async for item in self._iterator: + yield item + + async def _iter_results(self) -> AsyncIterator[ReturnType_co]: + async for event in self._decoder.aiter_events(self._response.iter_bytes()): + + result = self._deserialization_callback(self._response, event) + yield result + + async def __aexit__( + self, + exc_type: Optional[Type[BaseException]] = None, + exc_value: Optional[BaseException] = None, + traceback: Optional[TracebackType] = None, + ) -> None: + await self.close() + + async def __aenter__(self) -> Self: + return self + + async def close(self) -> None: + await self._response.close() diff --git a/sdk/core/azure-core/samples/sample_stream.py b/sdk/core/azure-core/samples/sample_stream.py new file mode 100644 index 000000000000..1e613596285b --- /dev/null +++ b/sdk/core/azure-core/samples/sample_stream.py @@ -0,0 +1,75 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- + +""" +FILE: sample_stream.py + +DESCRIPTION: + This sample demonstrates how to consume a JSON Lines (JSONL) streaming + response synchronously and asynchronously using the Stream/AsyncStream + iterators and JSONLDecoder/AsyncJSONLDecoder. + +USAGE: + python sample_stream.py +""" + +from typing import Any, MutableMapping +import asyncio + + +def sample_stream(): + # [START build_stream] + from azure.core import PipelineClient + from azure.core.rest import HttpRequest, HttpResponse + from azure.core.streaming import Stream, JSONLDecoder + + client: PipelineClient[HttpRequest, HttpResponse] = PipelineClient("https://example.com") + request = HttpRequest("GET", "https://example.com/stream") + response = client.send_request(request, stream=True) + + def deserialize(response: HttpResponse, event: MutableMapping[str, Any]) -> MutableMapping[str, Any]: + # Deserialize each decoded JSON object into a model. Here we just return it as-is. + return event + + stream = Stream( + response=response, + decoder=JSONLDecoder(), + deserialization_callback=deserialize, + ) + with stream: + for item in stream: + print(item) + # [END build_stream] + + +async def sample_stream_async(): + # [START build_stream_async] + from azure.core import AsyncPipelineClient + from azure.core.rest import HttpRequest, AsyncHttpResponse + from azure.core.streaming import AsyncStream, AsyncJSONLDecoder + + client: AsyncPipelineClient[HttpRequest, AsyncHttpResponse] = AsyncPipelineClient("https://example.com") + request = HttpRequest("GET", "https://example.com/stream") + response = await client.send_request(request, stream=True) + + def deserialize(response: AsyncHttpResponse, event: MutableMapping[str, Any]) -> MutableMapping[str, Any]: + # Deserialize each decoded JSON object into a model. Here we just return it as-is. + return event + + stream = AsyncStream( + response=response, + decoder=AsyncJSONLDecoder(), + deserialization_callback=deserialize, + ) + async with stream: + async for item in stream: + print(item) + # [END build_stream_async] + + +if __name__ == "__main__": + sample_stream() + asyncio.run(sample_stream_async()) diff --git a/sdk/core/azure-core/tests/async_tests/test_stream_async.py b/sdk/core/azure-core/tests/async_tests/test_stream_async.py new file mode 100644 index 000000000000..0a1c64197dbf --- /dev/null +++ b/sdk/core/azure-core/tests/async_tests/test_stream_async.py @@ -0,0 +1,202 @@ +# -------------------------------------------------------------------------- +# +# Copyright (c) Microsoft Corporation. All rights reserved. +# +# The MIT License (MIT) +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the ""Software""), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. +# +# -------------------------------------------------------------------------- + +import json + +import pytest + +from azure.core.rest import HttpRequest +from azure.core.streaming import AsyncStream, AsyncJSONLDecoder + + +@pytest.fixture +def deserialization_callback(): + def _callback(response, model_json): + return model_json + + return _callback + + +@pytest.fixture +def stream(client, deserialization_callback): + async def _callback(request, **kwargs): + http_response = await client.send_request(request=request, stream=True) + return AsyncStream( + deserialization_callback=deserialization_callback, response=http_response, decoder=AsyncJSONLDecoder() + ) + + return _callback + + +@pytest.mark.asyncio +async def test_stream_jsonl_basic(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_basic")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a message"}, + {"msg": "this is another message"}, + {"msg": "this is a third message"}, + {"msg": "this is a fourth message"}, + ] + + +@pytest.mark.asyncio +async def test_stream_jsonl_multiple_kv(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_multiple_kv")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a hello world message", "planet": {"earth": "hello earth", "mars": "hello mars"}}, + {"msg": "this is a hello world message", "planet": {"venus": "hello venus", "jupiter": "hello jupiter"}}, + ] + + +@pytest.mark.asyncio +async def test_stream_jsonl_no_final_line_separator(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_no_final_line_separator")) + async for s in jsonl_stream: + assert s == {"msg": "this is a message"} + + +@pytest.mark.asyncio +async def test_stream_jsonl_broken_up_data(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_broken_up_data")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "this is a second message"}, + ] + + +@pytest.mark.asyncio +async def test_stream_jsonl_broken_up_data_cr(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_broken_up_data_cr")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "this is a second message"}, + ] + + +@pytest.mark.asyncio +async def test_stream_jsonl_next(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_basic")) + message = await jsonl_stream.__anext__() + assert message == {"msg": "this is a message"} + message = await jsonl_stream.__anext__() + assert message == {"msg": "this is another message"} + message = await jsonl_stream.__anext__() + assert message == {"msg": "this is a third message"} + message = await jsonl_stream.__anext__() + assert message == {"msg": "this is a fourth message"} + + with pytest.raises(StopAsyncIteration): + await jsonl_stream.__anext__() + + +@pytest.mark.asyncio +async def test_stream_jsonl_context_manager(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_basic")) + async with jsonl_stream as streaming: + async for _ in streaming: + break + assert streaming._response.is_closed + + +@pytest.mark.asyncio +async def test_stream_jsonl_invalid_data(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_invalid_data")) + + with pytest.raises(json.decoder.JSONDecodeError): + async for _ in jsonl_stream: + ... + + +@pytest.mark.asyncio +async def test_stream_jsonl_escaped_newline_data(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_escaped_newline_data")) + + async for s in jsonl_stream: + assert s == {"msg": "this is a...\nmessage"} + + +@pytest.mark.asyncio +async def test_stream_jsonl_escaped_broken_newline_data(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_escaped_broken_newline_data")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "\nthis is a second message"}, + ] + + +@pytest.mark.asyncio +async def test_stream_jsonl_incomplete_char(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_broken_incomplete_char")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "𝜋this is a second message𝜋"}, + {"msg": "this is a third message"}, + ] + + +@pytest.mark.asyncio +async def test_stream_jsonl_list(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_list")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + ["this", "is", "a", "first", "message"], + ["this", "is", "a", "second", "message"], + ["this", "is", "a", "third", "message"], + ] + + +@pytest.mark.asyncio +async def test_stream_jsonl_string(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_string")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + "this", + "is", + "a", + "message", + ] diff --git a/sdk/core/azure-core/tests/test_stream.py b/sdk/core/azure-core/tests/test_stream.py new file mode 100644 index 000000000000..7dfe37a8d9e0 --- /dev/null +++ b/sdk/core/azure-core/tests/test_stream.py @@ -0,0 +1,187 @@ +# -------------------------------------------------------------------------- +# +# Copyright (c) Microsoft Corporation. All rights reserved. +# +# The MIT License (MIT) +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the ""Software""), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. +# +# -------------------------------------------------------------------------- + +import json + +import pytest + +from azure.core.rest import HttpRequest +from azure.core.streaming import Stream, JSONLDecoder + + +@pytest.fixture +def deserialization_callback(): + def _callback(response, model_json): + return model_json + + return _callback + + +@pytest.fixture +def stream(client, deserialization_callback): + def _callback(request, **kwargs): + http_response = client.send_request(request=request, stream=True) + return Stream(deserialization_callback=deserialization_callback, response=http_response, decoder=JSONLDecoder()) + + return _callback + + +def test_stream_jsonl_basic(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_basic")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a message"}, + {"msg": "this is another message"}, + {"msg": "this is a third message"}, + {"msg": "this is a fourth message"}, + ] + + +def test_stream_jsonl_multiple_kv(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_multiple_kv")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a hello world message", "planet": {"earth": "hello earth", "mars": "hello mars"}}, + {"msg": "this is a hello world message", "planet": {"venus": "hello venus", "jupiter": "hello jupiter"}}, + ] + + +def test_stream_jsonl_no_final_line_separator(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_no_final_line_separator")) + for s in jsonl_stream: + assert s == {"msg": "this is a message"} + + +def test_stream_jsonl_broken_up_data(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_broken_up_data")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "this is a second message"}, + ] + + +def test_stream_jsonl_broken_up_data_cr(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_broken_up_data_cr")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "this is a second message"}, + ] + + +def test_stream_jsonl_next(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_basic")) + message = next(jsonl_stream) + assert message == {"msg": "this is a message"} + message = next(jsonl_stream) + assert message == {"msg": "this is another message"} + message = next(jsonl_stream) + assert message == {"msg": "this is a third message"} + message = next(jsonl_stream) + assert message == {"msg": "this is a fourth message"} + + with pytest.raises(StopIteration): + next(jsonl_stream) + + +def test_stream_jsonl_context_manager(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_basic")) + with jsonl_stream as streaming: + for _ in streaming: + break + assert streaming._response.is_closed + + +def test_stream_jsonl_invalid_data(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_invalid_data")) + + with pytest.raises(json.decoder.JSONDecodeError): + for _ in jsonl_stream: + ... + + +def test_stream_jsonl_escaped_newline_data(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_escaped_newline_data")) + + for s in jsonl_stream: + assert s == {"msg": "this is a...\nmessage"} + + +def test_stream_jsonl_escaped_broken_newline_data(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_escaped_broken_newline_data")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "\nthis is a second message"}, + ] + + +def test_stream_jsonl_incomplete_char(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_broken_incomplete_char")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "𝜋this is a second message𝜋"}, + {"msg": "this is a third message"}, + ] + + +def test_stream_jsonl_list(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_list")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + ["this", "is", "a", "first", "message"], + ["this", "is", "a", "second", "message"], + ["this", "is", "a", "third", "message"], + ] + + +def test_stream_jsonl_string(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_string")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + "this", + "is", + "a", + "message", + ] diff --git a/sdk/core/azure-core/tests/testserver_tests/coretestserver/coretestserver/test_routes/streams.py b/sdk/core/azure-core/tests/testserver_tests/coretestserver/coretestserver/test_routes/streams.py index 7a4c6499cb06..9be9adfa4d27 100644 --- a/sdk/core/azure-core/tests/testserver_tests/coretestserver/coretestserver/test_routes/streams.py +++ b/sdk/core/azure-core/tests/testserver_tests/coretestserver/coretestserver/test_routes/streams.py @@ -104,3 +104,136 @@ def upload(): break byte_content += chunk return Response(byte_content, status=200) + + +def stream_jsonl_basic(): + data = [ + b'{"msg": "this is a message"}\n', + b'{"msg": "this is another message"}\n', + b'{"msg": "this is a third message"}\n{"msg": "this is a fourth message"}\n', + ] + yield from data + + +def stream_jsonl_multiple_kv(): + data = [ + b'{"msg": "this is a hello world message", "planet": {"earth": "hello earth", "mars": "hello mars"}}\n', + b'{"msg": "this is a hello world message", "planet": {"venus": "hello venus", "jupiter": "hello jupiter"}}\n', + ] + yield from data + + +def stream_jsonl_no_final_line_separator(): + data = b'{"msg": "this is a message"}' + yield data + + +def stream_jsonl_broken_up_data(): + data = [b'{"msg": "this is a first message"}\n{"msg": ', b'"this is a second message"}\n'] + yield from data + + +def stream_jsonl_broken_up_data_cr(): + data = [b'{"msg": "this is a first message"}\r\n{"msg": ', b'"this is a second message"}\r\n'] + yield from data + + +def stream_jsonl_invalid_data(): + data = [b'{"msg": "this is a third m'] + yield from data + + +def stream_jsonl_escaped_newline_data(): + data = b'{"msg": "this is a...\\nmessage"}\n' + yield data + + +def stream_jsonl_escaped_broken_newline_data(): + data = [b'{"msg": "this is a first message"}\n{"msg": "\\n', b'this is a second message"}\n'] + yield from data + + +def stream_jsonl_broken_incomplete_char(): + data = [ + b'{"msg": "this is a first message"}\n{"msg": "\xf0\x9d', + b"\x9c\x8bthis is a second message\xf0\x9d", + b'\x9c\x8b"}', + b'\n{"msg": "this is a third message"}', + ] + yield from data + + +def stream_jsonl_list(): + data = [ + b'["this", "is", "a", "first", "message"]\n', + b'["this", "is", "a", "second", "message"]\n', + b'["this", "is", "a", "third", "message"]\n', + ] + yield from data + + +def stream_jsonl_string(): + data = [ + b'"this"\n', + b'"is"\n', + b'"a"\n', + b'"message"\n', + ] + yield from data + + +@streams_api.route("/jsonl_basic", methods=["GET"]) +def jsonl_basic(): + return Response(stream_jsonl_basic(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_multiple_kv", methods=["GET"]) +def jsonl_multiple_kv(): + return Response(stream_jsonl_multiple_kv(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_no_final_line_separator", methods=["GET"]) +def jsonl_no_final_line_separator(): + return Response(stream_jsonl_no_final_line_separator(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_broken_up_data", methods=["GET"]) +def jsonl_broken_up_data(): + return Response(stream_jsonl_broken_up_data(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_broken_up_data_cr", methods=["GET"]) +def jsonl_broken_up_data_cr(): + return Response(stream_jsonl_broken_up_data_cr(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_invalid_data", methods=["GET"]) +def jsonl_invalid_data(): + return Response(stream_jsonl_invalid_data(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_escaped_newline_data", methods=["GET"]) +def jsonl_escaped_newline_data(): + return Response(stream_jsonl_escaped_newline_data(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_escaped_broken_newline_data", methods=["GET"]) +def jsonl_escaped_broken_newline_data(): + return Response( + stream_jsonl_escaped_broken_newline_data(), status=200, headers={"Content-Type": "application/jsonl"} + ) + + +@streams_api.route("/jsonl_broken_incomplete_char", methods=["GET"]) +def jsonl_broken_incomplete_char(): + return Response(stream_jsonl_broken_incomplete_char(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_list", methods=["GET"]) +def json_list(): + return Response(stream_jsonl_list(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_string", methods=["GET"]) +def json_string(): + return Response(stream_jsonl_string(), status=200, headers={"Content-Type": "application/jsonl"}) diff --git a/sdk/core/corehttp/CHANGELOG.md b/sdk/core/corehttp/CHANGELOG.md index 4b21b09f5872..9daf3b6137d2 100644 --- a/sdk/core/corehttp/CHANGELOG.md +++ b/sdk/core/corehttp/CHANGELOG.md @@ -4,6 +4,7 @@ ### Features Added +- Added JSONL streaming support via the `corehttp.streaming` module, including the `Stream`/`AsyncStream` iterators and `JSONLDecoder`/`AsyncJSONLDecoder`. [#38806](https://github.com/Azure/azure-sdk-for-python/issues/38806) - Introduced the keyword argument `additional_allowed_query_params` to `DistributedHttpTracingPolicy` to allow users to specify additional URL query parameters that should not be redacted in span attributes. [#46657](https://github.com/Azure/azure-sdk-for-python/pull/46657) ### Breaking Changes diff --git a/sdk/core/corehttp/README.md b/sdk/core/corehttp/README.md index 406da0fffe84..4099fbd81cbd 100644 --- a/sdk/core/corehttp/README.md +++ b/sdk/core/corehttp/README.md @@ -33,6 +33,27 @@ pip install corehttp[requests,httpx] If no transports are specified, `corehttp` will default to using `RequestsTransport` for synchronous pipeline requests and `AioHttpTransport` for asynchronous pipeline requests. +### Streaming + +`corehttp` provides a stream-agnostic `Stream` iterator for consuming streaming responses. Currently, JSON Lines (JSONL) streaming is supported via the `JSONLDecoder`. Pass the streamed response together with a decoder and a `deserialization_callback` that receives the response and each decoded event: + +```python +from corehttp.rest import HttpRequest, HttpResponse +from corehttp.streaming import Stream, JSONLDecoder + +request = HttpRequest("GET", "https://example.com/stream") +response = client.send_request(request, stream=True) + +def deserialize(response, event): + return event # or deserialize into a model + +with Stream(response=response, decoder=JSONLDecoder(), deserialization_callback=deserialize) as stream: + for item in stream: + print(item) +``` + +An asynchronous equivalent is available using `AsyncStream` and `AsyncJSONLDecoder`. + ## Contributing This project welcomes contributions and suggestions. Most contributions require diff --git a/sdk/core/corehttp/corehttp/streaming/__init__.py b/sdk/core/corehttp/corehttp/streaming/__init__.py new file mode 100644 index 000000000000..89be35e00564 --- /dev/null +++ b/sdk/core/corehttp/corehttp/streaming/__init__.py @@ -0,0 +1,38 @@ +# -------------------------------------------------------------------------- +# +# Copyright (c) Microsoft Corporation. All rights reserved. +# +# The MIT License (MIT) +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the ""Software""), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS +# IN THE SOFTWARE. +# +# -------------------------------------------------------------------------- + +from ._stream import Stream, AsyncStream +from ._decoders import StreamDecoder, AsyncStreamDecoder, JSONLDecoder, AsyncJSONLDecoder + + +__all__ = [ + "Stream", + "AsyncStream", + "StreamDecoder", + "AsyncStreamDecoder", + "JSONLDecoder", + "AsyncJSONLDecoder", +] diff --git a/sdk/core/corehttp/corehttp/streaming/_decoders.py b/sdk/core/corehttp/corehttp/streaming/_decoders.py new file mode 100644 index 000000000000..29df2c8d8b83 --- /dev/null +++ b/sdk/core/corehttp/corehttp/streaming/_decoders.py @@ -0,0 +1,154 @@ +# -------------------------------------------------------------------------- +# +# Copyright (c) Microsoft Corporation. All rights reserved. +# +# The MIT License (MIT) +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the ""Software""), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS +# IN THE SOFTWARE. +# +# -------------------------------------------------------------------------- + +import codecs +import json +from typing import Iterator, AsyncIterator, Protocol, Any, MutableMapping, Generic + +from typing_extensions import runtime_checkable, TypeVar + +T_co = TypeVar("T_co", covariant=True) +T = TypeVar("T", default=MutableMapping[str, Any]) + + +@runtime_checkable +class StreamDecoder(Protocol[T_co]): + """Protocol for stream decoders.""" + + def iter_events(self, iter_bytes: Iterator[bytes]) -> Iterator[T_co]: + """Iterate over events from a byte iterator. + + :param iter_bytes: An iterator of byte chunks. + :type iter_bytes: Iterator[bytes] + :return: An iterator of decoded data. + :rtype: Iterator[DecodedType_co] + """ + ... + + +@runtime_checkable +class AsyncStreamDecoder(Protocol[T_co]): + """Protocol for async stream decoders.""" + + # Why this isn't async def: https://mypy.readthedocs.io/en/stable/more_types.html#asynchronous-iterators + def aiter_events(self, iter_bytes: AsyncIterator[bytes]) -> AsyncIterator[T_co]: + """Asynchronously iterate over events from a byte iterator. + + :param iter_bytes: An asynchronous iterator of byte chunks. + :type iter_bytes: AsyncIterator[bytes] + :return: An asynchronous iterator of decoded data. + :rtype: AsyncIterator[DecodedType_co] + """ + ... + + +def iter_lines(iter_bytes: Iterator[bytes]) -> Iterator[str]: + """Iterate over lines from a byte iterator. + + :param iter_bytes: An iterator of byte chunks. + :type iter_bytes: Iterator[bytes] + :rtype: Iterator[str] + :return: An iterator of lines. + """ + decoder = codecs.getincrementaldecoder("utf-8")() + + decoded = "" + for chunk in iter_bytes: + decoded += decoder.decode(chunk) + if decoded: + decoded_lines = decoded.splitlines() + if decoded.endswith(("\n", "\r\n")): + yield from decoded_lines + decoded = "" + else: + yield from decoded_lines[:-1] + decoded = decoded_lines[-1] + + decoded += decoder.decode(b"", final=True) + if decoded: + yield from decoded.splitlines() + + +async def aiter_lines(iter_bytes: AsyncIterator[bytes]) -> AsyncIterator[str]: + """Iterate over lines from a byte iterator. + + :param iter_bytes: An iterator of byte chunks. + :type iter_bytes: Iterator[bytes] + :rtype: Iterator[str] + :return: An iterator of lines. + """ + decoder = codecs.getincrementaldecoder("utf-8")() + + decoded = "" + async for chunk in iter_bytes: + decoded += decoder.decode(chunk) + if decoded: + decoded_lines = decoded.splitlines() + if decoded.endswith(("\n", "\r\n")): + for line in decoded_lines: + yield line + decoded = "" + else: + for line in decoded_lines[:-1]: + yield line + decoded = decoded_lines[-1] + + decoded += decoder.decode(b"", final=True) + if decoded: + for line in decoded.splitlines(): + yield line + + +class JSONLDecoder(Generic[T]): + """Decoder for JSON Lines (JSONL) format. https://jsonlines.org/""" + + def iter_events(self, iter_bytes: Iterator[bytes]) -> Iterator[T]: + """Iterate over JSONL events from a byte iterator. + + :param iter_bytes: An iterator of byte chunks. + :type iter_bytes: Iterator[bytes] + :rtype: Iterator[T] + :return: An iterator of objects. + """ + + yield from (json.loads(line) for line in iter_lines(iter_bytes)) + + +class AsyncJSONLDecoder(Generic[T]): + """Asynchronous decoder for JSON Lines (JSONL) format. https://jsonlines.org/""" + + # pylint: disable=invalid-overridden-method + async def aiter_events(self, iter_bytes: AsyncIterator[bytes]) -> AsyncIterator[T]: + """Asynchronously iterate over JSONL events from a byte iterator. + + :param iter_bytes: An asynchronous iterator of byte chunks. + :type iter_bytes: AsyncIterator[bytes] + :rtype: AsyncIterator[T] + :return: An asynchronous iterator of objects. + """ + + async for line in aiter_lines(iter_bytes): + yield json.loads(line) diff --git a/sdk/core/corehttp/corehttp/streaming/_stream.py b/sdk/core/corehttp/corehttp/streaming/_stream.py new file mode 100644 index 000000000000..49091bca795e --- /dev/null +++ b/sdk/core/corehttp/corehttp/streaming/_stream.py @@ -0,0 +1,139 @@ +# -------------------------------------------------------------------------- +# +# Copyright (c) Microsoft Corporation. All rights reserved. +# +# The MIT License (MIT) +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the ""Software""), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS +# IN THE SOFTWARE. +# +# -------------------------------------------------------------------------- + +from types import TracebackType +from typing import Iterator, AsyncIterator, TypeVar, Callable, Optional, Type + +from typing_extensions import Self + +from ..rest import HttpResponse, AsyncHttpResponse +from ._decoders import StreamDecoder, AsyncStreamDecoder + +DecodedType = TypeVar("DecodedType") +ReturnType_co = TypeVar("ReturnType_co", covariant=True) + + +class Stream(Iterator[ReturnType_co]): + """Stream class for streaming JSONL. + + :keyword response: The response object. + :paramtype response: ~corehttp.rest.HttpResponse + :keyword decoder: A decoder to use for the stream. + :paramtype decoder: ~corehttp.streaming.decoders.StreamDecoder + :keyword deserialization_callback: A callback that takes the response and the decoded event and + returns a deserialized object. + :paramtype deserialization_callback: Callable[[~corehttp.rest.HttpResponse, Any], ReturnType] + """ + + def __init__( + self, + *, + response: HttpResponse, + decoder: StreamDecoder[DecodedType], + deserialization_callback: Callable[[HttpResponse, DecodedType], ReturnType_co], + ) -> None: + self._response = response + self._decoder = decoder + self._deserialization_callback = deserialization_callback + self._iterator = self._iter_results() + + def __next__(self) -> ReturnType_co: + return self._iterator.__next__() + + def __iter__(self) -> Iterator[ReturnType_co]: + yield from self._iterator + + def _iter_results(self) -> Iterator[ReturnType_co]: + for event in self._decoder.iter_events(self._response.iter_bytes()): + + result = self._deserialization_callback(self._response, event) + yield result + + def __exit__( + self, + exc_type: Optional[Type[BaseException]] = None, + exc_value: Optional[BaseException] = None, + traceback: Optional[TracebackType] = None, + ) -> None: + self.close() + + def __enter__(self) -> Self: + return self + + def close(self) -> None: + self._response.close() + + +class AsyncStream(AsyncIterator[ReturnType_co]): + """AsyncStream class for asynchronously streaming JSONL. + + :keyword response: The response object. + :paramtype response: ~corehttp.rest.AsyncHttpResponse + :keyword decoder: A decoder to use for the stream. + :paramtype decoder: ~corehttp.streaming.decoders.AsyncStreamDecoder + :keyword deserialization_callback: A callback that takes the response and the decoded event and + returns a deserialized object. + :paramtype deserialization_callback: Callable[[~corehttp.rest.AsyncHttpResponse, Any], ReturnType] + """ + + def __init__( + self, + *, + response: AsyncHttpResponse, + decoder: AsyncStreamDecoder[DecodedType], + deserialization_callback: Callable[[AsyncHttpResponse, DecodedType], ReturnType_co], + ) -> None: + self._response = response + self._decoder = decoder + self._deserialization_callback = deserialization_callback + self._iterator = self._iter_results() + + async def __anext__(self) -> ReturnType_co: + return await self._iterator.__anext__() + + async def __aiter__(self) -> AsyncIterator[ReturnType_co]: # pylint: disable=invalid-overridden-method + async for item in self._iterator: + yield item + + async def _iter_results(self) -> AsyncIterator[ReturnType_co]: + async for event in self._decoder.aiter_events(self._response.iter_bytes()): + + result = self._deserialization_callback(self._response, event) + yield result + + async def __aexit__( + self, + exc_type: Optional[Type[BaseException]] = None, + exc_value: Optional[BaseException] = None, + traceback: Optional[TracebackType] = None, + ) -> None: + await self.close() + + async def __aenter__(self) -> Self: + return self + + async def close(self) -> None: + await self._response.close() diff --git a/sdk/core/corehttp/samples/sample_stream.py b/sdk/core/corehttp/samples/sample_stream.py new file mode 100644 index 000000000000..936661087e33 --- /dev/null +++ b/sdk/core/corehttp/samples/sample_stream.py @@ -0,0 +1,75 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- + +""" +FILE: sample_stream.py + +DESCRIPTION: + This sample demonstrates how to consume a JSON Lines (JSONL) streaming + response synchronously and asynchronously using the Stream/AsyncStream + iterators and JSONLDecoder/AsyncJSONLDecoder. + +USAGE: + python sample_stream.py +""" + +from typing import Any, MutableMapping +import asyncio + + +def sample_stream(): + # [START build_stream] + from corehttp.runtime import PipelineClient + from corehttp.rest import HttpRequest, HttpResponse + from corehttp.streaming import Stream, JSONLDecoder + + client: PipelineClient[HttpRequest, HttpResponse] = PipelineClient("https://example.com") + request = HttpRequest("GET", "https://example.com/stream") + response = client.send_request(request, stream=True) + + def deserialize(response: HttpResponse, event: MutableMapping[str, Any]) -> MutableMapping[str, Any]: + # Deserialize each decoded JSON object into a model. Here we just return it as-is. + return event + + stream = Stream( + response=response, + decoder=JSONLDecoder(), + deserialization_callback=deserialize, + ) + with stream: + for item in stream: + print(item) + # [END build_stream] + + +async def sample_stream_async(): + # [START build_stream_async] + from corehttp.runtime import AsyncPipelineClient + from corehttp.rest import HttpRequest, AsyncHttpResponse + from corehttp.streaming import AsyncStream, AsyncJSONLDecoder + + client: AsyncPipelineClient[HttpRequest, AsyncHttpResponse] = AsyncPipelineClient("https://example.com") + request = HttpRequest("GET", "https://example.com/stream") + response = await client.send_request(request, stream=True) + + def deserialize(response: AsyncHttpResponse, event: MutableMapping[str, Any]) -> MutableMapping[str, Any]: + # Deserialize each decoded JSON object into a model. Here we just return it as-is. + return event + + stream = AsyncStream( + response=response, + decoder=AsyncJSONLDecoder(), + deserialization_callback=deserialize, + ) + async with stream: + async for item in stream: + print(item) + # [END build_stream_async] + + +if __name__ == "__main__": + sample_stream() + asyncio.run(sample_stream_async()) diff --git a/sdk/core/corehttp/tests/async_tests/test_stream_async.py b/sdk/core/corehttp/tests/async_tests/test_stream_async.py new file mode 100644 index 000000000000..b6b43dc0bee8 --- /dev/null +++ b/sdk/core/corehttp/tests/async_tests/test_stream_async.py @@ -0,0 +1,202 @@ +# -------------------------------------------------------------------------- +# +# Copyright (c) Microsoft Corporation. All rights reserved. +# +# The MIT License (MIT) +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the ""Software""), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. +# +# -------------------------------------------------------------------------- + +import json + +import pytest + +from corehttp.rest import HttpRequest +from corehttp.streaming import AsyncStream, AsyncJSONLDecoder + + +@pytest.fixture +def deserialization_callback(): + def _callback(response, model_json): + return model_json + + return _callback + + +@pytest.fixture +def stream(client, deserialization_callback): + async def _callback(request, **kwargs): + http_response = await client.send_request(request=request, stream=True) + return AsyncStream( + deserialization_callback=deserialization_callback, response=http_response, decoder=AsyncJSONLDecoder() + ) + + return _callback + + +@pytest.mark.asyncio +async def test_stream_jsonl_basic(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_basic")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a message"}, + {"msg": "this is another message"}, + {"msg": "this is a third message"}, + {"msg": "this is a fourth message"}, + ] + + +@pytest.mark.asyncio +async def test_stream_jsonl_multiple_kv(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_multiple_kv")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a hello world message", "planet": {"earth": "hello earth", "mars": "hello mars"}}, + {"msg": "this is a hello world message", "planet": {"venus": "hello venus", "jupiter": "hello jupiter"}}, + ] + + +@pytest.mark.asyncio +async def test_stream_jsonl_no_final_line_separator(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_no_final_line_separator")) + async for s in jsonl_stream: + assert s == {"msg": "this is a message"} + + +@pytest.mark.asyncio +async def test_stream_jsonl_broken_up_data(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_broken_up_data")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "this is a second message"}, + ] + + +@pytest.mark.asyncio +async def test_stream_jsonl_broken_up_data_cr(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_broken_up_data_cr")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "this is a second message"}, + ] + + +@pytest.mark.asyncio +async def test_stream_jsonl_next(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_basic")) + message = await jsonl_stream.__anext__() + assert message == {"msg": "this is a message"} + message = await jsonl_stream.__anext__() + assert message == {"msg": "this is another message"} + message = await jsonl_stream.__anext__() + assert message == {"msg": "this is a third message"} + message = await jsonl_stream.__anext__() + assert message == {"msg": "this is a fourth message"} + + with pytest.raises(StopAsyncIteration): + await jsonl_stream.__anext__() + + +@pytest.mark.asyncio +async def test_stream_jsonl_context_manager(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_basic")) + async with jsonl_stream as streaming: + async for _ in streaming: + break + assert streaming._response.is_closed + + +@pytest.mark.asyncio +async def test_stream_jsonl_invalid_data(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_invalid_data")) + + with pytest.raises(json.decoder.JSONDecodeError): + async for _ in jsonl_stream: + ... + + +@pytest.mark.asyncio +async def test_stream_jsonl_escaped_newline_data(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_escaped_newline_data")) + + async for s in jsonl_stream: + assert s == {"msg": "this is a...\nmessage"} + + +@pytest.mark.asyncio +async def test_stream_jsonl_escaped_broken_newline_data(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_escaped_broken_newline_data")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "\nthis is a second message"}, + ] + + +@pytest.mark.asyncio +async def test_stream_jsonl_incomplete_char(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_broken_incomplete_char")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "𝜋this is a second message𝜋"}, + {"msg": "this is a third message"}, + ] + + +@pytest.mark.asyncio +async def test_stream_jsonl_list(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_list")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + ["this", "is", "a", "first", "message"], + ["this", "is", "a", "second", "message"], + ["this", "is", "a", "third", "message"], + ] + + +@pytest.mark.asyncio +async def test_stream_jsonl_string(stream): + jsonl_stream = await stream(HttpRequest("GET", "/streams/jsonl_string")) + messages = [] + async for s in jsonl_stream: + messages.append(s) + assert messages == [ + "this", + "is", + "a", + "message", + ] diff --git a/sdk/core/corehttp/tests/test_stream.py b/sdk/core/corehttp/tests/test_stream.py new file mode 100644 index 000000000000..96f827437ce1 --- /dev/null +++ b/sdk/core/corehttp/tests/test_stream.py @@ -0,0 +1,187 @@ +# -------------------------------------------------------------------------- +# +# Copyright (c) Microsoft Corporation. All rights reserved. +# +# The MIT License (MIT) +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the ""Software""), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. +# +# -------------------------------------------------------------------------- + +import json + +import pytest + +from corehttp.rest import HttpRequest +from corehttp.streaming import Stream, JSONLDecoder + + +@pytest.fixture +def deserialization_callback(): + def _callback(response, model_json): + return model_json + + return _callback + + +@pytest.fixture +def stream(client, deserialization_callback): + def _callback(request, **kwargs): + http_response = client.send_request(request=request, stream=True) + return Stream(deserialization_callback=deserialization_callback, response=http_response, decoder=JSONLDecoder()) + + return _callback + + +def test_stream_jsonl_basic(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_basic")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a message"}, + {"msg": "this is another message"}, + {"msg": "this is a third message"}, + {"msg": "this is a fourth message"}, + ] + + +def test_stream_jsonl_multiple_kv(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_multiple_kv")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a hello world message", "planet": {"earth": "hello earth", "mars": "hello mars"}}, + {"msg": "this is a hello world message", "planet": {"venus": "hello venus", "jupiter": "hello jupiter"}}, + ] + + +def test_stream_jsonl_no_final_line_separator(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_no_final_line_separator")) + for s in jsonl_stream: + assert s == {"msg": "this is a message"} + + +def test_stream_jsonl_broken_up_data(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_broken_up_data")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "this is a second message"}, + ] + + +def test_stream_jsonl_broken_up_data_cr(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_broken_up_data_cr")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "this is a second message"}, + ] + + +def test_stream_jsonl_next(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_basic")) + message = next(jsonl_stream) + assert message == {"msg": "this is a message"} + message = next(jsonl_stream) + assert message == {"msg": "this is another message"} + message = next(jsonl_stream) + assert message == {"msg": "this is a third message"} + message = next(jsonl_stream) + assert message == {"msg": "this is a fourth message"} + + with pytest.raises(StopIteration): + next(jsonl_stream) + + +def test_stream_jsonl_context_manager(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_basic")) + with jsonl_stream as streaming: + for _ in streaming: + break + assert streaming._response.is_closed + + +def test_stream_jsonl_invalid_data(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_invalid_data")) + + with pytest.raises(json.decoder.JSONDecodeError): + for _ in jsonl_stream: + ... + + +def test_stream_jsonl_escaped_newline_data(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_escaped_newline_data")) + + for s in jsonl_stream: + assert s == {"msg": "this is a...\nmessage"} + + +def test_stream_jsonl_escaped_broken_newline_data(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_escaped_broken_newline_data")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "\nthis is a second message"}, + ] + + +def test_stream_jsonl_incomplete_char(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_broken_incomplete_char")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + {"msg": "this is a first message"}, + {"msg": "𝜋this is a second message𝜋"}, + {"msg": "this is a third message"}, + ] + + +def test_stream_jsonl_list(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_list")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + ["this", "is", "a", "first", "message"], + ["this", "is", "a", "second", "message"], + ["this", "is", "a", "third", "message"], + ] + + +def test_stream_jsonl_string(stream): + jsonl_stream = stream(HttpRequest("GET", "/streams/jsonl_string")) + messages = [] + for s in jsonl_stream: + messages.append(s) + assert messages == [ + "this", + "is", + "a", + "message", + ] diff --git a/sdk/core/corehttp/tests/testserver_tests/coretestserver/coretestserver/test_routes/streams.py b/sdk/core/corehttp/tests/testserver_tests/coretestserver/coretestserver/test_routes/streams.py index fa6f26aa9da3..6f2b385438bb 100644 --- a/sdk/core/corehttp/tests/testserver_tests/coretestserver/coretestserver/test_routes/streams.py +++ b/sdk/core/corehttp/tests/testserver_tests/coretestserver/coretestserver/test_routes/streams.py @@ -39,6 +39,82 @@ def stream_compressed_header_error(): yield b"test" +def stream_jsonl_basic(): + data = [ + b'{"msg": "this is a message"}\n', + b'{"msg": "this is another message"}\n', + b'{"msg": "this is a third message"}\n{"msg": "this is a fourth message"}\n', + ] + yield from data + + +def stream_jsonl_multiple_kv(): + data = [ + b'{"msg": "this is a hello world message", "planet": {"earth": "hello earth", "mars": "hello mars"}}\n', + b'{"msg": "this is a hello world message", "planet": {"venus": "hello venus", "jupiter": "hello jupiter"}}\n', + ] + yield from data + + +def stream_jsonl_no_final_line_separator(): + data = b'{"msg": "this is a message"}' + yield data + + +def stream_jsonl_broken_up_data(): + data = [b'{"msg": "this is a first message"}\n{"msg": ', b'"this is a second message"}\n'] + yield from data + + +def stream_jsonl_broken_up_data_cr(): + data = [b'{"msg": "this is a first message"}\r\n{"msg": ', b'"this is a second message"}\r\n'] + yield from data + + +def stream_jsonl_invalid_data(): + data = [b'{"msg": "this is a third m'] + yield from data + + +def stream_jsonl_escaped_newline_data(): + data = b'{"msg": "this is a...\\nmessage"}\n' + yield data + + +def stream_jsonl_escaped_broken_newline_data(): + data = [b'{"msg": "this is a first message"}\n{"msg": "\\n', b'this is a second message"}\n'] + yield from data + + +def stream_jsonl_broken_incomplete_char(): + data = [ + b'{"msg": "this is a first message"}\n{"msg": "\xf0\x9d', + b"\x9c\x8bthis is a second message\xf0\x9d", + b'\x9c\x8b"}', + b'\n{"msg": "this is a third message"}', + ] + yield from data + + +def stream_jsonl_list(): + data = [ + b'["this", "is", "a", "first", "message"]\n', + b'["this", "is", "a", "second", "message"]\n', + b'["this", "is", "a", "third", "message"]\n', + ] + yield from data + + +def stream_jsonl_string(): + data = [ + b'"this"\n', + b'"is"\n', + b'"a"\n', + b'"message"\n', + ] + yield from data + + @streams_api.route("/basic", methods=["GET"]) def basic(): return Response(streaming_body(), status=200) @@ -104,3 +180,60 @@ def upload(): break byte_content += chunk return Response(byte_content, status=200) + + +@streams_api.route("/jsonl_basic", methods=["GET"]) +def jsonl_basic(): + return Response(stream_jsonl_basic(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_multiple_kv", methods=["GET"]) +def jsonl_multiple_kv(): + return Response(stream_jsonl_multiple_kv(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_no_final_line_separator", methods=["GET"]) +def jsonl_no_final_line_separator(): + return Response(stream_jsonl_no_final_line_separator(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_broken_up_data", methods=["GET"]) +def jsonl_broken_up_data(): + return Response(stream_jsonl_broken_up_data(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_broken_up_data_cr", methods=["GET"]) +def jsonl_broken_up_data_cr(): + return Response(stream_jsonl_broken_up_data_cr(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_invalid_data", methods=["GET"]) +def jsonl_invalid_data(): + return Response(stream_jsonl_invalid_data(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_escaped_newline_data", methods=["GET"]) +def jsonl_escaped_newline_data(): + return Response(stream_jsonl_escaped_newline_data(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_escaped_broken_newline_data", methods=["GET"]) +def jsonl_escaped_broken_newline_data(): + return Response( + stream_jsonl_escaped_broken_newline_data(), status=200, headers={"Content-Type": "application/jsonl"} + ) + + +@streams_api.route("/jsonl_broken_incomplete_char", methods=["GET"]) +def jsonl_broken_incomplete_char(): + return Response(stream_jsonl_broken_incomplete_char(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_list", methods=["GET"]) +def json_list(): + return Response(stream_jsonl_list(), status=200, headers={"Content-Type": "application/jsonl"}) + + +@streams_api.route("/jsonl_string", methods=["GET"]) +def json_string(): + return Response(stream_jsonl_string(), status=200, headers={"Content-Type": "application/jsonl"})