fix(ometa): use requests instead of httpx for SSE transport

In some enterprise environments httpx's HTTP/1.1 wire pattern is reset
by network security middleboxes that pass requests/wget through. The
symptom is httpx.ReadError: [Errno 104] Connection reset by peer at
_receive_response_headers, with no server-side access log line because
the connection is reset before any HTTP response is committed. requests
and wget succeed against the same endpoint with the same JWT and body.

The SDK's REST client already uses requests, so this aligns the SSE
path with the proven transport. Behavior preserved: same retry loop,
same SSE parser, same Last-Event-ID resume on retry, same log lines.

Drive-by fix: the previous implementation built opts = { ...,
allow_redirects, verify, cookies, ... } but never passed those keys to
httpx.stream(...) - a latent bug. requests.Session.request now wires
them up properly (defaults: verify=True, allow_redirects=True when
unset).

No new dependency: requests is already used by the SDK's REST client.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: pmbrull

ingestion/src/metadata/ingestion/ometa/sse_client.py

@@ -17,7 +17,7 @@
 from logging import Logger  # noqa: TC003
 from typing import Any, Generator  # noqa: UP035
 
-import httpx
+import requests
 
 from metadata.ingestion.ometa.client import ClientConfig
 from metadata.ingestion.ometa.credentials import URL
@@ -87,35 +87,52 @@ def stream(self, method: str, path: str, data: None | dict[str, Any] = None) ->
                 if self.last_event_id:
                     headers["Last-Event-ID"] = self.last_event_id
 
-                with httpx.Client(timeout=None) as client:  # noqa: SIM117
-                    with client.stream(
-                        method,
-                        url,
-                        headers=headers,
-                        json=opts.get("json"),
-                        params=opts.get("params"),
-                    ) as response:
-                        response.raise_for_status()
-                        self.logger.info("Connected to SSE stream")
-
-                        event_buffer = []
-                        for line in response.iter_lines():
-                            if not line:
-                                if event_buffer:
-                                    parsed_event = self._parse_sse_event(event_buffer)
-                                    yield parsed_event
-                                    event_buffer = []
-
-                                    if self.stream_completed:
-                                        self.logger.info(
-                                            f"Stream terminated with event: {parsed_event.get('event', 'unknown')}"
-                                        )
-                                        return
-                            else:  # noqa: PLR5501
-                                if not line.startswith(":"):
-                                    event_buffer.append(line)
-
-            except httpx.HTTPStatusError as e:
+                # SSE transport uses `requests` rather than `httpx`: the SDK's REST client
+                # already uses requests, and some enterprise security middleboxes reject
+                # httpx's HTTP/1.1 wire pattern (lowercase header names, header/body in
+                # separate TCP segments) while passing requests/wget through unchanged.
+                request_kwargs = {
+                    "method": method,
+                    "url": str(url),
+                    "headers": headers,
+                    "json": opts.get("json"),
+                    "params": opts.get("params"),
+                    "stream": True,
+                    "timeout": None,
+                    "verify": (
+                        self.config.verify if self.config.verify is not None else True
+                    ),
+                    "allow_redirects": (
+                        self.config.allow_redirects
+                        if self.config.allow_redirects is not None
+                        else True
+                    ),
+                    "cookies": self.config.cookies,
+                }
+                with requests.Session() as session, session.request(
+                    **request_kwargs
+                ) as response:
+                    response.raise_for_status()
+                    self.logger.info("Connected to SSE stream")
+
+                    event_buffer = []
+                    for line in response.iter_lines(decode_unicode=True):
+                        if not line:
+                            if event_buffer:
+                                parsed_event = self._parse_sse_event(event_buffer)
+                                yield parsed_event
+                                event_buffer = []
+
+                                if self.stream_completed:
+                                    self.logger.info(
+                                        f"Stream terminated with event: {parsed_event.get('event', 'unknown')}"
+                                    )
+                                    return
+                        else:  # noqa: PLR5501
+                            if not line.startswith(":"):
+                                event_buffer.append(line)
+
+            except requests.exceptions.HTTPError as e:
                 self.logger.error(f"HTTP error: {e.response.status_code}")
                 raise
             except Exception as e: