feat(http-client-python): make SSE reconnection opt-in

.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 (including metadata-only blocks), 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
+```

cspell.yaml

@@ -280,6 +280,7 @@ words:
   - srnagar
   - ssdlrs
   - sses
+  - ssenoreconnect
   - ssvs
   - statment
   - strs

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`

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, "-");

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<PythonEmitterOptions> =
       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,

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`

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,14 +52,20 @@ export interface CodeModel {
  */
 export async function emitCodeModel(
   code: string,
-): Promise<{ codeModel: CodeModel; diagnostics: readonly Diagnostic[] }> {
+  options: PythonEmitterOptions = {},
+): Promise<{
+  codeModel: CodeModel;
+  commandArgs: Record<string, unknown>;
+  diagnostics: readonly Diagnostic[];
+}> {
   const outputDir = await mkdtemp(join(tmpdir(), "typespec-python-"));
   let yamlPath: string | undefined;
   try {
     const [, diagnostics] = await EmitterTester.compileAndDiagnose(code, {
       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) {

packages/http-client-python/eng/scripts/ci/regenerate-common.ts

@@ -80,10 +80,9 @@ export const SpecialFlags: Record<string, Record<string, any>> = {
 
 // ---- Spec-specific emitter option overrides ----
 
-export const AZURE_EMITTER_OPTIONS: Record<
-  string,
-  Record<string, string> | Record<string, string>[]
-> = {
+type EmitterOptions = Record<string, string | boolean>;
+
+export const AZURE_EMITTER_OPTIONS: Record<string, EmitterOptions | EmitterOptions[]> = {
   "azure/client-generator-core/access": {
     namespace: "specs.azure.clientgenerator.core.access",
   },
@@ -222,7 +221,16 @@ export const AZURE_EMITTER_OPTIONS: Record<
   },
 };
 
-export const EMITTER_OPTIONS: Record<string, Record<string, string> | Record<string, string>[]> = {
+export const EMITTER_OPTIONS: Record<string, EmitterOptions | EmitterOptions[]> = {
+  "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",
@@ -419,7 +427,7 @@ export function getEmitterOptions(
   spec: string,
   flavor: string,
   ctx: RegenerateContext,
-): Record<string, string>[] {
+): EmitterOptions[] {
   const specDir = isAzureSpec(spec) ? ctx.azureHttpSpecs : ctx.httpSpecs;
   const relativeSpec = toPosix(relative(specDir, spec));
   const key = relativeSpec.includes("resiliency/srv-driven/old.tsp")

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'")
 

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

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):"