From 09f0f92cddd203316a80a0764925ebde090f3254 Mon Sep 17 00:00:00 2001 From: Mothership Agent Date: Tue, 10 Mar 2026 17:38:48 +0000 Subject: [PATCH] fix(transport): move duplicate policy check after backoff sleep The duplicate check was running immediately after a timeout, before the backoff sleep. This meant the index often hadn't propagated the newly-created entity yet, causing the check to miss it and the retry to create a duplicate anyway. Also changes error handling in find_existing_policy to return None on failure instead of raising, so a search error doesn't break the retry loop. Fixes GOV-667. Co-Authored-By: Claude Opus 4.6 --- pyatlan/client/common/transport.py | 33 ++++++++++++++++-------------- pyatlan/client/transport.py | 22 ++++++++++++-------- tests/unit/test_transport.py | 26 +++++++++-------------- 3 files changed, 42 insertions(+), 39 deletions(-) diff --git a/pyatlan/client/common/transport.py b/pyatlan/client/common/transport.py index 585559940..d603f7bcd 100644 --- a/pyatlan/client/common/transport.py +++ b/pyatlan/client/common/transport.py @@ -16,7 +16,6 @@ import httpx from pyatlan.client.constants import BULK_UPDATE, INDEX_SEARCH -from pyatlan.errors import ErrorCode from pyatlan.model.search import DSL, Bool, IndexSearchRequest, Term logger = logging.getLogger(__name__) @@ -91,8 +90,7 @@ def find_existing_policy( """ Search for an existing AuthPolicy by name and persona GUID (synchronous). - Raises: - ErrorCode.UNABLE_TO_SEARCH_EXISTING_POLICY: if the search call fails. + Returns None on failure so the retry loop can proceed normally. """ try: search_request = build_policy_search_request(policy_name, persona_guid) @@ -101,9 +99,14 @@ def find_existing_policy( return raw_json["entities"][0] return None except Exception as e: - raise ErrorCode.UNABLE_TO_SEARCH_EXISTING_POLICY.exception_with_parameters( - policy_name, persona_guid, str(e) - ) from e + logger.warning( + "Duplicate policy search failed for '%s' (persona %s): %s. " + "Retry will proceed normally.", + policy_name, + persona_guid, + str(e), + ) + return None async def find_existing_policy_async( @@ -112,8 +115,7 @@ async def find_existing_policy_async( """ Search for an existing AuthPolicy by name and persona GUID (asynchronous). - Raises: - ErrorCode.UNABLE_TO_SEARCH_EXISTING_POLICY: if the search call fails. + Returns None on failure so the retry loop can proceed normally. """ try: search_request = build_policy_search_request(policy_name, persona_guid) @@ -122,9 +124,14 @@ async def find_existing_policy_async( return raw_json["entities"][0] return None except Exception as e: - raise ErrorCode.UNABLE_TO_SEARCH_EXISTING_POLICY.exception_with_parameters( - policy_name, persona_guid, str(e) - ) from e + logger.warning( + "Duplicate policy search failed for '%s' (persona %s): %s. " + "Retry will proceed normally.", + policy_name, + persona_guid, + str(e), + ) + return None def check_for_duplicate_policy( @@ -137,8 +144,6 @@ def check_for_duplicate_policy( Returns a mock response with the existing policy if a duplicate is found, or None to let the retry proceed normally. - Raises: - ErrorCode.UNABLE_TO_SEARCH_EXISTING_POLICY: if the duplicate search fails. """ parsed = parse_auth_policy_entity(request) if not parsed: @@ -165,8 +170,6 @@ async def check_for_duplicate_policy_async( Returns a mock response with the existing policy if a duplicate is found, or None to let the retry proceed normally. - Raises: - ErrorCode.UNABLE_TO_SEARCH_EXISTING_POLICY: if the duplicate search fails. """ parsed = parse_auth_policy_entity(request) if not parsed: diff --git a/pyatlan/client/transport.py b/pyatlan/client/transport.py index d5bd0a20b..eb462d06d 100644 --- a/pyatlan/client/transport.py +++ b/pyatlan/client/transport.py @@ -107,7 +107,13 @@ def _retry_operation( "_retry_operation retrying response=%s retry=%s", response, retry ) - # ONLY during retry: check if this is a policy creation and if duplicate exists + retry = retry.increment() + retry.sleep(response) + + # AFTER backoff: check if this is a policy creation and if duplicate exists. + # The sleep gives the index time to propagate the entity created by + # the previous request that may have succeeded server-side but timed + # out client-side. if self._client: duplicate_response = check_for_duplicate_policy( self._client, request @@ -119,9 +125,6 @@ def _retry_operation( ) return duplicate_response - retry = retry.increment() - retry.sleep(response) - try: response = send_method(request) except httpx.HTTPError as e: @@ -226,7 +229,13 @@ async def _retry_operation_async( retry, ) - # ONLY during retry: check if this is a policy creation and if duplicate exists + retry = retry.increment() + await retry.asleep(response) + + # AFTER backoff: check if this is a policy creation and if duplicate exists. + # The sleep gives the index time to propagate the entity created by + # the previous request that may have succeeded server-side but timed + # out client-side. if self._client: duplicate_response = await check_for_duplicate_policy_async( self._client, request @@ -238,9 +247,6 @@ async def _retry_operation_async( ) return duplicate_response - retry = retry.increment() - await retry.asleep(response) - try: response = await send_method(request) except httpx.HTTPError as e: diff --git a/tests/unit/test_transport.py b/tests/unit/test_transport.py index aa9220f30..6fb3bfacd 100644 --- a/tests/unit/test_transport.py +++ b/tests/unit/test_transport.py @@ -175,12 +175,11 @@ def test_returns_none_when_raw_json_is_none(self): result = find_existing_policy(client, POLICY_NAME, PERSONA_GUID) assert result is None - def test_raises_error_code_on_exception(self): + def test_returns_none_on_exception(self): client = MagicMock() client._call_api.side_effect = Exception("search failed") - with pytest.raises(Exception) as exc_info: - find_existing_policy(client, POLICY_NAME, PERSONA_GUID) - assert "ATLAN-PYTHON-500-007" in str(exc_info.value) + result = find_existing_policy(client, POLICY_NAME, PERSONA_GUID) + assert result is None # --------------------------------------------------------------------------- @@ -204,12 +203,11 @@ async def test_returns_none_when_no_entities(self): assert result is None @pytest.mark.asyncio - async def test_raises_error_code_on_exception(self): + async def test_returns_none_on_exception(self): client = MagicMock() client._call_api = AsyncMock(side_effect=Exception("async search failed")) - with pytest.raises(Exception) as exc_info: - await find_existing_policy_async(client, POLICY_NAME, PERSONA_GUID) - assert "ATLAN-PYTHON-500-007" in str(exc_info.value) + result = await find_existing_policy_async(client, POLICY_NAME, PERSONA_GUID) + assert result is None # --------------------------------------------------------------------------- @@ -244,13 +242,11 @@ def test_returns_mock_response_when_duplicate_found(self): body = resp.json() assert body["guidAssignments"][TEMP_GUID] == EXISTING_GUID - def test_propagates_search_error(self): + def test_returns_none_on_search_error(self): client = MagicMock() client._call_api.side_effect = Exception("search failed") req = _make_bulk_request() - with pytest.raises(Exception) as exc_info: - check_for_duplicate_policy(client, req) - assert "ATLAN-PYTHON-500-007" in str(exc_info.value) + assert check_for_duplicate_policy(client, req) is None # --------------------------------------------------------------------------- @@ -283,13 +279,11 @@ async def test_returns_mock_response_when_duplicate_found(self): assert resp.json()["guidAssignments"][TEMP_GUID] == EXISTING_GUID @pytest.mark.asyncio - async def test_propagates_search_error(self): + async def test_returns_none_on_search_error(self): client = MagicMock() client._call_api = AsyncMock(side_effect=Exception("async search failed")) req = _make_bulk_request() - with pytest.raises(Exception) as exc_info: - await check_for_duplicate_policy_async(client, req) - assert "ATLAN-PYTHON-500-007" in str(exc_info.value) + assert await check_for_duplicate_policy_async(client, req) is None # ---------------------------------------------------------------------------