[changes] Fixed AtlanClient event stream handling

Author: Aryamanz29

pyatlan/client/asset.py

@@ -25,15 +25,20 @@
 )
 from warnings import warn
 
-import httpx
 from pydantic.v1 import (
     StrictStr,
     ValidationError,
     constr,
     parse_obj_as,
     validate_arguments,
 )
-from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_fixed
+from tenacity import (
+    RetryError,
+    retry,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_fixed,
+)
 
 from pyatlan.client.common import ApiCaller
 from pyatlan.client.constants import (
@@ -844,7 +849,10 @@ def delete_by_guid(self, guid: Union[str, List[str]]) -> AssetMutationResponse:
         )
         response = AssetMutationResponse(**raw_json)
         for asset in response.assets_deleted(asset_type=Asset):
-            self._wait_till_deleted(asset)
+            try:
+                self._wait_till_deleted(asset)
+            except RetryError as err:
+                raise ErrorCode.RETRY_OVERRUN.exception_with_parameters() from err
         return response
 
     @retry(
@@ -854,12 +862,9 @@ def delete_by_guid(self, guid: Union[str, List[str]]) -> AssetMutationResponse:
         wait=wait_fixed(1),
     )
     def _wait_till_deleted(self, asset: Asset):
-        try:
-            asset = self.retrieve_minimal(guid=asset.guid, asset_type=Asset)
-            if asset.status == EntityStatus.DELETED:
-                return
-        except httpx.TransportError as err:
-            raise ErrorCode.RETRY_OVERRUN.exception_with_parameters() from err
+        asset = self.retrieve_minimal(guid=asset.guid, asset_type=Asset)
+        if asset.status == EntityStatus.DELETED:
+            return
 
     @validate_arguments
     def restore(self, asset_type: Type[A], qualified_name: str) -> bool:

pyatlan/client/atlan.py

@@ -7,8 +7,6 @@
 import copy
 import json
 import logging
-import os
-import shutil
 import uuid
 from contextvars import ContextVar
 from http import HTTPStatus
@@ -129,7 +127,7 @@ class AtlanClient(BaseSettings):
     read_timeout: float = 900.0  # 15 mins
     retry: Retry = DEFAULT_RETRY
     _401_has_retried: ContextVar[bool] = ContextVar("_401_has_retried", default=False)
-    _session: httpx.Client = PrivateAttr(default_factory=lambda: httpx.Client())
+    _session: httpx.Client = PrivateAttr()
     _request_params: dict = PrivateAttr()
     _user_id: Optional[str] = PrivateAttr(default=None)
     _workflow_client: Optional[WorkflowClient] = PrivateAttr(default=None)
@@ -169,7 +167,7 @@ def __init__(self, **data):
                 "authorization": f"Bearer {self.api_key}",
             }
         }
