diff --git a/CHANGELOG.md b/CHANGELOG.md index dbec268..007f850 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +## 0.13.0 + +- Synced with Apify OpenAPI spec `v2-2026-11-06T140128Z`: `ActorVersion::getActorPath()` and `Actor::getStandbyUrl()`. +- Added `ActorClient::server()` for Server Actors (formerly Actor Standby): `request()`, `getUrl()` and `getOpenApiSpec()`. +- Added `NoRedirectHttpClientInterface`, implemented by `GuzzleHttpClient` and needed by `request()`. +- **Breaking:** `PaginationList::getCount()` of a dataset items page is the scanned rows, or the items returned when larger. +- Offset iterators, dataset items included, step by the page's `count`, so Store and dataset filters neither repeat nor skip entries. +- Documented the page size of iteration without `$chunkSize`. + ## 0.12.0 - Synced with Apify OpenAPI spec `v2-2026-11-01T214220Z`. Response models return the new fields as diff --git a/README.md b/README.md index e13d6ae..95acc5d 100644 --- a/README.md +++ b/README.md @@ -151,6 +151,9 @@ $psr18 = new \GuzzleHttp\Client(['timeout' => 120]); // any Psr\Http\Message Cli $client = new ApifyClient(token: 'my-api-token', httpClient: new Psr18HttpClient($psr18)); ``` +`ActorClient::server()->request()` needs a transport that can return a redirect as it is, so it throws with `Psr18HttpClient` +or any custom transport that does not also implement `NoRedirectHttpClientInterface`. See [Server Actors](docs/actors.md). + ## Error handling Methods that fetch a single resource by ID return `null` when the resource does not exist (rather diff --git a/docs/README.md b/docs/README.md index 54f5223..719c256 100644 --- a/docs/README.md +++ b/docs/README.md @@ -20,7 +20,7 @@ Every class is under the `Apify\Client\` PSR-4 root. Use these when writing `use | `Apify\Client\` | The entry point and version constants. | `ApifyClient`, `Version` | | `Apify\Client\Model\` | Response models returned by the API. | `RequestQueueRequest`, `ActorEnvVar`, `Dataset`, `ActorRun`, `PaginationList` | | `Apify\Client\Options\` | Option objects (all the `*Options` classes) **and** enums. | e.g. `ActorListOptions`, `ActorStartOptions`, `TaskStartOptions`, `RunListOptions`, `RunIterateDatasetItemsOptions`, `RunResurrectOptions`, `StorageListOptions`, `StoreListOptions`, `DatasetListItemsOptions`, `ListKeysOptions`, `GetRecordOptions`, `ListRequestsOptions`, `BatchAddRequestsOptions`, `PaginateRequestsOptions`, `LogOptions`, `DownloadItemsFormat` — see [options reference](options.md) for the full list | -| `Apify\Client\Http\` | The replaceable transport, its adapters, and compressors. | `HttpClientInterface`, `GuzzleHttpClient`, `Psr18HttpClient`, `HttpCompressorInterface`, `BrotliHttpCompressor`, `GzipHttpCompressor` | +| `Apify\Client\Http\` | The replaceable transport, its adapters, and compressors. | `HttpClientInterface`, `NoRedirectHttpClientInterface`, `GuzzleHttpClient`, `Psr18HttpClient`, `HttpCompressorInterface`, `BrotliHttpCompressor`, `GzipHttpCompressor` | | `Apify\Client\Exception\` | Exceptions thrown by the client. | `ApifyApiException` and its status-specific subclasses (`NotFoundException`, `RateLimitException`, ...), `TransportException` | For example, to add requests to a queue you would import the model and (optionally) the batch options: @@ -133,6 +133,16 @@ $builds = $client->builds()->list(new ListOptions(limit: 50, desc: true)); `getOffset()`, `getLimit()`, `getCount()`, `isDesc()` and `getItems()`. Within-storage listers (`listKeys`, `listHead`) return their own page/head containers instead. +On a page of `DatasetClient::listItems()`, `getCount()` is the number of dataset rows the API scanned for the +page, or the number of items returned when that is larger. Filters such as `skipEmpty` or `clean` can leave it +above `count($page->getItems())`. The next page starts at `offset + min(count, limit)`; the cap matters only with +`unwind`, which can return more items than the rows scanned. + +The `iterate*()` helpers take a `$chunkSize`, the number of entries requested per API call. Without it, each call +gets the API's default page size, capped by the remaining `limit`. Dataset items have no default limit, so +iterating `iterateItems()` without `$chunkSize` loads the whole dataset, or all `limit` items, in one response. +Pass `$chunkSize` to iterate a large dataset. + ```php $page = $client->actors()->list(new ActorListOptions(limit: 5)); foreach ($page as $actor) { diff --git a/docs/actors.md b/docs/actors.md index 0dcc8f9..77c1d27 100644 --- a/docs/actors.md +++ b/docs/actors.md @@ -52,6 +52,7 @@ throws the last error. In `call()`, time spent retrying does not count toward `$ - `builds(): BuildCollectionClient`, `runs(): RunCollectionClient` - `version(string $versionNumber): ActorVersionClient`, `versions(): ActorVersionCollectionClient` - `webhooks(): NestedWebhookCollectionClient` — read-only. +- `server(): ActorServerClient` — the [server interface](#server-actors--client-actorid-server) of a Server Actor. Every call returns the same client. ```php $run = $client->actor('apify/hello-world')->call(['name' => 'world'], new ActorStartOptions(memoryMbytes: 512), 120); @@ -59,6 +60,46 @@ $isValid = $client->actor('apify/hello-world')->validateInput(['firstNumber' => $lastSucceeded = $client->actor('apify/hello-world')->lastRun(new LastRunOptions(status: 'SUCCEEDED'))->get(); ``` +## Server Actors — `$client->actor($id)->server()` + +A Server Actor (formerly Actor Standby) runs a web server that is called over HTTP without starting a run for +each call. `ActorServerClient` sends requests to that server and reads its URL and OpenAPI specification. + +- `request(string $path = '/', ?ActorServerRequestOptions $options = null, int|float|string|null $timeoutSecs = null): ActorServerResponse` — sends an HTTP request to the Actor's server. The timeout defaults to the `long` tier, which covers the cold start of an idle Actor. +- `getUrl(int|float|string|null $timeoutSecs = null): string` — the server URL (the Actor's `standbyUrl`). It is fetched once per client, by `getUrl()` or `getOpenApiSpec()`. +- `getOpenApiSpec(int|float|string|null $timeoutSecs = null): ?array` — the web server schema from the build that serves the server (`actorStandby.build`, or `latest`), or `null` if the Actor's definition has none. + +`request()` resolves `$path` against the server URL, so neither the request nor the token can reach another host. The +token goes in the `X-Apify-Authorization` header, which the platform removes before the request reaches the Actor, so +the headers you pass, `Authorization` included, reach the Actor as they are. Pass the query in the `params` option. + +It returns any response of the server, error statuses included. Only a rate limit (429) and Service Unavailable (503) +are retried, with the client's retry settings, and the last response is returned once the retries run out. A transport +error, a timeout included, throws `TransportException` without a retry, since the Actor may already have handled the +request. A redirect is returned as it is, since following it would send the token to a host the Actor's code picked. A +request body is sent uncompressed unless the `compression` option is set. + +The default `GuzzleHttpClient` can return a redirect as it is. A custom [`HttpClientInterface`](../README.md) must also +implement `NoRedirectHttpClientInterface` to be used for `request()`, which otherwise throws `ApifyClientException` +before it sends anything to the Actor's server. `Psr18HttpClient` does not implement it, since a PSR-18 client decides on its own whether to +follow redirects. + +`ActorServerResponse` offers `getStatusCode()`, `isSuccessful()`, `getHeaders()`, `getHeaderLine($name)`, `getBody()` +(the raw bytes), `getBodyStream()` (a PSR-7 stream, live with the `stream` option), `getData()` (the body parsed by its +content type: JSON is decoded, anything else is returned as a string) and `getResponse()` (the PSR-7 response). + +```php +$server = $client->actor('username/actor-name')->server(); + +$response = $server->request('/search', new ActorServerRequestOptions(params: ['q' => 'apify'])); +if ($response->isSuccessful()) { + print_r($response->getData()); +} + +$url = $server->getUrl(); +$spec = $server->getOpenApiSpec(); +``` + ## Actor versions — `$client->actor($id)->versions()` / `->version($n)` - Collection: `list(): PaginationList`, `iterate(): iterable`, `create(mixed $version): ActorVersion`. @@ -75,6 +116,27 @@ $version = $client->actor('me~my-actor')->versions()->create([ ]); ``` +A `SOURCE_FILES` or `TARBALL` version can set `actorPath`, the Actor's directory inside the uploaded source, to push +an Actor from an upload of a whole monorepo. It is a path relative to the upload root, with `/` separators and no +leading, trailing or repeated `/`, and no `.` or `..` segment (1 to 255 characters). The API rejects it for a +`GIT_REPO` or `GITHUB_GIST` version. An `update()` that sends `sourceType`, `sourceFiles` or `tarballUrl` without +`actorPath` removes the stored path, so send it again with every upload; `'actorPath' => null` removes it too, and an +update of other fields keeps it. `ActorVersion::getActorPath()` reads it back. + +```php +$version = $client->actor('me~my-actor')->versions()->create([ + 'versionNumber' => '0.2', + 'sourceType' => 'SOURCE_FILES', + 'sourceFiles' => [], + 'actorPath' => 'actors/typescript-actor', +]); +$client->actor('me~my-actor')->version('0.2')->update([ + 'sourceType' => 'SOURCE_FILES', + 'sourceFiles' => [], + 'actorPath' => 'actors/typescript-actor', +]); +``` + ## Environment variables — `->version($n)->envVars()` / `->envVar($name)` - Collection: `list(): PaginationList`, `iterate(): iterable`, `create(ActorEnvVar $envVar): ActorEnvVar`. diff --git a/docs/models.md b/docs/models.md index 5dfa214..56a3dcf 100644 --- a/docs/models.md +++ b/docs/models.md @@ -34,6 +34,7 @@ model is also used as an input object and therefore additionally exposes setters | `getTitle(): ?string` | Human-readable title. | | `getDescription(): ?string` | Free-text description. | | `isPublic(): ?bool` | Whether the Actor is public. | +| `getStandbyUrl(): ?string` | URL for sending requests to the Actor in Standby mode (a Server Actor), or `null` if Standby mode is not enabled. | | `getCreatedAt(): ?string` | ISO-8601 creation timestamp. | | `getModifiedAt(): ?string` | ISO-8601 last-modification timestamp. | @@ -72,6 +73,7 @@ model is also used as an input object and therefore additionally exposes setters |---|---| | `getVersionNumber(): ?string` | The version number (e.g. `0.0`). | | `getSourceType(): ?string` | Source type (e.g. `SOURCE_FILES`, `GIT_REPO`). | +| `getActorPath(): ?string` | The Actor's directory inside the uploaded source of a `SOURCE_FILES` or `TARBALL` version, or `null` when the upload root is the Actor's directory. | ### `Task` | Getter | Description | diff --git a/docs/options.md b/docs/options.md index 5fff686..91bffa6 100644 --- a/docs/options.md +++ b/docs/options.md @@ -115,6 +115,19 @@ For `actor()->lastRun()` / `task()->lastRun()`. | `status` | `?string` | Restrict to the last run with this status (e.g. `SUCCEEDED`). | | `origin` | `?string` | Restrict to the last run started via this origin. | +## Server Actors + +### `ActorServerRequestOptions` +For `actor($id)->server()->request()`. +| Field | Type | Description | +|---|---|---| +| `method` | `?string` | HTTP method (default `GET`). | +| `params` | `array\|null` | Query parameters, converted as for API calls: `null` values are left out, booleans are sent as `1`/`0` and a list is joined with commas. | +| `headers` | `array\|null` | Additional headers, `Authorization` included. They reach the Actor's server as they are. | +| `body` | `mixed` | The request body: an array (or any other JSON-serializable value) is sent as JSON, a string as it is. | +| `compression` | `string\|HttpCompressorInterface\|null` | `'brotli'`, `'gzip'` or a custom compressor for the request body. Unset sends the body uncompressed, since the Actor's server may not decode a compressed body. | +| `stream` | `bool` | Leave the response body unread, so it can be consumed incrementally with `ActorServerResponse::getBodyStream()`. The connection stays open until the body is read to the end or closed. | + ## Running Actors and tasks ### `ActorStartOptions` diff --git a/docs/storages.md b/docs/storages.md index a758672..cfea2ed 100644 --- a/docs/storages.md +++ b/docs/storages.md @@ -19,7 +19,7 @@ Single — `$client->dataset($id)`: - `get(): ?Dataset`, `update(mixed $newFields): Dataset`, `delete(): void` - `listItems(?DatasetListItemsOptions $options = null): PaginationList` — one page of items decoded to PHP values. -- `iterateItems(?DatasetListItemsOptions $options = null, ?int $chunkSize = null): iterable` — lazily iterate all items, paging on demand. The options' `limit` caps the total number of items yielded across all pages (unset = all); `$chunkSize` is the per-page size. The iterator advances and terminates by the number of rows the API scanned (the `X-Apify-Pagination-Count` response header), not by the number of items a page returns, matching the reference JS client exactly — see the method's doc comment for why that matters with `skipEmpty`/`clean`/`unwind`. +- `iterateItems(?DatasetListItemsOptions $options = null, ?int $chunkSize = null): iterable` — lazily iterate all items, paging on demand. The options' `limit` caps the total number of items yielded across all pages (unset = all); `$chunkSize` is the per-page size. Dataset items have no default limit, so without `$chunkSize` the iteration loads the whole dataset, or all `limit` items, in one response. The iterator advances and terminates by the page's `getCount()` (the rows the API scanned, or the items returned when that is larger), not by the number of items a page returns, matching the reference JS client exactly — see the method's doc comment for why that matters with `skipEmpty`/`clean`/`unwind`. - `downloadItems(DownloadItemsFormat $format, ?DatasetDownloadOptions $options = null): string` — raw export bytes. - `streamItems(DownloadItemsFormat $format, ?DatasetDownloadOptions $options = null): StreamInterface` — the same export as `downloadItems()`, as a PSR-7 stream instead of a fully-buffered string, so a large export can be read incrementally (e.g. piped to a file) without holding it all in memory. Bypasses the retrying transport, so a failed request is not retried. - `pushItems(mixed $items): void` @@ -59,7 +59,7 @@ Single — `$client->keyValueStore($id)`: - `get(): ?KeyValueStore`, `update(mixed $newFields): KeyValueStore`, `delete(): void` - `listKeys(?ListKeysOptions $options = null): KeyValueStoreKeysPage` -- `iterateKeys(?ListKeysOptions $options = null, ?int $chunkSize = null): iterable` — lazily iterate all keys, following cursor pagination (`exclusiveStartKey`/`nextExclusiveStartKey`). The options' `limit` caps the total number of keys yielded across all pages (unset = all); `$chunkSize` is the per-page size (unset = the remaining total cap, or the server default when unbounded). +- `iterateKeys(?ListKeysOptions $options = null, ?int $chunkSize = null): iterable` — lazily iterate all keys, following cursor pagination (`exclusiveStartKey`/`nextExclusiveStartKey`). The options' `limit` caps the total number of keys yielded across all pages (unset = all); `$chunkSize` is the per-page size (unset = the remaining total cap, or the API's default page size when unbounded). - `recordExists(string $key): bool` — reports whether a record with the given key exists, without downloading its value (a `HEAD` request). - `getRecord(string $key, ?GetRecordOptions $options = null): ?KeyValueStoreRecord` — the record value is never parsed: `getValue()` returns the exact bytes the API sent, whatever the content type, so a JSON record is decoded by the caller. - `setRecord(string $key, string $value, string $contentType, ?SetRecordOptions $options = null, ?string $contentEncoding = null): void` — `$value` is sent exactly as given, also under `application/json`, so it has to be valid JSON already (use `setRecordJson()` to have an array encoded). `$contentEncoding`, if given, is sent as the `Content-Encoding` header, for a `$value` that is already compressed (gzip, br, ...); the client's own best-effort request compression always backs off when `Content-Encoding` is already set, so the bytes reach the API exactly as given. diff --git a/src/ApifyClient.php b/src/ApifyClient.php index 12af463..21ca6e4 100644 --- a/src/ApifyClient.php +++ b/src/ApifyClient.php @@ -9,6 +9,7 @@ use Apify\Client\Http\GzipHttpCompressor; use Apify\Client\Http\HttpClientInterface; use Apify\Client\Http\HttpCompressorInterface; +use Apify\Client\Internal\Compression; use Apify\Client\Internal\HttpClientCore; use Apify\Client\Internal\Platform; use Apify\Client\Internal\RetryConfig; @@ -376,15 +377,7 @@ private static function trimTrailingSlash(string $value): string */ private static function resolveCompressor(string|HttpCompressorInterface|null $compression): ?HttpCompressorInterface { - return match (true) { - $compression === null => null, - $compression instanceof HttpCompressorInterface => $compression, - $compression === 'brotli' => new BrotliHttpCompressor(), - $compression === 'gzip' => new GzipHttpCompressor(), - default => throw new RuntimeException( - sprintf('ApifyClient: unknown compression "%s"; expected "brotli", "gzip", or a HttpCompressorInterface', $compression) - ), - }; + return Compression::resolveCompressor($compression, 'ApifyClient'); } /** diff --git a/src/Http/GuzzleHttpClient.php b/src/Http/GuzzleHttpClient.php index d7885fc..3c92e57 100644 --- a/src/Http/GuzzleHttpClient.php +++ b/src/Http/GuzzleHttpClient.php @@ -18,7 +18,7 @@ * are returned as normal responses ({@code http_errors} disabled); only connection/timeout failures * are thrown as {@see TransportException}. */ -final class GuzzleHttpClient implements HttpClientInterface +final class GuzzleHttpClient implements NoRedirectHttpClientInterface { /** Connection-establishment timeout (distinct from the per-request timeout the client applies). */ private const CONNECT_TIMEOUT_SECS = 30.0; @@ -32,20 +32,25 @@ public function __construct(?Guzzle $client = null) public function send(RequestInterface $request, float $timeoutSecs): ResponseInterface { - return $this->doSend($request, $timeoutSecs, false); + return $this->doSend($request, $timeoutSecs, false, true); } public function sendStreaming(RequestInterface $request, float $timeoutSecs): ResponseInterface { - return $this->doSend($request, $timeoutSecs, true); + return $this->doSend($request, $timeoutSecs, true, true); } - private function doSend(RequestInterface $request, float $timeoutSecs, bool $stream): ResponseInterface + public function sendWithoutRedirects(RequestInterface $request, float $timeoutSecs, bool $stream = false): ResponseInterface + { + return $this->doSend($request, $timeoutSecs, $stream, false); + } + + private function doSend(RequestInterface $request, float $timeoutSecs, bool $stream, bool $followRedirects): ResponseInterface { try { return $this->client->send($request, [ 'http_errors' => false, - 'allow_redirects' => true, + 'allow_redirects' => $followRedirects, 'connect_timeout' => self::CONNECT_TIMEOUT_SECS, 'timeout' => $timeoutSecs, 'stream' => $stream, diff --git a/src/Http/NoRedirectHttpClientInterface.php b/src/Http/NoRedirectHttpClientInterface.php new file mode 100644 index 0000000..5a3dc69 --- /dev/null +++ b/src/Http/NoRedirectHttpClientInterface.php @@ -0,0 +1,28 @@ + gzencode($body); } + /** + * Resolves a {@code compression} option ({@code 'brotli'}, {@code 'gzip'} or a custom compressor) to a + * concrete compressor, or {@code null} to keep the default best-effort choice. + * + * @param string $option the name of the option's owner, for the error message + * @throws RuntimeException if the option is an unknown algorithm name + */ + public static function resolveCompressor(string|HttpCompressorInterface|null $compression, string $option): ?HttpCompressorInterface + { + return match (true) { + $compression === null => null, + $compression instanceof HttpCompressorInterface => $compression, + $compression === 'brotli' => new BrotliHttpCompressor(), + $compression === 'gzip' => new GzipHttpCompressor(), + default => throw new RuntimeException( + sprintf('%s: unknown compression "%s"; expected "brotli", "gzip", or a HttpCompressorInterface', $option, $compression) + ), + }; + } + private function __construct() { } diff --git a/src/Internal/DatasetItemsPage.php b/src/Internal/DatasetItemsPage.php index 9fc1318..65c5397 100644 --- a/src/Internal/DatasetItemsPage.php +++ b/src/Internal/DatasetItemsPage.php @@ -8,8 +8,8 @@ use Apify\Client\Options\DatasetListItemsOptions; /** - * Reads one page of dataset items together with the number of rows the API scanned to produce it. - * Shared by {@see \Apify\Client\Resource\DatasetClient} and {@see \Apify\Client\Resource\RunClient}. + * Reads one page of dataset items. The page's count is the number of rows the API scanned to produce + * it, or the number of items returned when that is larger. Shared by {@see \Apify\Client\Resource\DatasetClient} and {@see \Apify\Client\Resource\RunClient}. * * @internal */ @@ -20,13 +20,14 @@ private function __construct() } /** - * Fetches one page of items and parses the pagination headers. Returns the page together with the - * {@code X-Apify-Pagination-Count} "scanned rows" count ({@code null} if the response omits the - * header), which the iterators use to advance. + * Fetches one page of items and parses the pagination headers. The page's {@code count} is the + * {@code X-Apify-Pagination-Count} "scanned rows" count, or the number of items returned when that is + * larger (or when the response omits the header). The header is derived from the dataset's item count, + * which lags a fresh push, so it can fall below the items returned on an unfiltered page. * - * @return array{0:PaginationList,1:?int} + * @return PaginationList */ - public static function fetch(HttpClientCore $http, ResourceContext $ctx, DatasetListItemsOptions $options, ?float $timeoutSecs = null): array + public static function fetch(HttpClientCore $http, ResourceContext $ctx, DatasetListItemsOptions $options, ?float $timeoutSecs = null): PaginationList { $params = new QueryParams(); $options->appendTo($params); @@ -35,24 +36,20 @@ public static function fetch(HttpClientCore $http, ResourceContext $ctx, Dataset $items = Json::decode((string) $response->getBody()); $items = is_array($items) ? array_values($items) : []; - $count = count($items); + $returned = count($items); + $scannedHeader = $response->getHeaderLine('X-Apify-Pagination-Count'); - $page = PaginationList::fromItems( + return PaginationList::fromItems( $items, - $response->getHeaderLine('X-Apify-Pagination-Total') === '' ? $count : (int) $response->getHeaderLine('X-Apify-Pagination-Total'), + $response->getHeaderLine('X-Apify-Pagination-Total') === '' ? $returned : (int) $response->getHeaderLine('X-Apify-Pagination-Total'), $response->getHeaderLine('X-Apify-Pagination-Offset') === '' ? 0 : (int) $response->getHeaderLine('X-Apify-Pagination-Offset'), - $response->getHeaderLine('X-Apify-Pagination-Limit') === '' ? $count : (int) $response->getHeaderLine('X-Apify-Pagination-Limit'), - $count, + $response->getHeaderLine('X-Apify-Pagination-Limit') === '' ? $returned : (int) $response->getHeaderLine('X-Apify-Pagination-Limit'), + max($scannedHeader === '' ? 0 : (int) $scannedHeader, $returned), // Prefer the server-reported X-Apify-Pagination-Desc header (matches the reference JS // client's `_createPaginationList`); fall back to the requested option when the header // is absent, e.g. against an older API version that predates it. self::headerBool($response->getHeaderLine('X-Apify-Pagination-Desc'), $options->desc ?? false), ); - - $scannedHeader = $response->getHeaderLine('X-Apify-Pagination-Count'); - $scanned = $scannedHeader === '' ? null : (int) $scannedHeader; - - return [$page, $scanned]; } private static function headerBool(string $value, bool $fallback): bool diff --git a/src/Internal/HttpClientCore.php b/src/Internal/HttpClientCore.php index 4c6c358..bd31d82 100644 --- a/src/Internal/HttpClientCore.php +++ b/src/Internal/HttpClientCore.php @@ -5,6 +5,7 @@ namespace Apify\Client\Internal; use Apify\Client\Exception\ApifyApiException; +use Apify\Client\Exception\ApifyClientException; use Apify\Client\Exception\ConflictException; use Apify\Client\Exception\ForbiddenException; use Apify\Client\Exception\InvalidRequestException; @@ -15,6 +16,7 @@ use Apify\Client\Exception\UnauthorizedException; use Apify\Client\Http\HttpClientInterface; use Apify\Client\Http\HttpCompressorInterface; +use Apify\Client\Http\NoRedirectHttpClientInterface; use Psr\Http\Message\RequestFactoryInterface; use Psr\Http\Message\RequestInterface; use Psr\Http\Message\ResponseInterface; @@ -77,6 +79,10 @@ public function timeoutTiers(): TimeoutTiers * Sends a request with auth, User-Agent and the retry policy applied. * * @param array $extraHeaders + * @param RequestPolicy|null $policy per-request deviations from the defaults of a call to the API + * ({@code null} keeps them) + * + * @throws ApifyClientException if the policy forbids following redirects and the transport cannot comply */ public function call( string $method, @@ -85,10 +91,23 @@ public function call( string $contentType = '', ?float $timeoutSecs = null, bool $doNotRetryTimeouts = false, - array $extraHeaders = [] + array $extraHeaders = [], + ?RequestPolicy $policy = null ): ResponseInterface { + $policy ??= new RequestPolicy(); + // Refuse before sending anything: a transport that follows redirects would hand the token to another host. + if (!$policy->followRedirects && !$this->transport instanceof NoRedirectHttpClientInterface) { + throw new ApifyClientException(sprintf( + 'The HTTP client %s cannot disable redirects, which this request needs to keep the API token from following one. Use a client that implements %s, such as the default GuzzleHttpClient.', + $this->transport::class, + NoRedirectHttpClientInterface::class + )); + } + // Compress the body once, up front, so every retry reuses the already-compressed payload. - [$body, $extraHeaders] = $this->maybeCompressBody($body, $contentType, $extraHeaders); + if ($policy->compression !== false) { + [$body, $extraHeaders] = $this->maybeCompressBody($body, $contentType, $extraHeaders, $policy->compression ?? $this->compressor); + } $delayMillis = $this->retry->minDelayMillis; $maxAttempts = $this->retry->maxRetries + 1; @@ -104,18 +123,25 @@ public function call( $body, $contentType, $extraHeaders, - $this->attemptTimeout($baseTimeout, $attempt) + $this->attemptTimeout($baseTimeout, $attempt), + $policy ); $status = $response->getStatusCode(); if ($status < self::MAX_SUCCESS_STATUS) { return $response; } + $retryable = self::isStatusRetryable($status, $policy->retryStatusCodes); + // The caller handles the error response itself once no retry follows, so its body stays unread. + if (!$policy->throwOnErrorStatus && (!$retryable || $attempt === $maxAttempts)) { + return $response; + } $lastError = self::buildApiError($status, (string) $response->getBody(), $attempt, $method, $path); - $retryable = self::isStatusRetryable($status); + $response->getBody()->close(); } catch (TransportException $e) { $lastError = $e; - // Network/timeout failures are retryable, unless the caller opted out of retrying timeouts. - $retryable = !($doNotRetryTimeouts && $e->isTimeout()); + // Network/timeout failures are retryable, unless the caller opted out of retrying timeouts or + // of retrying transport errors at all. + $retryable = $policy->retryTransportErrors && !($doNotRetryTimeouts && $e->isTimeout()); } if (!$retryable || $attempt === $maxAttempts) { @@ -137,34 +163,34 @@ public function call( * double-compressed. A body whose {@code $contentType} already carries its own compression (media, * archives, fonts, ...) is left uncompressed too — see {@see Compression::isCompressibleContentType()}. * - * Uses the configured {@see $compressor} when one was given to the constructor (the {@code - * compression} option of {@see \Apify\Client\ApifyClient}); otherwise falls back to the client's + * Uses {@code $compressor} when one is given: the one the caller configured on the constructor (the + * {@code compression} option of {@see \Apify\Client\ApifyClient}) or a per-request one; otherwise falls back to the client's * default best-effort choice ({@see Compression::maybeCompress()}: brotli, falling back to gzip). * * @param array $extraHeaders + * @param HttpCompressorInterface|null $compressor the compressor to use, {@code null} for the default choice * @return array{0: string|null, 1: array} */ - private function maybeCompressBody(?string $body, string $contentType, array $extraHeaders): array + private function maybeCompressBody(?string $body, string $contentType, array $extraHeaders, ?HttpCompressorInterface $compressor): array { if ($body === null || self::hasHeader($extraHeaders, 'Content-Encoding') - || !Compression::isCompressibleContentType($contentType) + || !Compression::isCompressibleContentType(self::headerValue($extraHeaders, 'Content-Type') ?? $contentType) ) { return [$body, $extraHeaders]; } - if ($this->compressor !== null) { + if ($compressor !== null) { // A configured compressor has no built-in size gate of its own (unlike // Compression::maybeCompress() below), so it is applied here. if (strlen($body) < Compression::MIN_COMPRESS_BYTES) { return [$body, $extraHeaders]; } - $compressedBody = $this->compressor->compress($body); + $compressedBody = $compressor->compress($body); if ($compressedBody === null) { return [$body, $extraHeaders]; } - $extraHeaders['Content-Encoding'] = $this->compressor->contentEncoding(); - return [$compressedBody, $extraHeaders]; + return [$compressedBody, self::withContentEncoding($extraHeaders, $compressor->contentEncoding())]; } $compressed = Compression::maybeCompress($body); @@ -173,8 +199,25 @@ private function maybeCompressBody(?string $body, string $contentType, array $ex } [$encoding, $compressedBody] = $compressed; - $extraHeaders['Content-Encoding'] = $encoding; - return [$compressedBody, $extraHeaders]; + return [$compressedBody, self::withContentEncoding($extraHeaders, $encoding)]; + } + + /** + * Marks a body as compressed and drops any caller-set {@code Content-Length}, which describes the + * uncompressed body; the transport computes the real one. + * + * @param array $headers + * @return array + */ + private static function withContentEncoding(array $headers, string $encoding): array + { + foreach (array_keys($headers) as $name) { + if (strcasecmp($name, 'Content-Length') === 0) { + unset($headers[$name]); + } + } + $headers['Content-Encoding'] = $encoding; + return $headers; } /** @@ -184,18 +227,28 @@ private function maybeCompressBody(?string $body, string $contentType, array $ex */ private static function hasHeader(array $headers, string $name): bool { - foreach (array_keys($headers) as $key) { + return self::headerValue($headers, $name) !== null; + } + + /** + * Case-insensitive lookup of a header value, {@code null} if the header is not set. + * + * @param array $headers + */ + private static function headerValue(array $headers, string $name): ?string + { + foreach ($headers as $key => $value) { if (strcasecmp($key, $name) === 0) { - return true; + return $value; } } - return false; + return null; } /** Opens a live streaming response (single attempt, no retry). Used by log streaming. */ public function stream(string $url): ResponseInterface { - $request = $this->buildRequest('GET', $url, null, '', []); + $request = $this->buildRequest('GET', $url, null, '', [], RequestPolicy::AUTHORIZATION_HEADER); return $this->transport->sendStreaming($request, $this->retry->timeoutSecs); } @@ -203,18 +256,20 @@ public function stream(string $url): ResponseInterface * Builds a fully-prepared PSR-7 request with auth, User-Agent, content type and extra headers. * * @param array $extraHeaders + * @param string $authorizationHeader the header that carries the token */ private function buildRequest( string $method, string $url, ?string $body, string $contentType, - array $extraHeaders + array $extraHeaders, + string $authorizationHeader ): RequestInterface { $request = $this->requestFactory->createRequest($method, $url) ->withHeader('User-Agent', $this->userAgent); if ($this->token !== null && $this->token !== '') { - $request = $request->withHeader('Authorization', 'Bearer ' . $this->token); + $request = $request->withHeader($authorizationHeader, 'Bearer ' . $this->token); } if ($contentType !== '') { $request = $request->withHeader('Content-Type', $contentType); @@ -237,10 +292,17 @@ private function doAttempt( ?string $body, string $contentType, array $extraHeaders, - float $timeoutSecs + float $timeoutSecs, + RequestPolicy $policy ): ResponseInterface { - $request = $this->buildRequest($method, $url, $body, $contentType, $extraHeaders); - return $this->transport->send($request, $timeoutSecs); + $request = $this->buildRequest($method, $url, $body, $contentType, $extraHeaders, $policy->authorizationHeader); + // `call()` has already checked that a transport asked not to follow redirects can do that. + if (!$policy->followRedirects && $this->transport instanceof NoRedirectHttpClientInterface) { + return $this->transport->sendWithoutRedirects($request, $timeoutSecs, $policy->stream); + } + return $policy->stream + ? $this->transport->sendStreaming($request, $timeoutSecs) + : $this->transport->send($request, $timeoutSecs); } /** @@ -264,8 +326,17 @@ private function attemptTimeout(float $base, int $attempt): float return min($scaled, $cap); } - private static function isStatusRetryable(int $status): bool + /** + * Whether a response status is retried: the statuses a request lists itself, or by default rate limits + * (429) and server errors (5xx). + * + * @param list|null $retryStatusCodes + */ + private static function isStatusRetryable(int $status, ?array $retryStatusCodes): bool { + if ($retryStatusCodes !== null) { + return in_array($status, $retryStatusCodes, true); + } return $status === self::RATE_LIMIT_EXCEEDED || $status >= self::MIN_SERVER_ERROR; } diff --git a/src/Internal/RequestPolicy.php b/src/Internal/RequestPolicy.php new file mode 100644 index 0000000..8834c16 --- /dev/null +++ b/src/Internal/RequestPolicy.php @@ -0,0 +1,47 @@ +|null $retryStatusCodes the statuses to retry, {@code null} for 429 and 5xx + * @param bool $retryTransportErrors whether a transport error, a timeout included, + * is retried + * @param bool $followRedirects whether the transport follows redirects + * @param bool $stream whether the response body is left unread + */ + public function __construct( + public readonly HttpCompressorInterface|false|null $compression = null, + public readonly string $authorizationHeader = self::AUTHORIZATION_HEADER, + public readonly bool $throwOnErrorStatus = true, + public readonly ?array $retryStatusCodes = null, + public readonly bool $retryTransportErrors = true, + public readonly bool $followRedirects = true, + public readonly bool $stream = false, + ) { + } +} diff --git a/src/Internal/ResourceContext.php b/src/Internal/ResourceContext.php index 3e108fd..6b68794 100644 --- a/src/Internal/ResourceContext.php +++ b/src/Internal/ResourceContext.php @@ -270,14 +270,16 @@ public function listResource(string $subPath, QueryParams $params, callable $hyd * Ports the reference client's paginated iterator ({@code _listPaginatedFromCallback}): * {@code $limit} caps the TOTAL number of items yielded across all pages ({@code null} = no cap, * i.e. all items), while {@code $chunkSize} caps how many items are requested per page - * ({@code null} = the server default). The two are independent — {@code $limit} is never reused - * as the page size. {@code $startOffset} is the offset of the first page. + * ({@code null} = the API's default page size, capped by the remaining {@code $limit}). The two are + * independent: {@code $limit} is never reused as the page size. {@code $startOffset} is the offset + * of the first page. * - * The next offset and the stop condition follow the number of items a page returned (capped at - * what that page's request asked for) — never the listing's reported {@code total}, which can grow - * while the iteration runs (e.g. a resource collection gaining entries). With no {@code $limit}, - * the loop therefore always ends with one request that finds no more items, matching the - * reference client exactly. + * The next offset and the stop condition follow the page's {@see PaginationList::getCount()} (or + * the items it returned when larger), capped at the limit requested for that page, never the + * listing's reported {@code total}, which can grow while the iteration runs. A page whose + * filtered-out entries were still counted (the Store, dataset filters) therefore neither repeats + * entries nor ends the iteration early. With no {@code $limit} the loop always ends with one + * request that advances by nothing, matching the reference client. * * @template T * @param callable(int,?int):PaginationList $fetchPage receives (offset, pageLimit) and returns that page @@ -285,58 +287,50 @@ public function listResource(string $subPath, QueryParams $params, callable $hyd */ public static function paginateOffset(int $startOffset, ?int $limit, ?int $chunkSize, callable $fetchPage): Generator { - // First page: request min(limit, chunkSize) items. A null/0 on either side means "unbounded", - // so the other bound wins (mirrors the reference client's minForLimitParam). - $firstPageLimit = self::minLimit($limit, $chunkSize); - $page = $fetchPage($startOffset, $firstPageLimit); - foreach ($page->getItems() as $item) { - yield $item; - } - - $pageCount = self::advancedBy($page); - $currentOffset = $startOffset + $pageCount; + $currentOffset = $startOffset; // Items still to yield, or null for "unbounded" (no $limit given). - $remaining = ($limit !== null && $limit > 0) ? $limit - $pageCount : null; - - // Two distinct ways to stop: a page that returns nothing (the listing is exhausted), or - // $remaining having already reached 0 (the caller's $limit is satisfied) — either ends the loop. - while ($pageCount > 0 && ($remaining === null || $remaining > 0)) { + $remaining = ($limit !== null && $limit > 0) ? $limit : null; + do { + // A null/0 on either bound means "unbounded", so the other bound wins (mirrors the + // reference client's minForLimitParam). $pageLimit = self::minLimit($remaining, $chunkSize); $page = $fetchPage($currentOffset, $pageLimit); foreach ($page->getItems() as $item) { yield $item; } - $pageCount = self::advancedBy($page); - $currentOffset += $pageCount; + $advance = self::advancedBy($page, $pageLimit); + $currentOffset += $advance; if ($remaining !== null) { - $remaining -= $pageCount; + $remaining -= $advance; } - } + // Two distinct ways to stop: a page that advances by nothing (the listing is exhausted), or + // $remaining having reached 0 (the caller's $limit is satisfied). + } while ($advance > 0 && ($remaining === null || $remaining > 0)); } /** - * How far a page advances the offset: the number of items it returned. Every endpoint still using - * this generic loop honors the {@code limit} it was sent — unlike the dataset-items endpoint (see - * {@see \Apify\Client\Resource\DatasetClient::scannedRows()}), none of them has an {@code unwind}- - * like transform that can make a page return more items than its requested limit, and an endpoint - * that ignores {@code offset}/{@code limit} entirely (env vars, Actor versions) bypasses this loop - * rather than going through it — so the returned count never needs capping to the requested limit. + * How far a page advances the offset: its count or the items it returned, whichever is larger, + * capped at the limit requested for it ({@code null} = uncapped). + * + * The count can exceed the returned items because the API drops entries after counting them (Store + * Actors, the dataset {@code skipEmpty} and {@code clean} filters). The items can exceed the count + * with {@code unwind}, which returns one item per unwound element. The server never walks more + * entries than the requested limit, so the cap keeps such an inflated item count from skipping the + * entries between the real and the inflated advance. * * @template T * @param PaginationList $page */ - private static function advancedBy(PaginationList $page): int + private static function advancedBy(PaginationList $page, ?int $requestedLimit): int { - return count($page->getItems()); + $advance = max($page->getCount(), count($page->getItems())); + return $requestedLimit !== null ? min($advance, $requestedLimit) : $advance; } /** * Returns the smaller of two optional positive bounds, treating {@code null} or {@code 0} as * "unbounded" (the API treats {@code limit=0} as unset). Mirrors the reference minForLimitParam. * - * Public so {@see \Apify\Client\Resource\DatasetClient::iterateItems()} can reuse it for its own - * scanned-count-aware pagination loop. - * * @internal */ public static function minLimit(?int $a, ?int $b): ?int diff --git a/src/Model/Actor.php b/src/Model/Actor.php index 65ba006..b6364ed 100644 --- a/src/Model/Actor.php +++ b/src/Model/Actor.php @@ -60,4 +60,13 @@ public function getModifiedAt(): ?string { return $this->getString('modifiedAt'); } + + /** + * The URL for sending requests to the Actor in Standby mode (a Server Actor), or {@code null} if Standby + * mode is not enabled. + */ + public function getStandbyUrl(): ?string + { + return $this->getString('standbyUrl'); + } } diff --git a/src/Model/ActorVersion.php b/src/Model/ActorVersion.php index 08b914c..ca1c8a3 100644 --- a/src/Model/ActorVersion.php +++ b/src/Model/ActorVersion.php @@ -18,4 +18,14 @@ public function getSourceType(): ?string { return $this->getString('sourceType'); } + + /** + * The Actor's directory inside the uploaded source, as a path relative to the upload root, for a + * {@code SOURCE_FILES} or {@code TARBALL} version, or {@code null} when the upload root is the Actor's + * directory. On a {@code TARBALL} version, only users with write access to the Actor get it. + */ + public function getActorPath(): ?string + { + return $this->getString('actorPath'); + } } diff --git a/src/Model/PaginationList.php b/src/Model/PaginationList.php index b7842a0..ee2dc6d 100644 --- a/src/Model/PaginationList.php +++ b/src/Model/PaginationList.php @@ -100,7 +100,13 @@ public function getLimit(): int return $this->limit; } - /** Number of items actually returned in this page. */ + /** + * Number of entries on this page. On a page from {@see \Apify\Client\Resource\DatasetClient::listItems()} + * it is the number of dataset rows the API scanned for it, or the number of items returned when that is + * larger: filters ({@code clean}, {@code skipEmpty}) can leave it above the number of items. The next + * page starts at {@code offset + min(count, limit)}; the cap matters only with {@code unwind}, which can + * return more items than the rows scanned. + */ public function getCount(): int { return $this->count; diff --git a/src/Options/ActorServerRequestOptions.php b/src/Options/ActorServerRequestOptions.php new file mode 100644 index 0000000..82b7f18 --- /dev/null +++ b/src/Options/ActorServerRequestOptions.php @@ -0,0 +1,92 @@ +|null $params query parameters of the request, converted the same way as for API + * calls: {@code null} values are left out, booleans are sent as + * {@code 1}/{@code 0} and a list is joined with commas + * @param array|null $headers additional headers of the request, {@code Authorization} included; + * they reach the Actor's server as they are + * @param string|HttpCompressorInterface|null $compression the compression of the request body: {@code 'brotli'}, + * {@code 'gzip'} or a custom compressor. The body is sent + * uncompressed when it is not set, since the Actor's server may + * not decode a compressed body. A body is compressed under the same + * conditions as for API calls, so set this only if the server + * decodes the resulting {@code Content-Encoding}. + */ + public function __construct( + /** The HTTP method of the request (default {@code GET}). */ + public readonly ?string $method = null, + public readonly ?array $params = null, + public readonly ?array $headers = null, + /** + * The request body: an array (or any other JSON-serializable value) is sent as JSON, with the content type + * {@code application/json} unless {@code headers} set another, and a string is sent as it is. + */ + public readonly mixed $body = null, + public readonly string|HttpCompressorInterface|null $compression = null, + /** + * If {@code true}, the response body is not read until the caller reads it, so a large or endless response + * can be consumed incrementally ({@see \Apify\Client\Resource\ActorServerResponse::getBodyStream()}). The + * connection stays open until the caller reads the body to the end or closes it. + */ + public readonly bool $stream = false, + ) { + } + + /** The HTTP method in upper case. */ + public function methodOrDefault(): string + { + return $this->method === null || $this->method === '' ? self::DEFAULT_METHOD : strtoupper($this->method); + } + + /** @internal */ + public function appendTo(QueryParams $q): void + { + foreach ($this->params ?? [] as $key => $value) { + if ($value === null) { + continue; + } + if (is_bool($value)) { + $q->addBool((string) $key, $value); + } elseif (is_int($value)) { + $q->addInt((string) $key, $value); + } elseif (is_float($value)) { + $q->addFloat((string) $key, $value); + } elseif (is_string($value)) { + $q->addString((string) $key, $value); + } elseif (is_array($value) && array_is_list($value) && self::isScalarList($value)) { + $q->addRaw((string) $key, implode(',', array_map(static fn (mixed $item): string => is_bool($item) ? ($item ? '1' : '0') : (string) $item, $value))); + } else { + throw new InvalidArgumentException(sprintf('ActorServerRequestOptions: the query parameter "%s" must be a scalar or a list of scalars', $key)); + } + } + } + + /** @param list $values */ + private static function isScalarList(array $values): bool + { + foreach ($values as $value) { + if (!is_scalar($value)) { + return false; + } + } + return true; + } +} diff --git a/src/Resource/AbstractWebhookCollectionClient.php b/src/Resource/AbstractWebhookCollectionClient.php index ab9448c..7739858 100644 --- a/src/Resource/AbstractWebhookCollectionClient.php +++ b/src/Resource/AbstractWebhookCollectionClient.php @@ -46,7 +46,7 @@ public function list(?ListOptions $options = null, int|float|string|null $timeou /** * Lazily iterates over webhooks, fetching pages on demand. The options' {@code limit} caps the * total number of webhooks yielded across all pages ({@code null} = all); {@code $chunkSize} is - * the per-page size ({@code null} = the server default). + * the per-page size ({@code null} = the API's default page size). * * @return Generator */ diff --git a/src/Resource/ActorClient.php b/src/Resource/ActorClient.php index cd93f95..49e60e5 100644 --- a/src/Resource/ActorClient.php +++ b/src/Resource/ActorClient.php @@ -29,6 +29,8 @@ final class ActorClient { private ResourceContext $ctx; + private ?ActorServerClient $serverClient = null; + /** @internal */ public function __construct( private ApifyClient $root, @@ -218,4 +220,18 @@ public function webhooks(): NestedWebhookCollectionClient { return new NestedWebhookCollectionClient($this->http, $this->ctx->subUrl('')); } + + /** + * Returns a client for the server interface of this Actor. + * + * A Server Actor (formerly Actor Standby) runs a web server, its server interface, that is called over + * HTTP without starting a run for each call. Every call returns the same client, so the server URL it + * fetches is reused. + * + * @see https://docs.apify.com/platform/actors/running/standby + */ + public function server(): ActorServerClient + { + return $this->serverClient ??= new ActorServerClient($this->http, $this->baseUrl, $this->id); + } } diff --git a/src/Resource/ActorCollectionClient.php b/src/Resource/ActorCollectionClient.php index 39c26c0..7c7146f 100644 --- a/src/Resource/ActorCollectionClient.php +++ b/src/Resource/ActorCollectionClient.php @@ -39,7 +39,7 @@ public function list(?ActorListOptions $options = null, int|float|string|null $t /** * Lazily iterates over the account's Actors, fetching pages on demand. The options' {@code limit} * caps the total number of Actors yielded across all pages ({@code null} = all); {@code $chunkSize} - * is the per-page size ({@code null} = the server default). + * is the per-page size ({@code null} = the API's default page size). * * @return Generator */ diff --git a/src/Resource/ActorServerClient.php b/src/Resource/ActorServerClient.php new file mode 100644 index 0000000..de67370 --- /dev/null +++ b/src/Resource/ActorServerClient.php @@ -0,0 +1,245 @@ +ctx = ResourceContext::single($http, $baseUrl, 'actors', $id); + } + + /** + * Returns the URL of the Actor's server. + * + * The URL is the Actor's {@code standbyUrl}, fetched from the API by the first call of this method or of + * {@see getOpenApiSpec()} 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. + * + * @param int|float|string|null $timeoutSecs timeout of the API request (default: the {@code short} tier) + * @throws ApifyClientException if the Actor is not a Server Actor + * @throws \Apify\Client\Exception\NotFoundException if the Actor does not exist + */ + public function getUrl(int|float|string|null $timeoutSecs = null): string + { + return $this->url ?? $this->fetchServerActor($timeoutSecs)[1]; + } + + /** + * Returns 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 {@code actorStandby.build}, or {@code latest} if none is set. Use it + * to generate an HTTP client for the server, and point that client at the URL returned by {@see getUrl()}. + * + * @param int|float|string|null $timeoutSecs timeout of each API request (default: the {@code short} tier) + * @return array|null the specification, or {@code null} if the Actor's definition does not + * include one + * @throws ApifyClientException if the Actor is not a Server Actor, or the build that serves its server + * interface does not exist + * @throws \Apify\Client\Exception\NotFoundException if the Actor does not exist + */ + public function getOpenApiSpec(int|float|string|null $timeoutSecs = null): ?array + { + [$actor] = $this->fetchServerActor($timeoutSecs); + $standby = $actor->get('actorStandby'); + $standbyBuild = is_array($standby) ? ($standby['build'] ?? null) : null; + $buildRef = is_string($standbyBuild) && $standbyBuild !== '' ? $standbyBuild : self::DEFAULT_BUILD_REF; + + $buildId = $this->resolveBuildId($actor, $buildRef, $timeoutSecs); + $build = $buildId === null ? null : (new BuildClient($this->http, $this->baseUrl, $buildId))->get(null, $timeoutSecs); + if ($build === null) { + throw new ApifyClientException(sprintf('Build %s of Actor %s does not exist.', $buildRef, $this->id)); + } + + $definition = $build->get('actorDefinition'); + $schema = is_array($definition) ? ($definition['webServerSchema'] ?? null) : null; + return is_array($schema) ? $schema : null; + } + + /** + * Sends an HTTP request to the Actor's server. + * + * The request goes to the URL returned by {@see getUrl()} and carries the client's token in the + * {@code X-Apify-Authorization} header, which the platform consumes, so it never reaches the Actor's code. + * Headers you pass, {@code 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 the + * {@code long} tier. + * + * 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. + * Other failures are not retried, since the Actor may already have handled the request: a transport error, + * a timeout included, throws {@see \Apify\Client\Exception\TransportException} right away. 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 the {@code compression} option is set, since the Actor's server may + * not decode a compressed body. + * + * @param string $path the path on the Actor's server, for example {@code /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 the {@code params} option. A fragment + * ({@code #...}) is dropped, since it is never sent to the server. + * @param int|float|string|null $timeoutSecs timeout of the HTTP request (default: the {@code long} tier). + * With the {@code stream} option, it bounds the wait for the server to start responding. + * @throws ApifyClientException if the Actor is not a Server Actor, or the HTTP client in use cannot disable + * redirects (see {@see \Apify\Client\Http\NoRedirectHttpClientInterface}) + * @throws \Apify\Client\Exception\NotFoundException if the Actor does not exist + * @throws \Apify\Client\Exception\TransportException if the request fails or times out before the server + * responds + * @throws \InvalidArgumentException if a query parameter is not a scalar or a list of scalars + */ + public function request(string $path = '/', ?ActorServerRequestOptions $options = null, int|float|string|null $timeoutSecs = null): ActorServerResponse + { + $options ??= new ActorServerRequestOptions(); + $params = new QueryParams(); + $options->appendTo($params); + [$body, $contentType] = self::encodeBody($options->body); + $extraHeaders = $options->headers ?? []; + + $policy = new RequestPolicy( + compression: Compression::resolveCompressor($options->compression, 'ActorServerRequestOptions') ?? false, + authorizationHeader: RequestPolicy::SERVER_AUTHORIZATION_HEADER, + throwOnErrorStatus: false, + retryStatusCodes: self::SERVER_RETRY_STATUS_CODES, + retryTransportErrors: false, + followRedirects: false, + stream: $options->stream, + ); + $timeout = $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_LONG); + + $url = $params->applyToUrl(self::serverRequestUrl($this->getUrl(), $path)); + $response = $this->http->call($options->methodOrDefault(), $url, $body, $contentType, $timeout, extraHeaders: $extraHeaders, policy: $policy); + return new ActorServerResponse($response); + } + + /** + * Fetches the Actor, checks that it is a Server Actor, and caches its server URL. + * + * @return array{0:Actor,1:string} the Actor and its server URL + * @throws ApifyClientException if the Actor is not a Server Actor + */ + private function fetchServerActor(int|float|string|null $timeoutSecs): array + { + $data = $this->ctx->getResourceRequired('', new QueryParams(), $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_SHORT)); + $actor = new Actor(is_array($data) ? $data : []); + $url = $actor->getStandbyUrl(); + if ($url === null || $url === '') { + throw new ApifyClientException(sprintf('Actor %s is not a Server Actor.', $this->id)); + } + $this->url = $url; + return [$actor, $url]; + } + + /** + * Finds the ID of the build a tag or build number refers to, or {@code null} if there is none. + */ + private function resolveBuildId(Actor $actor, string $buildRef, int|float|string|null $timeoutSecs): ?string + { + $taggedBuilds = $actor->get('taggedBuilds'); + $tagged = is_array($taggedBuilds) ? ($taggedBuilds[$buildRef] ?? null) : null; + $taggedId = is_array($tagged) ? ($tagged['buildId'] ?? null) : null; + if (is_string($taggedId)) { + return $taggedId; + } + + $target = self::parseBuildNumber($buildRef); + if ($target === null) { + return null; + } + // Not a tag, so look `$buildRef` up as a build number, which the API cannot resolve directly. Builds are + // listed newest first and the sequence numbers of a version grow over time, so meeting an older build of + // the same version means the target does not exist. + $builds = new BuildCollectionClient($this->http, $this->ctx->subUrl(''), 'builds'); + foreach ($builds->iterate(new ListOptions(desc: true), timeoutSecs: $timeoutSecs) as $build) { + if ($build->getBuildNumber() === $buildRef) { + return $build->getId(); + } + $number = self::parseBuildNumber($build->getBuildNumber()); + if ($number !== null && $number[0] === $target[0] && $number[1] === $target[1] && $number[2] < $target[2]) { + break; + } + } + return null; + } + + /** + * Splits a build number like {@code 0.1.23} into its parts, or returns {@code null} for anything else. + * + * @return array{0:int,1:int,2:int}|null + */ + private static function parseBuildNumber(?string $buildNumber): ?array + { + if (preg_match(self::BUILD_NUMBER_PATTERN, $buildNumber ?? '', $match) !== 1) { + return null; + } + return [(int) $match[1], (int) $match[2], (int) $match[3]]; + } + + /** Resolves {@code $path} against the server URL, so no path can address another host. */ + private static function serverRequestUrl(string $serverUrl, string $path): string + { + $withoutFragment = explode('#', $path, 2)[0]; + return rtrim($serverUrl, '/') . '/' . ltrim($withoutFragment, '/'); + } + + /** + * Serializes a request body: a string goes out as it is, and any other value as JSON. + * + * @return array{0:?string,1:string} the body and its content type + */ + private static function encodeBody(mixed $body): array + { + if ($body === null) { + return [null, '']; + } + if (is_string($body)) { + return [$body, '']; + } + return [Json::encode($body), ResourceContext::CONTENT_TYPE_JSON]; + } +} diff --git a/src/Resource/ActorServerResponse.php b/src/Resource/ActorServerResponse.php new file mode 100644 index 0000000..20da9d7 --- /dev/null +++ b/src/Resource/ActorServerResponse.php @@ -0,0 +1,102 @@ +response->getStatusCode(); + } + + /** Whether the status is a success, that is below 300. */ + public function isSuccessful(): bool + { + return $this->getStatusCode() < HttpClientCore::MAX_SUCCESS_STATUS; + } + + /** + * The response headers, by header name as the server sent it. + * + * @return array> + */ + public function getHeaders(): array + { + return $this->response->getHeaders(); + } + + /** The values of a header joined with commas, or an empty string if the response has none. The name is case-insensitive. */ + public function getHeaderLine(string $name): string + { + return $this->response->getHeaderLine($name); + } + + /** + * The response body as raw bytes. With the {@code stream} request option, this reads the rest of the stream. + */ + public function getBody(): string + { + return (string) $this->response->getBody(); + } + + /** + * The response body as a PSR-7 stream. It is live, and holds its connection open until it is read to the end + * or closed, when the request was made with the {@code stream} option. + */ + public function getBodyStream(): StreamInterface + { + return $this->response->getBody(); + } + + /** + * The response body parsed by its content type: JSON is decoded to PHP values (objects become associative + * arrays), and any other body is returned as a string. An error response whose body does not parse as its + * content type comes back as the raw string. With the {@code stream} request option, this reads the rest of the + * stream. + * + * @return mixed + * @throws InvalidResponseBodyException if the body of a success response is not valid JSON, although its + * content type says it is + */ + public function getData(): mixed + { + $body = $this->getBody(); + if (preg_match(self::JSON_CONTENT_TYPE_PATTERN, $this->getHeaderLine('Content-Type')) !== 1) { + return $body; + } + try { + return Json::decode($body); + } catch (InvalidResponseBodyException $e) { + if (!$this->isSuccessful()) { + return $body; + } + throw $e; + } + } + + /** The underlying PSR-7 response. */ + public function getResponse(): ResponseInterface + { + return $this->response; + } +} diff --git a/src/Resource/BuildCollectionClient.php b/src/Resource/BuildCollectionClient.php index 62e745c..ac87965 100644 --- a/src/Resource/BuildCollectionClient.php +++ b/src/Resource/BuildCollectionClient.php @@ -42,7 +42,7 @@ public function list(?ListOptions $options = null, int|float|string|null $timeou /** * Lazily iterates over builds, fetching pages on demand. The options' {@code limit} caps the total * number of builds yielded across all pages ({@code null} = all); {@code $chunkSize} is the - * per-page size ({@code null} = the server default). + * per-page size ({@code null} = the API's default page size). * * @return Generator */ diff --git a/src/Resource/DatasetClient.php b/src/Resource/DatasetClient.php index d7e909e..8666aac 100644 --- a/src/Resource/DatasetClient.php +++ b/src/Resource/DatasetClient.php @@ -84,14 +84,18 @@ public function delete(int|float|string|null $timeoutSecs = null): void * Lists items from the dataset, each decoded to a PHP value (associative array for objects). * * The dataset items endpoint returns a bare JSON array (not a data envelope) and reports - * pagination via {@code X-Apify-Pagination-*} headers, surfaced in the returned page. + * pagination via {@code X-Apify-Pagination-*} headers, surfaced in the returned page. The page's + * {@see PaginationList::getCount()} is the number of dataset rows the API scanned for it, or the number + * of items returned when that is larger: filters ({@code clean}, {@code skipEmpty}) can leave it above + * the number of items. The next page starts at {@code offset + min(count, limit)}; the cap matters only + * with {@code unwind}, which can return more items than the rows scanned. * * @return PaginationList */ public function listItems(?DatasetListItemsOptions $options = null, int|float|string|null $timeoutSecs = null): PaginationList { $resolved = $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_LONG); - return DatasetItemsPage::fetch($this->http, $this->ctx, $options ?? new DatasetListItemsOptions(), $resolved)[0]; + return DatasetItemsPage::fetch($this->http, $this->ctx, $options ?? new DatasetListItemsOptions(), $resolved); } /** @@ -99,87 +103,38 @@ public function listItems(?DatasetListItemsOptions $options = null, int|float|st * PHP value (an associative array for objects), like {@see listItems()}. * * The options' {@code limit} caps the total number of items yielded across all pages ({@code null} - * = all), {@code offset} is the starting offset, and {@code $chunkSize} is the per-page size - * ({@code null} = the server default). All other {@see DatasetListItemsOptions} fields (field - * selection, filtering, ordering) are applied to every page. + * = all), {@code offset} is the starting offset, and {@code $chunkSize} is the per-page size. Dataset + * items have no default limit, so without {@code $chunkSize} (and without a {@code limit}) the first + * request loads the whole dataset in one response: pass {@code $chunkSize} to iterate a large + * dataset. All other {@see DatasetListItemsOptions} fields (field selection, filtering, ordering) are + * applied to every page. * * Item-dropping filters ({@code skipEmpty}, and {@code clean} which implies it) and {@code unwind} * are applied after {@code offset}/{@code limit}, so a page's item count can land on either side of - * the number of rows the API actually scanned to produce it. The iterator therefore advances and - * terminates by the scanned count the API reports in the {@code X-Apify-Pagination-Count} header — - * never by the returned item count — matching the reference JS client exactly (including its - * fallback to the item count on a response that omits the header, e.g. behind a proxy that strips - * it, and its use of whichever of the two is larger, since the header can lag a fresh push). - * ({@code skipHidden} only strips hidden fields from each item, not whole items, so it does not - * affect paging.) - * - * The scanned count is also capped at what the page's request asked for — see {@see scannedRows()} - * for why {@code unwind} specifically makes that cap necessary here (unlike the offset/limit - * collection iterators in {@see \Apify\Client\Internal\ResourceContext::paginateOffset()}, which - * have no such cap because none of their endpoints has an item-multiplying transform). The - * listing's reported {@code total} is never consulted for the stop condition, there or here: it can - * grow while the iteration runs (e.g. a run still pushing to the dataset being read), so with no - * {@code $limit} the loop always ends with one request that scans nothing, rather than stopping at - * the size the dataset had when it started. + * the number of rows the API scanned to produce it. The iterator therefore pages like every other + * collection iterator (see {@see ResourceContext::paginateOffset()}): by the page's + * {@see PaginationList::getCount()}, the scanned rows or the returned items when larger, capped at the + * page's requested limit, never by the returned item count alone. ({@code skipHidden} only strips + * hidden fields from each item, not whole items, so it does not affect paging.) * * @return Generator */ public function iterateItems(?DatasetListItemsOptions $options = null, ?int $chunkSize = null, int|float|string|null $timeoutSecs = null): Generator { $options ??= new DatasetListItemsOptions(); - $startOffset = $options->offset ?? 0; - $limit = $options->limit; $resolved = $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_LONG); - $firstPageLimit = ResourceContext::minLimit($limit, $chunkSize); - [$page, $scanned] = DatasetItemsPage::fetch($this->http, $this->ctx, $options->withPagination($startOffset, $firstPageLimit), $resolved); - foreach ($page->getItems() as $item) { - yield $item; - } - - $pageScanned = self::scannedRows($page, $scanned, $firstPageLimit); - $currentOffset = $startOffset + $pageScanned; - // Items still to yield, or null for "unbounded" (no $limit given). - $remaining = ($limit !== null && $limit > 0) ? $limit - $pageScanned : null; - - // Two distinct ways to stop: a page that scans nothing (the dataset is exhausted), or - // $remaining having already reached 0 (the caller's $limit is satisfied) — either ends the loop. - while ($pageScanned > 0 && ($remaining === null || $remaining > 0)) { - $pageLimit = ResourceContext::minLimit($remaining, $chunkSize); - [$page, $scanned] = DatasetItemsPage::fetch($this->http, $this->ctx, $options->withPagination($currentOffset, $pageLimit), $resolved); - foreach ($page->getItems() as $item) { - yield $item; - } - $pageScanned = self::scannedRows($page, $scanned, $pageLimit); - $currentOffset += $pageScanned; - if ($remaining !== null) { - $remaining -= $pageScanned; - } - } - } - - /** - * The number of rows a page advances the offset by: the larger of its scanned-count header (if - * present) and its returned item count, capped at the limit requested for it ({@code null} = - * uncapped). - * - * Unlike {@see \Apify\Client\Internal\ResourceContext}'s plain {@code advancedBy()} (which has no - * such cap — see its docblock), this one is load-bearing whenever {@code unwind} is set: - * {@code unwind} turns each scanned row into one item per unwound array element, so the returned - * item count can legitimately exceed the number of rows the API scanned to produce it (the reverse - * of {@code skipEmpty}/{@code clean}, which only ever drop rows). The server never scans more rows - * for a page than that page's requested limit, so the true row advance can never exceed it either; - * capping the {@code max()} above at the requested limit keeps an unwound page's inflated item count - * (or, if the scanned-count header is also missing, its unwind-inflated fallback) from overstating - * the real advance, which would otherwise skip whatever rows lie between the true (smaller) advance - * and the inflated one. - * - * @param PaginationList $page - */ - private static function scannedRows(PaginationList $page, ?int $scannedHeader, ?int $requestedLimit): int - { - $scanned = max($scannedHeader ?? 0, count($page->getItems())); - return $requestedLimit !== null ? min($scanned, $requestedLimit) : $scanned; + return ResourceContext::paginateOffset( + $options->offset ?? 0, + $options->limit, + $chunkSize, + fn (int $offset, ?int $pageLimit) => DatasetItemsPage::fetch( + $this->http, + $this->ctx, + $options->withPagination($offset, $pageLimit), + $resolved, + ), + ); } /** diff --git a/src/Resource/DatasetCollectionClient.php b/src/Resource/DatasetCollectionClient.php index 20112bb..a144335 100644 --- a/src/Resource/DatasetCollectionClient.php +++ b/src/Resource/DatasetCollectionClient.php @@ -39,7 +39,7 @@ public function list(?StorageListOptions $options = null, int|float|string|null /** * Lazily iterates over datasets, fetching pages on demand. The options' {@code limit} caps the * total number of datasets yielded across all pages ({@code null} = all); {@code $chunkSize} is - * the per-page size ({@code null} = the server default). + * the per-page size ({@code null} = the API's default page size). * * @return Generator */ diff --git a/src/Resource/KeyValueStoreClient.php b/src/Resource/KeyValueStoreClient.php index 7c9a193..c6c125a 100644 --- a/src/Resource/KeyValueStoreClient.php +++ b/src/Resource/KeyValueStoreClient.php @@ -103,7 +103,7 @@ public function listKeys(?ListKeysOptions $options = null, int|float|string|null * The options' {@code limit} caps the total number of keys yielded across all pages ({@code null} * or {@code 0} = all); {@code exclusiveStartKey} starts the listing after a given key; {@code prefix} and * {@code collection} restrict which keys are listed. {@code $chunkSize} caps how many keys are - * requested per page ({@code null} = the remaining total cap, or the server default when + * requested per page ({@code null} = the remaining total cap, or the API's default page size when * unbounded) — the two are independent, matching the offset/limit collection iterators. * * @return Generator diff --git a/src/Resource/KeyValueStoreCollectionClient.php b/src/Resource/KeyValueStoreCollectionClient.php index bc44294..df81543 100644 --- a/src/Resource/KeyValueStoreCollectionClient.php +++ b/src/Resource/KeyValueStoreCollectionClient.php @@ -39,7 +39,7 @@ public function list(?StorageListOptions $options = null, int|float|string|null /** * Lazily iterates over key-value stores, fetching pages on demand. The options' {@code limit} * caps the total number of stores yielded across all pages ({@code null} = all); {@code $chunkSize} - * is the per-page size ({@code null} = the server default). + * is the per-page size ({@code null} = the API's default page size). * * @return Generator */ diff --git a/src/Resource/RequestQueueClient.php b/src/Resource/RequestQueueClient.php index 2b16c0e..ef8bd18 100644 --- a/src/Resource/RequestQueueClient.php +++ b/src/Resource/RequestQueueClient.php @@ -136,7 +136,7 @@ public function delete(int|float|string|null $timeoutSecs = null): void /** * Returns the requests at the head (front) of the queue, up to {@code $limit} ({@code null} for - * the server default). + * the API's default limit). */ public function listHead(?int $limit = null, int|float|string|null $timeoutSecs = null): RequestQueueHead { diff --git a/src/Resource/RequestQueueCollectionClient.php b/src/Resource/RequestQueueCollectionClient.php index 2d6e780..6893059 100644 --- a/src/Resource/RequestQueueCollectionClient.php +++ b/src/Resource/RequestQueueCollectionClient.php @@ -39,7 +39,7 @@ public function list(?StorageListOptions $options = null, int|float|string|null /** * Lazily iterates over request queues, fetching pages on demand. The options' {@code limit} caps * the total number of queues yielded across all pages ({@code null} = all); {@code $chunkSize} is - * the per-page size ({@code null} = the server default). + * the per-page size ({@code null} = the API's default page size). * * @return Generator */ diff --git a/src/Resource/RunClient.php b/src/Resource/RunClient.php index a860c5f..e403112 100644 --- a/src/Resource/RunClient.php +++ b/src/Resource/RunClient.php @@ -327,7 +327,7 @@ public function iterateDatasetItems(?RunIterateDatasetItemsOptions $options = nu while ($position < $itemCount) { $pageLimit = min($pageSize, $itemCount - $position); - [$page] = DatasetItemsPage::fetch($this->http, $itemsCtx, self::pageOptions($options, $position, $pageLimit), $requestTimeoutSecs); + $page = DatasetItemsPage::fetch($this->http, $itemsCtx, self::pageOptions($options, $position, $pageLimit), $requestTimeoutSecs); foreach ($page->getItems() as $item) { yield $item; } @@ -346,14 +346,14 @@ public function iterateDatasetItems(?RunIterateDatasetItemsOptions $options = nu // The run has finished but `itemCount` lags the pushed items: read on, a page at a time. while (true) { $pageLimit = $end !== null ? min($pageSize, $end - $position) : $pageSize; - [$page, $scanned] = DatasetItemsPage::fetch($this->http, $itemsCtx, self::pageOptions($options, $position, $pageLimit), $requestTimeoutSecs); + $page = DatasetItemsPage::fetch($this->http, $itemsCtx, self::pageOptions($options, $position, $pageLimit), $requestTimeoutSecs); foreach ($page->getItems() as $item) { yield $item; } // Only an empty page marks the end, as filters can shorten a full one. A page that `clean`, // `skipEmpty` or `unwind` emptied past a lagging `itemCount` reports no scanned rows either, // so a plain, unfiltered one-row read confirms the dataset really has nothing left. - $isEmpty = $page->getItems() === [] && ($scanned ?? 0) === 0; + $isEmpty = $page->getCount() === 0; if ($isEmpty && (!self::hasRowChangingFilter($options) || $this->hasNoItemsAt($datasetClient, $position, $timeoutSecs))) { return; } diff --git a/src/Resource/RunCollectionClient.php b/src/Resource/RunCollectionClient.php index aa81602..3df1a03 100644 --- a/src/Resource/RunCollectionClient.php +++ b/src/Resource/RunCollectionClient.php @@ -45,7 +45,7 @@ public function list(?ListOptions $options = null, ?RunListOptions $filter = nul /** * Lazily iterates over runs, fetching pages on demand and applying the run-specific filters to * every page. The options' {@code limit} caps the total number of runs yielded across all pages - * ({@code null} = all); {@code $chunkSize} is the per-page size ({@code null} = the server default). + * ({@code null} = all); {@code $chunkSize} is the per-page size ({@code null} = the API's default page size). * * @return Generator */ diff --git a/src/Resource/ScheduleCollectionClient.php b/src/Resource/ScheduleCollectionClient.php index d7cf4d7..103e284 100644 --- a/src/Resource/ScheduleCollectionClient.php +++ b/src/Resource/ScheduleCollectionClient.php @@ -39,7 +39,7 @@ public function list(?ListOptions $options = null, int|float|string|null $timeou /** * Lazily iterates over the account's schedules, fetching pages on demand. The options' * {@code limit} caps the total number of schedules yielded across all pages ({@code null} = all); - * {@code $chunkSize} is the per-page size ({@code null} = the server default). + * {@code $chunkSize} is the per-page size ({@code null} = the API's default page size). * * @return Generator */ diff --git a/src/Resource/StoreCollectionClient.php b/src/Resource/StoreCollectionClient.php index 8b55c10..821c56f 100644 --- a/src/Resource/StoreCollectionClient.php +++ b/src/Resource/StoreCollectionClient.php @@ -40,7 +40,7 @@ public function list(?StoreListOptions $options = null, int|float|string|null $t * Lazily iterates over Store Actors matching the options, fetching pages on demand. * * The options' {@code limit} caps the total number of Actors yielded across all pages ({@code - * null} = all); {@code $chunkSize} is the per-page size ({@code null} = the server default). + * null} = all); {@code $chunkSize} is the per-page size ({@code null} = the API's default page size). * * @return Generator */ diff --git a/src/Resource/TaskCollectionClient.php b/src/Resource/TaskCollectionClient.php index 50daafe..7c72556 100644 --- a/src/Resource/TaskCollectionClient.php +++ b/src/Resource/TaskCollectionClient.php @@ -39,7 +39,7 @@ public function list(?ListOptions $options = null, int|float|string|null $timeou /** * Lazily iterates over the account's tasks, fetching pages on demand. The options' {@code limit} * caps the total number of tasks yielded across all pages ({@code null} = all); {@code $chunkSize} - * is the per-page size ({@code null} = the server default). + * is the per-page size ({@code null} = the API's default page size). * * @return Generator */ diff --git a/src/Resource/WebhookDispatchCollectionClient.php b/src/Resource/WebhookDispatchCollectionClient.php index a6524ba..3d14233 100644 --- a/src/Resource/WebhookDispatchCollectionClient.php +++ b/src/Resource/WebhookDispatchCollectionClient.php @@ -42,7 +42,7 @@ public function list(?ListOptions $options = null, int|float|string|null $timeou /** * Lazily iterates over webhook dispatches, fetching pages on demand. The options' {@code limit} * caps the total number of dispatches yielded across all pages ({@code null} = all); - * {@code $chunkSize} is the per-page size ({@code null} = the server default). + * {@code $chunkSize} is the per-page size ({@code null} = the API's default page size). * * @return Generator */ diff --git a/src/Version.php b/src/Version.php index f968a2b..0772ed8 100644 --- a/src/Version.php +++ b/src/Version.php @@ -17,13 +17,13 @@ final class Version * The semantic version of this client library (see https://semver.org/). * Changes to the public interface other than additive ones are considered breaking changes. */ - public const CLIENT_VERSION = '0.12.0'; + public const CLIENT_VERSION = '0.13.0'; /** * The version of the Apify OpenAPI specification this client was generated and verified * against. Corresponds to the {@code info.version} field of the Apify OpenAPI document. */ - public const API_SPEC_VERSION = 'v2-2026-11-01T214220Z'; + public const API_SPEC_VERSION = 'v2-2026-11-06T140128Z'; private function __construct() { diff --git a/tests/Examples/DocSnippetsTest.php b/tests/Examples/DocSnippetsTest.php index a1fab47..ef80c53 100644 --- a/tests/Examples/DocSnippetsTest.php +++ b/tests/Examples/DocSnippetsTest.php @@ -82,6 +82,7 @@ private function imports(): string 'Apify\Client\Options\ActorListOptions', 'Apify\Client\Options\ActorStartOptions', 'Apify\Client\Options\ActorBuildOptions', + 'Apify\Client\Options\ActorServerRequestOptions', 'Apify\Client\Options\ListOptions', 'Apify\Client\Options\StorageListOptions', 'Apify\Client\Options\StoreListOptions', diff --git a/tests/Integration/ActorIntegrationTest.php b/tests/Integration/ActorIntegrationTest.php index ef41eaa..7241197 100644 --- a/tests/Integration/ActorIntegrationTest.php +++ b/tests/Integration/ActorIntegrationTest.php @@ -90,6 +90,34 @@ public function testActorVersionCrudFlow(): void } } + public function testActorVersionActorPathIsStoredAndRemoved(): void + { + $client = $this->requireClient(); + $created = $client->actors()->create(self::minimalActor(self::uniqueName('actor-path'))); + try { + $actor = $client->actor((string) $created->getId()); + $version = $actor->versions()->create([ + 'versionNumber' => '0.1', + 'sourceType' => 'SOURCE_FILES', + 'buildTag' => 'latest', + 'sourceFiles' => [], + 'actorPath' => 'actors/typescript-actor', + ]); + self::assertSame('actors/typescript-actor', $version->getActorPath()); + self::assertSame('actors/typescript-actor', $actor->version('0.1')->get()?->getActorPath()); + + // An update of other fields keeps the path. + $kept = $actor->version('0.1')->update(['buildTag' => 'beta']); + self::assertSame('actors/typescript-actor', $kept->getActorPath()); + + // An upload without `actorPath` removes it. + $removed = $actor->version('0.1')->update(['sourceType' => 'SOURCE_FILES', 'sourceFiles' => []]); + self::assertNull($removed->getActorPath()); + } finally { + $client->actor((string) $created->getId())->delete(); + } + } + public function testIterateActors(): void { $client = $this->requireClient(); diff --git a/tests/Integration/ActorServerIntegrationTest.php b/tests/Integration/ActorServerIntegrationTest.php new file mode 100644 index 0000000..8db6ea6 --- /dev/null +++ b/tests/Integration/ActorServerIntegrationTest.php @@ -0,0 +1,128 @@ + '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 {@code status} query parameter, or 200. + */ + private const ECHO_SERVER_SOURCE = <<<'JS' +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); +JS; + + private const DOCKERFILE = "FROM node:22-alpine\nCOPY main.js ./\nCMD [\"node\", \"main.js\"]\n"; + + public function testServerClientReturnsTheUrlAndSpecificationAndSendsRequestsToTheServer(): void + { + $client = $this->requireClient(); + $actorName = self::uniqueName('server'); + $created = $client->actors()->create([ + 'name' => $actorName, + 'versions' => [[ + 'versionNumber' => '0.0', + 'sourceType' => 'SOURCE_FILES', + 'buildTag' => 'latest', + 'sourceFiles' => [ + ['name' => 'Dockerfile', 'format' => 'TEXT', 'content' => self::DOCKERFILE], + ['name' => 'main.js', 'format' => 'TEXT', 'content' => self::ECHO_SERVER_SOURCE], + ['name' => '.actor/actor.json', 'format' => 'TEXT', 'content' => json_encode([ + 'actorSpecification' => 1, + 'name' => $actorName, + 'version' => '0.0', + 'usesStandbyMode' => true, + 'webServerSchema' => self::OPENAPI_SPEC, + ], JSON_THROW_ON_ERROR)], + ], + ]], + 'actorStandby' => ['isEnabled' => true, 'memoryMbytes' => 256, 'idleTimeoutSecs' => 60], + ]); + $actorId = (string) $created->getId(); + try { + $build = $client->actor($actorId)->build('0.0'); + $finished = $client->build((string) $build->getId())->waitForFinish(self::BUILD_WAIT_SECS); + self::assertSame('SUCCEEDED', $finished->getStatus()); + + $actor = $client->actor($actorId)->get(); + self::assertNotNull($actor); + $server = $client->actor($actorId)->server(); + + self::assertSame($actor->getStandbyUrl(), $server->getUrl()); + self::assertSame(self::OPENAPI_SPEC, $server->getOpenApiSpec()); + + $response = $server->request('/echo', new ActorServerRequestOptions(method: 'POST', params: ['q' => 'apify'], body: ['limit' => 5])); + $echoed = $response->getData(); + self::assertIsArray($echoed); + self::assertSame(200, $response->getStatusCode()); + self::assertSame('POST', $echoed['method']); + self::assertSame('/echo?q=apify', $echoed['url']); + self::assertSame(['limit' => 5], json_decode($echoed['body'], true)); + // The platform consumes the client's token, so it never reaches the Actor's code. + self::assertArrayNotHasKey('x-apify-authorization', $echoed['headers']); + self::assertArrayNotHasKey('authorization', $echoed['headers']); + + $own = $server->request('/echo', new ActorServerRequestOptions(headers: ['Authorization' => 'Bearer actor-token']))->getData(); + self::assertIsArray($own); + self::assertSame('Bearer actor-token', $own['headers']['authorization']); + + $error = $server->request('/echo', new ActorServerRequestOptions(params: ['status' => 404])); + self::assertSame(404, $error->getStatusCode()); + $errorBody = $error->getData(); + self::assertIsArray($errorBody); + self::assertSame('/echo?status=404', $errorBody['url']); + } finally { + $client->actor($actorId)->delete(); + } + } + + public function testServerClientRejectsAnActorThatIsNotAServerActor(): void + { + $client = $this->requireClient(); + $created = $client->actors()->create(self::minimalActor(self::uniqueName('not-server'))); + try { + $this->expectException(ApifyClientException::class); + $this->expectExceptionMessageMatches('/is not a Server Actor/'); + $client->actor((string) $created->getId())->server()->getUrl(); + } finally { + $client->actor((string) $created->getId())->delete(); + } + } + + public function testServerClientThrowsNotFoundForAMissingActor(): void + { + $client = $this->requireClient(); + + $this->expectException(NotFoundException::class); + $client->actor(self::uniqueName('missing'))->server()->getUrl(); + } +} diff --git a/tests/Integration/DatasetIntegrationTest.php b/tests/Integration/DatasetIntegrationTest.php index 2f6015f..1c3bdb6 100644 --- a/tests/Integration/DatasetIntegrationTest.php +++ b/tests/Integration/DatasetIntegrationTest.php @@ -64,16 +64,8 @@ public function testIterateDatasetItems(): void $dataset = $client->dataset((string) $ds->getId()); $dataset->pushItems([['n' => 0], ['n' => 1], ['n' => 2], ['n' => 3], ['n' => 4]]); - // The dataset's item total is computed asynchronously and can briefly lag a write. - // iterateItems() pages by the reported total (matching the reference client), so wait for - // the count to settle before iterating; otherwise a stale total would stop it early. - $deadline = microtime(true) + 30.0; - while ( - $dataset->listItems(new DatasetListItemsOptions())->getTotal() < 5 - && microtime(true) < $deadline - ) { - usleep(500_000); - } + // The dataset's item count lags a write, so wait for it to settle before iterating. + self::waitForDatasetItemCount($dataset, 5); $values = []; // chunkSize=2 across 5 items => three pages (2, 2, 1). @@ -109,6 +101,28 @@ public function testStreamItemsMatchesDownloadItems(): void } } + public function testListItemsCountReportsTheScannedRowsWhenAFilterDropsItems(): void + { + $client = $this->requireClient(); + $ds = $client->datasets()->getOrCreate(self::uniqueName('ds-count')); + try { + $dataset = $client->dataset((string) $ds->getId()); + $dataset->pushItems([['n' => 1], new \stdClass(), ['n' => 3], new \stdClass(), ['n' => 5]]); + + self::waitForDatasetItemCount($dataset, 5); + $page = $dataset->listItems(new DatasetListItemsOptions(skipEmpty: true)); + self::assertSame([1, 3, 5], array_column($page->getItems(), 'n')); + // `count` is the five rows scanned, not the three items returned. + self::assertSame(5, $page->getCount()); + + // The iterator steps by the scanned rows, so it neither repeats nor skips an item. + $iterated = iterator_to_array($dataset->iterateItems(new DatasetListItemsOptions(skipEmpty: true), 2), false); + self::assertSame([1, 3, 5], array_column($iterated, 'n')); + } finally { + $client->dataset((string) $ds->getId())->delete(); + } + } + public function testDatasetCrudFlow(): void { $client = $this->requireClient(); diff --git a/tests/Integration/IntegrationTestCase.php b/tests/Integration/IntegrationTestCase.php index 3ea6a68..8779a7f 100644 --- a/tests/Integration/IntegrationTestCase.php +++ b/tests/Integration/IntegrationTestCase.php @@ -5,6 +5,8 @@ namespace Apify\Client\Tests\Integration; use Apify\Client\ApifyClient; +use Apify\Client\Options\DatasetListItemsOptions; +use Apify\Client\Resource\DatasetClient; use PHPUnit\Framework\TestCase; /** @@ -105,6 +107,21 @@ protected static function assertEventuallyIterated(array $expectedIds, callable } } + /** + * Waits until the dataset reports at least {@code $expectedItems} items. The dataset's item count, + * which its total and its scanned-rows count come from, is computed asynchronously and lags a fresh + * push. It re-reads as often as {@see assertEventuallyIterated()} does. + */ + protected static function waitForDatasetItemCount(DatasetClient $dataset, int $expectedItems): void + { + for ($attempt = 1; $attempt <= self::LISTING_MAX_ATTEMPTS; $attempt++) { + if ($dataset->listItems(new DatasetListItemsOptions())->getTotal() >= $expectedItems) { + return; + } + sleep(self::LISTING_RETRY_DELAY_SECS); + } + } + /** * A minimal Actor definition; the API requires at least one version. * diff --git a/tests/Unit/ActorServerClientTest.php b/tests/Unit/ActorServerClientTest.php new file mode 100644 index 0000000..86b437b --- /dev/null +++ b/tests/Unit/ActorServerClientTest.php @@ -0,0 +1,531 @@ + $extra */ + private static function actorBody(?string $standbyUrl = self::SERVER_URL, array $extra = []): string + { + return Json::encode(['data' => array_merge(['id' => 'a1', 'standbyUrl' => $standbyUrl], $extra)]); + } + + /** @param array|null $actorDefinition */ + private static function buildBody(string $id, string $buildNumber, ?array $actorDefinition = null): string + { + $build = ['id' => $id, 'buildNumber' => $buildNumber]; + if ($actorDefinition !== null) { + $build['actorDefinition'] = $actorDefinition; + } + return Json::encode(['data' => $build]); + } + + // ----- getUrl() ---------------------------------------------------------------- + + public function testGetUrlFetchesTheActorOnceAndCachesTheUrl(): void + { + $transport = (new MockTransport())->queueResponse(200, self::actorBody()); + $server = $this->client($transport)->actor('jane/my-actor')->server(); + + self::assertSame(self::SERVER_URL, $server->getUrl()); + self::assertSame(self::SERVER_URL, $server->getUrl()); + + self::assertSame(1, $transport->callCount()); + self::assertSame('GET', $transport->lastRequest()->getMethod()); + self::assertStringEndsWith('/v2/actors/jane~my-actor', (string) $transport->lastRequest()->getUri()); + } + + public function testServerReturnsTheSameClientSoTheUrlIsSharedAcrossCalls(): void + { + $transport = (new MockTransport())->queueResponse(200, self::actorBody()); + $actor = $this->client($transport)->actor('a1'); + + self::assertSame($actor->server(), $actor->server()); + $actor->server()->getUrl(); + $actor->server()->getUrl(); + self::assertSame(1, $transport->callCount()); + } + + public function testGetUrlRejectsAnActorThatIsNotAServerActor(): void + { + $transport = (new MockTransport())->queueResponse(200, self::actorBody(null)); + + $this->expectException(ApifyClientException::class); + $this->expectExceptionMessage('Actor a1 is not a Server Actor.'); + $this->client($transport)->actor('a1')->server()->getUrl(); + } + + public function testAFailedLookupIsNotCached(): void + { + $transport = (new MockTransport()) + ->queueResponse(200, self::actorBody(null)) + ->queueResponse(200, self::actorBody()); + $server = $this->client($transport)->actor('a1')->server(); + + try { + $server->getUrl(); + self::fail('expected an ApifyClientException'); + } catch (ApifyClientException) { + } + + self::assertSame(self::SERVER_URL, $server->getUrl()); + } + + public function testGetUrlThrowsNotFoundForAMissingActor(): void + { + $transport = (new MockTransport())->queueResponse(404, Json::encode(['error' => ['type' => 'record-not-found', 'message' => 'Actor was not found']])); + + $this->expectException(NotFoundException::class); + $this->client($transport)->actor('missing')->server()->getUrl(); + } + + // ----- getOpenApiSpec() ----------------------------------------------------------- + + public function testGetOpenApiSpecReadsTheWebServerSchemaOfTheLatestBuildByDefault(): void + { + $schema = ['openapi' => '3.0.0', 'paths' => ['/search' => []]]; + $transport = (new MockTransport()) + ->queueResponse(200, self::actorBody(extra: ['taggedBuilds' => ['latest' => ['buildId' => 'b-latest'], 'beta' => ['buildId' => 'b-beta']]])) + ->queueResponse(200, self::buildBody('b-latest', '0.1.5', ['webServerSchema' => $schema])); + + $spec = $this->client($transport)->actor('a1')->server()->getOpenApiSpec(); + + self::assertSame($schema, $spec); + self::assertStringEndsWith('/v2/actor-builds/b-latest', (string) $transport->lastRequest()->getUri()); + } + + public function testGetOpenApiSpecUsesTheBuildTagTheActorStandbyNames(): void + { + $transport = (new MockTransport()) + ->queueResponse(200, self::actorBody(extra: [ + 'actorStandby' => ['build' => 'beta'], + 'taggedBuilds' => ['latest' => ['buildId' => 'b-latest'], 'beta' => ['buildId' => 'b-beta']], + ])) + ->queueResponse(200, self::buildBody('b-beta', '0.2.0', ['webServerSchema' => ['openapi' => 'beta']])); + + $spec = $this->client($transport)->actor('a1')->server()->getOpenApiSpec(); + + self::assertSame(['openapi' => 'beta'], $spec); + self::assertStringEndsWith('/v2/actor-builds/b-beta', (string) $transport->lastRequest()->getUri()); + } + + public function testGetOpenApiSpecLooksUpABuildNumberInTheBuildList(): void + { + $page = Json::encode(['data' => ['items' => [ + ['id' => 'b3', 'buildNumber' => '0.1.7'], + ['id' => 'b2', 'buildNumber' => '0.1.6'], + ['id' => 'b1', 'buildNumber' => '0.1.5'], + ], 'total' => 3, 'offset' => 0, 'limit' => 1000, 'count' => 3, 'desc' => true]]); + $transport = (new MockTransport()) + ->queueResponse(200, self::actorBody(extra: ['actorStandby' => ['build' => '0.1.6'], 'taggedBuilds' => []])) + ->queueResponse(200, $page) + ->queueResponse(200, self::buildBody('b2', '0.1.6', ['webServerSchema' => ['openapi' => 'by-number']])); + + $spec = $this->client($transport)->actor('a1')->server()->getOpenApiSpec(); + + self::assertSame(['openapi' => 'by-number'], $spec); + self::assertStringContainsString('/v2/actors/a1/builds?', (string) $transport->received[1]->getUri()); + self::assertStringContainsString('desc=1', (string) $transport->received[1]->getUri()); + self::assertStringEndsWith('/v2/actor-builds/b2', (string) $transport->lastRequest()->getUri()); + } + + public function testGetOpenApiSpecStopsLookingAtAnOlderBuildOfTheSameVersion(): void + { + $page = Json::encode(['data' => ['items' => [ + ['id' => 'b3', 'buildNumber' => '0.1.7'], + ['id' => 'b1', 'buildNumber' => '0.1.5'], + ], 'total' => 2, 'offset' => 0, 'limit' => 1000, 'count' => 2, 'desc' => true]]); + // Only one page is queued: meeting 0.1.5 means 0.1.6 does not exist, so the next page is never read. + $transport = (new MockTransport()) + ->queueResponse(200, self::actorBody(extra: ['actorStandby' => ['build' => '0.1.6']])) + ->queueResponse(200, $page); + + try { + $this->client($transport)->actor('a1')->server()->getOpenApiSpec(); + self::fail('expected an ApifyClientException'); + } catch (ApifyClientException $e) { + self::assertSame('Build 0.1.6 of Actor a1 does not exist.', $e->getMessage()); + } + self::assertSame(2, $transport->callCount()); + } + + public function testGetOpenApiSpecRejectsAnUnknownTag(): void + { + $transport = (new MockTransport())->queueResponse(200, self::actorBody(extra: ['actorStandby' => ['build' => 'nope'], 'taggedBuilds' => []])); + + $this->expectException(ApifyClientException::class); + $this->expectExceptionMessage('Build nope of Actor a1 does not exist.'); + $this->client($transport)->actor('a1')->server()->getOpenApiSpec(); + } + + public function testGetOpenApiSpecReturnsNullWhenTheDefinitionHasNoWebServerSchema(): void + { + $transport = (new MockTransport()) + ->queueResponse(200, self::actorBody(extra: ['taggedBuilds' => ['latest' => ['buildId' => 'b1']]])) + ->queueResponse(200, self::buildBody('b1', '0.1.1', ['actorSpecification' => 1])); + + self::assertNull($this->client($transport)->actor('a1')->server()->getOpenApiSpec()); + } + + public function testGetOpenApiSpecAlsoCachesTheServerUrl(): void + { + $transport = (new MockTransport()) + ->queueResponse(200, self::actorBody(extra: ['taggedBuilds' => ['latest' => ['buildId' => 'b1']]])) + ->queueResponse(200, self::buildBody('b1', '0.1.1', ['webServerSchema' => []])); + $server = $this->client($transport)->actor('a1')->server(); + + $server->getOpenApiSpec(); + + self::assertSame(self::SERVER_URL, $server->getUrl()); + self::assertSame(2, $transport->callCount()); + } + + // ----- request() ------------------------------------------------------------------- + + private function serverTransport(): MockTransport + { + return (new MockTransport())->queueResponse(200, self::actorBody()); + } + + public function testRequestGoesToTheServerUrlWithTheTokenInTheServerAuthorizationHeader(): void + { + $transport = $this->serverTransport()->queueResponse(200, '{"hits":3}', ['Content-Type' => 'application/json']); + + $response = $this->client($transport)->actor('a1')->server()->request('/search', new ActorServerRequestOptions(params: ['q' => 'apify web', 'page' => 2, 'exact' => true, 'tags' => ['a', 'b'], 'skipped' => null])); + + $request = $transport->lastRequest(); + self::assertSame('GET', $request->getMethod()); + self::assertSame(self::SERVER_URL . '/search?q=apify%20web&page=2&exact=1&tags=a%2Cb', (string) $request->getUri()); + self::assertSame('Bearer secret-token', $request->getHeaderLine('X-Apify-Authorization')); + self::assertFalse($request->hasHeader('Authorization')); + self::assertStringStartsWith('ApifyClient/', $request->getHeaderLine('User-Agent')); + self::assertSame(200, $response->getStatusCode()); + self::assertTrue($response->isSuccessful()); + self::assertSame(['hits' => 3], $response->getData()); + self::assertSame('{"hits":3}', $response->getBody()); + self::assertSame('application/json', $response->getHeaderLine('content-type')); + } + + public function testRequestSendsNoRedirectsAndTheUserHeadersAsTheyAre(): void + { + $transport = $this->serverTransport()->queueResponse(200, 'ok'); + + $this->client($transport)->actor('a1')->server()->request('/', new ActorServerRequestOptions(headers: ['Authorization' => 'Bearer actor-secret', 'X-Custom' => 'yes'])); + + $request = $transport->lastRequest(); + self::assertSame('Bearer actor-secret', $request->getHeaderLine('Authorization')); + self::assertSame('Bearer secret-token', $request->getHeaderLine('X-Apify-Authorization')); + self::assertSame('yes', $request->getHeaderLine('X-Custom')); + // The lookup of the Actor is an API call and follows redirects, the request to the server does not. + self::assertSame([false, true], $transport->withoutRedirects); + } + + public function testRequestUsesTheLongTierTimeoutByDefaultAndAnExplicitOne(): void + { + $transport = $this->serverTransport()->queueResponse(200, 'a')->queueResponse(200, 'b'); + $client = new ApifyClient(token: 't', timeoutSecs: 5, timeoutShortSecs: 3, timeoutLongSecs: 90, timeoutMaxSecs: 120, httpClient: $transport); + $server = $client->actor('a1')->server(); + + $server->request('/'); + $server->request('/', timeoutSecs: 7); + + self::assertSame([3.0, 90.0, 7.0], $transport->timeouts); + } + + /** @return array */ + public static function pathProvider(): array + { + return [ + 'empty path is the root' => ['', self::SERVER_URL . '/'], + 'slash is the root' => ['/', self::SERVER_URL . '/'], + 'relative path' => ['search/items', self::SERVER_URL . '/search/items'], + 'fragment is dropped' => ['/search#top', self::SERVER_URL . '/search'], + 'leading slashes cannot make a host' => ['//evil.example/steal', self::SERVER_URL . '/evil.example/steal'], + 'an absolute URL stays a path' => ['https://evil.example/steal', self::SERVER_URL . '/https://evil.example/steal'], + ]; + } + + #[DataProvider('pathProvider')] + public function testRequestAlwaysResolvesThePathAgainstTheServerUrl(string $path, string $expectedUrl): void + { + $transport = $this->serverTransport()->queueResponse(200, 'ok'); + + $this->client($transport)->actor('a1')->server()->request($path); + + self::assertSame($expectedUrl, (string) $transport->lastRequest()->getUri()); + } + + public function testRequestIgnoresATrailingSlashOfTheServerUrl(): void + { + $transport = (new MockTransport())->queueResponse(200, self::actorBody(self::SERVER_URL . '/'))->queueResponse(200, 'ok'); + + $this->client($transport)->actor('a1')->server()->request('/x'); + + self::assertSame(self::SERVER_URL . '/x', (string) $transport->lastRequest()->getUri()); + } + + public function testRequestSerializesAnArrayBodyAsJsonAndAStringAsItIs(): void + { + $transport = $this->serverTransport()->queueResponse(200, 'a')->queueResponse(200, 'b')->queueResponse(200, 'c'); + $server = $this->client($transport)->actor('a1')->server(); + + $server->request('/json', new ActorServerRequestOptions(method: 'post', body: ['query' => 'apify'])); + $json = $transport->lastRequest(); + $server->request('/text', new ActorServerRequestOptions(method: 'PUT', body: 'raw bytes', headers: ['content-type' => 'text/plain'])); + $text = $transport->lastRequest(); + $server->request('/plain', new ActorServerRequestOptions(method: 'POST', body: 'no content type')); + $plain = $transport->lastRequest(); + + self::assertSame('POST', $json->getMethod()); + self::assertSame('application/json', $json->getHeaderLine('Content-Type')); + self::assertSame('{"query":"apify"}', (string) $json->getBody()); + self::assertSame('PUT', $text->getMethod()); + self::assertSame(['text/plain'], $text->getHeader('Content-Type')); + self::assertSame('raw bytes', (string) $text->getBody()); + self::assertSame('', $plain->getHeaderLine('Content-Type')); + self::assertSame('no content type', (string) $plain->getBody()); + } + + public function testRequestBodiesAreSentUncompressedUnlessCompressionIsSet(): void + { + $large = str_repeat('{"key":"value"},', 400); + $transport = $this->serverTransport()->queueResponse(200, 'a')->queueResponse(200, 'b'); + $server = $this->client($transport)->actor('a1')->server(); + + $server->request('/plain', new ActorServerRequestOptions(method: 'POST', body: $large, headers: ['Content-Type' => 'application/json'])); + $plain = $transport->lastRequest(); + $server->request('/gzip', new ActorServerRequestOptions(method: 'POST', body: $large, headers: ['Content-Type' => 'application/json'], compression: 'gzip')); + $gzip = $transport->lastRequest(); + + self::assertFalse($plain->hasHeader('Content-Encoding')); + self::assertSame($large, (string) $plain->getBody()); + self::assertSame('gzip', $gzip->getHeaderLine('Content-Encoding')); + self::assertSame($large, gzdecode((string) $gzip->getBody())); + } + + public function testACallerContentTypeThatIsAlreadyCompressedKeepsTheBodyUncompressed(): void + { + $png = str_repeat("\x89PNG", 800); + $transport = $this->serverTransport()->queueResponse(200, 'a'); + + $this->client($transport)->actor('a1')->server()->request('/', new ActorServerRequestOptions(method: 'POST', body: $png, headers: ['content-type' => 'image/png'], compression: 'gzip')); + + self::assertFalse($transport->lastRequest()->hasHeader('Content-Encoding')); + self::assertSame($png, (string) $transport->lastRequest()->getBody()); + self::assertSame('image/png', $transport->lastRequest()->getHeaderLine('Content-Type')); + } + + public function testACompressedBodyDropsTheCallerContentLength(): void + { + $large = str_repeat('{"key":"value"},', 400); + $transport = $this->serverTransport()->queueResponse(200, 'a'); + + $this->client($transport)->actor('a1')->server()->request('/', new ActorServerRequestOptions(method: 'POST', body: $large, headers: ['content-length' => (string) strlen($large), 'Content-Type' => 'application/json'], compression: 'gzip')); + + $sent = $transport->lastRequest(); + self::assertSame('gzip', $sent->getHeaderLine('Content-Encoding')); + self::assertFalse($sent->hasHeader('Content-Length')); + } + + public function testTheClientWideCompressionDoesNotApplyToServerRequests(): void + { + $large = str_repeat('{"key":"value"},', 400); + $transport = $this->serverTransport()->queueResponse(200, 'a'); + $client = new ApifyClient(token: 't', compression: 'gzip', httpClient: $transport); + + $client->actor('a1')->server()->request('/', new ActorServerRequestOptions(method: 'POST', body: $large, headers: ['Content-Type' => 'application/json'])); + + self::assertFalse($transport->lastRequest()->hasHeader('Content-Encoding')); + } + + public function testRequestRejectsAnUnknownCompressionAndABadQueryParameter(): void + { + $transport = $this->serverTransport()->queueResponse(200, 'a'); + $server = $this->client($transport)->actor('a1')->server(); + + try { + $server->request('/', new ActorServerRequestOptions(compression: 'zstd')); + self::fail('expected a RuntimeException'); + } catch (\RuntimeException $e) { + self::assertStringContainsString('unknown compression "zstd"', $e->getMessage()); + } + + $this->expectException(InvalidArgumentException::class); + $server->request('/', new ActorServerRequestOptions(params: ['nested' => ['a' => 1]])); + } + + public function testErrorStatusesAreReturnedNotThrown(): void + { + $transport = $this->serverTransport() + ->queueResponse(404, '{"error":"no such item"}', ['Content-Type' => 'application/json']) + ->queueResponse(500, 'boom', ['Content-Type' => 'text/plain']); + $server = $this->client($transport)->actor('a1')->server(); + + $notFound = $server->request('/missing'); + $failure = $server->request('/broken'); + + self::assertSame(404, $notFound->getStatusCode()); + self::assertFalse($notFound->isSuccessful()); + self::assertSame(['error' => 'no such item'], $notFound->getData()); + self::assertSame(500, $failure->getStatusCode()); + self::assertSame('boom', $failure->getData()); + // A 500 comes from the Actor's code, which may have handled the request: no retry. + self::assertSame(3, $transport->callCount()); + } + + public function testOnlyRateLimitAndServiceUnavailableAreRetriedAndTheLastOneIsReturned(): void + { + $transport = $this->serverTransport() + ->queueResponse(429, 'slow down') + ->queueResponse(503, 'starting') + ->queueResponse(200, 'finally'); + $server = $this->client($transport)->actor('a1')->server(); + + self::assertSame('finally', $server->request('/')->getBody()); + self::assertSame(4, $transport->callCount()); + + $exhausted = $this->serverTransport()->queueResponse(503, 'one')->queueResponse(503, 'two')->queueResponse(503, 'three'); + $response = $this->client($exhausted, maxRetries: 2)->actor('a1')->server()->request('/'); + + self::assertSame(503, $response->getStatusCode()); + self::assertSame('three', $response->getBody()); + self::assertSame(4, $exhausted->callCount()); + } + + public function testATransportErrorIsNotRetried(): void + { + $transport = $this->serverTransport()->queueError(timeout: true)->queueResponse(200, 'never sent'); + $server = $this->client($transport)->actor('a1')->server(); + + try { + $server->request('/slow'); + self::fail('expected a TransportException'); + } catch (TransportException $e) { + self::assertTrue($e->isTimeout()); + } + self::assertSame(2, $transport->callCount()); + } + + public function testARedirectIsReturnedAsItIs(): void + { + $transport = $this->serverTransport()->queueResponse(302, '', ['Location' => 'https://elsewhere.example/']); + + $response = $this->client($transport)->actor('a1')->server()->request('/old'); + + self::assertSame(302, $response->getStatusCode()); + self::assertSame('https://elsewhere.example/', $response->getHeaderLine('Location')); + self::assertSame(2, $transport->callCount()); + } + + public function testTheStreamOptionReadsTheBodyOnDemand(): void + { + $transport = $this->serverTransport()->queueResponse(200, 'line one'); + + $response = $this->client($transport)->actor('a1')->server()->request('/events', new ActorServerRequestOptions(stream: true)); + + self::assertSame([false, true], $transport->streamed); + self::assertSame('line one', (string) $response->getBodyStream()); + } + + public function testTheRequestFailsBeforeSendingWhenTheTransportCannotDisableRedirects(): void + { + $actor = new Response(200, [], self::actorBody()); + $psr18 = new StubPsr18Client($actor); + $client = new ApifyClient(token: 't', httpClient: new Psr18HttpClient($psr18)); + + try { + $client->actor('a1')->server()->request('/'); + self::fail('expected an ApifyClientException'); + } catch (ApifyClientException $e) { + self::assertStringContainsString('cannot disable redirects', $e->getMessage()); + } + // Only the lookup of the Actor went out. + self::assertSame(1, $psr18->calls); + } + + // ----- ActorServerResponse::getData() ---------------------------------------------- + + public function testGetDataParsesByContentType(): void + { + $transport = $this->serverTransport() + ->queueResponse(200, '{"a":[1,2]}', ['Content-Type' => 'application/vnd.api+json; charset=utf-8']) + ->queueResponse(200, 'hello', ['Content-Type' => 'text/plain']) + ->queueResponse(200, "\x00\x01", ['Content-Type' => 'application/octet-stream']) + ->queueResponse(200, '') + ->queueResponse(200, '{"a":1}' . "\n", ['Content-Type' => 'application/json-seq']); + $server = $this->client($transport)->actor('a1')->server(); + + self::assertSame(['a' => [1, 2]], $server->request('/1')->getData()); + self::assertSame('hello', $server->request('/2')->getData()); + self::assertSame("\x00\x01", $server->request('/3')->getData()); + self::assertSame('', $server->request('/4')->getData()); + self::assertSame('{"a":1}' . "\n", $server->request('/5')->getData()); + } + + public function testGetDataThrowsForAnInvalidJsonSuccessBodyAndReturnsRawBytesForAnErrorBody(): void + { + $transport = $this->serverTransport() + ->queueResponse(200, '{broken', ['Content-Type' => 'application/json']) + ->queueResponse(502, 'bad gateway', ['Content-Type' => 'application/json']); + $server = $this->client($transport)->actor('a1')->server(); + + $failure = $server->request('/ok'); + try { + $failure->getData(); + self::fail('expected an InvalidResponseBodyException'); + } catch (InvalidResponseBodyException $e) { + self::assertSame('{broken', $e->getRawBody()); + } + + self::assertSame('bad gateway', $server->request('/bad')->getData()); + } + + // ----- GuzzleHttpClient --------------------------------------------------------------- + + public function testGuzzleTransportFollowsRedirectsUnlessAskedNotTo(): void + { + $redirect = new Response(302, ['Location' => 'https://elsewhere.example/']); + $request = new Request('GET', 'https://example.com/start'); + + $followingMock = new MockHandler([$redirect, new Response(200, [], 'landed')]); + $following = new GuzzleHttpClient(new Guzzle(['handler' => HandlerStack::create($followingMock)])); + self::assertSame('landed', (string) $following->send($request, 5.0)->getBody()); + + $stoppingMock = new MockHandler([$redirect, new Response(200, [], 'landed')]); + $stopping = new GuzzleHttpClient(new Guzzle(['handler' => HandlerStack::create($stoppingMock)])); + $response = $stopping->sendWithoutRedirects($request, 5.0); + self::assertSame(302, $response->getStatusCode()); + self::assertSame(1, $stoppingMock->count(), 'the redirect target must not be requested'); + } +} diff --git a/tests/Unit/IterationTest.php b/tests/Unit/IterationTest.php index 0a7e949..5c503dd 100644 --- a/tests/Unit/IterationTest.php +++ b/tests/Unit/IterationTest.php @@ -44,6 +44,19 @@ private static function page(array $items, int $total, int $offset): string ]]); } + /** + * Builds a list-page envelope whose {@code count} differs from the number of items, as the API reports + * when it drops entries after counting them. + * + * @param list> $items + */ + private static function pageWithCount(array $items, int $count, int $offset): string + { + return Json::encode(['data' => [ + 'items' => $items, 'total' => 100, 'offset' => $offset, 'limit' => $count, 'count' => $count, 'desc' => false, + ]]); + } + /** * @param int ...$ids * @return list> @@ -126,6 +139,41 @@ public function testLimitIsTotalCapAndChunkSizeIsPageSize(): void self::assertStringContainsString('limit=1', $uris[2]); } + public function testStorePagesStepByTheCountEvenWhenTheApiDroppedItems(): void + { + // The Store counts Actors before it drops some (for example, hidden ones), so a page can carry fewer + // items than its `count`, or none. The iterator steps by the count: no repeats, no early end. + $transport = (new MockTransport()) + ->queueResponse(200, self::pageWithCount(self::actors(1, 2), 4, 0)) + ->queueResponse(200, self::pageWithCount([], 3, 4)) + ->queueResponse(200, self::pageWithCount(self::actors(3), 1, 7)) + ->queueResponse(200, self::pageWithCount([], 0, 8)); + $ids = []; + foreach ($this->client($transport)->store()->iterate() as $item) { + $ids[] = $item->getId(); + } + self::assertSame(['a1', 'a2', 'a3'], $ids); + self::assertSame(4, $transport->callCount()); + $uris = array_map(static fn ($r) => (string) $r->getUri(), $transport->received); + self::assertStringContainsString('offset=4', $uris[1]); + self::assertStringContainsString('offset=7', $uris[2]); + self::assertStringContainsString('offset=8', $uris[3]); + } + + public function testStorePageCountIsCappedAtTheRequestedLimit(): void + { + // A count above the limit asked for the page cannot advance the offset past it. + $transport = (new MockTransport()) + ->queueResponse(200, self::pageWithCount(self::actors(1), 50, 0)) + ->queueResponse(200, self::pageWithCount(self::actors(2), 50, 2)); + $ids = []; + foreach ($this->client($transport)->store()->iterate(new StoreListOptions(limit: 4), 2) as $item) { + $ids[] = $item->getId(); + } + self::assertSame(['a1', 'a2'], $ids); + self::assertStringContainsString('offset=2', (string) $transport->received[1]->getUri()); + } + public function testLimitCapStopsBeforeExhaustingPages(): void { // limit=2 with a big first page: only 2 items are yielded and no second request is made. diff --git a/tests/Unit/MockTransport.php b/tests/Unit/MockTransport.php index 4f7c65b..c714d89 100644 --- a/tests/Unit/MockTransport.php +++ b/tests/Unit/MockTransport.php @@ -6,6 +6,7 @@ use Apify\Client\Exception\TransportException; use Apify\Client\Http\HttpClientInterface; +use Apify\Client\Http\NoRedirectHttpClientInterface; use GuzzleHttp\Psr7\Response; use Psr\Http\Message\RequestInterface; use Psr\Http\Message\ResponseInterface; @@ -14,9 +15,10 @@ /** * A scripted {@see HttpClientInterface} for offline unit tests. Each queued entry is either a * {@see ResponseInterface} to return or a {@see TransportException} to throw, consumed in order. - * All received requests are recorded for assertions. + * All received requests are recorded for assertions, together with whether each one was sent streaming + * and whether it was sent with redirects disabled. */ -final class MockTransport implements HttpClientInterface +final class MockTransport implements NoRedirectHttpClientInterface { /** @var list */ private array $queue = []; @@ -24,6 +26,12 @@ final class MockTransport implements HttpClientInterface /** @var list */ public array $received = []; + /** @var list Whether each received request was sent with redirects disabled, in order. */ + public array $withoutRedirects = []; + + /** @var list Whether each received request was sent streaming, in order. */ + public array $streamed = []; + /** @var list The per-request timeout (seconds) each call was made with, in order. */ public array $timeouts = []; @@ -91,9 +99,16 @@ public function callCount(): int } public function send(RequestInterface $request, float $timeoutSecs): ResponseInterface + { + return $this->record($request, $timeoutSecs, false, false); + } + + private function record(RequestInterface $request, float $timeoutSecs, bool $stream, bool $withoutRedirects): ResponseInterface { $this->received[] = $request; $this->timeouts[] = $timeoutSecs; + $this->streamed[] = $stream; + $this->withoutRedirects[] = $withoutRedirects; if ($this->queue === []) { throw new RuntimeException('MockTransport queue is empty'); } @@ -106,6 +121,11 @@ public function send(RequestInterface $request, float $timeoutSecs): ResponseInt public function sendStreaming(RequestInterface $request, float $timeoutSecs): ResponseInterface { - return $this->send($request, $timeoutSecs); + return $this->record($request, $timeoutSecs, true, false); + } + + public function sendWithoutRedirects(RequestInterface $request, float $timeoutSecs, bool $stream = false): ResponseInterface + { + return $this->record($request, $timeoutSecs, $stream, true); } } diff --git a/tests/Unit/SpecSyncTest.php b/tests/Unit/SpecSyncTest.php index 622ac34..1011ff6 100644 --- a/tests/Unit/SpecSyncTest.php +++ b/tests/Unit/SpecSyncTest.php @@ -6,6 +6,7 @@ use Apify\Client\ApifyClient; use Apify\Client\Internal\Json; +use Apify\Client\Options\DatasetListItemsOptions; use Apify\Client\Options\RunListOptions; use Apify\Client\Options\StoreListOptions; use PHPUnit\Framework\Attributes\DataProvider; @@ -157,4 +158,90 @@ public function testPendingStoredRequestWithANullHandledAtIsReadAsIs(): void self::assertSame('r1', $request->getId()); self::assertNull($request->get('handledAt')); } + + /** @return iterable,int}> */ + public static function datasetPageCounts(): iterable + { + yield 'scanned rows above the items returned (skipEmpty/clean)' => [['X-Apify-Pagination-Count' => '5'], 5]; + yield 'items returned above the scanned rows (a lagging item count)' => [['X-Apify-Pagination-Count' => '1'], 3]; + yield 'equal' => [['X-Apify-Pagination-Count' => '3'], 3]; + yield 'header missing' => [[], 3]; + } + + /** @param array $headers */ + #[DataProvider('datasetPageCounts')] + public function testDatasetItemsPageCountIsTheScannedRowsOrTheItemsReturnedWhicheverIsLarger(array $headers, int $expected): void + { + $transport = (new MockTransport())->queueResponse(200, Json::encode([['n' => 0], ['n' => 1], ['n' => 2]]), $headers); + + $page = $this->client($transport)->dataset('ds')->listItems(); + + self::assertCount(3, $page->getItems()); + self::assertSame($expected, $page->getCount()); + } + + public function testDatasetIterationStepsByTheScannedRowsCappedAtTheRequestedLimit(): void + { + // Page 1 scans 3 rows and returns 1 item (the filter dropped two). Page 2 is an unwound page that + // returns 4 items for the 2 rows it was asked for, so the step is capped at its limit of 2. + $transport = (new MockTransport()) + ->queueResponse(200, Json::encode([['n' => 0]]), ['X-Apify-Pagination-Count' => '3']) + ->queueResponse(200, Json::encode([['n' => 3], ['n' => 3], ['n' => 4], ['n' => 4]]), ['X-Apify-Pagination-Count' => '2']) + ->queueResponse(200, Json::encode([]), ['X-Apify-Pagination-Count' => '0']); + + $items = iterator_to_array($this->client($transport)->dataset('ds')->iterateItems(new DatasetListItemsOptions(limit: 5), 3), false); + + self::assertCount(5, $items); + self::assertStringContainsString('offset=3', (string) $transport->received[1]->getUri()); + self::assertStringContainsString('limit=2', (string) $transport->received[1]->getUri()); + self::assertSame(2, $transport->callCount()); + } + + public function testActorVersionActorPathIsSentOnCreateAndUpdateAndReadBack(): void + { + $created = ['versionNumber' => '0.1', 'sourceType' => 'SOURCE_FILES', 'actorPath' => 'actors/typescript-actor']; + $transport = (new MockTransport()) + ->queueResponse(201, Json::encode(['data' => $created])) + ->queueResponse(200, Json::encode(['data' => ['versionNumber' => '0.1', 'sourceType' => 'SOURCE_FILES', 'actorPath' => null]])) + ->queueResponse(200, Json::encode(['data' => ['versionNumber' => '0.1', 'sourceType' => 'GIT_REPO']])); + $actor = $this->client($transport)->actor('a1'); + + $version = $actor->versions()->create($created); + self::assertSame($created, Json::decode(MockTransport::readBody($transport->lastRequest()))); + self::assertSame('actors/typescript-actor', $version->getActorPath()); + + // A null actorPath removes the stored path, so it is sent as an explicit null. + $updated = $actor->version('0.1')->update(['actorPath' => null]); + self::assertSame('{"actorPath":null}', MockTransport::readBody($transport->lastRequest())); + self::assertSame('PUT', $transport->lastRequest()->getMethod()); + self::assertNull($updated->getActorPath()); + + self::assertNull($actor->version('0.1')->update(['buildTag' => 'beta'])->getActorPath()); + } + + public function testActorStandbyUrlAndTheStandbyFieldsAreReadAsReceived(): void + { + $standby = ['isEnabled' => true, 'tenancy' => 'MULTI_TENANT', 'isConsoleAuthEnabled' => false, 'isTokenlessEnabled' => null]; + $transport = (new MockTransport())->queueResponse(200, Json::encode(['data' => [ + 'id' => 'a1', + 'standbyUrl' => 'https://jane--my-actor.apify.actor', + 'actorStandby' => $standby, + ]])); + + $actor = $this->client($transport)->actor('a1')->get(); + + self::assertNotNull($actor); + self::assertSame('https://jane--my-actor.apify.actor', $actor->getStandbyUrl()); + self::assertSame($standby, $actor->get('actorStandby')); + } + + public function testActorWithoutStandbyHasNoStandbyUrl(): void + { + $transport = (new MockTransport())->queueResponse(200, Json::encode(['data' => ['id' => 'a1', 'standbyUrl' => null]])); + + $actor = $this->client($transport)->actor('a1')->get(); + + self::assertNotNull($actor); + self::assertNull($actor->getStandbyUrl()); + } }