From 88a79dd3b3e0274c5af7827196366dd406b63eaa Mon Sep 17 00:00:00 2001 From: Libba Lawrence Date: Fri, 9 Oct 2026 10:27:14 -0700 Subject: [PATCH 1/2] feat(http-client-python): make SSE reconnection opt-in Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../changes/python-sse-reconnect-option.md | 17 ++ packages/http-client-python/README.md | 18 ++ .../http-client-python/emitter/src/emitter.ts | 1 + .../http-client-python/emitter/src/lib.ts | 8 + .../emitter/test/emitter.test.ts | 36 ++- .../emitter/test/test-host.ts | 17 +- .../eng/scripts/ci/regenerate-common.ts | 20 +- .../generator/pygen/__init__.py | 4 + .../pygen/codegen/models/response.py | 11 +- .../codegen/serializers/builder_serializer.py | 8 +- .../templates/streaming_base.py.jinja2 | 69 ++++- .../asynctests/test_streaming_sse_async.py | 17 ++ .../mock_api/shared/test_streaming_sse.py | 15 + .../tests/unit/test_options_dict.py | 24 ++ .../tests/unit/test_sse_reconnect_option.py | 276 ++++++++++++++++++ .../tests/unit/test_sse_streaming.py | 27 +- .../http-client-python/reference/emitter.md | 18 ++ 17 files changed, 554 insertions(+), 32 deletions(-) create mode 100644 .chronus/changes/python-sse-reconnect-option.md create mode 100644 packages/http-client-python/tests/unit/test_sse_reconnect_option.py diff --git a/.chronus/changes/python-sse-reconnect-option.md b/.chronus/changes/python-sse-reconnect-option.md new file mode 100644 index 00000000000..805bd54228a --- /dev/null +++ b/.chronus/changes/python-sse-reconnect-option.md @@ -0,0 +1,17 @@ +--- +changeKind: feature +packages: + - "@typespec/http-client-python" +--- + +Make automatic SSE reconnection opt-in with the `enable-sse-reconnect` emitter option, which defaults to `false`. Without this option, SSE streams end at EOF without reconnect callbacks, retry delays, or `Last-Event-ID` request plumbing. SSE parsing, event ID and retry metadata, terminal events, and stream cleanup remain available. + +Enable reconnection when generating the client: + +```yaml +emit: + - "@typespec/http-client-python" +options: + "@typespec/http-client-python": + enable-sse-reconnect: true +``` diff --git a/packages/http-client-python/README.md b/packages/http-client-python/README.md index 0c3335df5fa..058356da6ec 100644 --- a/packages/http-client-python/README.md +++ b/packages/http-client-python/README.md @@ -96,6 +96,24 @@ The name of the package to be used in pretty-printing. Will be the name of the p Whether to return responses from HEAD requests as boolean. Defaults to `true`. +### `enable-sse-reconnect` + +**Type:** `boolean` + +**Default:** `false` + +Whether to generate automatic reconnection support for Server-Sent Events (SSE). Defaults to `false`: SSE streams end when the response reaches EOF, without sending another request. Set to `true` to generate reconnect callbacks, retry delays, and `Last-Event-ID` resumption. SSE event IDs and retry metadata are parsed in either mode. Transport errors still propagate to the caller. + +Enable this generation-time option in `tspconfig.yaml`: + +```yaml +emit: + - "@typespec/http-client-python" +options: + "@typespec/http-client-python": + enable-sse-reconnect: true +``` + ### `use-pyodide` **Type:** `boolean` diff --git a/packages/http-client-python/emitter/src/emitter.ts b/packages/http-client-python/emitter/src/emitter.ts index c68517f99f4..9af976be426 100644 --- a/packages/http-client-python/emitter/src/emitter.ts +++ b/packages/http-client-python/emitter/src/emitter.ts @@ -33,6 +33,7 @@ function addDefaultOptions(sdkContext: PythonSdkContext) { ...sdkContext.emitContext.options, }; const options = sdkContext.emitContext.options; + options["enable-sse-reconnect"] ??= false; if (!options["package-name"]) { const namespace = getRootNamespace(sdkContext); const packageName = namespace.replace(/\./g, "-"); diff --git a/packages/http-client-python/emitter/src/lib.ts b/packages/http-client-python/emitter/src/lib.ts index 0ee8f1bf41f..c7974a621ad 100644 --- a/packages/http-client-python/emitter/src/lib.ts +++ b/packages/http-client-python/emitter/src/lib.ts @@ -20,6 +20,7 @@ export interface PythonEmitterOptions { "packaging-files-config"?: object; "package-pprint-name"?: string; "head-as-boolean"?: boolean; + "enable-sse-reconnect"?: boolean; "use-pyodide"?: boolean; "keep-setup-py"?: boolean; "generate-typeddict"?: boolean; @@ -86,6 +87,13 @@ export const PythonEmitterOptionsSchema: JSONSchemaType = nullable: true, description: "Whether to return responses from HEAD requests as boolean. Defaults to `true`.", }, + "enable-sse-reconnect": { + type: "boolean", + nullable: true, + default: false, + description: + 'Whether to generate automatic reconnection support for Server-Sent Events (SSE). Defaults to `false`: SSE streams end when the response reaches EOF, without sending another request. Set to `true` to generate reconnect callbacks, retry delays, and `Last-Event-ID` resumption. SSE event IDs and retry metadata are parsed in either mode. Transport errors still propagate to the caller.\n\nEnable this generation-time option in `tspconfig.yaml`:\n\n```yaml\nemit:\n - "@typespec/http-client-python"\noptions:\n "@typespec/http-client-python":\n enable-sse-reconnect: true\n```', + }, "use-pyodide": { type: "boolean", nullable: true, diff --git a/packages/http-client-python/emitter/test/emitter.test.ts b/packages/http-client-python/emitter/test/emitter.test.ts index ecb2081d5a1..fbbc7a37690 100644 --- a/packages/http-client-python/emitter/test/emitter.test.ts +++ b/packages/http-client-python/emitter/test/emitter.test.ts @@ -1,6 +1,40 @@ import { expectDiagnostics, t } from "@typespec/compiler/testing"; import { expect, it } from "vitest"; -import { emitCodeModel } from "./test-host.js"; +import { EmitterTester, emitCodeModel } from "./test-host.js"; + +it.each([undefined, false, true])( + "propagates enable-sse-reconnect=%s to the Python generator", + async (enabled) => { + const { commandArgs, diagnostics } = await emitCodeModel( + "@service namespace Test; op read(): string;", + enabled === undefined ? {} : { "enable-sse-reconnect": enabled }, + ); + + expectDiagnostics(diagnostics, []); + expect(commandArgs["enable-sse-reconnect"]).toBe(enabled ?? false); + }, +); + +it.each(["yes", "no", 2, {}, []])("rejects invalid enable-sse-reconnect=%s", async (value) => { + const [, diagnostics] = await EmitterTester.compileAndDiagnose( + "@service namespace Test; op read(): string;", + { + compilerOptions: { + options: { + "@typespec/http-client-python": { + "emit-yaml-only": true, + "enable-sse-reconnect": value, + }, + }, + }, + }, + ); + + expectDiagnostics(diagnostics, { + code: "invalid-schema", + message: /enable-sse-reconnect/, + }); +}); it("targets the service namespace when no SDK clients are found", async () => { const { diagnostics } = await emitCodeModel(t.code` diff --git a/packages/http-client-python/emitter/test/test-host.ts b/packages/http-client-python/emitter/test/test-host.ts index 07a762f578d..7e93659a4e1 100644 --- a/packages/http-client-python/emitter/test/test-host.ts +++ b/packages/http-client-python/emitter/test/test-host.ts @@ -6,6 +6,7 @@ import jsyaml from "js-yaml"; import { tmpdir } from "os"; import { join } from "path"; import { $onEmit } from "../src/emitter.js"; +import { $lib, type PythonEmitterOptions } from "../src/lib.js"; const PythonTester = createTester(resolvePath(import.meta.dirname, "../.."), { libraries: ["@azure-tools/typespec-client-generator-core"], @@ -20,10 +21,13 @@ export const EmitterTester = PythonTester.files({ "node_modules/@typespec/http-client-python/package.json": JSON.stringify({ name: "@typespec/http-client-python", version: "0.0.0", + tspMain: "./lib.tsp", exports: { ".": "./index.js" }, }), + "node_modules/@typespec/http-client-python/lib.tsp": 'import "./index.js";', "node_modules/@typespec/http-client-python/index.js": mockFile.js({ $onEmit, + $lib, }), }).emit("@typespec/http-client-python", { "generate-packaging-files": false, @@ -48,7 +52,12 @@ export interface CodeModel { */ export async function emitCodeModel( code: string, -): Promise<{ codeModel: CodeModel; diagnostics: readonly Diagnostic[] }> { + options: PythonEmitterOptions = {}, +): Promise<{ + codeModel: CodeModel; + commandArgs: Record; + diagnostics: readonly Diagnostic[]; +}> { const outputDir = await mkdtemp(join(tmpdir(), "typespec-python-")); let yamlPath: string | undefined; try { @@ -56,6 +65,7 @@ export async function emitCodeModel( compilerOptions: { options: { "@typespec/http-client-python": { + ...options, "emit-yaml-only": true, "emitter-output-dir": outputDir, }, @@ -71,9 +81,10 @@ export async function emitCodeModel( if (!pointerName) { throw new Error("Emitter did not produce a code model."); } - ({ yamlPath } = JSON.parse(await readFile(join(outputDir, pointerName), "utf-8"))); + const pointer = JSON.parse(await readFile(join(outputDir, pointerName), "utf-8")); + ({ yamlPath } = pointer); const codeModel = jsyaml.load(await readFile(yamlPath!, "utf-8")) as CodeModel; - return { codeModel, diagnostics }; + return { codeModel, commandArgs: pointer.commandArgs, diagnostics }; } finally { await rm(outputDir, { recursive: true, force: true }); if (yamlPath) { diff --git a/packages/http-client-python/eng/scripts/ci/regenerate-common.ts b/packages/http-client-python/eng/scripts/ci/regenerate-common.ts index e96933c5bf5..bc6833f4df4 100644 --- a/packages/http-client-python/eng/scripts/ci/regenerate-common.ts +++ b/packages/http-client-python/eng/scripts/ci/regenerate-common.ts @@ -80,10 +80,9 @@ export const SpecialFlags: Record> = { // ---- Spec-specific emitter option overrides ---- -export const AZURE_EMITTER_OPTIONS: Record< - string, - Record | Record[] -> = { +type EmitterOptions = Record; + +export const AZURE_EMITTER_OPTIONS: Record = { "azure/client-generator-core/access": { namespace: "specs.azure.clientgenerator.core.access", }, @@ -219,7 +218,16 @@ export const AZURE_EMITTER_OPTIONS: Record< }, }; -export const EMITTER_OPTIONS: Record | Record[]> = { +export const EMITTER_OPTIONS: Record = { + "streaming/sse": [ + { + "enable-sse-reconnect": true, + }, + { + "package-name": "streaming-sse-no-reconnect", + namespace: "streaming.ssenoreconnect", + }, + ], "resiliency/srv-driven/old.tsp": { "package-name": "resiliency-srv-driven1", namespace: "resiliency.srv.driven1", @@ -416,7 +424,7 @@ export function getEmitterOptions( spec: string, flavor: string, ctx: RegenerateContext, -): Record[] { +): EmitterOptions[] { const specDir = isAzureSpec(spec) ? ctx.azureHttpSpecs : ctx.httpSpecs; const relativeSpec = toPosix(relative(specDir, spec)); const key = relativeSpec.includes("resiliency/srv-driven/old.tsp") diff --git a/packages/http-client-python/generator/pygen/__init__.py b/packages/http-client-python/generator/pygen/__init__.py index 5b77d715d8d..5d5d2cebb1d 100644 --- a/packages/http-client-python/generator/pygen/__init__.py +++ b/packages/http-client-python/generator/pygen/__init__.py @@ -32,6 +32,7 @@ class OptionsDict(MutableMapping): "keep-pyproject-fields": "", "generate-test": False, "head-as-boolean": True, + "enable-sse-reconnect": False, "keep-version-file": False, "low-level-client": False, "no-async": False, @@ -175,6 +176,9 @@ def _validate_combinations(self) -> None: raise ValueError("Can only have tracing turned on for Azure SDKs.") def _validate_and_transform(self, key: str, value: Any) -> Any: + if key == "enable-sse-reconnect" and not isinstance(value, bool): + raise ValueError("--enable-sse-reconnect must be a boolean") + if key == "builders-visibility" and value not in ["public", "hidden", "embedded"]: raise ValueError("The value of --builders-visibility must be either 'public', 'hidden', or 'embedded'") diff --git a/packages/http-client-python/generator/pygen/codegen/models/response.py b/packages/http-client-python/generator/pygen/codegen/models/response.py index 4a23fd21412..ba6014eb69d 100644 --- a/packages/http-client-python/generator/pygen/codegen/models/response.py +++ b/packages/http-client-python/generator/pygen/codegen/models/response.py @@ -267,11 +267,12 @@ def imports(self, **kwargs: Any) -> FileImport: file_import.add_submodule_import(relative_path, stream_class, ImportType.LOCAL) if self.streaming_kind == "sse": file_import.add_import("json", ImportType.STDLIB) - read_response_helper = ( - "_read_sse_response_async" if kwargs.get("async_mode", False) else "_read_sse_response" - ) - file_import.add_submodule_import(relative_path, read_response_helper, ImportType.LOCAL) - file_import.add_submodule_import(relative_path, "_update_sse_request_headers", ImportType.LOCAL) + if self.code_model.options["enable-sse-reconnect"]: + read_response_helper = ( + "_read_sse_response_async" if kwargs.get("async_mode", False) else "_read_sse_response" + ) + file_import.add_submodule_import(relative_path, read_response_helper, ImportType.LOCAL) + file_import.add_submodule_import(relative_path, "_update_sse_request_headers", ImportType.LOCAL) for event in self.streaming_events: file_import.merge(event.payload_type.imports(**kwargs)) return file_import diff --git a/packages/http-client-python/generator/pygen/codegen/serializers/builder_serializer.py b/packages/http-client-python/generator/pygen/codegen/serializers/builder_serializer.py index 3bb0007d263..8d26d6bc358 100644 --- a/packages/http-client-python/generator/pygen/codegen/serializers/builder_serializer.py +++ b/packages/http-client-python/generator/pygen/codegen/serializers/builder_serializer.py @@ -1043,8 +1043,10 @@ def _call_request_builder_helper( def call_request_builder(self, builder: OperationType, is_paging: bool = False) -> list[str]: retval = self._call_request_builder_helper(builder, builder.request_builder, is_paging=is_paging) - if builder.has_structured_stream_response and any( - response.streaming_kind == "sse" for response in builder.responses + if ( + self.code_model.options["enable-sse-reconnect"] + and builder.has_structured_stream_response + and any(response.streaming_kind == "sse" for response in builder.responses) ): retval.insert(0, '_last_event_id = kwargs.pop("last_event_id", None)') retval.extend( @@ -1396,7 +1398,7 @@ def emit_message_branch(keyword: str) -> None: stream_kwargs.append(f"terminal_event={terminal_event!r}") if terminal_event_names: stream_kwargs.append(f"terminal_event_names={terminal_event_names!r}") - if response.streaming_kind == "sse": # type: ignore[attr-defined] + if response.streaming_kind == "sse" and self.code_model.options["enable-sse-reconnect"]: retval.append("") retval.append( "async def _reconnect(_last_event_id, _reconnect_delay):" diff --git a/packages/http-client-python/generator/pygen/codegen/templates/streaming_base.py.jinja2 b/packages/http-client-python/generator/pygen/codegen/templates/streaming_base.py.jinja2 index 8ba3bc2fbbd..67f2abaf832 100644 --- a/packages/http-client-python/generator/pygen/codegen/templates/streaming_base.py.jinja2 +++ b/packages/http-client-python/generator/pygen/codegen/templates/streaming_base.py.jinja2 @@ -1,4 +1,5 @@ # coding=utf-8 +{% set enable_sse_reconnect = code_model.options["enable-sse-reconnect"] %} {% if code_model.license_header %} {{ code_model.license_header }} {% endif %} @@ -33,8 +34,11 @@ from typing import ( from typing_extensions import Self +{% if enable_sse_reconnect %} from {{ code_model.core_library }}.exceptions import StreamClosedError, StreamConsumedError -from {{ code_model.core_library }}.rest import AsyncHttpResponse, HttpRequest, HttpResponse +from {{ code_model.core_library }}.rest import HttpRequest +{% endif %} +from {{ code_model.core_library }}.rest import AsyncHttpResponse, HttpResponse DecodedType = TypeVar("DecodedType") ReturnType_co = TypeVar("ReturnType_co", covariant=True) @@ -436,6 +440,7 @@ async def _aiter_sse_lines(iter_bytes: AsyncIterator[bytes]) -> AsyncGenerator[s yield line +{% if enable_sse_reconnect %} def _read_sse_response(response: HttpResponse) -> None: """Read an SSE response body, ignoring responses that are already consumed or closed. @@ -474,14 +479,22 @@ def _update_sse_request_headers(request: HttpRequest, last_event_id: Optional[st request.headers.pop("Last-Event-ID", None) +{% endif %} class _SSEEventBuilder: """Accumulates SSE field lines and builds :class:`ServerSentEvent` instances. +{% if enable_sse_reconnect %} :param last_event_id: The initial value of the SSE last event ID buffer. :type last_event_id: str +{% endif %} """ +{% if enable_sse_reconnect %} def __init__(self, last_event_id: str = "") -> None: +{% else %} + def __init__(self) -> None: + last_event_id = "" +{% endif %} self._data: List[str] = [] self._event_type = "" self._last_id = last_event_id @@ -565,12 +578,19 @@ class SSEDecoder: https://html.spec.whatwg.org/multipage/server-sent-events.html +{% if enable_sse_reconnect %} :keyword last_event_id: The initial value of the SSE last event ID buffer. :paramtype last_event_id: str or None +{% endif %} """ +{% if enable_sse_reconnect %} def __init__(self, *, last_event_id: Optional[str] = None) -> None: self._builder = _SSEEventBuilder(last_event_id or "") +{% else %} + def __init__(self) -> None: + self._builder = _SSEEventBuilder() +{% endif %} @property def last_event_id(self) -> str: @@ -609,12 +629,19 @@ class AsyncSSEDecoder: https://html.spec.whatwg.org/multipage/server-sent-events.html +{% if enable_sse_reconnect %} :keyword last_event_id: The initial value of the SSE last event ID buffer. :paramtype last_event_id: str or None +{% endif %} """ +{% if enable_sse_reconnect %} def __init__(self, *, last_event_id: Optional[str] = None) -> None: self._builder = _SSEEventBuilder(last_event_id or "") +{% else %} + def __init__(self) -> None: + self._builder = _SSEEventBuilder() +{% endif %} @property def last_event_id(self) -> str: @@ -672,6 +699,7 @@ class Stream(Iterator[ReturnType_co]): :keyword terminal_event_predicate: Optional predicate that identifies a payload-bearing terminal event. The event is yielded before iteration stops. :paramtype terminal_event_predicate: Callable[[Any], bool] or None +{% if enable_sse_reconnect %} :keyword last_event_id: Optional event ID from which this stream is resuming. :paramtype last_event_id: str or None :keyword reconnect_callback: Optional callback used to re-establish the underlying response @@ -680,6 +708,7 @@ class Stream(Iterator[ReturnType_co]): called with the last-seen SSE event ID (or ``None``) and the reconnect delay in seconds, and must return a new response. :paramtype reconnect_callback: Callable[[str or None, float], ~{{ code_model.core_library }}.rest.HttpResponse] or None +{% endif %} """ def __init__( @@ -691,8 +720,10 @@ class Stream(Iterator[ReturnType_co]): terminal_event: Optional[str] = None, terminal_event_names: Optional[Sequence[str]] = None, terminal_event_predicate: Optional[Callable[[DecodedType], bool]] = None, +{% if enable_sse_reconnect %} last_event_id: Optional[str] = None, reconnect_callback: Optional[Callable[[Optional[str], float], HttpResponse]] = None, +{% endif %} ) -> None: self._response = response content_type = response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower() @@ -700,7 +731,7 @@ class Stream(Iterator[ReturnType_co]): decoder if decoder is not None else ( - SSEDecoder(last_event_id=last_event_id) + SSEDecoder({% if enable_sse_reconnect %}last_event_id=last_event_id{% endif %}) if content_type == "text/event-stream" else JSONLDecoder() ) @@ -709,8 +740,12 @@ class Stream(Iterator[ReturnType_co]): self._terminal_event = terminal_event self._terminal_event_names = frozenset(terminal_event_names or ()) self._terminal_event_predicate = terminal_event_predicate +{% if enable_sse_reconnect %} self._reconnect_callback = reconnect_callback self._last_event_id = last_event_id +{% else %} + self._last_event_id: Optional[str] = None +{% endif %} self._retry: Optional[int] = None self._iterator = self._iter_results() @@ -731,7 +766,9 @@ class Stream(Iterator[ReturnType_co]): return self def _iter_results(self) -> Generator[ReturnType_co, None, None]: +{% if enable_sse_reconnect %} terminal = False +{% endif %} try: while True: if getattr(self._response, "status_code", None) == 204: @@ -744,7 +781,9 @@ class Stream(Iterator[ReturnType_co]): if event_retry is not None: self._retry = event_retry if self._terminal_event is not None and getattr(event, "data", None) == self._terminal_event: +{% if enable_sse_reconnect %} terminal = True +{% endif %} break result = self._deserialization_callback(self._response, cast(Any, event)) yield result @@ -755,8 +794,11 @@ class Stream(Iterator[ReturnType_co]): self._terminal_event_predicate is not None and self._terminal_event_predicate(cast(Any, event)) ): +{% if enable_sse_reconnect %} terminal = True +{% endif %} break +{% if enable_sse_reconnect %} if terminal or self._reconnect_callback is None: break if isinstance(self._decoder, SSEDecoder): @@ -771,6 +813,9 @@ class Stream(Iterator[ReturnType_co]): if isinstance(self._decoder, SSEDecoder) else type(self._decoder)() ) +{% else %} + break +{% endif %} finally: self._response.close() @@ -814,6 +859,7 @@ class AsyncStream(AsyncIterator[ReturnType_co]): :keyword terminal_event_predicate: Optional predicate that identifies a payload-bearing terminal event. The event is yielded before iteration stops. :paramtype terminal_event_predicate: Callable[[Any], bool] or None +{% if enable_sse_reconnect %} :keyword last_event_id: Optional event ID from which this stream is resuming. :paramtype last_event_id: str or None :keyword reconnect_callback: Optional callback used to re-establish the underlying response @@ -822,6 +868,7 @@ class AsyncStream(AsyncIterator[ReturnType_co]): called with the last-seen SSE event ID (or ``None``) and the reconnect delay in seconds, and must resolve to a new response. :paramtype reconnect_callback: Callable[[str or None, float], Any] or None +{% endif %} """ def __init__( @@ -833,8 +880,10 @@ class AsyncStream(AsyncIterator[ReturnType_co]): terminal_event: Optional[str] = None, terminal_event_names: Optional[Sequence[str]] = None, terminal_event_predicate: Optional[Callable[[DecodedType], bool]] = None, +{% if enable_sse_reconnect %} last_event_id: Optional[str] = None, reconnect_callback: Optional[Callable[[Optional[str], float], Any]] = None, +{% endif %} ) -> None: self._response = response content_type = response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower() @@ -842,7 +891,7 @@ class AsyncStream(AsyncIterator[ReturnType_co]): decoder if decoder is not None else ( - AsyncSSEDecoder(last_event_id=last_event_id) + AsyncSSEDecoder({% if enable_sse_reconnect %}last_event_id=last_event_id{% endif %}) if content_type == "text/event-stream" else AsyncJSONLDecoder() ) @@ -851,8 +900,12 @@ class AsyncStream(AsyncIterator[ReturnType_co]): self._terminal_event = terminal_event self._terminal_event_names = frozenset(terminal_event_names or ()) self._terminal_event_predicate = terminal_event_predicate +{% if enable_sse_reconnect %} self._reconnect_callback = reconnect_callback self._last_event_id = last_event_id +{% else %} + self._last_event_id: Optional[str] = None +{% endif %} self._retry: Optional[int] = None self._iterator = self._iter_results() @@ -873,7 +926,9 @@ class AsyncStream(AsyncIterator[ReturnType_co]): return self async def _iter_results(self) -> AsyncGenerator[ReturnType_co, None]: +{% if enable_sse_reconnect %} terminal = False +{% endif %} try: while True: if getattr(self._response, "status_code", None) == 204: @@ -888,7 +943,9 @@ class AsyncStream(AsyncIterator[ReturnType_co]): if event_retry is not None: self._retry = event_retry if self._terminal_event is not None and getattr(event, "data", None) == self._terminal_event: +{% if enable_sse_reconnect %} terminal = True +{% endif %} break result = self._deserialization_callback(self._response, cast(Any, event)) yield result @@ -899,12 +956,15 @@ class AsyncStream(AsyncIterator[ReturnType_co]): self._terminal_event_predicate is not None and self._terminal_event_predicate(cast(Any, event)) ): +{% if enable_sse_reconnect %} terminal = True +{% endif %} break finally: aclose = getattr(events, "aclose", None) if aclose is not None: await aclose() +{% if enable_sse_reconnect %} if terminal or self._reconnect_callback is None: break if isinstance(self._decoder, AsyncSSEDecoder): @@ -919,6 +979,9 @@ class AsyncStream(AsyncIterator[ReturnType_co]): if isinstance(self._decoder, AsyncSSEDecoder) else type(self._decoder)() ) +{% else %} + break +{% endif %} finally: await self._response.close() diff --git a/packages/http-client-python/tests/mock_api/shared/asynctests/test_streaming_sse_async.py b/packages/http-client-python/tests/mock_api/shared/asynctests/test_streaming_sse_async.py index 546cdaa552a..6ef0fd806a4 100644 --- a/packages/http-client-python/tests/mock_api/shared/asynctests/test_streaming_sse_async.py +++ b/packages/http-client-python/tests/mock_api/shared/asynctests/test_streaming_sse_async.py @@ -15,6 +15,7 @@ from streaming.sse.protocol.models import ProtocolInfo from streaming.sse.retrieve.models import FinalResult, PartialResult, RetrievalRequest from streaming.sse.unnamed.models import Info +from streaming.ssenoreconnect.aio import SseClient as NoReconnectSseClient @pytest_asyncio.fixture @@ -118,6 +119,22 @@ async def test_protocol_reconnect(client: SseClient): assert stream.last_event_id == "event-2" +@pytest.mark.asyncio +async def test_default_sse_does_not_reconnect(): + async with NoReconnectSseClient(endpoint="http://localhost:3000") as client: + async with await client.protocol.reconnect() as stream: + assert [item.message async for item in stream] == ["hello"] + assert stream.last_event_id == "event-1" + + +@pytest.mark.asyncio +async def test_default_sse_parses_retry_metadata(): + async with NoReconnectSseClient(endpoint="http://localhost:3000") as client: + async with await client.protocol.retry() as stream: + assert [item.message async for item in stream] == ["hello"] + assert stream.retry == 1000 + + # --------------------------------------------------------------------------- # Named / model terminal events (yield-then-stop) -- see the sync test module # for the rationale. Driven through the generated ``AsyncStream`` with a fake diff --git a/packages/http-client-python/tests/mock_api/shared/test_streaming_sse.py b/packages/http-client-python/tests/mock_api/shared/test_streaming_sse.py index 1134252d930..5dafe81fd85 100644 --- a/packages/http-client-python/tests/mock_api/shared/test_streaming_sse.py +++ b/packages/http-client-python/tests/mock_api/shared/test_streaming_sse.py @@ -14,6 +14,7 @@ from streaming.sse.protocol.models import ProtocolInfo from streaming.sse.retrieve.models import FinalResult, PartialResult, RetrievalRequest from streaming.sse.unnamed.models import Info +from streaming.ssenoreconnect import SseClient as NoReconnectSseClient @pytest.fixture @@ -107,6 +108,20 @@ def test_protocol_reconnect(client: SseClient): assert stream.last_event_id == "event-2" +def test_default_sse_does_not_reconnect(): + with NoReconnectSseClient(endpoint="http://localhost:3000") as client: + with client.protocol.reconnect() as stream: + assert [item.message for item in stream] == ["hello"] + assert stream.last_event_id == "event-1" + + +def test_default_sse_parses_retry_metadata(): + with NoReconnectSseClient(endpoint="http://localhost:3000") as client: + with client.protocol.retry() as stream: + assert [item.message for item in stream] == ["hello"] + assert stream.retry == 1000 + + # --------------------------------------------------------------------------- # Named / model terminal events (yield-then-stop). # diff --git a/packages/http-client-python/tests/unit/test_options_dict.py b/packages/http-client-python/tests/unit/test_options_dict.py index cd7a2451dbb..5822aa6ae88 100644 --- a/packages/http-client-python/tests/unit/test_options_dict.py +++ b/packages/http-client-python/tests/unit/test_options_dict.py @@ -7,6 +7,7 @@ import pytest from pygen import OptionsDict +from pygen.utils import parse_args def test_models_mode_none_normalized_via_constructor(): @@ -37,6 +38,29 @@ def test_generate_typeddict_can_be_disabled(): assert OptionsDict({"generate-typeddict": False})["generate-typeddict"] is False +@pytest.mark.parametrize("options", [{}, {"enable-sse-reconnect": False}, {"enable-sse-reconnect": True}]) +def test_enable_sse_reconnect_is_opt_in(options): + assert OptionsDict(options)["enable-sse-reconnect"] is options.get("enable-sse-reconnect", False) + + +@pytest.mark.parametrize("value", ["true", "false", 0, 1, None]) +def test_enable_sse_reconnect_requires_boolean(value): + with pytest.raises(ValueError, match="enable-sse-reconnect.*boolean"): + OptionsDict({"enable-sse-reconnect": value}) + with pytest.raises(ValueError, match="enable-sse-reconnect.*boolean"): + OptionsDict()["enable-sse-reconnect"] = value + + +@pytest.mark.parametrize("value, expected", [("false", False), ("true", True)]) +def test_enable_sse_reconnect_cli_propagation(monkeypatch, value, expected): + monkeypatch.setattr( + "sys.argv", + ["pygen", "--output-folder=out", "--tsp-file=main.yaml", f"--enable-sse-reconnect={value}"], + ) + _, options = parse_args() + assert OptionsDict(options)["enable-sse-reconnect"] is expected + + def test_models_mode_none_with_tsp_generates_typeddict_by_default(): # For TypeSpec input, models-mode=none keeps TypedDict generation on by # default. Internally models-mode stays 'none' (falsy); TypedDict-only diff --git a/packages/http-client-python/tests/unit/test_sse_reconnect_option.py b/packages/http-client-python/tests/unit/test_sse_reconnect_option.py new file mode 100644 index 00000000000..0cbc7e4bcd6 --- /dev/null +++ b/packages/http-client-python/tests/unit/test_sse_reconnect_option.py @@ -0,0 +1,276 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +import asyncio +import inspect +from types import ModuleType, SimpleNamespace + +import pytest +from jinja2 import Environment, PackageLoader + +from pygen import OptionsDict +from pygen.codegen.models import CodeModel +from pygen.codegen.models.response import Response +from pygen.codegen.serializers.builder_serializer import OperationSerializer +from pygen.codegen.serializers.general_serializer import GeneralSerializer + + +@pytest.fixture(params=["azure", "unbranded"]) +def flavor(request): + return request.param + + +@pytest.fixture(params=[{}, {"enable-sse-reconnect": False}, {"enable-sse-reconnect": True}]) +def reconnect_options(request): + return request.param + + +@pytest.fixture +def code_model(flavor, reconnect_options): + return CodeModel( + { + "namespace": "test", + "clients": [{"name": "Client", "namespace": "test", "parameters": [], "url": "", "operationGroups": []}], + }, + options=OptionsDict({"flavor": flavor, "models-mode": "dpg", **reconnect_options}), + ) + + +@pytest.fixture +def runtime(code_model): + env = Environment(loader=PackageLoader("pygen.codegen", "templates"), trim_blocks=True, lstrip_blocks=True) + source = GeneralSerializer(code_model=code_model, env=env).serialize_streaming_base_file() + module = ModuleType("generated_streaming") + exec(compile(source, "", "exec"), module.__dict__) + return SimpleNamespace(module=module, source=source) + + +class FakeResponse: + def __init__(self, body, *, content_type="text/event-stream", status_code=200, error=None): + self.headers = {"Content-Type": content_type} + self.body = body + self.status_code = status_code + self.error = error + self.closed = False + self.byte_iterator_closed = False + + def iter_bytes(self): + try: + for index in range(0, len(self.body), 7): + yield self.body[index : index + 7] + if self.error: + raise self.error + finally: + self.byte_iterator_closed = True + + def close(self): + self.closed = True + + +class AsyncFakeResponse(FakeResponse): + async def iter_bytes(self): + try: + for chunk in super().iter_bytes(): + yield chunk + finally: + self.byte_iterator_closed = True + + async def close(self): + self.closed = True + + +def _consume(runtime, body, *, async_mode=False, stream_kwargs=None, response_kwargs=None): + async def consume_async(): + response = AsyncFakeResponse(body, **(response_kwargs or {})) + stream = runtime.module.AsyncStream( + response=response, + deserialization_callback=lambda _response, event: event, + **(stream_kwargs or {}), + ) + return [event async for event in stream], stream, response + + if async_mode: + return asyncio.run(consume_async()) + response = FakeResponse(body, **(response_kwargs or {})) + stream = runtime.module.Stream( + response=response, + deserialization_callback=lambda _response, event: event, + **(stream_kwargs or {}), + ) + return list(stream), stream, response + + +def test_runtime_reconnect_support_is_generated_only_when_enabled(runtime, reconnect_options): + enabled = reconnect_options.get("enable-sse-reconnect", False) + for name in ["Stream", "AsyncStream"]: + parameters = inspect.signature(getattr(runtime.module, name)).parameters + assert ("reconnect_callback" in parameters) is enabled + assert ("last_event_id" in parameters) is enabled + for name in ["SSEDecoder", "AsyncSSEDecoder"]: + assert ("last_event_id" in inspect.signature(getattr(runtime.module, name)).parameters) is enabled + for symbol in ["_read_sse_response", "_read_sse_response_async", "_update_sse_request_headers", "HttpRequest"]: + assert hasattr(runtime.module, symbol) is enabled + if not enabled: + assert "reconnect_callback" not in runtime.source + assert "reconnect_delay" not in runtime.source + assert "Last-Event-ID" not in runtime.source + + +@pytest.mark.parametrize("async_mode", [False, True]) +def test_normal_sse_eof_keeps_metadata_and_closes(runtime, async_mode): + events, stream, response = _consume( + runtime, b"id: first\nretry: 1000\ndata: one\n\ndata: two\n\n", async_mode=async_mode + ) + assert [event.data for event in events] == ["one", "two"] + assert [event.id for event in events] == ["first", "first"] + assert [event.retry for event in events] == [1000, 1000] + assert stream.last_event_id == "first" + assert stream.retry == 1000 + assert response.closed + assert response.byte_iterator_closed + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize( + ("body", "kwargs", "expected"), + [ + (b"data: one\n\ndata: [DONE]\n\ndata: after\n\n", {"terminal_event": "[DONE]"}, ["one"]), + ( + b"event: complete\ndata: done\n\ndata: after\n\n", + {"terminal_event_names": ["complete"]}, + ["done"], + ), + ( + b"data: done\n\ndata: after\n\n", + {"terminal_event_predicate": lambda event: event.data == "done"}, + ["done"], + ), + ], +) +def test_terminal_events_are_preserved(runtime, async_mode, body, kwargs, expected): + events, _, response = _consume(runtime, body, async_mode=async_mode, stream_kwargs=kwargs) + assert [event.data for event in events] == expected + assert response.closed + assert response.byte_iterator_closed + + +@pytest.mark.parametrize("async_mode", [False, True]) +def test_transport_error_propagates_without_reconnect(runtime, async_mode): + response = (AsyncFakeResponse if async_mode else FakeResponse)(b"data: one\n\n", error=OSError("disconnected")) + + async def consume_async(): + stream = runtime.module.AsyncStream(response=response, deserialization_callback=lambda _response, event: event) + return [event async for event in stream] + + with pytest.raises(OSError, match="disconnected"): + if async_mode: + asyncio.run(consume_async()) + else: + list(runtime.module.Stream(response=response, deserialization_callback=lambda _response, event: event)) + assert response.closed + assert response.byte_iterator_closed + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize("started", [False, True]) +def test_explicit_close_releases_response(runtime, async_mode, started): + response = (AsyncFakeResponse if async_mode else FakeResponse)(b"data: one\n\ndata: two\n\n") + + async def consume_async(): + stream = runtime.module.AsyncStream(response=response, deserialization_callback=lambda _response, event: event) + if started: + await stream.__anext__() + await stream.close() + + if async_mode: + asyncio.run(consume_async()) + else: + stream = runtime.module.Stream(response=response, deserialization_callback=lambda _response, event: event) + if started: + next(stream) + stream.close() + assert response.closed + if started: + assert response.byte_iterator_closed + + +@pytest.mark.parametrize("async_mode", [False, True]) +def test_jsonl_is_unchanged(runtime, async_mode): + events, _, response = _consume( + runtime, + b'{"message": "one"}\n{"message": "two"}\n', + async_mode=async_mode, + response_kwargs={"content_type": "application/jsonl"}, + ) + assert [event.json() for event in events] == [{"message": "one"}, {"message": "two"}] + assert response.closed + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize("kind", ["sse", "jsonl", None]) +def test_initial_request_resumption_is_gated(code_model, reconnect_options, async_mode, kind, monkeypatch): + serializer = OperationSerializer(code_model, async_mode=async_mode, client_namespace="test") + monkeypatch.setattr( + serializer, "_call_request_builder_helper", lambda *args, **kwargs: ["_request = build_request()"] + ) + builder = SimpleNamespace( + request_builder=object(), + has_structured_stream_response=kind is not None, + responses=[SimpleNamespace(streaming_kind=kind)], + ) + generated = "\n".join(serializer.call_request_builder(builder)) + enabled = reconnect_options.get("enable-sse-reconnect", False) and kind == "sse" + assert ('kwargs.pop("last_event_id", None)' in generated) is enabled + assert ("Last-Event-ID" in generated) is enabled + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize("kind", ["sse", "jsonl"]) +def test_reconnect_only_operation_imports_are_gated(code_model, reconnect_options, async_mode, kind): + response = Response({"statusCodes": [200], "streaming": {"kind": kind}}, code_model) + imports = response.imports(async_mode=async_mode).imports + names = {item.submodule_name for item in imports} + enabled = reconnect_options.get("enable-sse-reconnect", False) and kind == "sse" + assert ("_update_sse_request_headers" in names) is enabled + assert (("_read_sse_response_async" if async_mode else "_read_sse_response") in names) is enabled + assert ("AsyncStream" if async_mode else "Stream") in names + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize("metadata, expected", [(b"", ("", 3.0)), (b"id: first\nretry: 1000\n", ("first", 1.0))]) +@pytest.mark.parametrize("reconnect_options", [{"enable-sse-reconnect": True}]) +def test_enabled_reconnect_resumes_and_stops_at_204(runtime, reconnect_options, async_mode, metadata, expected): + calls = [] + first = (AsyncFakeResponse if async_mode else FakeResponse)(metadata + b"data: one\n\n") + second = (AsyncFakeResponse if async_mode else FakeResponse)(b"", status_code=204) + + def reconnect(last_event_id, delay): + calls.append((last_event_id, delay)) + return second + + async def reconnect_async(last_event_id, delay): + return reconnect(last_event_id, delay) + + async def consume_async(): + stream = runtime.module.AsyncStream( + response=first, + deserialization_callback=lambda _response, event: event.data, + reconnect_callback=reconnect_async, + ) + return [event async for event in stream] + + if async_mode: + items = asyncio.run(consume_async()) + else: + items = list( + runtime.module.Stream( + response=first, + deserialization_callback=lambda _response, event: event.data, + reconnect_callback=reconnect, + ) + ) + assert items == ["one"] + assert calls == [expected] + assert first.closed and second.closed diff --git a/packages/http-client-python/tests/unit/test_sse_streaming.py b/packages/http-client-python/tests/unit/test_sse_streaming.py index 7683e664979..f68216e7fb8 100644 --- a/packages/http-client-python/tests/unit/test_sse_streaming.py +++ b/packages/http-client-python/tests/unit/test_sse_streaming.py @@ -7,6 +7,7 @@ import pytest +from pygen import OptionsDict from pygen.codegen.models.response import ( StreamingEvent, _get_terminal_event_names, @@ -84,7 +85,8 @@ def test_inconsistent_unnamed_discriminators_are_ambiguous(): ("async_mode", "stream_class"), [(False, "Stream"), (True, "AsyncStream")], ) -def test_generated_unnamed_discriminator_dispatch_and_terminal_predicate(async_mode, stream_class): +@pytest.mark.parametrize("reconnect_options", [{}, {"enable-sse-reconnect": False}, {"enable-sse-reconnect": True}]) +def test_generated_unnamed_discriminator_dispatch_and_terminal_predicate(async_mode, stream_class, reconnect_options): connected = _streaming_event(payload_type=_discriminated_payload("connected", "_models.Connected")) disconnected = _streaming_event( payload_type=_discriminated_payload("disconnected", "_models.Disconnected"), @@ -101,7 +103,7 @@ def test_generated_unnamed_discriminator_dispatch_and_terminal_predicate(async_m ) code_model = SimpleNamespace( is_azure_flavor=False, - options={"models-mode": "dpg", "azure-arm": False}, + options=OptionsDict({"models-mode": "dpg", **reconnect_options}), generate_typeddict_only=False, get_serialize_namespace=lambda *args, **kwargs: "test", ) @@ -131,6 +133,15 @@ def test_generated_unnamed_discriminator_dispatch_and_terminal_predicate(async_m assert "_event_json.get('kind') in ['disconnected']" in generated assert f"deserialized: {stream_class}[" in generated assert "terminal_event_predicate=_is_terminal_event" in generated + assert generated.count("return cls(pipeline_response, deserialized, {})") == 1 + assert 'raise ValueError(f"Unknown SSE event type: {_event.event!r}")' in generated + assert not any(line.strip().startswith("_event.event =") for line in generated.splitlines()) + if not reconnect_options.get("enable-sse-reconnect"): + assert "_reconnect" not in generated + assert "_last_event_id" not in generated + assert "_transport" not in generated + assert "_read_sse_response" not in generated + return assert ( "async def _reconnect(_last_event_id, _reconnect_delay):" if async_mode @@ -143,8 +154,7 @@ def test_generated_unnamed_discriminator_dispatch_and_terminal_predicate(async_m "self._client.send_request(_request, stream=True, **kwargs)" ) in generated assert ( - f"{'await ' if async_mode else ''}_read_sse_response" - f"{'_async' if async_mode else ''}(_reconnect_response)" + f"{'await ' if async_mode else ''}_read_sse_response" f"{'_async' if async_mode else ''}(_reconnect_response)" ) in generated assert "_update_sse_request_headers(_request, _last_event_id)" in generated assert "if _reconnect_response.status_code not in [200, 204]:" in generated @@ -154,9 +164,6 @@ def test_generated_unnamed_discriminator_dispatch_and_terminal_predicate(async_m assert "raise HttpResponseError(response=_reconnect_response, model=error)" in generated assert "last_event_id=_last_event_id" in generated assert "reconnect_callback=_reconnect" in generated - assert generated.count("return cls(pipeline_response, deserialized, {})") == 1 - assert 'raise ValueError(f"Unknown SSE event type: {_event.event!r}")' in generated - assert not any(line.strip().startswith("_event.event =") for line in generated.splitlines()) def test_named_terminal_event_uses_explicit_event_name(): @@ -180,7 +187,7 @@ def test_reconnect_error_uses_reconnect_pipeline_response_for_legacy_models(): ) code_model = SimpleNamespace( is_azure_flavor=False, - options={"models-mode": "msrest", "azure-arm": False}, + options=OptionsDict({"models-mode": "msrest", "enable-sse-reconnect": True}), generate_typeddict_only=False, get_serialize_namespace=lambda *args, **kwargs: "test", ) @@ -197,9 +204,7 @@ def test_reconnect_error_uses_reconnect_pipeline_response_for_legacy_models(): generated = "\n".join(serializer.handle_structured_stream_response(builder)) assert "if _reconnect_response.status_code not in [200, 204]:" in generated - assert ( - "PipelineResponse(_request, _reconnect_response, pipeline_response.context)," in generated - ) + assert "PipelineResponse(_request, _reconnect_response, pipeline_response.context)," in generated assert "raise HttpResponseError(response=_reconnect_response, model=error)" in generated diff --git a/website/src/content/docs/docs/emitters/clients/http-client-python/reference/emitter.md b/website/src/content/docs/docs/emitters/clients/http-client-python/reference/emitter.md index c2a86182af5..f08d9c3aa31 100644 --- a/website/src/content/docs/docs/emitters/clients/http-client-python/reference/emitter.md +++ b/website/src/content/docs/docs/emitters/clients/http-client-python/reference/emitter.md @@ -90,6 +90,24 @@ The name of the package to be used in pretty-printing. Will be the name of the p Whether to return responses from HEAD requests as boolean. Defaults to `true`. +### `enable-sse-reconnect` + +**Type:** `boolean` + +**Default:** `false` + +Whether to generate automatic reconnection support for Server-Sent Events (SSE). Defaults to `false`: SSE streams end when the response reaches EOF, without sending another request. Set to `true` to generate reconnect callbacks, retry delays, and `Last-Event-ID` resumption. SSE event IDs and retry metadata are parsed in either mode. Transport errors still propagate to the caller. + +Enable this generation-time option in `tspconfig.yaml`: + +```yaml +emit: + - "@typespec/http-client-python" +options: + "@typespec/http-client-python": + enable-sse-reconnect: true +``` + ### `use-pyodide` **Type:** `boolean` From b440a2666ea29b63a2b1c81c0226d76ee6c5b924 Mon Sep 17 00:00:00 2001 From: Libba Lawrence Date: Mon, 12 Oct 2026 09:02:25 -0700 Subject: [PATCH 2/2] fix(http-client-python): preserve metadata-only SSE state Copy decoder metadata before EOF exit and reconnect decisions in sync and async streams. Cover metadata-only blocks across reconnect options and flavors while preserving enabled resumption and HTTP 204 termination. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../changes/python-sse-reconnect-option.md | 2 +- .../templates/streaming_base.py.jinja2 | 12 ++--- .../tests/unit/test_sse_reconnect_option.py | 54 +++++++++++++++++-- 3 files changed, 57 insertions(+), 11 deletions(-) diff --git a/.chronus/changes/python-sse-reconnect-option.md b/.chronus/changes/python-sse-reconnect-option.md index 805bd54228a..b0f9a45c1b1 100644 --- a/.chronus/changes/python-sse-reconnect-option.md +++ b/.chronus/changes/python-sse-reconnect-option.md @@ -4,7 +4,7 @@ packages: - "@typespec/http-client-python" --- -Make automatic SSE reconnection opt-in with the `enable-sse-reconnect` emitter option, which defaults to `false`. Without this option, SSE streams end at EOF without reconnect callbacks, retry delays, or `Last-Event-ID` request plumbing. SSE parsing, event ID and retry metadata, terminal events, and stream cleanup remain available. +Make automatic SSE reconnection opt-in with the `enable-sse-reconnect` emitter option, which defaults to `false`. Without this option, SSE streams end at EOF without reconnect callbacks, retry delays, or `Last-Event-ID` request plumbing. SSE parsing, event ID and retry metadata (including metadata-only blocks), terminal events, and stream cleanup remain available. Enable reconnection when generating the client: diff --git a/packages/http-client-python/generator/pygen/codegen/templates/streaming_base.py.jinja2 b/packages/http-client-python/generator/pygen/codegen/templates/streaming_base.py.jinja2 index 67f2abaf832..814138c5815 100644 --- a/packages/http-client-python/generator/pygen/codegen/templates/streaming_base.py.jinja2 +++ b/packages/http-client-python/generator/pygen/codegen/templates/streaming_base.py.jinja2 @@ -798,13 +798,13 @@ class Stream(Iterator[ReturnType_co]): terminal = True {% endif %} break -{% if enable_sse_reconnect %} - if terminal or self._reconnect_callback is None: - break if isinstance(self._decoder, SSEDecoder): self._last_event_id = self._decoder.last_event_id if self._decoder.retry is not None: self._retry = self._decoder.retry +{% if enable_sse_reconnect %} + if terminal or self._reconnect_callback is None: + break self._response.close() reconnect_delay = (self._retry if self._retry is not None else 3000) / 1000 self._response = self._reconnect_callback(self._last_event_id, reconnect_delay) @@ -964,13 +964,13 @@ class AsyncStream(AsyncIterator[ReturnType_co]): aclose = getattr(events, "aclose", None) if aclose is not None: await aclose() -{% if enable_sse_reconnect %} - if terminal or self._reconnect_callback is None: - break if isinstance(self._decoder, AsyncSSEDecoder): self._last_event_id = self._decoder.last_event_id if self._decoder.retry is not None: self._retry = self._decoder.retry +{% if enable_sse_reconnect %} + if terminal or self._reconnect_callback is None: + break await self._response.close() reconnect_delay = (self._retry if self._retry is not None else 3000) / 1000 self._response = await self._reconnect_callback(self._last_event_id, reconnect_delay) diff --git a/packages/http-client-python/tests/unit/test_sse_reconnect_option.py b/packages/http-client-python/tests/unit/test_sse_reconnect_option.py index 0cbc7e4bcd6..4ae11abeaa1 100644 --- a/packages/http-client-python/tests/unit/test_sse_reconnect_option.py +++ b/packages/http-client-python/tests/unit/test_sse_reconnect_option.py @@ -132,6 +132,43 @@ def test_normal_sse_eof_keeps_metadata_and_closes(runtime, async_mode): assert response.byte_iterator_closed +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize( + ("body", "expected_data", "expected_id", "expected_retry"), + [ + (b"id: latest\nretry: 1000\n\n", [], "latest", 1000), + ( + b"id: first\nretry: 3000\ndata: one\n\nid: latest\nretry: 1000\n\n", + ["one"], + "latest", + 1000, + ), + (b"id: first\nretry: 3000\ndata: one\n\nid:\nretry: 0\n\n", ["one"], "", 0), + ( + b"id: first\nretry: 3000\ndata: one\n\nid: invalid\x00id\nretry: invalid\n\n", + ["one"], + "first", + 3000, + ), + ( + b"id: first\nretry: 3000\ndata: one\n\nid: incomplete\n", + ["one"], + "first", + 3000, + ), + ], +) +def test_sse_eof_keeps_metadata_without_data_event( + runtime, async_mode, body, expected_data, expected_id, expected_retry +): + events, stream, response = _consume(runtime, body, async_mode=async_mode) + assert [event.data for event in events] == expected_data + assert stream.last_event_id == expected_id + assert stream.retry == expected_retry + assert response.closed + assert response.byte_iterator_closed + + @pytest.mark.parametrize("async_mode", [False, True]) @pytest.mark.parametrize( ("body", "kwargs", "expected"), @@ -239,11 +276,20 @@ def test_reconnect_only_operation_imports_are_gated(code_model, reconnect_option @pytest.mark.parametrize("async_mode", [False, True]) -@pytest.mark.parametrize("metadata, expected", [(b"", ("", 3.0)), (b"id: first\nretry: 1000\n", ("first", 1.0))]) +@pytest.mark.parametrize( + ("body", "expected_items", "expected"), + [ + (b"data: one\n\n", ["one"], ("", 3.0)), + (b"id: first\nretry: 1000\ndata: one\n\n", ["one"], ("first", 1.0)), + (b"id: first\nretry: 1000\n\n", [], ("first", 1.0)), + ], +) @pytest.mark.parametrize("reconnect_options", [{"enable-sse-reconnect": True}]) -def test_enabled_reconnect_resumes_and_stops_at_204(runtime, reconnect_options, async_mode, metadata, expected): +def test_enabled_reconnect_resumes_and_stops_at_204( + runtime, reconnect_options, async_mode, body, expected_items, expected +): calls = [] - first = (AsyncFakeResponse if async_mode else FakeResponse)(metadata + b"data: one\n\n") + first = (AsyncFakeResponse if async_mode else FakeResponse)(body) second = (AsyncFakeResponse if async_mode else FakeResponse)(b"", status_code=204) def reconnect(last_event_id, delay): @@ -271,6 +317,6 @@ async def consume_async(): reconnect_callback=reconnect, ) ) - assert items == ["one"] + assert items == expected_items assert calls == [expected] assert first.closed and second.closed