-        # Configure httpx client with retry transport
+        # Configure httpx client with the provided retry settings
         self._session = httpx.Client(
             transport=RetryTransport(retry=self.retry),
             headers={
@@ -409,8 +407,9 @@ def update_headers(self, header: Dict[str, str]):
 
     def _handle_file_download(self, raw_response: Any, file_path: str) -> str:
         try:
-            download_file = open(file_path, "wb")
-            shutil.copyfileobj(raw_response, download_file)
+            with open(file_path, "wb") as download_file:
+                for chunk in raw_response:
+                    download_file.write(chunk)
         except Exception as err:
             raise ErrorCode.UNABLE_TO_DOWNLOAD_FILE.exception_with_parameters(
                 str((hasattr(err, "strerror") and err.strerror) or err), file_path
@@ -441,15 +440,49 @@ def _call_api_internal(
                     timeout=timeout,
                 )
             elif api.consumes == EVENT_STREAM and api.produces == EVENT_STREAM:
-                response = self._session.request(
+                with self._session.stream(
                     api.method.value,
                     path,
                     **params,
-                    stream=True,
                     timeout=timeout,
-                )
-                if download_file_path:
-                    return self._handle_file_download(response.raw, download_file_path)
+                ) as stream_response:
+                    if download_file_path:
+                        return self._handle_file_download(
+                            stream_response.iter_raw(), download_file_path
+                        )
+
+                    # For event streams, we need to read the content while the stream is open
+                    # Store the response data and create a mock response object for common processing
+                    content = stream_response.read()
+                    text = content.decode("utf-8") if content else ""
+                    lines = []
+
+                    # Only process lines for successful responses to avoid errors on error responses
+                    if stream_response.status_code == api.expected_status:
+                        # Reset stream position and get lines
+                        lines = text.splitlines() if text else []
+
+                    response_data = {
+                        "status_code": stream_response.status_code,
+                        "headers": stream_response.headers,
+                        "text": text,
+                        "content": content,
+                        "lines": lines,
+                    }
+
+                    # Create a simple namespace object to mimic the response interface
+                    response = SimpleNamespace(
+                        status_code=response_data["status_code"],
+                        headers=response_data["headers"],
+                        text=response_data["text"],
+                        content=response_data["content"],
+                        _stream_lines=response_data[
+                            "lines"
+                        ],  # Store lines for event processing
+                        json=lambda: json.loads(response_data["text"])
+                        if response_data["text"]
+                        else {},
+                    )
             else:
                 response = self._session.request(
                     api.method.value,
@@ -496,14 +529,16 @@ def _call_api_internal(
                             response,
                         )
                     if api.consumes == EVENT_STREAM and api.produces == EVENT_STREAM:
-                        for line in response.iter_lines(decode_unicode=True):
-                            if not line:
-                                continue
-                            if not line.startswith("data: "):
-                                raise ErrorCode.UNABLE_TO_DESERIALIZE.exception_with_parameters(
-                                    line
-                                )
-                            events.append(json.loads(line.split("data: ")[1]))
+                        # Process event stream using stored lines from the streaming response
+                        if hasattr(response, "_stream_lines"):
+                            for line in response._stream_lines:
+                                if not line:
+                                    continue
+                                if not line.startswith("data: "):
+                                    raise ErrorCode.UNABLE_TO_DESERIALIZE.exception_with_parameters(
+                                        line
+                                    )
+                                events.append(json.loads(line.split("data: ")[1]))
                     if text_response:
                         response_ = response.text
                     else:

tests/integration/test_index_search.py

@@ -10,6 +10,7 @@
 import httpx
 import pytest
 from httpx_retries import Retry
+from pydantic.v1 import HttpUrl
 
 from pyatlan.cache.source_tag_cache import SourceTagName
 from pyatlan.client.asset import LOGGER, IndexSearchResults, Persona, Purpose
@@ -854,12 +855,15 @@ def test_read_timeout(client: AtlanClient):
 
 
 def test_connect_timeout(client: AtlanClient):
-    request = (FluentSearch().select()).to_request()
+    request = FluentSearch().select().to_request()
+
+    # Use a non-routable IP that will definitely timeout
+    # 192.0.2.1 is reserved for documentation/testing
     with client_connection(
-        client=client, connect_timeout=0.0001, retry=Retry(total=0)
+        client=client,
+        base_url=HttpUrl("http://192.0.2.1:80", scheme="http"),  # Non-routable test IP
+        connect_timeout=0.001,
+        retry=Retry(total=1),
     ) as timed_client:
-        with pytest.raises(
-            httpx.ConnectTimeout,
-            match="timed out",
-        ):
+        with pytest.raises(httpx.ConnectTimeout):
             timed_client.asset.search(criteria=request)

tests/unit/test_file_client.py

@@ -79,7 +79,22 @@ def mock_session():
         mock_response = Mock()
         mock_response.status_code = 200
         mock_response.raw = open(UPLOAD_FILE_PATH, "rb")
-        mock_session.request.return_value = mock_response
+        mock_response.headers = {}
+
+        # Mock the methods our streaming code expects
+        mock_response.read.return_value = b"test content"
+
+        def mock_iter_raw(chunk_size=None):
+            # Use the actual expected content from upload.txt
+            content = b"test data 12345.\n"
+            yield content
+
+        mock_response.iter_raw = mock_iter_raw
+
+        # Use Mock's context manager support
+        mock_session.stream.return_value.__enter__.return_value = mock_response
+        mock_session.stream.return_value.__exit__.return_value = None
+
         yield mock_session
     assert os.path.exists(DOWNLOAD_FILE_PATH)
     os.remove(DOWNLOAD_FILE_PATH)
@@ -91,10 +106,34 @@ def mock_session_invalid():
         mock_response = Mock()
         mock_response.status_code = 200
         mock_response.raw = "not a bytes-like object"
-        mock_session.request.return_value = mock_response
+        mock_response.headers = {}
+
+        # Mock the methods our streaming code expects
+        mock_response.read.return_value = b"test content"
+
+        def mock_iter_raw(chunk_size=None):
+            # Return a generator that will fail during iteration
+            # This simulates a case where the response object is invalid
+            class BadIterator:
+                def __iter__(self):
+                    return self
+
+                def __next__(self):
+                    # Simulate the error that would happen in real scenario
+                    raise AttributeError("'str' object has no attribute 'read'")
+
+            return BadIterator()
+
+        mock_response.iter_raw = mock_iter_raw
+
+        # Use Mock's context manager support
+        mock_session.stream.return_value.__enter__.return_value = mock_response
+        mock_session.stream.return_value.__exit__.return_value = None
+
         yield mock_session
-    assert os.path.exists(DOWNLOAD_FILE_PATH)
-    os.remove(DOWNLOAD_FILE_PATH)
+    # Don't assert file exists for invalid case since error should prevent creation
+    if os.path.exists(DOWNLOAD_FILE_PATH):
+        os.remove(DOWNLOAD_FILE_PATH)
 
 
 @pytest.mark.parametrize("method, params", TEST_FILE_CLIENT_METHODS.items())
@@ -200,7 +239,7 @@ def test_file_client_download_file(client, s3_presigned_url, mock_session):
         presigned_url=s3_presigned_url, file_path=DOWNLOAD_FILE_PATH
     )
     assert response == DOWNLOAD_FILE_PATH
-    assert mock_session.request.call_count == 1
+    assert mock_session.stream.call_count == 1
     # The file should exist after calling the method
     assert os.path.exists(DOWNLOAD_FILE_PATH)
     assert open(DOWNLOAD_FILE_PATH, "r").read() == "test data 12345.\n"

tests/unit/test_query_client.py

@@ -44,10 +44,23 @@ def mock_session():
         mock_response = Mock()
         mock_response.status_code = 200
         mock_response.content = "test-content"
+        mock_response.headers = {}
+
         with open(QUERY_RESPONSES, "r", encoding="utf-8") as file:
             lines_from_file = [line.strip() for line in file.readlines()]
         mock_response.iter_lines.return_value = lines_from_file
+
+        # Mock the methods our streaming code expects
+        file_content = "\n".join(lines_from_file)
+        mock_response.read.return_value = file_content.encode("utf-8")
+        mock_response.text = file_content
+
+        # Support both old request-style and new stream-style
         mock_session.request.return_value = mock_response
+
+        # Use Mock's context manager support for streaming
+        mock_session.stream.return_value.__enter__.return_value = mock_response
+        mock_session.stream.return_value.__exit__.return_value = None
         yield mock_session
 
 
@@ -90,8 +103,21 @@ def test_stream_get_raises_error(
     mock_response = Mock()
     mock_response.status_code = 200
     mock_response.content = "test-content"
+    mock_response.headers = {}
     mock_response.iter_lines.return_value = test_response
+
+    # Mock the methods our streaming code expects
+    file_content = "\n".join(test_response)
+    mock_response.read.return_value = file_content.encode("utf-8")
+    mock_response.text = file_content
+
+    # Support both old request-style and new stream-style
     mock_session.request.return_value = mock_response
+
+    # Use Mock's context manager support for streaming
+    mock_session.stream.return_value.__enter__.return_value = mock_response
+    mock_session.stream.return_value.__exit__.return_value = None
+
     with pytest.raises(test_error) as err:
         client.queries.stream(request=query_request)
     assert error_msg in str(err.value)