diff --git a/README.md b/README.md index 9cb58d86..bdce7968 100644 --- a/README.md +++ b/README.md @@ -137,6 +137,7 @@ For a guided walkthrough — authenticating, running an Actor, and reading its r - **Convenience methods** — `call()`, `wait_for_finish()`, nested resource access, and other shortcuts that hide platform quirks ([Convenience methods](https://docs.apify.com/api/client/python/docs/concepts/convenience-methods)). - **Pluggable HTTP layer** — use the default [Impit](https://github.com/apify/impit)-based client, opt in to the built-in [HTTPX2](https://github.com/pydantic/httpx2) client, or plug in any custom implementation ([HTTP clients](https://docs.apify.com/api/client/python/docs/concepts/custom-http-clients)). - **Structured errors** — every API error surfaces as an [`ApifyApiError`](https://docs.apify.com/api/client/python/reference/class/ApifyApiError) with HTTP-specific subclasses for precise handling ([Error handling](https://docs.apify.com/api/client/python/docs/concepts/error-handling)). +- **Server Actors** — send HTTP requests to an Actor's server interface (formerly Actor Standby), and get the server's URL and OpenAPI specification ([Server Actors](https://docs.apify.com/api/client/python/docs/concepts/server-actors)). - **Debug logging** — opt-in structured logging on the `apify_client` logger captures request URLs, status codes, retry attempts, and more ([Logging](https://docs.apify.com/api/client/python/docs/concepts/logging)). ## Usage examples diff --git a/docs/02_concepts/10_custom_http_clients.mdx b/docs/02_concepts/10_custom_http_clients.mdx index ad05f9f1..be62d5e4 100644 --- a/docs/02_concepts/10_custom_http_clients.mdx +++ b/docs/02_concepts/10_custom_http_clients.mdx @@ -100,7 +100,7 @@ The base classes and the response protocol are available from the `apify_client. The public `call` method provides the shared request pipeline. A concrete transport implements these hooks: -- `send_request(...)` sends one prepared request and returns an `HttpResponse`, error statuses included. It receives the URL with the query parameters already encoded into it, the headers with the client's default headers already merged in, the body, and the timeout for this attempt in seconds. The body is either `bytes`, already serialized and compressed, or an iterator of `bytes` chunks for a streamed body, which is an async iterator in the asynchronous client. The inherited `call` needs it, so every transport adapter has to implement it. Let the HTTP library's exceptions propagate unwrapped, and leave status handling and `ApifyApiError` to `call`. +- `send_request(...)` sends one prepared request and returns an `HttpResponse`, error statuses included. It receives the URL with the query parameters already encoded into it, the headers with the client's default headers already merged in, the body, and the timeout for this attempt in seconds. The body is either `bytes`, already serialized and compressed, or an iterator of `bytes` chunks for a streamed body, which is an async iterator in the asynchronous client. The inherited `call` needs it, so every transport adapter has to implement it. Let the HTTP library's exceptions propagate unwrapped, and leave status handling and `ApifyApiError` to `call`. It also takes `follow_redirects`, which `call` passes as `False` for a request to an Actor's server: return a redirect response as it is then, since the request carries the API token. - `is_retryable_transport_error(exc)` classifies transport failures for the shared retry loop. The default classifies nothing as retryable, so a transport that doesn't override it gives up on the first connection failure. - `is_timeout_error(exc)` identifies transport-specific timeout exceptions for higher-level client features. The default recognizes Python's `TimeoutError`. Timeout classification is independent of retryability, so a timeout the retry loop should retry has to be listed in `is_retryable_transport_error` too. - `close()` or `aclose()` closes resources owned by the transport. The default does nothing, which is correct for a transport that owns no pool or session. @@ -151,7 +151,7 @@ Use the `ApifyClient.wit After that, all API calls made through the client will go through your custom HTTP client. :::warning -If you override `call` itself, your implementation becomes responsible for request preparation, retries, timeouts, API error conversion, logging, and statistics. Implementing the transport hooks and inheriting `call` keeps the shared behavior. +If you override `call` itself, your implementation becomes responsible for request preparation, retries, timeouts, API error conversion, logging, and statistics. It also has to honor the `compression`, `authorization_header`, `raise_for_status`, `retry_status_codes`, `retry_transport_errors`, and `follow_redirects` arguments. `ActorServerClient.request` relies on them to keep the API token away from the Actor's code, to return the server's error responses, and to avoid sending a request to the Actor twice. Implementing the transport hooks and inheriting `call` keeps the shared behavior. ::: If you override `call` and prepare requests with the inherited helpers, a streamed body arrives as a `StreamedRequestBody`. Send its `iter_bytes()` or `aiter_bytes()`, and send the body a second time only when `rewindable` is true, after calling `rewind()`. For what the client streams and how, see [Streaming uploads](./09_streaming.mdx#streaming-uploads). diff --git a/docs/02_concepts/14_server_actors.mdx b/docs/02_concepts/14_server_actors.mdx new file mode 100644 index 00000000..506a3d70 --- /dev/null +++ b/docs/02_concepts/14_server_actors.mdx @@ -0,0 +1,87 @@ +--- +id: server-actors +title: Server Actors +description: Send HTTP requests to Server Actors (formerly Actor Standby), and get their URL and OpenAPI specification. +keywords: [server actor, server interface, standby, actor standby, standby_url, ActorServerClient] +--- + +import Tabs from '@theme/Tabs'; +import TabItem from '@theme/TabItem'; +import CodeBlock from '@theme/CodeBlock'; +import ApiLink from '@theme/ApiLink'; + +import ServerAsyncExample from '!!raw-loader!./code/14_server_async.py'; +import ServerSyncExample from '!!raw-loader!./code/14_server_sync.py'; +import ServerUrlSpecAsyncExample from '!!raw-loader!./code/14_server_url_spec_async.py'; +import ServerUrlSpecSyncExample from '!!raw-loader!./code/14_server_url_spec_sync.py'; + +A [Server Actor](https://docs.apify.com/platform/actors/running/standby) (formerly Actor Standby) runs a web server. You call it over HTTP, and the platform keeps a run ready to answer, so each call skips the run startup. + +:::note Formerly Actor Standby + +Actor Standby is the former name of Server Actors, and it'll be deprecated. The platform documentation and the API still use it, for example in the Actor's `standby_url` and `actor_standby` fields, and in the `actor_standby_*` arguments of `ActorClient.update`. In the client, `ActorClient.standby` is a deprecated alias of `ActorClient.server` that will be removed in v4. + +::: + +To work with a Server Actor, get its server client with `ActorClient.server`. To send a request, use `ActorServerClient.request`. It takes the path on the Actor's server, and optionally the HTTP method, query parameters, headers, and a body: + + + + + {ServerAsyncExample} + + + + + {ServerSyncExample} + + + + +The method returns the server's response. To read a large body in chunks, pass `stream=True` and close the response when you're done with it. To find out which endpoints an Actor serves, see its page in [Apify Store](https://apify.com/store), its README, or its OpenAPI specification. For details, see [Server URL and OpenAPI specification](#server-url-and-openapi-specification). + +## How the request is sent + +Note that: + +- The client takes the server's URL from the Actor's `standby_url`. It fetches the Actor from the API on the first call and reuses the URL for later calls through the same `ActorClient`. If the Actor isn't a Server Actor, the server client raises an `ApifyClientError`. +- The path is always resolved against that URL, so a path can't send the request to another host. +- By default, the request body is sent uncompressed, since the Actor's server may not decode a compressed body. If the server decodes it, pass `compression`, for example `compression='gzip'`. The client then compresses large bodies the same way as for API calls. + +## Server URL and OpenAPI specification + +The Actor server client also gives you the server's URL and its OpenAPI specification: + +- `ActorServerClient.get_url` returns the server's URL. It fetches and reuses the URL the same way as `request`. +- `ActorServerClient.get_openapi_spec` returns the server's OpenAPI specification as a `dict`. It's the [web server schema](https://docs.apify.com/platform/actors/development/actor-definition/web-server-schema) from the Actor's definition, taken from the build that serves the server interface. That's the build set in the Actor's `actor_standby.build` field, or `latest` if none is set. If the Actor doesn't define a web server schema, the method returns `None`. + +The following example prints the server's URL and the paths of its endpoints: + + + + + {ServerUrlSpecAsyncExample} + + + + + {ServerUrlSpecSyncExample} + + + + +## Authentication + +The request carries the client's API token in the `X-Apify-Authorization` header. The platform removes the token before it forwards the request, so the token doesn't reach the Actor's code. The method doesn't follow redirects and returns a redirect response as it is, since following it would send the token to a host the Actor's code picked. + +The headers you pass in `headers` reach the Actor's server as they are. If the Actor reads the `Authorization` header for its own authentication, pass that header in `headers`. + +For details, see [How do I authenticate my requests](https://docs.apify.com/platform/actors/running/standby#how-do-i-authenticate-my-requests) in the platform documentation. + +## Timeouts, retries, and errors + +When no server run is ready, the first request waits for the platform to start one. The method uses the `long` timeout tier by default for that reason. You can pass a different `timeout`, the same way as for API calls. For details, see [Timeouts](./11_timeouts.mdx). + +The method returns the server's response whatever its status, so check `status_code` before you use the body. Only rate-limit (429) and Service Unavailable (503) responses are retried, with the client's retry settings. Once the retries run out, the method returns the last response. The platform sends these when it can't pass the request to a server run, but the Actor's code can return them too, and then a retry sends the request to the Actor again. Network errors and timeouts aren't retried, since the Actor may have already handled the request. For details, see [Retries](./05_retries.mdx). + +The method raises only when it doesn't get a response from the server: an `ApifyApiError` when the API can't return the Actor, for example because it doesn't exist, an `ApifyClientError` when it isn't a Server Actor, or the HTTP library's error when the request fails or times out. For details, see [Error handling](./04_error_handling.mdx). diff --git a/docs/02_concepts/code/10_plugging_in_async.py b/docs/02_concepts/code/10_plugging_in_async.py index 69837312..f911804d 100644 --- a/docs/02_concepts/code/10_plugging_in_async.py +++ b/docs/02_concepts/code/10_plugging_in_async.py @@ -21,6 +21,7 @@ async def send_request( content: bytes | AsyncIterator[bytes] | None, timeout: float | None, stream: bool, + follow_redirects: bool = True, ) -> HttpResponse: """Send one request through the custom transport.""" raise NotImplementedError diff --git a/docs/02_concepts/code/10_plugging_in_sync.py b/docs/02_concepts/code/10_plugging_in_sync.py index 1a8ae9c6..09fab7fe 100644 --- a/docs/02_concepts/code/10_plugging_in_sync.py +++ b/docs/02_concepts/code/10_plugging_in_sync.py @@ -21,6 +21,7 @@ def send_request( content: bytes | Iterator[bytes] | None, timeout: float | None, stream: bool, + follow_redirects: bool = True, ) -> HttpResponse: """Send one request through the custom transport.""" raise NotImplementedError diff --git a/docs/02_concepts/code/14_server_async.py b/docs/02_concepts/code/14_server_async.py new file mode 100644 index 00000000..a47b9619 --- /dev/null +++ b/docs/02_concepts/code/14_server_async.py @@ -0,0 +1,25 @@ +from http import HTTPStatus + +from apify_client import ApifyClientAsync + +TOKEN = 'MY-APIFY-TOKEN' + + +async def main() -> None: + apify_client = ApifyClientAsync(TOKEN) + actor_server_client = apify_client.actor('username/actor-name').server() + + # Send a GET request to the /search endpoint of the Actor's server. The method + # returns error responses too, so check the status first. + response = await actor_server_client.request('/search', params={'q': 'apify'}) + if response.status_code != HTTPStatus.OK: + raise RuntimeError(f'Search failed with status {response.status_code}') + print(response.json()) + + # Send a POST request with a JSON body. + response = await actor_server_client.request( + '/items', + method='POST', + json={'url': 'https://apify.com'}, + ) + print(response.status_code) diff --git a/docs/02_concepts/code/14_server_sync.py b/docs/02_concepts/code/14_server_sync.py new file mode 100644 index 00000000..b31742e8 --- /dev/null +++ b/docs/02_concepts/code/14_server_sync.py @@ -0,0 +1,25 @@ +from http import HTTPStatus + +from apify_client import ApifyClient + +TOKEN = 'MY-APIFY-TOKEN' + + +def main() -> None: + apify_client = ApifyClient(TOKEN) + actor_server_client = apify_client.actor('username/actor-name').server() + + # Send a GET request to the /search endpoint of the Actor's server. The method + # returns error responses too, so check the status first. + response = actor_server_client.request('/search', params={'q': 'apify'}) + if response.status_code != HTTPStatus.OK: + raise RuntimeError(f'Search failed with status {response.status_code}') + print(response.json()) + + # Send a POST request with a JSON body. + response = actor_server_client.request( + '/items', + method='POST', + json={'url': 'https://apify.com'}, + ) + print(response.status_code) diff --git a/docs/02_concepts/code/14_server_url_spec_async.py b/docs/02_concepts/code/14_server_url_spec_async.py new file mode 100644 index 00000000..7acc4cd7 --- /dev/null +++ b/docs/02_concepts/code/14_server_url_spec_async.py @@ -0,0 +1,17 @@ +from apify_client import ApifyClientAsync + +TOKEN = 'MY-APIFY-TOKEN' + + +async def main() -> None: + apify_client = ApifyClientAsync(TOKEN) + actor_server_client = apify_client.actor('username/actor-name').server() + + # Get the URL of the Actor's server. + server_url = await actor_server_client.get_url() + print(server_url) + + # Get the OpenAPI specification and list the server's endpoints. + spec = await actor_server_client.get_openapi_spec() + if spec is not None: + print(list(spec['paths'])) diff --git a/docs/02_concepts/code/14_server_url_spec_sync.py b/docs/02_concepts/code/14_server_url_spec_sync.py new file mode 100644 index 00000000..c34a7dff --- /dev/null +++ b/docs/02_concepts/code/14_server_url_spec_sync.py @@ -0,0 +1,17 @@ +from apify_client import ApifyClient + +TOKEN = 'MY-APIFY-TOKEN' + + +def main() -> None: + apify_client = ApifyClient(TOKEN) + actor_server_client = apify_client.actor('username/actor-name').server() + + # Get the URL of the Actor's server. + server_url = actor_server_client.get_url() + print(server_url) + + # Get the OpenAPI specification and list the server's endpoints. + spec = actor_server_client.get_openapi_spec() + if spec is not None: + print(list(spec['paths'])) diff --git a/docs/03_guides/code/05_custom_http_client_async.py b/docs/03_guides/code/05_custom_http_client_async.py index 944743fa..74a1a283 100644 --- a/docs/03_guides/code/05_custom_http_client_async.py +++ b/docs/03_guides/code/05_custom_http_client_async.py @@ -100,6 +100,7 @@ async def send_request( content: bytes | AsyncIterator[bytes] | None, timeout: float | None, stream: bool, + follow_redirects: bool = True, ) -> HttpResponse: response = await self._session.request( method=method, @@ -107,6 +108,7 @@ async def send_request( headers=headers, data=content, timeout=aiohttp.ClientTimeout(total=timeout), + allow_redirects=follow_redirects, ) adapted_response = AiohttpResponse(response) diff --git a/docs/03_guides/code/05_custom_http_client_sync.py b/docs/03_guides/code/05_custom_http_client_sync.py index d88cce0e..49cfabb4 100644 --- a/docs/03_guides/code/05_custom_http_client_sync.py +++ b/docs/03_guides/code/05_custom_http_client_sync.py @@ -108,6 +108,7 @@ def send_request( content: bytes | Iterator[bytes] | None, timeout: float | None, stream: bool, + follow_redirects: bool = True, ) -> HttpResponse: response = self._session.request( method=method, @@ -116,6 +117,7 @@ def send_request( data=content, timeout=timeout, stream=stream, + allow_redirects=follow_redirects, ) adapted_response = RequestsResponse(response) diff --git a/docs/uv.lock b/docs/uv.lock new file mode 100644 index 00000000..e1c93aca --- /dev/null +++ b/docs/uv.lock @@ -0,0 +1,14 @@ +version = 1 +revision = 5 +requires-python = ">=3.15" + +[options] +exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values. +exclude-newer-span = "PT24H" + +[options.exclude-newer-package] +apify-client = false +apify-fingerprint-datapoints = false +apify-shared = false +crawlee = false +impit = false diff --git a/scripts/_utils.py b/scripts/_utils.py index 48612c54..5be77090 100644 --- a/scripts/_utils.py +++ b/scripts/_utils.py @@ -20,7 +20,7 @@ """Path to the source code of the apify_client package.""" _SUBSTITUTIONS = [ - (re.compile(r'Client'), 'ClientAsync'), + (re.compile(r'Client(?!Error)'), 'ClientAsync'), (re.compile(r'\bsynchronously\b'), 'asynchronously'), (re.compile(r'\bSynchronously\b'), 'Asynchronously'), (re.compile(r'\bsynchronous\b'), 'asynchronous'), diff --git a/src/apify_client/_apify_client.py b/src/apify_client/_apify_client.py index c15add93..aac0da66 100644 --- a/src/apify_client/_apify_client.py +++ b/src/apify_client/_apify_client.py @@ -23,6 +23,8 @@ ActorEnvVarClientAsync, ActorEnvVarCollectionClient, ActorEnvVarCollectionClientAsync, + ActorServerClient, + ActorServerClientAsync, ActorVersionClient, ActorVersionClientAsync, ActorVersionCollectionClient, @@ -188,6 +190,7 @@ def __init__( actor_collection_client=ActorCollectionClient, actor_env_var_client=ActorEnvVarClient, actor_env_var_collection_client=ActorEnvVarCollectionClient, + actor_server_client=ActorServerClient, actor_version_client=ActorVersionClient, actor_version_collection_client=ActorVersionCollectionClient, build_client=BuildClient, @@ -567,6 +570,7 @@ def __init__( actor_collection_client=ActorCollectionClientAsync, actor_env_var_client=ActorEnvVarClientAsync, actor_env_var_collection_client=ActorEnvVarCollectionClientAsync, + actor_server_client=ActorServerClientAsync, actor_version_client=ActorVersionClientAsync, actor_version_collection_client=ActorVersionCollectionClientAsync, build_client=BuildClientAsync, diff --git a/src/apify_client/_client_registry.py b/src/apify_client/_client_registry.py index 5f3c7994..a99d3008 100644 --- a/src/apify_client/_client_registry.py +++ b/src/apify_client/_client_registry.py @@ -13,6 +13,8 @@ ActorEnvVarClientAsync, ActorEnvVarCollectionClient, ActorEnvVarCollectionClientAsync, + ActorServerClient, + ActorServerClientAsync, ActorVersionClient, ActorVersionClientAsync, ActorVersionCollectionClient, @@ -74,6 +76,7 @@ class ClientRegistry: actor_collection_client: type[ActorCollectionClient] actor_env_var_client: type[ActorEnvVarClient] actor_env_var_collection_client: type[ActorEnvVarCollectionClient] + actor_server_client: type[ActorServerClient] actor_version_client: type[ActorVersionClient] actor_version_collection_client: type[ActorVersionCollectionClient] build_client: type[BuildClient] @@ -111,6 +114,7 @@ class ClientRegistryAsync: actor_collection_client: type[ActorCollectionClientAsync] actor_env_var_client: type[ActorEnvVarClientAsync] actor_env_var_collection_client: type[ActorEnvVarCollectionClientAsync] + actor_server_client: type[ActorServerClientAsync] actor_version_client: type[ActorVersionClientAsync] actor_version_collection_client: type[ActorVersionCollectionClientAsync] build_client: type[BuildClientAsync] diff --git a/src/apify_client/_resource_clients/__init__.py b/src/apify_client/_resource_clients/__init__.py index e818ce34..3c978c89 100644 --- a/src/apify_client/_resource_clients/__init__.py +++ b/src/apify_client/_resource_clients/__init__.py @@ -2,6 +2,7 @@ from .actor_collection import ActorCollectionClient, ActorCollectionClientAsync from .actor_env_var import ActorEnvVarClient, ActorEnvVarClientAsync from .actor_env_var_collection import ActorEnvVarCollectionClient, ActorEnvVarCollectionClientAsync +from .actor_server import ActorServerClient, ActorServerClientAsync from .actor_version import ActorVersionClient, ActorVersionClientAsync from .actor_version_collection import ActorVersionCollectionClient, ActorVersionCollectionClientAsync from .build import BuildClient, BuildClientAsync @@ -35,6 +36,8 @@ 'ActorEnvVarClientAsync', 'ActorEnvVarCollectionClient', 'ActorEnvVarCollectionClientAsync', + 'ActorServerClient', + 'ActorServerClientAsync', 'ActorVersionClient', 'ActorVersionClientAsync', 'ActorVersionCollectionClient', diff --git a/src/apify_client/_resource_clients/actor.py b/src/apify_client/_resource_clients/actor.py index 4d524eb2..f5d5a430 100644 --- a/src/apify_client/_resource_clients/actor.py +++ b/src/apify_client/_resource_clients/actor.py @@ -40,6 +40,8 @@ from apify_client._literals import ActorJobStatus, ActorPermissionLevel, RunOrigin from apify_client._resource_clients import ( + ActorServerClient, + ActorServerClientAsync, ActorVersionClient, ActorVersionClientAsync, ActorVersionCollectionClient, @@ -86,6 +88,7 @@ def __init__( resource_path=resource_path, **kwargs, ) + self._server_client: ActorServerClient | None = None def get(self, *, timeout: Timeout = 'short') -> ActorResource | None: """Retrieve the Actor. @@ -627,6 +630,45 @@ def validate_input( return True + def server(self) -> ActorServerClient: + """Retrieve a client for the server interface of this Actor. + + A Server Actor runs a web server, its server interface, that is called over HTTP without starting a run for each + call. Server Actors were formerly called Actor Standby, and `standby` is a deprecated alias of this method. + Every call returns the same client, so the server URL it fetches is reused. + + https://docs.apify.com/platform/actors/running/standby + + Returns: + The client for the server interface of this Actor. + """ + if self._server_client is None: + self._server_client = self._client_registry.actor_server_client( + resource_id=self._resource_id, + resource_path=self._resource_path, + base_url=self._base_url, + api_base_url=self._api_base_url, + public_base_url=self._public_base_url, + http_client=self._http_client, + client_registry=self._client_registry, + ) + return self._server_client + + def standby(self) -> ActorServerClient: + """Retrieve a client for the server interface of this Actor in Standby mode. + + Deprecated alias of `server`, named after Actor Standby, the former name of Server Actors. Will be removed in + v4. Use `server` instead. + + https://docs.apify.com/platform/actors/running/standby + """ + warnings.warn( + '`standby` is deprecated and will be removed in v4. Use `server` instead.', + DeprecationWarning, + stacklevel=3, + ) + return self.server() + @docs_group('Resource clients') class ActorClientAsync(ResourceClientAsync): @@ -648,6 +690,7 @@ def __init__( resource_path=resource_path, **kwargs, ) + self._server_client: ActorServerClientAsync | None = None async def get(self, *, timeout: Timeout = 'short') -> ActorResource | None: """Retrieve the Actor. @@ -1189,3 +1232,42 @@ async def validate_input( ) return True + + def server(self) -> ActorServerClientAsync: + """Retrieve a client for the server interface of this Actor. + + A Server Actor runs a web server, its server interface, that is called over HTTP without starting a run for each + call. Server Actors were formerly called Actor Standby, and `standby` is a deprecated alias of this method. + Every call returns the same client, so the server URL it fetches is reused. + + https://docs.apify.com/platform/actors/running/standby + + Returns: + The client for the server interface of this Actor. + """ + if self._server_client is None: + self._server_client = self._client_registry.actor_server_client( + resource_id=self._resource_id, + resource_path=self._resource_path, + base_url=self._base_url, + api_base_url=self._api_base_url, + public_base_url=self._public_base_url, + http_client=self._http_client, + client_registry=self._client_registry, + ) + return self._server_client + + def standby(self) -> ActorServerClientAsync: + """Retrieve a client for the server interface of this Actor in Standby mode. + + Deprecated alias of `server`, named after Actor Standby, the former name of Server Actors. Will be removed in + v4. Use `server` instead. + + https://docs.apify.com/platform/actors/running/standby + """ + warnings.warn( + '`standby` is deprecated and will be removed in v4. Use `server` instead.', + DeprecationWarning, + stacklevel=3, + ) + return self.server() diff --git a/src/apify_client/_resource_clients/actor_server.py b/src/apify_client/_resource_clients/actor_server.py new file mode 100644 index 00000000..29219305 --- /dev/null +++ b/src/apify_client/_resource_clients/actor_server.py @@ -0,0 +1,436 @@ +from __future__ import annotations + +import re +from typing import TYPE_CHECKING, Any + +from apify_client._docs import docs_group +from apify_client._models import ActorResponse +from apify_client._resource_clients._resource_client import ResourceClient, ResourceClientAsync +from apify_client._utils.http import response_to_dict +from apify_client.errors import ApifyClientError +from apify_client.http_compressors._resolve import resolve_compressor + +if TYPE_CHECKING: + from apify_client._literals import HttpMethod + from apify_client._models import ActorResource + from apify_client.http_clients import HttpResponse + from apify_client.http_compressors import HttpCompressor + from apify_client.types import HttpCompressionAlgorithm, JsonSerializable, QueryParams, Timeout + + +SERVER_RETRY_STATUS_CODES = frozenset({429, 503}) +"""Statuses a server request is retried on. The platform answers with them when it can't pass a request to a server run, +so a retry usually doesn't run the Actor's code again. An Actor that returns one itself gets the request again.""" + +_BUILD_NUMBER_PATTERN = re.compile(r'(\d+)\.(\d+)\.(\d+)') + + +def _parse_build_number(build_number: str | None) -> tuple[int, int, int] | None: + """Split a build number like `0.1.23` into its version and sequence number, or return `None` for anything else.""" + match = _BUILD_NUMBER_PATTERN.fullmatch(build_number or '') + if match is None: + return None + major, minor, sequence = match.groups() + return int(major), int(minor), int(sequence) + + +def _is_older_build(build_number: str | None, target: tuple[int, int, int]) -> bool: + """Return whether `build_number` belongs to the same version as `target` and comes before it. + + Builds are listed newest first and the sequence numbers of a version grow over time, so meeting such a build means + the target build doesn't exist. + """ + parsed = _parse_build_number(build_number) + return parsed is not None and parsed[:2] == target[:2] and parsed < target + + +@docs_group('Resource clients') +class ActorServerClient(ResourceClient): + """Sub-client for the server interface of a specific Actor. + + Provides methods to work with the web server of a Server Actor, e.g. send it requests or get its OpenAPI + specification. Obtain an instance via an appropriate method on the `ActorClient` class. + """ + + def __init__( + self, + *, + resource_id: str, + resource_path: str = 'actors', + **kwargs: Any, + ) -> None: + super().__init__( + resource_id=resource_id, + resource_path=resource_path, + **kwargs, + ) + self._url: str | None = None + """The server URL, cached for `get_url` and `request`, since getting it takes an API call for the Actor.""" + + def get_url(self, *, timeout: Timeout = 'short') -> str: + """Return the URL of the Actor's server. + + The URL is the Actor's `standby_url`, fetched from the API by the first call of this method or + `get_openapi_spec` and reused by later calls on this client. Use it to call the server with another HTTP client, + for example one generated from the server's OpenAPI specification. + + https://docs.apify.com/platform/actors/running/standby + + Args: + timeout: Timeout for the API HTTP request. + + Returns: + The URL of the Actor's server. + + Raises: + ApifyClientError: If the Actor is not a Server Actor. + ApifyApiError: If the Actor does not exist. + """ + if self._url is not None: + return self._url + _, url = self._get_server_actor(timeout=timeout) + return url + + def get_openapi_spec(self, *, timeout: Timeout = 'short') -> dict[str, Any] | None: + """Return the OpenAPI specification of the Actor's server. + + The specification is the web server schema from the Actor's definition, taken from the build that serves the + server interface: the build set in `actor_standby.build`, or `latest` if none is set. Use it to generate an HTTP + client for the server, and point that client at the URL returned by `get_url`. + + https://docs.apify.com/platform/actors/development/actor-definition/web-server-schema + + Args: + timeout: Timeout for each API HTTP request. + + Returns: + The OpenAPI specification of the Actor's server, or `None` if the Actor's definition doesn't include one. + + Raises: + ApifyClientError: If the Actor is not a Server Actor, or the build that serves its server interface doesn't + exist. + ApifyApiError: If the Actor does not exist. + """ + actor, _ = self._get_server_actor(timeout=timeout) + + build_ref = (actor.actor_standby.build if actor.actor_standby else None) or 'latest' + + build_id = None + tagged_build = (actor.tagged_builds or {}).get(build_ref) + target_build_number = _parse_build_number(build_ref) + if tagged_build is not None: + build_id = tagged_build.build_id + elif target_build_number is not None: + # Not a tag, so look `build_ref` up as a build number, which the API can't resolve directly. + builds_client = self._client_registry.build_collection_client( + resource_path='builds', + **self._base_client_kwargs, + ) + for build_item in builds_client.iterate(desc=True, timeout=timeout): + if build_item.build_number == build_ref: + build_id = build_item.id + break + if _is_older_build(build_item.build_number, target_build_number): + break + + build = None + if build_id is not None: + build_client = self._client_registry.build_client( + resource_id=build_id, + base_url=self._base_url, + public_base_url=self._public_base_url, + http_client=self._http_client, + client_registry=self._client_registry, + ) + build = build_client.get(timeout=timeout) + if build is None: + raise ApifyClientError(f'Build {build_ref} of Actor {self.resource_id} does not exist.') + + if build.actor_definition is None: + return None + return build.actor_definition.web_server_schema + + def request( + self, + path: str = '/', + *, + method: HttpMethod = 'GET', + params: QueryParams | None = None, + headers: dict[str, str] | None = None, + data: str | bytes | None = None, + json: JsonSerializable | None = None, + compression: HttpCompressionAlgorithm | HttpCompressor | None = None, + stream: bool = False, + timeout: Timeout = 'long', + ) -> HttpResponse: + """Send an HTTP request to the Actor's server. + + The request goes to the URL returned by `get_url` and carries the client's token in the `X-Apify-Authorization` + header, which the platform consumes, so it never reaches the Actor's code. Headers you pass, `Authorization` + included, reach the Actor's server as they are. The first request to an idle Actor waits for the platform to + start a server run, which is why the timeout defaults to `long`. + + The method returns any response of the server, error statuses included. A rate-limit (429) or Service + Unavailable (503) response is retried with the client's retry settings, and the last one is returned once the + retries run out. The Actor's code can return these statuses too, and then it gets the request again. A redirect + is returned as it is, since following it would send the token to a host the Actor's code picked. Request + bodies are sent uncompressed unless `compression` is set, since the Actor's server may not decode a compressed + body. + + https://docs.apify.com/platform/actors/running/standby + + Args: + path: The path on the Actor's server, for example `/search`. It is always resolved against the server URL, + so the request, and the token with it, cannot be sent to another host. Pass the query in `params`. A + fragment (`#...`) is dropped, since it is never sent to the server. + method: HTTP method of the request. + params: Query parameters of the request, converted the same way as for API calls: booleans become `true` + or `false`, a list of strings becomes one comma-separated value, a datetime becomes an ISO 8601 string + in UTC, and `None` values are left out. + headers: Additional headers of the request. + data: Raw request body. Cannot be used together with `json`. + json: JSON-serializable request body. Cannot be used together with `data`. + compression: Compression algorithm for the request body, as a string literal or an `HttpCompressor` + instance. The body is compressed under the same conditions as for API calls. Set it only if the + Actor's server decodes the resulting `Content-Encoding`. + stream: Whether to return the response with the body unread, so it can be read in chunks with `iter_bytes` + (`aiter_bytes` in the async client). The response holds its connection open until you close it. + timeout: Timeout for the HTTP request. With `stream`, it bounds the wait for the server to start responding. + + Returns: + The response of the Actor's server. + + Raises: + ApifyClientError: If the Actor is not a Server Actor. + ApifyApiError: If the Actor does not exist. + ImportError: If `compression` needs an optional extra that is not installed. + ValueError: If both `data` and `json` are passed, or `compression` is not a supported algorithm. + """ + server_url = self.get_url() + server_path = path.partition('#')[0].lstrip('/') + return self._http_client.call( + url=f'{server_url.rstrip("/")}/{server_path}', + method=method, + params=params, + headers=headers, + data=data, + json=json, + timeout=timeout, + compression=resolve_compressor(compression) if compression is not None else False, + authorization_header='X-Apify-Authorization', + raise_for_status=False, + retry_status_codes=SERVER_RETRY_STATUS_CODES, + retry_transport_errors=False, + follow_redirects=False, + stream=stream, + ) + + def _get_server_actor(self, *, timeout: Timeout) -> tuple[ActorResource, str]: + """Fetch the Actor, check that it is a Server Actor, and cache its server URL. + + Returns: + The Actor and the URL of its server. + """ + response = self._http_client.call(url=self._build_url(), method='GET', timeout=timeout) + actor = ActorResponse.model_validate(response_to_dict(response)).data + url = actor.standby_url + if url is None: + raise ApifyClientError(f'Actor {self.resource_id} is not a Server Actor.') + self._url = url + return actor, url + + +@docs_group('Resource clients') +class ActorServerClientAsync(ResourceClientAsync): + """Sub-client for the server interface of a specific Actor. + + Provides methods to work with the web server of a Server Actor, e.g. send it requests or get its OpenAPI + specification. Obtain an instance via an appropriate method on the `ActorClientAsync` class. + """ + + def __init__( + self, + *, + resource_id: str, + resource_path: str = 'actors', + **kwargs: Any, + ) -> None: + super().__init__( + resource_id=resource_id, + resource_path=resource_path, + **kwargs, + ) + self._url: str | None = None + """The server URL, cached for `get_url` and `request`, since getting it takes an API call for the Actor.""" + + async def get_url(self, *, timeout: Timeout = 'short') -> str: + """Return the URL of the Actor's server. + + The URL is the Actor's `standby_url`, fetched from the API by the first call of this method or + `get_openapi_spec` and reused by later calls on this client. Use it to call the server with another HTTP client, + for example one generated from the server's OpenAPI specification. + + https://docs.apify.com/platform/actors/running/standby + + Args: + timeout: Timeout for the API HTTP request. + + Returns: + The URL of the Actor's server. + + Raises: + ApifyClientError: If the Actor is not a Server Actor. + ApifyApiError: If the Actor does not exist. + """ + if self._url is not None: + return self._url + _, url = await self._get_server_actor(timeout=timeout) + return url + + async def get_openapi_spec(self, *, timeout: Timeout = 'short') -> dict[str, Any] | None: + """Return the OpenAPI specification of the Actor's server. + + The specification is the web server schema from the Actor's definition, taken from the build that serves the + server interface: the build set in `actor_standby.build`, or `latest` if none is set. Use it to generate an HTTP + client for the server, and point that client at the URL returned by `get_url`. + + https://docs.apify.com/platform/actors/development/actor-definition/web-server-schema + + Args: + timeout: Timeout for each API HTTP request. + + Returns: + The OpenAPI specification of the Actor's server, or `None` if the Actor's definition doesn't include one. + + Raises: + ApifyClientError: If the Actor is not a Server Actor, or the build that serves its server interface doesn't + exist. + ApifyApiError: If the Actor does not exist. + """ + actor, _ = await self._get_server_actor(timeout=timeout) + + build_ref = (actor.actor_standby.build if actor.actor_standby else None) or 'latest' + + build_id = None + tagged_build = (actor.tagged_builds or {}).get(build_ref) + target_build_number = _parse_build_number(build_ref) + if tagged_build is not None: + build_id = tagged_build.build_id + elif target_build_number is not None: + # Not a tag, so look `build_ref` up as a build number, which the API can't resolve directly. + builds_client = self._client_registry.build_collection_client( + resource_path='builds', + **self._base_client_kwargs, + ) + async for build_item in builds_client.iterate(desc=True, timeout=timeout): + if build_item.build_number == build_ref: + build_id = build_item.id + break + if _is_older_build(build_item.build_number, target_build_number): + break + + build = None + if build_id is not None: + build_client = self._client_registry.build_client( + resource_id=build_id, + base_url=self._base_url, + public_base_url=self._public_base_url, + http_client=self._http_client, + client_registry=self._client_registry, + ) + build = await build_client.get(timeout=timeout) + if build is None: + raise ApifyClientError(f'Build {build_ref} of Actor {self.resource_id} does not exist.') + + if build.actor_definition is None: + return None + return build.actor_definition.web_server_schema + + async def request( + self, + path: str = '/', + *, + method: HttpMethod = 'GET', + params: QueryParams | None = None, + headers: dict[str, str] | None = None, + data: str | bytes | None = None, + json: JsonSerializable | None = None, + compression: HttpCompressionAlgorithm | HttpCompressor | None = None, + stream: bool = False, + timeout: Timeout = 'long', + ) -> HttpResponse: + """Send an HTTP request to the Actor's server. + + The request goes to the URL returned by `get_url` and carries the client's token in the `X-Apify-Authorization` + header, which the platform consumes, so it never reaches the Actor's code. Headers you pass, `Authorization` + included, reach the Actor's server as they are. The first request to an idle Actor waits for the platform to + start a server run, which is why the timeout defaults to `long`. + + The method returns any response of the server, error statuses included. A rate-limit (429) or Service + Unavailable (503) response is retried with the client's retry settings, and the last one is returned once the + retries run out. The Actor's code can return these statuses too, and then it gets the request again. A redirect + is returned as it is, since following it would send the token to a host the Actor's code picked. Request + bodies are sent uncompressed unless `compression` is set, since the Actor's server may not decode a compressed + body. + + https://docs.apify.com/platform/actors/running/standby + + Args: + path: The path on the Actor's server, for example `/search`. It is always resolved against the server URL, + so the request, and the token with it, cannot be sent to another host. Pass the query in `params`. A + fragment (`#...`) is dropped, since it is never sent to the server. + method: HTTP method of the request. + params: Query parameters of the request, converted the same way as for API calls: booleans become `true` + or `false`, a list of strings becomes one comma-separated value, a datetime becomes an ISO 8601 string + in UTC, and `None` values are left out. + headers: Additional headers of the request. + data: Raw request body. Cannot be used together with `json`. + json: JSON-serializable request body. Cannot be used together with `data`. + compression: Compression algorithm for the request body, as a string literal or an `HttpCompressor` + instance. The body is compressed under the same conditions as for API calls. Set it only if the + Actor's server decodes the resulting `Content-Encoding`. + stream: Whether to return the response with the body unread, so it can be read in chunks with `iter_bytes` + (`aiter_bytes` in the async client). The response holds its connection open until you close it. + timeout: Timeout for the HTTP request. With `stream`, it bounds the wait for the server to start responding. + + Returns: + The response of the Actor's server. + + Raises: + ApifyClientError: If the Actor is not a Server Actor. + ApifyApiError: If the Actor does not exist. + ImportError: If `compression` needs an optional extra that is not installed. + ValueError: If both `data` and `json` are passed, or `compression` is not a supported algorithm. + """ + server_url = await self.get_url() + server_path = path.partition('#')[0].lstrip('/') + return await self._http_client.call( + url=f'{server_url.rstrip("/")}/{server_path}', + method=method, + params=params, + headers=headers, + data=data, + json=json, + timeout=timeout, + compression=resolve_compressor(compression) if compression is not None else False, + authorization_header='X-Apify-Authorization', + raise_for_status=False, + retry_status_codes=SERVER_RETRY_STATUS_CODES, + retry_transport_errors=False, + follow_redirects=False, + stream=stream, + ) + + async def _get_server_actor(self, *, timeout: Timeout) -> tuple[ActorResource, str]: + """Fetch the Actor, check that it is a Server Actor, and cache its server URL. + + Returns: + The Actor and the URL of its server. + """ + response = await self._http_client.call(url=self._build_url(), method='GET', timeout=timeout) + actor = ActorResponse.model_validate(response_to_dict(response)).data + url = actor.standby_url + if url is None: + raise ApifyClientError(f'Actor {self.resource_id} is not a Server Actor.') + self._url = url + return actor, url diff --git a/src/apify_client/http_clients/_base.py b/src/apify_client/http_clients/_base.py index 378c570d..26fa2718 100644 --- a/src/apify_client/http_clients/_base.py +++ b/src/apify_client/http_clients/_base.py @@ -38,7 +38,7 @@ from apify_client.http_compressors._gzip import GzipHttpCompressor if TYPE_CHECKING: - from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Mapping + from collections.abc import AsyncIterator, Awaitable, Callable, Collection, Iterator, Mapping from types import TracebackType from typing import Self @@ -103,6 +103,15 @@ def aiter_bytes(self) -> AsyncIterator[bytes]: """Iterate over the response body in bytes chunks asynchronously.""" +def _redirect_kwargs(*, follow_redirects: bool) -> dict[str, bool]: + """Return the `send_request` arguments for `follow_redirects`. + + The argument is left out when redirects are followed, so a transport written before `send_request` took it still + serves every other request. + """ + return {} if follow_redirects else {'follow_redirects': False} + + @docs_group('HTTP clients') class HttpClientBase: """Shared configuration and utilities for HTTP clients. @@ -312,18 +321,21 @@ def _prepare_request_call( params: QueryParams | None = None, data: str | bytes | bytearray | StreamedBodySource | None = None, json: JsonSerializable | None = None, + compression: HttpCompressor | bool = True, + authorization_header: str | None = None, ) -> tuple[dict[str, str], dict[str, Any] | None, bytes | StreamedRequestBody | None]: """Prepare headers, params, and body for an HTTP request. Merges the client's default headers (including authorization) with per-request headers and serializes a JSON body. Header names are treated case-insensitively and per-request values win over the client - defaults. For JSON bodies, a `Content-Type` header is set unless the caller supplied one. + defaults. For JSON bodies, a `Content-Type` header is set unless the caller supplied one. With + `authorization_header`, the client's `Authorization` value goes out in the header of that name instead. - The body is compressed unless a `Content-Encoding` header is already set, the body is smaller than + The body is compressed with the client's compressor, or with the one passed in `compression`, unless + `compression` is `False`, a `Content-Encoding` header is already set, the body is smaller than `MIN_COMPRESSION_SIZE`, or its content type says the payload is already compressed. A caller-supplied `Content-Encoding` is forwarded verbatim, which is how a pre-encoded body is uploaded - including one in - an encoding the client ships no compressor for. `Content-Encoding: identity` therefore opts a single - request out of compression. + an encoding the client ships no compressor for. A body streamed from a file-like object, an iterable of byte chunks, or a streamed response is wrapped in `StreamedRequestBody` and never compressed, since its chunks go out as they are produced. A caller-supplied @@ -332,7 +344,15 @@ def _prepare_request_call( if json is not None and data is not None: raise ValueError('Cannot pass both "json" and "data" parameters at the same time!') - headers = self._merge_headers(self._headers, headers) + default_headers = self._headers + authorization = self._get_header(default_headers, 'authorization') + if authorization_header is not None and authorization is not None: + without_authorization = { + key: value for key, value in default_headers.items() if key.lower() != 'authorization' + } + default_headers = self._merge_headers(without_authorization, {authorization_header: authorization}) + + headers = self._merge_headers(default_headers, headers) # Dump JSON data to a string so it can be sent as a request body. if json is not None: @@ -351,15 +371,18 @@ def _prepare_request_call( if isinstance(data, (str, bytes, bytearray)): content = data.encode('utf-8') if isinstance(data, str) else bytes(data) + compressor = self._http_compressor if compression is True else compression + # A caller-supplied encoding says the body arrives already encoded, so compressing it here would # both mislabel it and waste the work. if ( - self._get_header(headers, 'content-encoding') is None + compressor is not False + and self._get_header(headers, 'content-encoding') is None and len(content) >= MIN_COMPRESSION_SIZE and is_compressible_content_type(self._get_header(headers, 'content-type')) ): - content = self._http_compressor.compress(content) - headers = self._merge_headers(headers, {'Content-Encoding': self._http_compressor.content_encoding}) + content = compressor.compress(content) + headers = self._merge_headers(headers, {'Content-Encoding': compressor.content_encoding}) elif data is not None: # Without this the request would go out with no body at all, losing the payload without a word. @@ -392,8 +415,9 @@ def _handle_request_exception( *, content: bytes | StreamedRequestBody | None, stop_retrying: Callable[[], None], + retry_transport_errors: bool, ) -> None: - """Stop retrying when an exception is not a retryable transport failure. + """Stop retrying when an exception is not a retryable transport failure, or the request retries none. A failure of a streamed body's source stops retrying too, since sending the body again cannot fix it. The transport reports such a failure as its own error, which may wrap the cause beyond recognition and pass as @@ -409,7 +433,7 @@ def _handle_request_exception( raise source_error from exc return - if not self.is_retryable_transport_error(exc): + if not retry_transport_errors or not self.is_retryable_transport_error(exc): logger.debug('Exception is not retryable', exc_info=exc) stop_retrying() @@ -440,8 +464,13 @@ def _handle_response_status( attempt: int, streamed: bool, stop_retrying: Callable[[], None], + retry_status_codes: Collection[int] | None, + raise_for_status: bool, ) -> bool: - """Record the response status and stop retrying unless it is a server error or a rate limit. + """Record the response status and stop retrying unless the status is retryable. + + A rate limit counts toward the client's statistics only for a request that raises for an error status, since + the statistics track the API's rate limits and only API calls raise. Returns whether the response is a success, so the caller can hand it back instead of raising. """ @@ -449,7 +478,7 @@ def _handle_response_status( logger.debug('Request successful', extra={'status_code': response.status_code}) return True - if response.status_code == HTTPStatus.TOO_MANY_REQUESTS: + if raise_for_status and response.status_code == HTTPStatus.TOO_MANY_REQUESTS: self._statistics.add_rate_limit_error(attempt) logger.debug('Request unsuccessful', extra={'status_code': response.status_code}) @@ -460,15 +489,37 @@ def _handle_response_status( 'within its time limit of about 5 minutes. Write a source that produces data slowly to a temporary ' 'file first, and upload the file.' ) - if ( - response.status_code < HTTPStatus.INTERNAL_SERVER_ERROR - and response.status_code != HTTPStatus.TOO_MANY_REQUESTS - ): + if not self._is_status_code_retryable(response.status_code, retry_status_codes): logger.debug('Status code is not retryable', extra={'status_code': response.status_code}) stop_retrying() return False + @staticmethod + def _is_status_code_retryable(status_code: int, retry_status_codes: Collection[int] | None) -> bool: + """Return whether a status is retried: one in `retry_status_codes`, or else a rate limit or server error.""" + if retry_status_codes is not None: + return status_code in retry_status_codes + return status_code >= HTTPStatus.INTERNAL_SERVER_ERROR or status_code == HTTPStatus.TOO_MANY_REQUESTS + + def _should_return_error_response( + self, + response: HttpResponse, + *, + attempt: int, + content: bytes | StreamedRequestBody | None, + retry_status_codes: Collection[int] | None, + ) -> bool: + """Return whether no retry follows an error response, so a call with `raise_for_status=False` returns it. + + A streamed body that cannot be rewound rules out a retry too. + """ + if not self._is_status_code_retryable(response.status_code, retry_status_codes): + return True + if isinstance(content, StreamedRequestBody) and not content.rewindable: + return True + return attempt > self._max_retries + @docs_group('HTTP clients') class HttpClient(HttpClientBase): @@ -483,6 +534,10 @@ class HttpClient(HttpClientBase): The client's default headers from `self._headers` have to go out with every request, otherwise the `Authorization` header never reaches the API. The inherited `call` merges them into the per-request headers through `_prepare_request_call`, so only a client that replaces `call` has to do it itself. + + A replaced `call` has to accept every argument of the inherited one and honor `compression`, + `authorization_header`, `raise_for_status`, `retry_status_codes`, `retry_transport_errors`, and + `follow_redirects`. """ def __enter__(self) -> Self: @@ -509,6 +564,12 @@ def call( json: JsonSerializable | None = None, stream: bool | None = None, timeout: Timeout = 'medium', + compression: HttpCompressor | bool = True, + authorization_header: str | None = None, + raise_for_status: bool = True, + retry_status_codes: Collection[int] | None = None, + retry_transport_errors: bool = True, + follow_redirects: bool = True, ) -> HttpResponse: """Make an HTTP request with automatic retry and exponential backoff. @@ -527,12 +588,27 @@ def call( timeout: Timeout for the API HTTP request. Use `short`, `medium`, or `long` tier literals for preconfigured timeouts. A `timedelta` overrides it for this call (capped at `timeout_max`), and `no_timeout` disables the timeout entirely. + compression: How to compress the request body. `True` uses the client's compressor, an `HttpCompressor` + is used for this request only, and `False` sends the body uncompressed. + authorization_header: Name of the header that carries the client's `Authorization` value for this + request, which then goes out without the `Authorization` header. A per-request header of either name + still wins. `ActorServerClient.request` sets it to `X-Apify-Authorization`, which the platform + consumes, so the `Authorization` header stays free for the Actor's own authentication. + raise_for_status: Whether an error status raises `ApifyApiError`. With `False`, the error response is + returned once no retry follows, and a rate limit (429) doesn't count toward the client's statistics, + which track the API's rate limits. + retry_status_codes: Status codes to retry. Defaults to rate limits (429) and server errors (5xx). + retry_transport_errors: Whether transport errors, timeouts included, are retried. With `False`, the + first one is raised, since the request may have reached the server before it failed. + follow_redirects: Whether the transport follows redirects. With `False`, a redirect response is handled + like an error status, so headers that carry the token never go to a host the server picked. Returns: The HTTP response object. Raises: - ApifyApiError: If the request fails after all retries or returns a non-retryable error status. + ApifyApiError: If the request fails after all retries or returns a non-retryable error status, unless + `raise_for_status` is `False`. ValueError: If both json and data are provided. TypeError: If data is neither bytes-like nor a value the client can stream. """ @@ -546,6 +622,8 @@ def call( params=params, data=data, json=json, + compression=compression, + authorization_header=authorization_header, ) return self._retry_with_exp_backoff( @@ -559,6 +637,10 @@ def call( content=content, stream=stream, timeout=timeout, + raise_for_status=raise_for_status, + retry_status_codes=retry_status_codes, + retry_transport_errors=retry_transport_errors, + follow_redirects=follow_redirects, ), max_retries=self._max_retries, backoff_base=self._min_delay_between_retries, @@ -579,6 +661,7 @@ def send_request( content: bytes | Iterator[bytes] | None, timeout: float | None, stream: bool, + follow_redirects: bool = True, ) -> HttpResponse: """Send one request through the underlying HTTP library. @@ -594,6 +677,9 @@ def send_request( (an async iterator in the asynchronous client), or None for a request without a body. timeout: Timeout for this attempt in seconds, or None for no timeout at all. stream: Whether to return the response with the body unread, so the caller can stream it. + follow_redirects: Whether to follow redirects. With `False`, return a redirect response as it is. `call` + passes this argument only as `False`, for a request whose headers carry a token that must not follow + a redirect to another host, so a transport that doesn't accept it fails such a request. Returns: The HTTP response object. @@ -644,6 +730,10 @@ def _make_request( content: bytes | StreamedRequestBody | None, stream: bool | None, timeout: Timeout, + raise_for_status: bool, + retry_status_codes: Collection[int] | None, + retry_transport_errors: bool, + follow_redirects: bool, ) -> HttpResponse: """Execute one request attempt through the transport adapter.""" log_context.attempt.set(attempt) @@ -660,16 +750,29 @@ def _make_request( content=content.iter_bytes() if isinstance(content, StreamedRequestBody) else content, timeout=self._compute_timeout(timeout, attempt=attempt), stream=stream or False, + **_redirect_kwargs(follow_redirects=follow_redirects), ) except Exception as exc: - self._handle_request_exception(exc, content=content, stop_retrying=stop_retrying) + self._handle_request_exception( + exc, content=content, stop_retrying=stop_retrying, retry_transport_errors=retry_transport_errors + ) raise finally: if isinstance(content, StreamedRequestBody): content.close_chunks() if self._handle_response_status( - response, attempt=attempt, streamed=isinstance(content, StreamedRequestBody), stop_retrying=stop_retrying + response, + attempt=attempt, + streamed=isinstance(content, StreamedRequestBody), + stop_retrying=stop_retrying, + retry_status_codes=retry_status_codes, + raise_for_status=raise_for_status, + ): + return response + + if not raise_for_status and self._should_return_error_response( + response, attempt=attempt, content=content, retry_status_codes=retry_status_codes ): return response @@ -721,6 +824,12 @@ async def call( json: JsonSerializable | None = None, stream: bool | None = None, timeout: Timeout = 'medium', + compression: HttpCompressor | bool = True, + authorization_header: str | None = None, + raise_for_status: bool = True, + retry_status_codes: Collection[int] | None = None, + retry_transport_errors: bool = True, + follow_redirects: bool = True, ) -> HttpResponse: """Make an HTTP request with automatic retry and exponential backoff. @@ -739,12 +848,27 @@ async def call( timeout: Timeout for the API HTTP request. Use `short`, `medium`, or `long` tier literals for preconfigured timeouts. A `timedelta` overrides it for this call (capped at `timeout_max`), and `no_timeout` disables the timeout entirely. + compression: How to compress the request body. `True` uses the client's compressor, an `HttpCompressor` + is used for this request only, and `False` sends the body uncompressed. + authorization_header: Name of the header that carries the client's `Authorization` value for this + request, which then goes out without the `Authorization` header. A per-request header of either name + still wins. `ActorServerClientAsync.request` sets it to `X-Apify-Authorization`, which the platform + consumes, so the `Authorization` header stays free for the Actor's own authentication. + raise_for_status: Whether an error status raises `ApifyApiError`. With `False`, the error response is + returned once no retry follows, and a rate limit (429) doesn't count toward the client's statistics, + which track the API's rate limits. + retry_status_codes: Status codes to retry. Defaults to rate limits (429) and server errors (5xx). + retry_transport_errors: Whether transport errors, timeouts included, are retried. With `False`, the + first one is raised, since the request may have reached the server before it failed. + follow_redirects: Whether the transport follows redirects. With `False`, a redirect response is handled + like an error status, so headers that carry the token never go to a host the server picked. Returns: The HTTP response object. Raises: - ApifyApiError: If the request fails after all retries or returns a non-retryable error status. + ApifyApiError: If the request fails after all retries or returns a non-retryable error status, unless + `raise_for_status` is `False`. ValueError: If both json and data are provided. TypeError: If data is neither bytes-like nor a value the client can stream. """ @@ -757,13 +881,15 @@ async def call( # offload preparation to a worker thread whenever there is something to compress. A body the # client sends as it is costs less to prepare inline than the hop itself. A `json` body always # hops, as its size is only known once serialized. - if json is not None or self._is_body_worth_compressing(data): + if json is not None or (compression is not False and self._is_body_worth_compressing(data)): prepared_headers, prepared_params, content = await asyncio.to_thread( self._prepare_request_call, headers=headers, params=params, data=data, json=json, + compression=compression, + authorization_header=authorization_header, ) else: prepared_headers, prepared_params, content = self._prepare_request_call( @@ -771,6 +897,8 @@ async def call( params=params, data=data, json=json, + compression=compression, + authorization_header=authorization_header, ) return await self._retry_with_exp_backoff( @@ -784,6 +912,10 @@ async def call( content=content, stream=stream, timeout=timeout, + raise_for_status=raise_for_status, + retry_status_codes=retry_status_codes, + retry_transport_errors=retry_transport_errors, + follow_redirects=follow_redirects, ), max_retries=self._max_retries, backoff_base=self._min_delay_between_retries, @@ -804,6 +936,7 @@ async def send_request( content: bytes | AsyncIterator[bytes] | None, timeout: float | None, stream: bool, + follow_redirects: bool = True, ) -> HttpResponse: """Send one request through the underlying HTTP library. @@ -819,6 +952,9 @@ async def send_request( (an async iterator in the asynchronous client), or None for a request without a body. timeout: Timeout for this attempt in seconds, or None for no timeout at all. stream: Whether to return the response with the body unread, so the caller can stream it. + follow_redirects: Whether to follow redirects. With `False`, return a redirect response as it is. `call` + passes this argument only as `False`, for a request whose headers carry a token that must not follow + a redirect to another host, so a transport that doesn't accept it fails such a request. Returns: The HTTP response object. @@ -869,6 +1005,10 @@ async def _make_request( content: bytes | StreamedRequestBody | None, stream: bool | None, timeout: Timeout, + raise_for_status: bool, + retry_status_codes: Collection[int] | None, + retry_transport_errors: bool, + follow_redirects: bool, ) -> HttpResponse: """Execute one request attempt through the transport adapter.""" log_context.attempt.set(attempt) @@ -885,16 +1025,29 @@ async def _make_request( content=content.aiter_bytes() if isinstance(content, StreamedRequestBody) else content, timeout=self._compute_timeout(timeout, attempt=attempt), stream=stream or False, + **_redirect_kwargs(follow_redirects=follow_redirects), ) except Exception as exc: - self._handle_request_exception(exc, content=content, stop_retrying=stop_retrying) + self._handle_request_exception( + exc, content=content, stop_retrying=stop_retrying, retry_transport_errors=retry_transport_errors + ) raise finally: if isinstance(content, StreamedRequestBody): await content.aclose_chunks() if self._handle_response_status( - response, attempt=attempt, streamed=isinstance(content, StreamedRequestBody), stop_retrying=stop_retrying + response, + attempt=attempt, + streamed=isinstance(content, StreamedRequestBody), + stop_retrying=stop_retrying, + retry_status_codes=retry_status_codes, + raise_for_status=raise_for_status, + ): + return response + + if not raise_for_status and self._should_return_error_response( + response, attempt=attempt, content=content, retry_status_codes=retry_status_codes ): return response diff --git a/src/apify_client/http_clients/_httpx2.py b/src/apify_client/http_clients/_httpx2.py index abf9761f..c8d3eb26 100644 --- a/src/apify_client/http_clients/_httpx2.py +++ b/src/apify_client/http_clients/_httpx2.py @@ -129,6 +129,7 @@ def send_request( content: bytes | Iterator[bytes] | None, timeout: float | None, stream: bool, + follow_redirects: bool = True, ) -> httpx2.Response: request = self._httpx2_client.build_request( method=method, @@ -138,7 +139,7 @@ def send_request( timeout=timeout, ) _restore_explicit_cookie_header(request, headers) - return self._httpx2_client.send(request, stream=stream) + return self._httpx2_client.send(request, stream=stream, follow_redirects=follow_redirects) def _clear_response_cookies(self, _response: httpx2.Response) -> None: """Prevent HTTPX2's shared cookie jar from leaking server cookies into later API requests.""" @@ -236,6 +237,7 @@ async def send_request( content: bytes | AsyncIterator[bytes] | None, timeout: float | None, stream: bool, + follow_redirects: bool = True, ) -> httpx2.Response: request = self._httpx2_async_client.build_request( method=method, @@ -245,7 +247,7 @@ async def send_request( timeout=timeout, ) _restore_explicit_cookie_header(request, headers) - return await self._httpx2_async_client.send(request, stream=stream) + return await self._httpx2_async_client.send(request, stream=stream, follow_redirects=follow_redirects) async def _clear_response_cookies(self, _response: httpx2.Response) -> None: """Prevent HTTPX2's shared cookie jar from leaking server cookies into later API requests.""" diff --git a/src/apify_client/http_clients/_impit.py b/src/apify_client/http_clients/_impit.py index 039fa86b..e518589e 100644 --- a/src/apify_client/http_clients/_impit.py +++ b/src/apify_client/http_clients/_impit.py @@ -94,6 +94,8 @@ def __init__( ) self._impit_client = impit.Client(follow_redirects=True) + # Impit sets redirect handling per client, so a request that must not follow redirects needs its own. + self._impit_client_without_redirects = impit.Client(follow_redirects=False) @override def is_timeout_error(self, exc: Exception) -> bool: @@ -117,6 +119,7 @@ def close(self) -> None: Impit does. """ self._impit_client.__exit__(None, None, None) + self._impit_client_without_redirects.__exit__(None, None, None) @override def send_request( @@ -128,12 +131,14 @@ def send_request( content: bytes | Iterator[bytes] | None, timeout: float | None, stream: bool, + follow_redirects: bool = True, ) -> impit.Response: # Impit treats timeout=None as "use client default (30s)", not "no timeout". # Use a large value (24 hours) to effectively disable the timeout. # This can be removed once impit updates its behaviour: https://github.com/apify/impit/issues/401 impit_timeout = 86_400 if timeout is None else timeout - return self._impit_client.request( + impit_client = self._impit_client if follow_redirects else self._impit_client_without_redirects + return impit_client.request( method=method, url=url, headers=headers, @@ -197,6 +202,8 @@ def __init__( ) self._impit_async_client = impit.AsyncClient(follow_redirects=True) + # See the synchronous client for why a second client is needed. + self._impit_async_client_without_redirects = impit.AsyncClient(follow_redirects=False) @override def is_timeout_error(self, exc: Exception) -> bool: @@ -218,6 +225,7 @@ async def aclose(self) -> None: See `ImpitHttpClient.close` for what Impit's teardown does. """ await self._impit_async_client.__aexit__(None, None, None) + await self._impit_async_client_without_redirects.__aexit__(None, None, None) @override async def send_request( @@ -229,10 +237,12 @@ async def send_request( content: bytes | AsyncIterator[bytes] | None, timeout: float | None, stream: bool, + follow_redirects: bool = True, ) -> impit.Response: # See the synchronous implementation for why None maps to 24 hours. impit_timeout = 86_400 if timeout is None else timeout - return await self._impit_async_client.request( + impit_client = self._impit_async_client if follow_redirects else self._impit_async_client_without_redirects + return await impit_client.request( method=method, url=url, headers=headers, diff --git a/tests/integration/test_actor_server.py b/tests/integration/test_actor_server.py new file mode 100644 index 00000000..9045cb78 --- /dev/null +++ b/tests/integration/test_actor_server.py @@ -0,0 +1,121 @@ +"""Unified tests for the server interface of an Actor (sync + async).""" + +from __future__ import annotations + +import json +import os +from datetime import timedelta +from typing import TYPE_CHECKING + +import pytest + +from .._utils import API_URL_ENV_VAR, get_random_resource_name, maybe_await +from apify_client import ApifyClient +from apify_client._consts import DEFAULT_API_URL + +if TYPE_CHECKING: + from collections.abc import Generator + + from apify_client import ApifyClientAsync + +OPENAPI_SPEC = { + 'openapi': '3.0.0', + 'info': {'title': 'Echo server', 'version': '1.0.0'}, + 'paths': {'/echo': {'post': {'responses': {'200': {'description': 'The echoed request.'}}}}}, +} + +# A web server that answers every request, including the platform's readiness probe, with the request it received, +# under the status in the `status` query parameter, or 200. +ECHO_SERVER_SOURCE = """ +const http = require('http'); + +http.createServer((req, res) => { + let body = ''; + req.on('data', (chunk) => { body += chunk; }); + req.on('end', () => { + const status = Number(new URL(req.url, 'http://localhost').searchParams.get('status')) || 200; + res.writeHead(status, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ method: req.method, url: req.url, headers: req.headers, body })); + }); +}).listen(process.env.ACTOR_WEB_SERVER_PORT); +""" + +DOCKERFILE = """ +FROM node:22-alpine +COPY main.js ./ +CMD ["node", "main.js"] +""" + + +@pytest.fixture(scope='module') +def server_actor_id(api_token: str) -> Generator[str]: + """Create and build a Server Actor running an echo server, and delete it after the module's tests.""" + client = ApifyClient(api_token, api_url=os.getenv(API_URL_ENV_VAR) or DEFAULT_API_URL) + actor_name = get_random_resource_name('server') + actor_json = { + 'actorSpecification': 1, + 'name': actor_name, + 'version': '0.0', + 'usesStandbyMode': True, + 'webServerSchema': OPENAPI_SPEC, + } + actor = client.actors().create( + name=actor_name, + versions=[ + { + 'versionNumber': '0.0', + 'sourceType': 'SOURCE_FILES', + 'buildTag': 'latest', + 'sourceFiles': [ + {'name': 'Dockerfile', 'format': 'TEXT', 'content': DOCKERFILE}, + {'name': 'main.js', 'format': 'TEXT', 'content': ECHO_SERVER_SOURCE}, + {'name': '.actor/actor.json', 'format': 'TEXT', 'content': json.dumps(actor_json)}, + ], + } + ], + actor_standby_is_enabled=True, + actor_standby_memory_mbytes=256, + actor_standby_idle_timeout=timedelta(minutes=1), + ) + actor_client = client.actor(actor.id) + + try: + build = actor_client.build(version_number='0.0') + finished_build = client.build(build.id).wait_for_finish(wait_duration=timedelta(minutes=5)) + assert finished_build is not None + assert finished_build.status == 'SUCCEEDED' + yield actor.id + finally: + actor_client.delete() + + +async def test_actor_server(client: ApifyClient | ApifyClientAsync, server_actor_id: str) -> None: + """The server client returns the Actor's URL and OpenAPI specification, and sends requests to its server.""" + actor = await maybe_await(client.actor(server_actor_id).get()) + assert actor is not None + actor_server_client = client.actor(server_actor_id).server() + + assert await maybe_await(actor_server_client.get_url()) == actor.standby_url + assert await maybe_await(actor_server_client.get_openapi_spec()) == OPENAPI_SPEC + + response = await maybe_await( + actor_server_client.request('/echo', method='POST', params={'q': 'apify'}, json={'limit': 5}) + ) + + assert response.status_code == 200 + echoed = response.json() + assert echoed['method'] == 'POST' + assert echoed['url'] == '/echo?q=apify' + assert json.loads(echoed['body']) == {'limit': 5} + # The platform consumes the client's token, so it never reaches the Actor's code. + assert 'x-apify-authorization' not in echoed['headers'] + assert 'authorization' not in echoed['headers'] + + with_own_authorization = await maybe_await( + actor_server_client.request('/echo', headers={'Authorization': 'Bearer actor-token'}) + ) + assert with_own_authorization.json()['headers']['authorization'] == 'Bearer actor-token' + + error_response = await maybe_await(actor_server_client.request('/echo', params={'status': 404})) + assert error_response.status_code == 404 + assert error_response.json()['url'] == '/echo?status=404' diff --git a/tests/unit/test_actor_server.py b/tests/unit/test_actor_server.py new file mode 100644 index 00000000..75b02508 --- /dev/null +++ b/tests/unit/test_actor_server.py @@ -0,0 +1,738 @@ +from __future__ import annotations + +import gzip +import json +import threading +from datetime import timedelta +from typing import TYPE_CHECKING, Any +from unittest.mock import Mock + +import pytest +from werkzeug import Request, Response + +from apify_client import ApifyClient +from apify_client.errors import ApifyClientError, NotFoundError +from apify_client.http_compressors import GzipHttpCompressor + +if TYPE_CHECKING: + from pytest_httpserver import HTTPServer + + from apify_client import ApifyClientAsync + from apify_client.http_compressors import HttpCompressor + from apify_client.types import HttpCompressionAlgorithm + +pytestmark = pytest.mark.usefixtures('http_client_classes') + +_MOCKED_ACTOR_ID = 'test_actor_id' +_ACTOR_PATH = f'/v2/actors/{_MOCKED_ACTOR_ID}' +_SERVER_PREFIX = '/server' + + +def actor_response(server_url: str | None, **extra_fields: Any) -> dict[str, Any]: + return { + 'data': { + 'id': _MOCKED_ACTOR_ID, + 'userId': 'test_user_id', + 'name': 'test-actor', + 'username': 'test-user', + 'isPublic': False, + 'createdAt': '2026-08-01T10:00:00.000Z', + 'modifiedAt': '2026-08-01T10:00:00.000Z', + 'stats': {}, + 'versions': [], + 'defaultRunOptions': {}, + 'standbyUrl': server_url, + **extra_fields, + } + } + + +def expect_server_actor(httpserver: HTTPServer) -> None: + server_url = httpserver.url_for(_SERVER_PREFIX) + httpserver.expect_request(_ACTOR_PATH, method='GET').respond_with_json(actor_response(server_url)) + + +def server_requests(httpserver: HTTPServer) -> list[Request]: + return [request for request, _ in httpserver.log if request.path.startswith(f'{_SERVER_PREFIX}/')] + + +def test_request_sends_request_to_server_url(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """The request goes to the server URL with the method, path, query, body, and the client's token.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/search', method='POST').respond_with_json({'ok': True}) + + response = ( + sync_client.actor(_MOCKED_ACTOR_ID) + .server() + .request('/search', method='POST', params={'q': 'apify'}, json={'limit': 5}) + ) + + assert response.status_code == 200 + assert response.json() == {'ok': True} + captured = server_requests(httpserver) + assert len(captured) == 1 + assert captured[0].args.to_dict() == {'q': 'apify'} + assert json.loads(captured[0].get_data()) == {'limit': 5} + assert captured[0].headers['X-Apify-Authorization'] == 'Bearer test' + assert 'Authorization' not in captured[0].headers + + +async def test_request_sends_request_to_server_url_async( + httpserver: HTTPServer, async_client: ApifyClientAsync +) -> None: + """Async variant of `test_request_sends_request_to_server_url`.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/search', method='POST').respond_with_json({'ok': True}) + + response = ( + await async_client.actor(_MOCKED_ACTOR_ID) + .server() + .request('/search', method='POST', params={'q': 'apify'}, json={'limit': 5}) + ) + + assert response.status_code == 200 + assert response.json() == {'ok': True} + captured = server_requests(httpserver) + assert len(captured) == 1 + assert captured[0].args.to_dict() == {'q': 'apify'} + assert json.loads(captured[0].get_data()) == {'limit': 5} + assert captured[0].headers['X-Apify-Authorization'] == 'Bearer test' + assert 'Authorization' not in captured[0].headers + + +def test_request_sends_caller_authorization_header(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """An `Authorization` header passed by the caller reaches the server as it is, next to the client's token.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/', method='GET').respond_with_json({'ok': True}) + + sync_client.actor(_MOCKED_ACTOR_ID).server().request(headers={'Authorization': 'Bearer actor-token'}) + + captured = server_requests(httpserver) + assert captured[0].headers['Authorization'] == 'Bearer actor-token' + assert captured[0].headers['X-Apify-Authorization'] == 'Bearer test' + + +def test_request_fetches_server_url_once(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """Later calls on the same client reuse the server URL fetched by the first one.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/', method='GET').respond_with_json({'ok': True}) + actor_client = sync_client.actor(_MOCKED_ACTOR_ID) + + actor_client.server().request() + actor_client.server().request() + + actor_gets = [request for request, _ in httpserver.log if request.path == _ACTOR_PATH] + assert len(actor_gets) == 1 + + +async def test_request_fetches_server_url_once_async(httpserver: HTTPServer, async_client: ApifyClientAsync) -> None: + """Async variant of `test_request_fetches_server_url_once`.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/', method='GET').respond_with_json({'ok': True}) + actor_client = async_client.actor(_MOCKED_ACTOR_ID) + + await actor_client.server().request() + await actor_client.server().request() + + actor_gets = [request for request, _ in httpserver.log if request.path == _ACTOR_PATH] + assert len(actor_gets) == 1 + + +def test_get_url_returns_standby_url(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """`get_url` returns the Actor's `standby_url` and fetches it from the API only once.""" + expect_server_actor(httpserver) + actor_client = sync_client.actor(_MOCKED_ACTOR_ID) + + assert actor_client.server().get_url() == httpserver.url_for(_SERVER_PREFIX) + assert actor_client.server().get_url() == httpserver.url_for(_SERVER_PREFIX) + + actor_gets = [request for request, _ in httpserver.log if request.path == _ACTOR_PATH] + assert len(actor_gets) == 1 + + +async def test_get_url_returns_standby_url_async(httpserver: HTTPServer, async_client: ApifyClientAsync) -> None: + """Async variant of `test_get_url_returns_standby_url`.""" + expect_server_actor(httpserver) + actor_client = async_client.actor(_MOCKED_ACTOR_ID) + + assert await actor_client.server().get_url() == httpserver.url_for(_SERVER_PREFIX) + assert await actor_client.server().get_url() == httpserver.url_for(_SERVER_PREFIX) + + actor_gets = [request for request, _ in httpserver.log if request.path == _ACTOR_PATH] + assert len(actor_gets) == 1 + + +def test_get_url_without_server_mode_raises(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """`get_url` raises for an Actor without a server interface.""" + httpserver.expect_request(_ACTOR_PATH, method='GET').respond_with_json(actor_response(None)) + + with pytest.raises(ApifyClientError, match='is not a Server Actor'): + sync_client.actor(_MOCKED_ACTOR_ID).server().get_url() + + +@pytest.mark.parametrize( + 'path', + [ + pytest.param('//evil.example/steal', id='protocol-relative'), + pytest.param('@evil.example/steal', id='userinfo'), + pytest.param('https://evil.example/steal', id='absolute url'), + ], +) +def test_request_path_stays_on_server_host(httpserver: HTTPServer, sync_client: ApifyClient, path: str) -> None: + """A path that looks like another host is still sent to the server, as a path on it.""" + expect_server_actor(httpserver) + httpserver.no_handler_status_code = 200 + + response = sync_client.actor(_MOCKED_ACTOR_ID).server().request(path) + + assert response.status_code == 200 + assert len(server_requests(httpserver)) == 1 + + +def test_request_drops_path_fragment(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """A fragment in the path is dropped, so `params` still reach the server.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/search').respond_with_json({'ok': True}) + + sync_client.actor(_MOCKED_ACTOR_ID).server().request('/search#results', params={'q': 'apify'}) + + captured = server_requests(httpserver) + assert len(captured) == 1 + assert captured[0].path == f'{_SERVER_PREFIX}/search' + assert captured[0].args.to_dict() == {'q': 'apify'} + + +def test_request_sends_body_uncompressed(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """A body large enough for the client to compress is sent as is, since servers may not decode it.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/items', method='POST').respond_with_json({'ok': True}) + payload = {'items': ['x' * 100] * 100} + + sync_client.actor(_MOCKED_ACTOR_ID).server().request('/items', method='POST', json=payload) + + captured = server_requests(httpserver) + assert len(captured) == 1 + assert 'Content-Encoding' not in captured[0].headers + assert json.loads(captured[0].get_data()) == payload + + +async def test_request_sends_body_uncompressed_async(httpserver: HTTPServer, async_client: ApifyClientAsync) -> None: + """Async variant of `test_request_sends_body_uncompressed`.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/items', method='POST').respond_with_json({'ok': True}) + payload = {'items': ['x' * 100] * 100} + + await async_client.actor(_MOCKED_ACTOR_ID).server().request('/items', method='POST', json=payload) + + captured = server_requests(httpserver) + assert len(captured) == 1 + assert 'Content-Encoding' not in captured[0].headers + assert json.loads(captured[0].get_data()) == payload + + +@pytest.mark.parametrize( + 'compression', + [ + pytest.param('gzip', id='algorithm name'), + pytest.param(GzipHttpCompressor(quality=1), id='compressor instance'), + ], +) +def test_request_compresses_body_when_asked( + httpserver: HTTPServer, sync_client: ApifyClient, compression: HttpCompressionAlgorithm | HttpCompressor +) -> None: + """With `compression` set, a body large enough to compress is sent compressed and labeled with its encoding.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/items', method='POST').respond_with_json({'ok': True}) + payload = {'items': ['x' * 100] * 100} + + sync_client.actor(_MOCKED_ACTOR_ID).server().request('/items', method='POST', json=payload, compression=compression) + + captured = server_requests(httpserver) + assert len(captured) == 1 + assert captured[0].headers.get('Content-Encoding') == 'gzip' + assert json.loads(gzip.decompress(captured[0].get_data())) == payload + + +async def test_request_compresses_body_when_asked_async(httpserver: HTTPServer, async_client: ApifyClientAsync) -> None: + """Async variant of `test_request_compresses_body_when_asked`.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/items', method='POST').respond_with_json({'ok': True}) + payload = {'items': ['x' * 100] * 100} + + await ( + async_client.actor(_MOCKED_ACTOR_ID).server().request('/items', method='POST', json=payload, compression='gzip') + ) + + captured = server_requests(httpserver) + assert len(captured) == 1 + assert captured[0].headers.get('Content-Encoding') == 'gzip' + assert json.loads(gzip.decompress(captured[0].get_data())) == payload + + +def test_request_rejects_unknown_compression(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """An unsupported compression algorithm raises a `ValueError` without sending a request to the server.""" + expect_server_actor(httpserver) + + with pytest.raises(ValueError, match='Unsupported compression algorithm'): + sync_client.actor(_MOCKED_ACTOR_ID).server().request('/items', method='POST', data='x', compression='zip') # ty: ignore[invalid-argument-type] + + assert server_requests(httpserver) == [] + + +def test_request_keeps_caller_content_encoding(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """A `Content-Encoding` header passed by the caller is sent as given, with the body untouched.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/items', method='POST').respond_with_json({'ok': True}) + body = gzip.compress(b'x' * 2000) + + sync_client.actor(_MOCKED_ACTOR_ID).server().request( + '/items', method='POST', headers={'content-encoding': 'gzip'}, data=body + ) + + captured = server_requests(httpserver) + assert len(captured) == 1 + assert captured[0].headers.get('Content-Encoding') == 'gzip' + assert captured[0].get_data() == body + + +def test_request_without_server_mode_raises(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """An Actor without a server interface raises an error before any request to a server.""" + httpserver.expect_request(_ACTOR_PATH, method='GET').respond_with_json(actor_response(None)) + + with pytest.raises(ApifyClientError, match='is not a Server Actor'): + sync_client.actor(_MOCKED_ACTOR_ID).server().request() + + +async def test_request_without_server_mode_raises_async(httpserver: HTTPServer, async_client: ApifyClientAsync) -> None: + """Async variant of `test_request_without_server_mode_raises`.""" + httpserver.expect_request(_ACTOR_PATH, method='GET').respond_with_json(actor_response(None)) + + with pytest.raises(ApifyClientError, match='is not a Server Actor'): + await async_client.actor(_MOCKED_ACTOR_ID).server().request() + + +def test_request_missing_actor_raises_not_found(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """A missing Actor raises `NotFoundError`.""" + httpserver.expect_request(_ACTOR_PATH, method='GET').respond_with_json( + {'error': {'type': 'record-not-found', 'message': 'Actor was not found'}}, status=404 + ) + + with pytest.raises(NotFoundError): + sync_client.actor(_MOCKED_ACTOR_ID).server().request() + + +def test_request_returns_error_response(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """An error response of the server comes back as it is.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/missing', method='GET').respond_with_data('Not here', status=404) + + response = sync_client.actor(_MOCKED_ACTOR_ID).server().request('/missing') + + assert response.status_code == 404 + assert response.text == 'Not here' + + +async def test_request_returns_error_response_async(httpserver: HTTPServer, async_client: ApifyClientAsync) -> None: + """Async variant of `test_request_returns_error_response`.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/missing', method='GET').respond_with_data('Not here', status=404) + + response = await async_client.actor(_MOCKED_ACTOR_ID).server().request('/missing') + + assert response.status_code == 404 + assert response.text == 'Not here' + + +def test_request_returns_redirect_unfollowed(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """A redirect from the server comes back as it is, so the token never reaches the host it points to.""" + expect_server_actor(httpserver) + target = httpserver.url_for('/elsewhere') + httpserver.expect_request(f'{_SERVER_PREFIX}/moved').respond_with_response( + Response(status=302, headers={'Location': target}) + ) + httpserver.expect_request('/elsewhere').respond_with_data('followed') + + response = sync_client.actor(_MOCKED_ACTOR_ID).server().request('/moved') + + assert response.status_code == 302 + assert response.headers['location'] == target + assert not any(request.path == '/elsewhere' for request, _ in httpserver.log) + + +async def test_request_returns_redirect_unfollowed_async( + httpserver: HTTPServer, async_client: ApifyClientAsync +) -> None: + """Async variant of `test_request_returns_redirect_unfollowed`.""" + expect_server_actor(httpserver) + target = httpserver.url_for('/elsewhere') + httpserver.expect_request(f'{_SERVER_PREFIX}/moved').respond_with_response( + Response(status=302, headers={'Location': target}) + ) + httpserver.expect_request('/elsewhere').respond_with_data('followed') + + response = await async_client.actor(_MOCKED_ACTOR_ID).server().request('/moved') + + assert response.status_code == 302 + assert not any(request.path == '/elsewhere' for request, _ in httpserver.log) + + +def test_request_returns_response_unread_with_stream(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """With `stream`, the body can be read in chunks.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/', method='GET').respond_with_data(b'x' * 10_000) + + response = sync_client.actor(_MOCKED_ACTOR_ID).server().request(stream=True) + try: + assert b''.join(response.iter_bytes()) == b'x' * 10_000 + finally: + response.close() + + +async def test_request_returns_response_unread_with_stream_async( + httpserver: HTTPServer, async_client: ApifyClientAsync +) -> None: + """Async variant of `test_request_returns_response_unread_with_stream`.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/', method='GET').respond_with_data(b'x' * 10_000) + + response = await async_client.actor(_MOCKED_ACTOR_ID).server().request(stream=True) + try: + assert b''.join([chunk async for chunk in response.aiter_bytes()]) == b'x' * 10_000 + finally: + await response.aclose() + + +def expect_flaky_server(httpserver: HTTPServer, statuses: list[int]) -> None: + """Answer `/flaky` with each of `statuses` in turn, then with 200.""" + remaining = list(statuses) + + def handler(_request: Request) -> Response: + status = remaining.pop(0) if remaining else 200 + return Response(json.dumps({'status': status}), status=status, content_type='application/json') + + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/flaky').respond_with_handler(handler) + + +def retrying_client(httpserver: HTTPServer, max_retries: int) -> ApifyClient: + api_url = httpserver.url_for('/').removesuffix('/') + return ApifyClient(token='test', api_url=api_url, max_retries=max_retries, min_delay_between_retries=timedelta(0)) + + +@pytest.mark.parametrize('status', [pytest.param(429, id='rate limit'), pytest.param(503, id='service unavailable')]) +def test_request_retries_platform_statuses(httpserver: HTTPServer, status: int) -> None: + """A rate-limit or Service Unavailable response is retried.""" + expect_flaky_server(httpserver, [status, status]) + + response = retrying_client(httpserver, max_retries=2).actor(_MOCKED_ACTOR_ID).server().request('/flaky') + + assert response.status_code == 200 + assert len(server_requests(httpserver)) == 3 + + +def test_request_rate_limit_is_left_out_of_statistics(httpserver: HTTPServer) -> None: + """A 429 from the Actor's server doesn't count toward the client's API rate-limit statistics.""" + expect_flaky_server(httpserver, [429]) + client = retrying_client(httpserver, max_retries=1) + + client.actor(_MOCKED_ACTOR_ID).server().request('/flaky') + + assert client._statistics.rate_limit_errors == {} + + +def test_request_returns_last_retried_response(httpserver: HTTPServer) -> None: + """Once the retries run out, the last retried response comes back.""" + expect_flaky_server(httpserver, [503, 503, 503]) + + response = retrying_client(httpserver, max_retries=1).actor(_MOCKED_ACTOR_ID).server().request('/flaky') + + assert response.status_code == 503 + assert len(server_requests(httpserver)) == 2 + + +def test_request_does_not_retry_other_server_errors(httpserver: HTTPServer) -> None: + """A server error other than 503 comes back after a single attempt.""" + expect_flaky_server(httpserver, [500]) + + client = retrying_client(httpserver, max_retries=2) + response = client.actor(_MOCKED_ACTOR_ID).server().request('/flaky', method='POST', json={'a': 1}) + + assert response.status_code == 500 + assert len(server_requests(httpserver)) == 1 + + +def test_request_does_not_retry_a_timeout(httpserver: HTTPServer) -> None: + """A request that times out is not sent again, since the Actor may have already handled it.""" + release = threading.Event() + + def wait_for_release(_request: Request) -> Response: + release.wait(timeout=5) + return Response('late') + + handler = Mock(side_effect=wait_for_release) + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/slow').respond_with_handler(handler) + client = retrying_client(httpserver, max_retries=2) + + try: + with pytest.raises(Exception) as exc_info: # noqa: PT011 - the error type depends on the transport + client.actor(_MOCKED_ACTOR_ID).server().request('/slow', method='POST', timeout=timedelta(milliseconds=200)) + finally: + release.set() + + assert client.http_client.is_timeout_error(exc_info.value) + handler.assert_called_once() + + +_OPENAPI_SPEC = {'openapi': '3.0.0', 'info': {'title': 'Test server', 'version': '1.0.0'}, 'paths': {}} +_TAGGED_BUILDS = { + 'latest': {'buildId': 'latest_build_id', 'buildNumber': '0.0.2'}, + 'beta': {'buildId': 'beta_build_id', 'buildNumber': '0.0.3'}, +} + + +def build_response(build_id: str, actor_definition: dict[str, Any] | None) -> dict[str, Any]: + return { + 'data': { + 'id': build_id, + 'actId': _MOCKED_ACTOR_ID, + 'userId': 'test_user_id', + 'startedAt': '2026-08-01T10:00:00.000Z', + 'status': 'SUCCEEDED', + 'meta': {'origin': 'API'}, + 'buildNumber': '0.0.1', + 'actorDefinition': actor_definition, + } + } + + +def expect_builds_list(httpserver: HTTPServer, builds: dict[str, str]) -> None: + """Serve the Actor's builds list as one page, so a request for a later offset gets no items.""" + items = [ + { + 'id': build_id, + 'status': 'SUCCEEDED', + 'startedAt': '2026-08-01T10:00:00.000Z', + 'usageTotalUsd': 0.01, + 'buildNumber': build_number, + } + for build_id, build_number in builds.items() + ] + + def handler(request: Request) -> Response: + offset = int(request.args.get('offset', 0)) + page = items[offset:] + data = {'total': len(items), 'offset': offset, 'limit': 1000, 'desc': True, 'count': len(page), 'items': page} + return Response(json.dumps({'data': data}), content_type='application/json') + + httpserver.expect_request(f'{_ACTOR_PATH}/builds', method='GET').respond_with_handler(handler) + + +def expect_server_actor_with_build( + httpserver: HTTPServer, *, standby_build: str | None, build_id: str, actor_definition: dict[str, Any] | None +) -> None: + actor_standby = {'isEnabled': True, 'build': standby_build} + httpserver.expect_request(_ACTOR_PATH, method='GET').respond_with_json( + actor_response(httpserver.url_for(_SERVER_PREFIX), taggedBuilds=_TAGGED_BUILDS, actorStandby=actor_standby) + ) + httpserver.expect_request(f'/v2/actor-builds/{build_id}', method='GET').respond_with_json( + build_response(build_id, actor_definition) + ) + + +@pytest.mark.parametrize( + ('standby_build', 'build_id'), + [ + pytest.param(None, 'latest_build_id', id='no build set uses latest'), + pytest.param('beta', 'beta_build_id', id='build tag'), + ], +) +def test_get_openapi_spec_from_tagged_build( + httpserver: HTTPServer, sync_client: ApifyClient, standby_build: str | None, build_id: str +) -> None: + """The specification comes from the web server schema of the tagged build that serves the server interface.""" + expect_server_actor_with_build( + httpserver, standby_build=standby_build, build_id=build_id, actor_definition={'webServerSchema': _OPENAPI_SPEC} + ) + + assert sync_client.actor(_MOCKED_ACTOR_ID).server().get_openapi_spec() == _OPENAPI_SPEC + + +async def test_get_openapi_spec_from_tagged_build_async(httpserver: HTTPServer, async_client: ApifyClientAsync) -> None: + """Async variant of `test_get_openapi_spec_from_tagged_build`.""" + expect_server_actor_with_build( + httpserver, standby_build='beta', build_id='beta_build_id', actor_definition={'webServerSchema': _OPENAPI_SPEC} + ) + + assert await async_client.actor(_MOCKED_ACTOR_ID).server().get_openapi_spec() == _OPENAPI_SPEC + + +def test_get_openapi_spec_from_build_number(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """A build number set for the server interface is looked up in the Actor's builds.""" + expect_server_actor_with_build( + httpserver, standby_build='0.0.1', build_id='old_build_id', actor_definition={'webServerSchema': _OPENAPI_SPEC} + ) + expect_builds_list(httpserver, {'latest_build_id': '0.0.2', 'old_build_id': '0.0.1'}) + + assert sync_client.actor(_MOCKED_ACTOR_ID).server().get_openapi_spec() == _OPENAPI_SPEC + + +def test_get_openapi_spec_passes_timeout_to_every_api_call( + httpserver: HTTPServer, sync_client: ApifyClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """The timeout of `get_openapi_spec` applies to fetching the Actor, listing its builds, and fetching the build.""" + expect_server_actor_with_build( + httpserver, standby_build='0.0.1', build_id='old_build_id', actor_definition={'webServerSchema': _OPENAPI_SPEC} + ) + expect_builds_list(httpserver, {'old_build_id': '0.0.1'}) + call = Mock(wraps=sync_client.http_client.call) + monkeypatch.setattr(sync_client.http_client, 'call', call) + + sync_client.actor(_MOCKED_ACTOR_ID).server().get_openapi_spec(timeout='long') + + assert [call_args.kwargs['timeout'] for call_args in call.call_args_list] == ['long', 'long', 'long'] + + +def test_request_reuses_server_url_fetched_by_get_openapi_spec( + httpserver: HTTPServer, sync_client: ApifyClient +) -> None: + """A request after `get_openapi_spec` reuses the server URL it fetched.""" + expect_server_actor_with_build( + httpserver, standby_build=None, build_id='latest_build_id', actor_definition={'webServerSchema': _OPENAPI_SPEC} + ) + httpserver.expect_request(f'{_SERVER_PREFIX}/', method='GET').respond_with_json({'ok': True}) + actor_server_client = sync_client.actor(_MOCKED_ACTOR_ID).server() + + actor_server_client.get_openapi_spec() + actor_server_client.request() + + actor_gets = [request for request, _ in httpserver.log if request.path == _ACTOR_PATH] + assert len(actor_gets) == 1 + + +def test_get_url_passes_timeout( + httpserver: HTTPServer, sync_client: ApifyClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """The timeout of `get_url` applies to fetching the Actor.""" + expect_server_actor(httpserver) + call = Mock(wraps=sync_client.http_client.call) + monkeypatch.setattr(sync_client.http_client, 'call', call) + + sync_client.actor(_MOCKED_ACTOR_ID).server().get_url(timeout='medium') + + assert call.call_args.kwargs['timeout'] == 'medium' + + +async def test_get_openapi_spec_from_build_number_async(httpserver: HTTPServer, async_client: ApifyClientAsync) -> None: + """Async variant of `test_get_openapi_spec_from_build_number`.""" + expect_server_actor_with_build( + httpserver, standby_build='0.0.1', build_id='old_build_id', actor_definition={'webServerSchema': _OPENAPI_SPEC} + ) + expect_builds_list(httpserver, {'latest_build_id': '0.0.2', 'old_build_id': '0.0.1'}) + + assert await async_client.actor(_MOCKED_ACTOR_ID).server().get_openapi_spec() == _OPENAPI_SPEC + + +@pytest.mark.parametrize( + 'actor_definition', + [ + pytest.param({'name': 'test-actor'}, id='no web server schema'), + pytest.param(None, id='no actor definition'), + ], +) +def test_get_openapi_spec_without_schema_returns_none( + httpserver: HTTPServer, sync_client: ApifyClient, actor_definition: dict[str, Any] | None +) -> None: + """A build without a web server schema has no specification to return.""" + expect_server_actor_with_build( + httpserver, standby_build=None, build_id='latest_build_id', actor_definition=actor_definition + ) + + assert sync_client.actor(_MOCKED_ACTOR_ID).server().get_openapi_spec() is None + + +def test_get_openapi_spec_unknown_build_raises(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """A server interface build that is neither a tag nor a listed build number raises an error.""" + expect_server_actor_with_build(httpserver, standby_build='9.9.9', build_id='unused_build_id', actor_definition=None) + expect_builds_list(httpserver, {'latest_build_id': '0.0.2'}) + + with pytest.raises(ApifyClientError, match='does not exist'): + sync_client.actor(_MOCKED_ACTOR_ID).server().get_openapi_spec() + + +def test_get_openapi_spec_missing_build_number_stops_at_older_build( + httpserver: HTTPServer, sync_client: ApifyClient +) -> None: + """The build number lookup stops at an older build of the same version instead of listing every build.""" + builds = [{'id': f'build_{number}', 'buildNumber': f'0.0.{number}'} for number in (5, 3, 2, 1)] + + def one_build_per_page(request: Request) -> Response: + offset = int(request.args.get('offset', 0)) + page = [ + {'status': 'SUCCEEDED', 'startedAt': '2026-08-01T10:00:00.000Z', 'usageTotalUsd': 0.01, **build} + for build in builds[offset : offset + 1] + ] + data = {'total': len(builds), 'offset': offset, 'limit': 1, 'desc': True, 'count': len(page), 'items': page} + return Response(json.dumps({'data': data}), content_type='application/json') + + handler = Mock(side_effect=one_build_per_page) + expect_server_actor_with_build(httpserver, standby_build='0.0.4', build_id='unused_build_id', actor_definition=None) + httpserver.expect_request(f'{_ACTOR_PATH}/builds', method='GET').respond_with_handler(handler) + + with pytest.raises(ApifyClientError, match='does not exist'): + sync_client.actor(_MOCKED_ACTOR_ID).server().get_openapi_spec() + + assert handler.call_count == 2 + + +def test_get_openapi_spec_unknown_tag_raises_without_listing_builds( + httpserver: HTTPServer, sync_client: ApifyClient +) -> None: + """A server interface build that is neither a tag nor a build number raises without listing the builds.""" + expect_server_actor_with_build( + httpserver, standby_build='nightly', build_id='unused_build_id', actor_definition=None + ) + + with pytest.raises(ApifyClientError, match='Build nightly of Actor'): + sync_client.actor(_MOCKED_ACTOR_ID).server().get_openapi_spec() + + assert not any(request.path.endswith('/builds') for request, _ in httpserver.log) + + +def test_get_openapi_spec_without_server_mode_raises(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """`get_openapi_spec` raises for an Actor without a server interface.""" + httpserver.expect_request(_ACTOR_PATH, method='GET').respond_with_json(actor_response(None)) + + with pytest.raises(ApifyClientError, match='is not a Server Actor'): + sync_client.actor(_MOCKED_ACTOR_ID).server().get_openapi_spec() + + +def test_standby_alias_warns_and_returns_server_client(httpserver: HTTPServer, sync_client: ApifyClient) -> None: + """The deprecated `standby` emits a `DeprecationWarning` pointing at the caller and returns the server client.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/search', method='GET').respond_with_json({'ok': True}) + actor_client = sync_client.actor(_MOCKED_ACTOR_ID) + + with pytest.warns(DeprecationWarning, match='`standby` is deprecated') as record: + standby_client = actor_client.standby() + + assert record[0].filename == __file__ + assert standby_client is actor_client.server() + assert standby_client.request('/search').json() == {'ok': True} + + +async def test_standby_alias_warns_and_returns_server_client_async( + httpserver: HTTPServer, async_client: ApifyClientAsync +) -> None: + """Async variant of `test_standby_alias_warns_and_returns_server_client`.""" + expect_server_actor(httpserver) + httpserver.expect_request(f'{_SERVER_PREFIX}/search', method='GET').respond_with_json({'ok': True}) + actor_client = async_client.actor(_MOCKED_ACTOR_ID) + + with pytest.warns(DeprecationWarning, match='`standby` is deprecated') as record: + standby_client = actor_client.standby() + + assert record[0].filename == __file__ + assert standby_client is actor_client.server() + response = await standby_client.request('/search') + assert response.json() == {'ok': True} diff --git a/tests/unit/test_http_clients.py b/tests/unit/test_http_clients.py index 00d07b9a..8f37b7b3 100644 --- a/tests/unit/test_http_clients.py +++ b/tests/unit/test_http_clients.py @@ -49,6 +49,10 @@ class ConcreteHttpClient(HttpClient): """Minimal concrete HttpClient for testing base class helpers, relying on the hook defaults.""" +class ConcreteHttpClientAsync(HttpClientAsync): + """Minimal concrete HttpClientAsync for testing base class helpers, relying on the hook defaults.""" + + class CallOnlyHttpClient(HttpClient): """Custom client written against the pre-hook contract, overriding `call` and nothing else.""" @@ -539,6 +543,30 @@ def test_transient_transport_error_is_retried() -> None: assert request.call_count == 3 +def test_transport_error_is_not_retried_with_retry_transport_errors_false() -> None: + """With `retry_transport_errors=False`, a transient transport failure is raised on the first attempt.""" + client = ImpitHttpClient(token='test_token', max_retries=2, min_delay_between_retries=timedelta(0)) + request = Mock(side_effect=impit.TimeoutException('timeout')) + client._impit_client = Mock(request=request) + + with pytest.raises(impit.TimeoutException): + client.call(method='POST', url='https://api.test.com/endpoint', retry_transport_errors=False) + + request.assert_called_once() + + +async def test_transport_error_is_not_retried_with_retry_transport_errors_false_async() -> None: + """The async client also raises the first transport failure with `retry_transport_errors=False`.""" + client = ImpitHttpClientAsync(token='test_token', max_retries=2, min_delay_between_retries=timedelta(0)) + request = AsyncMock(side_effect=impit.TimeoutException('timeout')) + client._impit_async_client = Mock(request=request) + + with pytest.raises(impit.TimeoutException): + await client.call(method='POST', url='https://api.test.com/endpoint', retry_transport_errors=False) + + request.assert_awaited_once() + + def test_httpx2_permanent_transport_error_is_not_retried(monkeypatch: pytest.MonkeyPatch) -> None: """The HTTPX2 adapter feeds the same fail-fast classification into the shared pipeline.""" with Httpx2HttpClient(token='test_token', min_delay_between_retries=timedelta(0)) as client: @@ -668,6 +696,86 @@ def test_error_response_read_failure_on_non_retryable_status_is_not_retried() -> response.close.assert_called_once() +def error_response(status_code: int) -> Mock: + return Mock(status_code=status_code, text='', headers={}, json=Mock(side_effect=ValueError), aread=AsyncMock()) + + +def test_call_returns_a_non_retryable_error_response_with_raise_for_status_false( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """With `raise_for_status=False`, a non-retryable error response comes back after one attempt, body unread.""" + client = ConcreteHttpClient(min_delay_between_retries=timedelta(0)) + response = error_response(404) + send_request = Mock(return_value=response) + monkeypatch.setattr(client, 'send_request', send_request) + + result = client.call(method='GET', url='https://api.test.com/endpoint', stream=True, raise_for_status=False) + + assert result is response + send_request.assert_called_once() + response.read.assert_not_called() + + +def test_call_returns_the_last_error_response_once_the_retries_run_out(monkeypatch: pytest.MonkeyPatch) -> None: + """With `raise_for_status=False`, a retryable error response is retried, and the last one comes back.""" + client = ConcreteHttpClient(max_retries=2, min_delay_between_retries=timedelta(0)) + responses = [error_response(500) for _ in range(3)] + send_request = Mock(side_effect=responses) + monkeypatch.setattr(client, 'send_request', send_request) + + result = client.call(method='GET', url='https://api.test.com/endpoint', raise_for_status=False) + + assert result is responses[-1] + assert send_request.call_count == 3 + + +async def test_call_returns_the_last_error_response_once_the_retries_run_out_async( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The async client also returns the last retried error response with `raise_for_status=False`.""" + client = ConcreteHttpClientAsync(max_retries=2, min_delay_between_retries=timedelta(0)) + responses = [error_response(500) for _ in range(3)] + send_request = AsyncMock(side_effect=responses) + monkeypatch.setattr(client, 'send_request', send_request) + + result = await client.call(method='GET', url='https://api.test.com/endpoint', raise_for_status=False) + + assert result is responses[-1] + assert send_request.await_count == 3 + + +def test_call_returns_a_retryable_error_response_for_a_body_it_cannot_resend(monkeypatch: pytest.MonkeyPatch) -> None: + """With `raise_for_status=False`, a streamed body that cannot be rewound gets its error response back at once.""" + client = ConcreteHttpClient(min_delay_between_retries=timedelta(0)) + response = error_response(503) + send_request = Mock(return_value=response) + monkeypatch.setattr(client, 'send_request', send_request) + + result = client.call( + method='POST', url='https://api.test.com/endpoint', data=iter([b'chunk']), raise_for_status=False + ) + + assert result is response + send_request.assert_called_once() + + +def test_call_retries_only_the_statuses_in_retry_status_codes(monkeypatch: pytest.MonkeyPatch) -> None: + """`retry_status_codes` replaces the default retried statuses.""" + client = ConcreteHttpClient(min_delay_between_retries=timedelta(0)) + send_request = Mock( + side_effect=[error_response(503), error_response(503), Mock(status_code=200), error_response(500)] + ) + monkeypatch.setattr(client, 'send_request', send_request) + + response = client.call(method='GET', url='https://api.test.com/endpoint', retry_status_codes=[503]) + assert response.status_code == 200 + assert send_request.call_count == 3 + + with pytest.raises(ApifyApiError): + client.call(method='GET', url='https://api.test.com/endpoint', retry_status_codes=[503]) + assert send_request.call_count == 4 + + @pytest.fixture( params=[ pytest.param((GzipHttpCompressor(), 'gzip', gzip.decompress), id='gzip'), @@ -783,6 +891,43 @@ def test_prepare_request_call_skips_compression_below_threshold(compressor_case: assert not any(key.lower() == 'content-encoding' for key in headers) +def test_prepare_request_call_skips_compression_when_disabled() -> None: + """With `compression=False`, a body above the threshold is sent verbatim with no `Content-Encoding`.""" + client = ConcreteHttpClient() + body = b'x' * MIN_COMPRESSION_SIZE + + headers, _params, data = client._prepare_request_call(data=body, compression=False) + + assert data == body + assert not any(key.lower() == 'content-encoding' for key in headers) + + +def test_prepare_request_call_uses_per_request_compressor() -> None: + """A compressor passed in `compression` is used for the request in place of the client's compressor.""" + client = ConcreteHttpClient(http_compressor=GzipHttpCompressor()) + body = b'x' * MIN_COMPRESSION_SIZE + + headers, _params, data = client._prepare_request_call(data=body, compression=BrotliHttpCompressor()) + + assert headers['Content-Encoding'] == 'br' + assert brotli.decompress(data) == body + + +def test_prepare_request_call_moves_authorization_to_the_named_header() -> None: + """`authorization_header` sends the client's `Authorization` value in that header, and a caller's own one wins.""" + client = ConcreteHttpClient(token='test_token') + + headers, _params, _data = client._prepare_request_call(authorization_header='X-Apify-Authorization') + own_headers, _params, _data = client._prepare_request_call( + headers={'authorization': 'Basic own'}, authorization_header='X-Apify-Authorization' + ) + + assert headers['X-Apify-Authorization'] == 'Bearer test_token' + assert 'Authorization' not in headers + assert own_headers['X-Apify-Authorization'] == 'Bearer test_token' + assert own_headers['authorization'] == 'Basic own' + + def test_prepare_request_call_compresses_bytearray_data(compressor_case: tuple) -> None: """A `bytearray` body above the threshold is compressed without error (regression: needs bytes conversion).""" compressor, content_encoding, decompress = compressor_case @@ -1219,6 +1364,22 @@ async def test_async_call_skips_thread_offload_for_a_body_it_cannot_compress( spy.assert_not_called() +async def test_async_call_skips_thread_offload_when_compression_is_disabled( + http_client_async_class: type[HttpClientAsync], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A raw body is prepared inline when compression is disabled, whatever its size.""" + client = http_client_async_class(token='test_token') + monkeypatch.setattr(client, 'send_request', AsyncMock(return_value=Mock(status_code=200))) + spy = _to_thread_spy(monkeypatch) + + await client.call( + method='PUT', url='https://api.test.com/endpoint', data=b'x' * MIN_COMPRESSION_SIZE, compression=False + ) + + spy.assert_not_called() + + class StreamingTransport(HttpClient): """A transport that drains every body it is handed and answers from a script of statuses and exceptions.""" @@ -1240,8 +1401,9 @@ def send_request( content: bytes | Iterator[bytes] | None, timeout: float | None, stream: bool, + follow_redirects: bool = True, ) -> HttpResponse: - _ = method, url, headers, timeout, stream + _ = method, url, headers, timeout, stream, follow_redirects self.attempts_with_iterator.append(not isinstance(content, (bytes, type(None)))) self.bodies.append(b''.join(content) if isinstance(content, Iterator) else content) outcome = self._outcomes.pop(0) @@ -1271,8 +1433,9 @@ async def send_request( content: bytes | AsyncIterator[bytes] | None, timeout: float | None, stream: bool, + follow_redirects: bool = True, ) -> HttpResponse: - _ = method, url, headers, timeout, stream + _ = method, url, headers, timeout, stream, follow_redirects self.attempts_with_iterator.append(not isinstance(content, (bytes, type(None)))) self.bodies.append( b''.join([chunk async for chunk in content]) if isinstance(content, AsyncIterator) else content diff --git a/tests/unit/test_pluggable_http_client.py b/tests/unit/test_pluggable_http_client.py index 2034b832..1492a767 100644 --- a/tests/unit/test_pluggable_http_client.py +++ b/tests/unit/test_pluggable_http_client.py @@ -31,10 +31,11 @@ ) if TYPE_CHECKING: - from collections.abc import Iterator + from collections.abc import Collection, Iterator from pytest_httpserver import HTTPServer + from apify_client.http_compressors import HttpCompressor from apify_client.types import QueryParams, StreamedBodySource, Timeout @@ -99,6 +100,12 @@ def call( json: Any = None, stream: bool | None = None, timeout: Timeout = 'medium', + compression: HttpCompressor | bool = True, + authorization_header: str | None = None, + raise_for_status: bool = True, + retry_status_codes: Collection[int] | None = None, + retry_transport_errors: bool = True, + follow_redirects: bool = True, ) -> HttpResponse: self.calls.append( { @@ -110,6 +117,12 @@ def call( 'json': json, 'stream': stream, 'timeout': timeout, + 'compression': compression, + 'authorization_header': authorization_header, + 'raise_for_status': raise_for_status, + 'retry_status_codes': retry_status_codes, + 'retry_transport_errors': retry_transport_errors, + 'follow_redirects': follow_redirects, } ) return _make_fake_response() @@ -133,6 +146,12 @@ async def call( json: Any = None, stream: bool | None = None, timeout: Timeout = 'medium', + compression: HttpCompressor | bool = True, + authorization_header: str | None = None, + raise_for_status: bool = True, + retry_status_codes: Collection[int] | None = None, + retry_transport_errors: bool = True, + follow_redirects: bool = True, ) -> HttpResponse: self.calls.append( { @@ -144,6 +163,12 @@ async def call( 'json': json, 'stream': stream, 'timeout': timeout, + 'compression': compression, + 'authorization_header': authorization_header, + 'raise_for_status': raise_for_status, + 'retry_status_codes': retry_status_codes, + 'retry_transport_errors': retry_transport_errors, + 'follow_redirects': follow_redirects, } ) return _make_fake_response() @@ -192,8 +217,9 @@ def send_request( content: bytes | Iterator[bytes] | None, timeout: float | None, stream: bool, + follow_redirects: bool = True, ) -> HttpResponse: - _ = stream + _ = stream, follow_redirects return _stdlib_fetch(method=method, url=url, headers=headers, content=content, timeout=timeout) @@ -209,8 +235,9 @@ async def send_request( content: bytes | AsyncIterator[bytes] | None, timeout: float | None, stream: bool, + follow_redirects: bool = True, ) -> HttpResponse: - _ = stream + _ = stream, follow_redirects if isinstance(content, AsyncIterator): content = b''.join([chunk async for chunk in content]) return await asyncio.to_thread( @@ -665,6 +692,37 @@ def test_custom_http_client_keeps_differently_cased_authorization() -> None: assert http_client._headers['authorization'] == 'Bearer client_token' +def test_transport_without_follow_redirects_refuses_unfollowed_requests() -> None: + """A transport without the `follow_redirects` argument sends other requests but fails an unfollowed one.""" + send_request = Mock(return_value=_make_fake_response()) + + class LegacyHttpClient(HttpClient): + # The signature `send_request` had before it took `follow_redirects`. + def send_request( # ty: ignore[invalid-method-override] + self, + *, + method: str, + url: str, + headers: dict[str, str], + content: bytes | Iterator[bytes] | None, + timeout: float | None, + stream: bool, + ) -> HttpResponse: + return send_request( + method=method, url=url, headers=headers, content=content, timeout=timeout, stream=stream + ) + + http_client = LegacyHttpClient() + + http_client.call(url='https://api.example/v2/actors', method='GET') + send_request.assert_called_once() + send_request.reset_mock() + + with pytest.raises(TypeError, match='follow_redirects'): + http_client.call(url='https://server.example/', method='GET', follow_redirects=False) + send_request.assert_not_called() + + def test_hooks_only_client_retries_and_sends_default_headers(httpserver: HTTPServer) -> None: """A client implementing only `send_request` inherits header merging and 5xx retries from the shared `call`.""" auth_headers: list[str | None] = []