diff --git a/packages/uipath-platform/pyproject.toml b/packages/uipath-platform/pyproject.toml index c3e51954e..512c3a940 100644 --- a/packages/uipath-platform/pyproject.toml +++ b/packages/uipath-platform/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "uipath-platform" -version = "0.2.41" +version = "0.2.42" description = "HTTP client library for programmatic access to UiPath Platform" readme = { file = "README.md", content-type = "text/markdown" } requires-python = ">=3.11" diff --git a/packages/uipath-platform/src/uipath/platform/connections/_connections_service.py b/packages/uipath-platform/src/uipath/platform/connections/_connections_service.py index 3e6aee31c..405b9683c 100644 --- a/packages/uipath-platform/src/uipath/platform/connections/_connections_service.py +++ b/packages/uipath-platform/src/uipath/platform/connections/_connections_service.py @@ -3,7 +3,8 @@ from typing import Any, Dict, List, Optional from urllib.parse import parse_qsl, quote, urlsplit -from httpx import Response +from httpx import ReadTimeout, Response, Timeout +from tenacity import retry_if_exception, retry_if_result from uipath.core.tracing import traced from ..common._base_service import BaseService @@ -12,6 +13,7 @@ from ..common._execution_context import UiPathExecutionContext from ..common._folder_context import header_folder from ..common._models import Endpoint, RequestSpec +from ..common.retry import is_retryable_platform_exception, is_retryable_response from ..orchestrator._folder_service import FolderService from .connections import ( ActivityMetadata, @@ -31,6 +33,30 @@ HEADER_ACTIVITY_JOB_ID = "x-uipath-job-id" _ORIGINATOR_VALUE = "uipath-python" +# Integration Service waits up to ~120 s on the vendor behind an activity (a customer's +# Kusto query, a web read), so the default 30 s client timeout hangs up while it is still +# working. Wait longer than Integration Service does for the response. +_ACTIVITY_TIMEOUT = Timeout(30.0, read=150.0) + + +def _is_retryable_activity_exception(exception: BaseException) -> bool: + """Like `is_retryable_platform_exception`, but never retry a read timeout. + + A read timeout means Integration Service received the activity and may still run + it: sending it again would run a non-idempotent activity twice. + """ + if isinstance(exception, ReadTimeout): + return False + return is_retryable_platform_exception(exception) + + +_ACTIVITY_RETRY = retry_if_exception( + _is_retryable_activity_exception +) | retry_if_result(is_retryable_response) +# BaseService.request with its wait and stop strategies, retrying `_ACTIVITY_RETRY` only. +_request_activity = BaseService.request.retry_with(retry=_ACTIVITY_RETRY) # type: ignore[attr-defined] +_request_activity_async = BaseService.request_async.retry_with(retry=_ACTIVITY_RETRY) # type: ignore[attr-defined] + def _to_param_str(value: Any) -> str: """Convert an activity parameter value to its wire string; booleans become lowercase.""" @@ -678,6 +704,8 @@ def invoke_activity( Raises: ValueError: If required parameters are missing or invalid RuntimeError: If the HTTP request fails or returns an error status + httpx.ReadTimeout: If Integration Service does not answer within 150 s. + It is not retried: Integration Service may still be running the activity. """ connection = self.retrieve(connection_id) folder_key = connection.folder.get("key") if connection.folder else None @@ -686,13 +714,15 @@ def invoke_activity( activity_metadata, connection.id, activity_input, folder_key ) - response = self.request( + response = _request_activity( + self, spec.method, url=spec.endpoint, headers=spec.headers, params=spec.params, json=spec.json, files=spec.files, + timeout=_ACTIVITY_TIMEOUT, ) return response.json() @@ -720,6 +750,8 @@ async def invoke_activity_async( Raises: ValueError: If required parameters are missing or invalid RuntimeError: If the HTTP request fails or returns an error status + httpx.ReadTimeout: If Integration Service does not answer within 150 s. + It is not retried: Integration Service may still be running the activity. """ connection = await self.retrieve_async(connection_id) folder_key = connection.folder.get("key") if connection.folder else None @@ -728,13 +760,15 @@ async def invoke_activity_async( activity_metadata, connection.id, activity_input, folder_key ) - response = await self.request_async( + response = await _request_activity_async( + self, spec.method, url=spec.endpoint, headers=spec.headers, params=spec.params, json=spec.json, files=spec.files, + timeout=_ACTIVITY_TIMEOUT, ) return response.json() diff --git a/packages/uipath-platform/tests/services/test_connections_service.py b/packages/uipath-platform/tests/services/test_connections_service.py index c9fa45441..e7163e44a 100644 --- a/packages/uipath-platform/tests/services/test_connections_service.py +++ b/packages/uipath-platform/tests/services/test_connections_service.py @@ -2,6 +2,7 @@ from unittest.mock import AsyncMock, MagicMock from urllib.parse import unquote_plus +import httpx import pytest from pydantic import ValidationError from pytest_httpx import HTTPXMock @@ -1478,6 +1479,123 @@ def test_invoke_activity_omits_job_id_header_when_unset( sent_request = httpx_mock.get_requests()[1] assert HEADER_ACTIVITY_JOB_ID not in sent_request.headers + def _add_connection_response( + self, httpx_mock: HTTPXMock, connection_id: str + ) -> None: + httpx_mock.add_response( + method="GET", + status_code=200, + json={"id": connection_id, "name": "Test", "elementInstanceId": 1}, + ) + + def test_invoke_activity_waits_longer_than_integration_service( + self, + httpx_mock: HTTPXMock, + service: ConnectionsService, + simple_activity_metadata: ActivityMetadata, + ) -> None: + """The activity call waits past Integration Service's ~120 s vendor budget, not the 30 s default.""" + connection_id = "test-connection-123" + self._add_connection_response(httpx_mock, connection_id) + httpx_mock.add_response(method="POST", status_code=200, json={}) + + service.invoke_activity( + activity_metadata=simple_activity_metadata, + connection_id=connection_id, + activity_input={"body_field1": "x"}, + ) + + connection_request, activity_request = httpx_mock.get_requests() + assert activity_request.extensions["timeout"]["read"] == 150.0 + assert activity_request.extensions["timeout"]["connect"] == 30.0 + assert connection_request.extensions["timeout"]["read"] == 30.0 + + def test_invoke_activity_read_timeout_not_retried( + self, + httpx_mock: HTTPXMock, + service: ConnectionsService, + simple_activity_metadata: ActivityMetadata, + ) -> None: + """A read timeout means Integration Service got the activity: it is not sent again.""" + connection_id = "test-connection-123" + self._add_connection_response(httpx_mock, connection_id) + httpx_mock.add_exception(httpx.ReadTimeout("timed out"), method="POST") + + with pytest.raises(httpx.ReadTimeout): + service.invoke_activity( + activity_metadata=simple_activity_metadata, + connection_id=connection_id, + activity_input={"body_field1": "x"}, + ) + + assert [r.method for r in httpx_mock.get_requests()] == ["GET", "POST"] + + @pytest.mark.anyio + async def test_invoke_activity_async_read_timeout_not_retried( + self, + httpx_mock: HTTPXMock, + service: ConnectionsService, + simple_activity_metadata: ActivityMetadata, + ) -> None: + connection_id = "test-connection-123" + self._add_connection_response(httpx_mock, connection_id) + httpx_mock.add_exception(httpx.ReadTimeout("timed out"), method="POST") + + with pytest.raises(httpx.ReadTimeout): + await service.invoke_activity_async( + activity_metadata=simple_activity_metadata, + connection_id=connection_id, + activity_input={"body_field1": "x"}, + ) + + activity_requests = [r for r in httpx_mock.get_requests() if r.method == "POST"] + assert len(activity_requests) == 1 + assert activity_requests[0].extensions["timeout"]["read"] == 150.0 + + @pytest.mark.anyio + async def test_invoke_activity_async_connect_error_still_retried( + self, + httpx_mock: HTTPXMock, + service: ConnectionsService, + simple_activity_metadata: ActivityMetadata, + ) -> None: + """A failed connect never reached Integration Service, so it is still retried.""" + connection_id = "test-connection-123" + self._add_connection_response(httpx_mock, connection_id) + httpx_mock.add_exception(httpx.ConnectTimeout("timed out"), method="POST") + httpx_mock.add_response(method="POST", status_code=200, json={"ok": True}) + + result = await service.invoke_activity_async( + activity_metadata=simple_activity_metadata, + connection_id=connection_id, + activity_input={"body_field1": "x"}, + ) + + assert result == {"ok": True} + assert len([r for r in httpx_mock.get_requests() if r.method == "POST"]) == 2 + + def test_invoke_activity_retryable_status_still_retried( + self, + httpx_mock: HTTPXMock, + service: ConnectionsService, + simple_activity_metadata: ActivityMetadata, + ) -> None: + connection_id = "test-connection-123" + self._add_connection_response(httpx_mock, connection_id) + httpx_mock.add_response( + method="POST", status_code=429, headers={"retry-after": "0"} + ) + httpx_mock.add_response(method="POST", status_code=200, json={"ok": True}) + + result = service.invoke_activity( + activity_metadata=simple_activity_metadata, + connection_id=connection_id, + activity_input={"body_field1": "x"}, + ) + + assert result == {"ok": True} + assert len([r for r in httpx_mock.get_requests() if r.method == "POST"]) == 2 + def test_invoke_activity_with_body_fields( self, httpx_mock: HTTPXMock, diff --git a/packages/uipath-platform/uv.lock b/packages/uipath-platform/uv.lock index 5c59bdb8e..3ef58613d 100644 --- a/packages/uipath-platform/uv.lock +++ b/packages/uipath-platform/uv.lock @@ -1095,7 +1095,7 @@ dev = [ [[package]] name = "uipath-platform" -version = "0.2.41" +version = "0.2.42" source = { editable = "." } dependencies = [ { name = "anyio" }, diff --git a/packages/uipath/uv.lock b/packages/uipath/uv.lock index e6ae1076e..25fa35fa2 100644 --- a/packages/uipath/uv.lock +++ b/packages/uipath/uv.lock @@ -2762,7 +2762,7 @@ wheels = [ [[package]] name = "uipath-platform" -version = "0.2.41" +version = "0.2.42" source = { editable = "../uipath-platform" } dependencies = [ { name = "anyio" },