Skip to content

Commit 00901b8

Browse files
aditya-arolkar-swefern-support
aditya-arolkar-swe
and
fern-support
authored
feat(python): eliminate SSE dependency 'httpx_sse' by hard-forking into core_utilities (#9784)
* Hard forked httpx_sse, moved into core_utilities, made fixes to line parsing bugs and improved idiomatic class definition. * updated parser to: copy http_sse code into output, use new in-house SSE EventStream object and better handle failed SSE parsing * updated changelog * Automated update of seed files * skip terminator sse events * typing fixes * Fix formatting issues: remove trailing whitespace and apply ruff formatting * Automated update of seed files * improved changelog language * SSE Support arbitrary charsets from the headers - defaulting to utf-8 * added SSE unit tests * properly copy core/http_sse to src/.../core/http_sse * better typing in test_http_sse * formatting * refactored 'server-sent-event-examples' output (seed gen was broken), and added a test file 'tests/utils/test_sse_streaming.py' * Hard forked httpx_sse, moved into core_utilities, made fixes to line parsing bugs and improved idiomatic class definition. * updated parser to: copy http_sse code into output, use new in-house SSE EventStream object and better handle failed SSE parsing * updated changelog * skip terminator sse events * Automated update of seed files * typing fixes * Fix formatting issues: remove trailing whitespace and apply ruff formatting * improved changelog language * Automated update of seed files * SSE Support arbitrary charsets from the headers - defaulting to utf-8 * added SSE unit tests * properly copy core/http_sse to src/.../core/http_sse * better typing in test_http_sse * formatting * refactored 'server-sent-event-examples' output (seed gen was broken), and added a test file 'tests/utils/test_sse_streaming.py' * Updated AsyncGenerator typing * Properly generator stream terminator as a string i.e. was returning [[DONE]] before instead of '[[DONE]]' * updated http_sse's __init__.py generation * manual seed updates to fixture 'server-sent-event-examples' * Automated update of seed files * updated sse tests * formatting fixes --------- Co-authored-by: aditya-arolkar-swe <aditya-arolkar-swe@users.noreply.github.com> Co-authored-by: fern-support <fern-support@users.noreply.github.com>
1 parent 5934288 commit 00901b8

654 files changed

Lines changed: 32175 additions & 62 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎generators/python/core_utilities/shared/http_sse/__init__.py‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
from ._api import EventSource, aconnect_sse, connect_sse
2+
from ._exceptions import SSEError
3+
from ._models import ServerSentEvent
4+
5+
__version__ = "0.4.1"
6+
7+
__all__ = [
8+
"__version__",
9+
"EventSource",
10+
"connect_sse",
11+
"aconnect_sse",
12+
"ServerSentEvent",
13+
"SSEError",
14+
]

‎generators/python/core_utilities/shared/http_sse/_api.py‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
import re
2+
from contextlib import asynccontextmanager, contextmanager
3+
from typing import Any, AsyncGenerator, AsyncIterator, Iterator, cast
4+
5+
import httpx
6+
from ._decoders import SSEDecoder
7+
from ._exceptions import SSEError
8+
from ._models import ServerSentEvent
9+
10+
11+
class EventSource:
12+
def __init__(self, response: httpx.Response) -> None:
13+
self._response = response
14+
15+
def _check_content_type(self) -> None:
16+
content_type = self._response.headers.get("content-type", "").partition(";")[0]
17+
if "text/event-stream" not in content_type:
18+
raise SSEError(
19+
f"Expected response header Content-Type to contain 'text/event-stream', got {content_type!r}"
20+
)
21+
22+
def _get_charset(self) -> str:
23+
"""Extract charset from Content-Type header, fallback to UTF-8."""
24+
content_type = self._response.headers.get("content-type", "")
25+
26+
# Parse charset parameter using regex
27+
charset_match = re.search(r"charset=([^;\s]+)", content_type, re.IGNORECASE)
28+
if charset_match:
29+
charset = charset_match.group(1).strip("\"'")
30+
# Validate that it's a known encoding
31+
try:
32+
# Test if the charset is valid by trying to encode/decode
33+
"test".encode(charset).decode(charset)
34+
return charset
35+
except (LookupError, UnicodeError):
36+
# If charset is invalid, fall back to UTF-8
37+
pass
38+
39+
# Default to UTF-8 if no charset specified or invalid charset
40+
return "utf-8"
41+
42+
@property
43+
def response(self) -> httpx.Response:
44+
return self._response
45+
46+
def iter_sse(self) -> Iterator[ServerSentEvent]:
47+
self._check_content_type()
48+
decoder = SSEDecoder()
49+
charset = self._get_charset()
50+
51+
buffer = ""
52+
for chunk in self._response.iter_bytes():
53+
# Decode chunk using detected charset
54+
text_chunk = chunk.decode(charset, errors="replace")
55+
buffer += text_chunk
56+
57+
# Process complete lines
58+
while "\n" in buffer:
59+
line, buffer = buffer.split("\n", 1)
60+
line = line.rstrip("\r")
61+
sse = decoder.decode(line)
62+
# when we reach a "\n\n" => line = ''
63+
# => decoder will attempt to return an SSE Event
64+
if sse is not None:
65+
yield sse
66+
67+
# Process any remaining data in buffer
68+
if buffer.strip():
69+
line = buffer.rstrip("\r")
70+
sse = decoder.decode(line)
71+
if sse is not None:
72+
yield sse
73+
74+
async def aiter_sse(self) -> AsyncGenerator[ServerSentEvent, None]:
75+
self._check_content_type()
76+
decoder = SSEDecoder()
77+
lines = cast(AsyncGenerator[str, None], self._response.aiter_lines())
78+
try:
79+
async for line in lines:
80+
line = line.rstrip("\n")
81+
sse = decoder.decode(line)
82+
if sse is not None:
83+
yield sse
84+
finally:
85+
await lines.aclose()
86+
87+
88+
@contextmanager
89+
def connect_sse(client: httpx.Client, method: str, url: str, **kwargs: Any) -> Iterator[EventSource]:
90+
headers = kwargs.pop("headers", {})
91+
headers["Accept"] = "text/event-stream"
92+
headers["Cache-Control"] = "no-store"
93+
94+
with client.stream(method, url, headers=headers, **kwargs) as response:
95+
yield EventSource(response)
96+
97+
98+
@asynccontextmanager
99+
async def aconnect_sse(
100+
client: httpx.AsyncClient,
101+
method: str,
102+
url: str,
103+
**kwargs: Any,
104+
) -> AsyncIterator[EventSource]:
105+
headers = kwargs.pop("headers", {})
106+
headers["Accept"] = "text/event-stream"
107+
headers["Cache-Control"] = "no-store"
108+
109+
async with client.stream(method, url, headers=headers, **kwargs) as response:
110+
yield EventSource(response)

‎generators/python/core_utilities/shared/http_sse/_decoders.py‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
from typing import List, Optional
2+
3+
from ._models import ServerSentEvent
4+
5+
6+
class SSEDecoder:
7+
def __init__(self) -> None:
8+
self._event = ""
9+
self._data: List[str] = []
10+
self._last_event_id = ""
11+
self._retry: Optional[int] = None
12+
13+
def decode(self, line: str) -> Optional[ServerSentEvent]:
14+
# See: https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation # noqa: E501
15+
16+
if not line:
17+
if not self._event and not self._data and not self._last_event_id and self._retry is None:
18+
return None
19+
20+
sse = ServerSentEvent(
21+
event=self._event,
22+
data="\n".join(self._data),
23+
id=self._last_event_id,
24+
retry=self._retry,
25+
)
26+
27+
# NOTE: as per the SSE spec, do not reset last_event_id.
28+
self._event = ""
29+
self._data = []
30+
self._retry = None
31+
32+
return sse
33+
34+
if line.startswith(":"):
35+
return None
36+
37+
fieldname, _, value = line.partition(":")
38+
39+
if value.startswith(" "):
40+
value = value[1:]
41+
42+
if fieldname == "event":
43+
self._event = value
44+
elif fieldname == "data":
45+
self._data.append(value)
46+
elif fieldname == "id":
47+
if "\0" in value:
48+
pass
49+
else:
50+
self._last_event_id = value
51+
elif fieldname == "retry":
52+
try:
53+
self._retry = int(value)
54+
except (TypeError, ValueError):
55+
pass
56+
else:
57+
pass # Field is ignored.
58+
59+
return None

‎generators/python/core_utilities/shared/http_sse/_exceptions.py‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
import httpx
2+
3+
4+
class SSEError(httpx.TransportError):
5+
pass

‎generators/python/core_utilities/shared/http_sse/_models.py‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
import json
2+
from dataclasses import dataclass
3+
from typing import Any, Optional
4+
5+
6+
@dataclass(frozen=True)
7+
class ServerSentEvent:
8+
event: str = "message"
9+
data: str = ""
10+
id: str = ""
11+
retry: Optional[int] = None
12+
13+
def json(self) -> Any:
14+
"""Parse the data field as JSON."""
15+
return json.loads(self.data)

‎generators/python/sdk/versions.yml‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,14 @@
11
# yaml-language-server: $schema=../../../fern-versions-yml.schema.json
22
# For unreleased changes, use unreleased.yml
3+
- version: 4.31.0
4+
changelogEntry:
5+
- summary: |
6+
- Removed external dependency on httpx-sse by bringing SSE handling in-house.
7+
- Fixed SSE handling of events longer than or containing escaped newlines.
8+
type: feat
9+
createdAt: "2025-10-06"
10+
irVersion: 60
11+
312
- version: 4.30.4-rc1
413
changelogEntry:
514
- summary: |

‎generators/python/src/fern_python/generators/sdk/client_generator/endpoint_response_code_writer.py‎

Lines changed: 81 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@
22

33
from ..context.sdk_generator_context import SdkGeneratorContext
44
from fern_python.codegen import AST
5-
from fern_python.external_dependencies.httpx_sse import HttpxSSE
65
from fern_python.external_dependencies.json import Json
76
from fern_python.generators.sdk.client_generator.constants import CHUNK_VARIABLE, RESPONSE_VARIABLE
87
from fern_python.generators.sdk.client_generator.pagination.abstract_paginator import (
@@ -99,7 +98,18 @@ def _handle_success_stream(self, *, writer: AST.NodeWriter, stream_response: ir_
9998
AST.VariableDeclaration(
10099
name=EndpointResponseCodeWriter.EVENT_SOURCE_VARIABLE,
101100
initializer=AST.Expression(
102-
AST.ClassInstantiation(HttpxSSE.EVENT_SOURCE, [AST.Expression(RESPONSE_VARIABLE)])
101+
AST.ClassInstantiation(
102+
class_=AST.ClassReference(
103+
qualified_name_excluding_import=(),
104+
import_=AST.ReferenceImport(
105+
module=AST.Module.local(
106+
*self._context.core_utilities._module_path, "http_sse", "_api"
107+
),
108+
named_import="EventSource",
109+
),
110+
),
111+
args=[AST.Expression(RESPONSE_VARIABLE)],
112+
)
103113
),
104114
),
105115
AST.ForStatement(
@@ -119,7 +129,7 @@ def _handle_success_stream(self, *, writer: AST.NodeWriter, stream_response: ir_
119129
conditions=[
120130
AST.IfConditionLeaf(
121131
condition=AST.Expression(
122-
f"{EndpointResponseCodeWriter.SSE_VARIABLE}.data == {stream_response_union.terminator}"
132+
f"{EndpointResponseCodeWriter.SSE_VARIABLE}.data == {repr(stream_response_union.terminator)}"
123133
),
124134
code=[AST.ReturnStatement()],
125135
),
@@ -131,16 +141,77 @@ def _handle_success_stream(self, *, writer: AST.NodeWriter, stream_response: ir_
131141
AST.YieldStatement(
132142
self._context.core_utilities.get_construct(
133143
self._get_streaming_response_data_type(stream_response),
134-
AST.Expression(
135-
Json.loads(
136-
AST.Expression(f"{EndpointResponseCodeWriter.SSE_VARIABLE}.data")
137-
)
138-
),
144+
AST.Expression(f"{EndpointResponseCodeWriter.SSE_VARIABLE}.json()"),
139145
),
140146
),
141147
],
142148
handlers=[
143-
noop_except_handler,
149+
AST.ExceptHandler(
150+
body=[
151+
AST.Expression(
152+
AST.FunctionInvocation(
153+
function_definition=AST.Reference(
154+
qualified_name_excluding_import=(),
155+
import_=AST.ReferenceImport(
156+
module=AST.Module.built_in(("logging",)),
157+
named_import="warning",
158+
),
159+
),
160+
args=[
161+
AST.Expression(
162+
f'f"Skipping SSE event with invalid JSON: {{e}}, sse: {{{EndpointResponseCodeWriter.SSE_VARIABLE}!r}}"'
163+
)
164+
],
165+
)
166+
),
167+
],
168+
exception_type="JSONDecodeError",
169+
name="e",
170+
),
171+
AST.ExceptHandler(
172+
body=[
173+
AST.Expression(
174+
AST.FunctionInvocation(
175+
function_definition=AST.Reference(
176+
qualified_name_excluding_import=(),
177+
import_=AST.ReferenceImport(
178+
module=AST.Module.built_in(("logging",)),
179+
named_import="warning",
180+
),
181+
),
182+
args=[
183+
AST.Expression(
184+
f'f"Skipping SSE event due to model construction error: {{type(e).__name__}}: {{e}}, sse: {{{EndpointResponseCodeWriter.SSE_VARIABLE}!r}}"'
185+
)
186+
],
187+
)
188+
),
189+
],
190+
exception_type="(TypeError, ValueError, KeyError, AttributeError)",
191+
name="e",
192+
),
193+
AST.ExceptHandler(
194+
body=[
195+
AST.Expression(
196+
AST.FunctionInvocation(
197+
function_definition=AST.Reference(
198+
qualified_name_excluding_import=(),
199+
import_=AST.ReferenceImport(
200+
module=AST.Module.built_in(("logging",)),
201+
named_import="error",
202+
),
203+
),
204+
args=[
205+
AST.Expression(
206+
f'f"Unexpected error processing SSE event: {{type(e).__name__}}: {{e}}, sse: {{{EndpointResponseCodeWriter.SSE_VARIABLE}!r}}"'
207+
)
208+
],
209+
)
210+
),
211+
],
212+
exception_type="Exception",
213+
name="e",
214+
),
144215
],
145216
),
146217
],
@@ -156,7 +227,7 @@ def _handle_success_stream(self, *, writer: AST.NodeWriter, stream_response: ir_
156227
conditions=[
157228
AST.IfConditionLeaf(
158229
condition=AST.Expression(
159-
f"{EndpointResponseCodeWriter.STREAM_TEXT_VARIABLE} == {stream_response_union.terminator}"
230+
f"{EndpointResponseCodeWriter.STREAM_TEXT_VARIABLE} == {repr(stream_response_union.terminator)}"
160231
),
161232
code=[AST.ReturnStatement()],
162233
),

0 commit comments

Comments
 (0)