diff --git a/scripts/postprocess_generated_models.py b/scripts/postprocess_generated_models.py index 50a51f56..6bc26f21 100644 --- a/scripts/postprocess_generated_models.py +++ b/scripts/postprocess_generated_models.py @@ -56,7 +56,7 @@ RESOURCE_INPUT_TYPEDDICTS: frozenset[str] = frozenset( { 'Request', # RequestQueueClient.update_request - 'RequestDraft', # RequestQueueClient.add_request, batch_add_requests + 'RequestWithoutId', # RequestQueueClient.add_request, batch_add_requests 'RequestDraftDelete', # RequestQueueClient.batch_delete_requests 'TaskInput', # Actor/Task start/call/update default input 'WebhookCreate', # Actor/Task start/call webhook list element diff --git a/src/apify_client/_resource_clients/request_queue.py b/src/apify_client/_resource_clients/request_queue.py index 789bcc83..5ce8c6b6 100644 --- a/src/apify_client/_resource_clients/request_queue.py +++ b/src/apify_client/_resource_clients/request_queue.py @@ -32,6 +32,7 @@ RequestQueueResponse, RequestRegistration, RequestResponse, + RequestWithoutId, UnlockRequestsResponse, UnlockRequestsResult, ) @@ -50,10 +51,10 @@ from apify_client._typeddicts import ( RequestCamelDict, RequestDict, - RequestDraftCamelDict, RequestDraftDeleteCamelDict, RequestDraftDeleteDict, - RequestDraftDict, + RequestWithoutIdCamelDict, + RequestWithoutIdDict, ) from apify_client.types import Timeout @@ -68,7 +69,7 @@ def _serialize_requests( - requests: list[RequestDraft] | list[RequestDraftDict] | list[RequestDraftCamelDict], + requests: list[RequestWithoutId] | list[RequestWithoutIdDict] | list[RequestWithoutIdCamelDict], ) -> list[bytes]: """Validate requests and serialize each one into the JSON bytes it will occupy in the batch request body. @@ -80,8 +81,8 @@ def _serialize_requests( """ return [ json.dumps( - (request if isinstance(request, RequestDraft) else RequestDraft.model_validate(request)).model_dump( - by_alias=True, exclude_none=True + (request if isinstance(request, RequestWithoutId) else RequestWithoutId.model_validate(request)).model_dump( + mode='json', by_alias=True, exclude_none=True, fallback=str ), ensure_ascii=False, allow_nan=False, @@ -228,7 +229,7 @@ def list_and_lock_head( def add_request( self, - request: RequestDraftDict | RequestDraftCamelDict | RequestDraft, + request: RequestWithoutIdDict | RequestWithoutIdCamelDict | RequestWithoutId, *, forefront: bool | None = None, timeout: Timeout = 'short', @@ -238,22 +239,22 @@ def add_request( https://docs.apify.com/api/v2#/reference/request-queues/request-collection/add-request Args: - request: The request to add to the queue. + request: The request to add to the queue. Must carry a `unique_key` and a `url`. forefront: Whether to add the request to the head or the end of the queue. timeout: Timeout for the API HTTP request. Returns: The added request. """ - if not isinstance(request, RequestDraft): - request = RequestDraft.model_validate(request) + if not isinstance(request, RequestWithoutId): + request = RequestWithoutId.model_validate(request) request_params = self._build_params(forefront=forefront, clientKey=self.client_key) response = self._http_client.call( url=self._build_url('requests'), method='POST', - json=request.model_dump(by_alias=True, exclude_none=True), + json=request.model_dump(mode='json', by_alias=True, exclude_none=True, fallback=str), params=request_params, timeout=timeout, ) @@ -321,7 +322,7 @@ def update_request( response = self._http_client.call( url=self._build_url(f'requests/{to_path_segment(request.id)}'), method='PUT', - json=request.model_dump(by_alias=True, exclude_none=True), + json=request.model_dump(mode='json', by_alias=True, exclude_none=True, fallback=str), params=request_params, timeout=timeout, ) @@ -410,7 +411,7 @@ def delete_request_lock( def batch_add_requests( self, - requests: list[RequestDraft] | list[RequestDraftDict] | list[RequestDraftCamelDict], + requests: list[RequestWithoutId] | list[RequestWithoutIdDict] | list[RequestWithoutIdCamelDict], *, forefront: bool = False, max_parallel: int = 1, @@ -423,7 +424,7 @@ def batch_add_requests( https://docs.apify.com/api/v2#/reference/request-queues/batch-request-operations/add-requests Args: - requests: List of requests to be added to the queue. + requests: List of requests to be added to the queue. Each must carry a `unique_key` and a `url`. forefront: Whether to add requests to the front of the queue. max_parallel: Specifies the maximum number of parallel tasks for API calls. This is only applicable to the async client. For the sync client, this value must be set to 1, as parallel execution @@ -510,7 +511,7 @@ def batch_delete_requests( else RequestDraftDelete.model_validate( request, ) - ).model_dump(by_alias=True, exclude_none=True) + ).root.model_dump(mode='json', by_alias=True, exclude_none=True, fallback=str) for request in requests ] @@ -761,7 +762,7 @@ async def list_and_lock_head( async def add_request( self, - request: RequestDraftDict | RequestDraftCamelDict | RequestDraft, + request: RequestWithoutIdDict | RequestWithoutIdCamelDict | RequestWithoutId, *, forefront: bool | None = None, timeout: Timeout = 'short', @@ -771,22 +772,22 @@ async def add_request( https://docs.apify.com/api/v2#/reference/request-queues/request-collection/add-request Args: - request: The request to add to the queue. + request: The request to add to the queue. Must carry a `unique_key` and a `url`. forefront: Whether to add the request to the head or the end of the queue. timeout: Timeout for the API HTTP request. Returns: The added request. """ - if not isinstance(request, RequestDraft): - request = RequestDraft.model_validate(request) + if not isinstance(request, RequestWithoutId): + request = RequestWithoutId.model_validate(request) request_params = self._build_params(forefront=forefront, clientKey=self.client_key) response = await self._http_client.call( url=self._build_url('requests'), method='POST', - json=request.model_dump(by_alias=True, exclude_none=True), + json=request.model_dump(mode='json', by_alias=True, exclude_none=True, fallback=str), params=request_params, timeout=timeout, ) @@ -852,7 +853,7 @@ async def update_request( response = await self._http_client.call( url=self._build_url(f'requests/{to_path_segment(request.id)}'), method='PUT', - json=request.model_dump(by_alias=True, exclude_none=True), + json=request.model_dump(mode='json', by_alias=True, exclude_none=True, fallback=str), params=request_params, timeout=timeout, ) @@ -989,7 +990,7 @@ async def _batch_add_requests_worker( async def batch_add_requests( self, - requests: list[RequestDraft] | list[RequestDraftDict] | list[RequestDraftCamelDict], + requests: list[RequestWithoutId] | list[RequestWithoutIdDict] | list[RequestWithoutIdCamelDict], *, forefront: bool = False, max_parallel: int = 5, @@ -1002,7 +1003,7 @@ async def batch_add_requests( https://docs.apify.com/api/v2#/reference/request-queues/batch-request-operations/add-requests Args: - requests: List of requests to be added to the queue. + requests: List of requests to be added to the queue. Each must carry a `unique_key` and a `url`. forefront: Whether to add requests to the front of the queue. max_parallel: Specifies the maximum number of parallel tasks for API calls. This is only applicable to the async client. For the sync client, this value must be set to 1, as parallel execution @@ -1094,7 +1095,7 @@ async def batch_delete_requests( else RequestDraftDelete.model_validate( request, ) - ).model_dump(by_alias=True, exclude_none=True) + ).root.model_dump(mode='json', by_alias=True, exclude_none=True, fallback=str) for request in requests ] diff --git a/src/apify_client/_typeddicts.py b/src/apify_client/_typeddicts.py index ec967615..222e056e 100644 --- a/src/apify_client/_typeddicts.py +++ b/src/apify_client/_typeddicts.py @@ -111,44 +111,6 @@ class RequestCamelDict(RequestBaseCamelDict): """ -@docs_group('Typed dicts') -class RequestDraftDict(TypedDict): - """A request that failed to be processed during a request queue operation and can be retried.""" - - id: NotRequired[str] - """ - A unique identifier assigned to the request. - """ - unique_key: str - """ - A unique key used for request de-duplication. Requests with the same unique key are considered identical. - """ - url: str - """ - The URL of the request. - """ - method: NotRequired[Literal['GET', 'HEAD', 'POST', 'PUT', 'DELETE', 'CONNECT', 'OPTIONS', 'TRACE', 'PATCH']] - - -@docs_group('Typed dicts') -class RequestDraftCamelDict(TypedDict): - """A request that failed to be processed during a request queue operation and can be retried.""" - - id: NotRequired[str] - """ - A unique identifier assigned to the request. - """ - uniqueKey: str - """ - A unique key used for request de-duplication. Requests with the same unique key are considered identical. - """ - url: str - """ - The URL of the request. - """ - method: NotRequired[Literal['GET', 'HEAD', 'POST', 'PUT', 'DELETE', 'CONNECT', 'OPTIONS', 'TRACE', 'PATCH']] - - @docs_group('Typed dicts') class RequestDraftDeleteByIdDict(TypedDict): """A request that should be deleted, identified by its ID.""" @@ -221,6 +183,16 @@ class RequestDraftDeleteByUniqueKeyCamelDict(TypedDict): RequestUserDataCamelDict: TypeAlias = dict[str, Any] +@docs_group('Typed dicts') +class RequestWithoutIdDict(RequestBaseDict): + """A request stored in the request queue, including its metadata and processing state, without the assigned ID.""" + + +@docs_group('Typed dicts') +class RequestWithoutIdCamelDict(RequestBaseCamelDict): + """A request stored in the request queue, including its metadata and processing state, without the assigned ID.""" + + TaskInputDict: TypeAlias = dict[str, Any] TaskInputCamelDict: TypeAlias = dict[str, Any] diff --git a/tests/integration/test_request_queue.py b/tests/integration/test_request_queue.py index c09e2d7d..33446d7f 100644 --- a/tests/integration/test_request_queue.py +++ b/tests/integration/test_request_queue.py @@ -3,8 +3,10 @@ from __future__ import annotations from collections.abc import AsyncIterator, Iterator -from datetime import timedelta -from typing import TYPE_CHECKING +from datetime import UTC, datetime, timedelta +from typing import TYPE_CHECKING, Any + +import pytest from .._utils import ( collect_iterate_until_present, @@ -20,14 +22,15 @@ ListOfRequests, LockedRequestQueueHead, Request, - RequestDraft, RequestLockInfo, RequestQueue, RequestQueueHead, RequestQueueShort, RequestRegistration, + RequestWithoutId, UnlockRequestsResult, ) +from apify_client.errors import ApifyApiError if TYPE_CHECKING: from apify_client import ApifyClient, ApifyClientAsync @@ -35,10 +38,68 @@ from apify_client._typeddicts import ( RequestDict, RequestDraftDeleteDict, - RequestDraftDict, + RequestWithoutIdCamelDict, + RequestWithoutIdDict, ) +# The wire format the API declares for `handled_at`, and the instant it denotes. The client validates the string +# into a datetime, so it only sends ISO 8601 back out while it serializes in JSON mode. +HANDLED_AT_ISO = '2019-06-16T10:23:31.607Z' +HANDLED_AT = datetime(2019, 6, 16, 10, 23, 31, 607000, tzinfo=UTC) + +# Every request field beyond `id`/`unique_key`/`url`, snake_cased. The API declares its write bodies with +# `additionalProperties: false`, so each of these has to reach it camelCased to be stored at all. A fragment, not +# a full `RequestDict`: every use spreads it alongside `unique_key`/`url` supplied separately. +ALL_REQUEST_FIELDS: dict[str, Any] = { + 'method': 'POST', + 'user_data': {'label': 'DETAIL', 'depth': 2}, + 'no_retry': True, + 'retry_count': 3, + 'headers': {'X-Test': 'yes'}, + 'payload': '{"a": 1}', + 'loaded_url': 'https://example.com/loaded', + 'error_messages': ['boom'], + 'handled_at': HANDLED_AT_ISO, +} + + +async def fetch_stored_request( + rq_client: RequestQueueClient | RequestQueueClientAsync, + request_id: str, +) -> Request: + """Poll until `request_id` is readable back from the queue, then return it.""" + + async def get_request() -> Request | None: + return await maybe_await(rq_client.get_request(request_id)) + + stored = await poll_until_condition(get_request, lambda request: request is not None) + assert isinstance(stored, Request) + return stored + + +def non_identity_fields(request: Request) -> dict[str, Any]: + """Return a stored request's fields without the ones identifying it, so two write paths can be compared.""" + dumped = request.model_dump(by_alias=True) + for key in ('id', 'uniqueKey', 'url'): + dumped.pop(key, None) + return dumped + + +def assert_all_fields_stored(request: Request) -> None: + """Assert the stored request carries every value of `ALL_REQUEST_FIELDS`.""" + assert request.method == 'POST' + assert request.user_data is not None + assert request.user_data.model_dump() == {'label': 'DETAIL', 'depth': 2} + assert request.no_retry is True + assert request.retry_count == 3 + assert request.headers == {'X-Test': 'yes'} + assert request.payload == '{"a": 1}' + assert str(request.loaded_url) == 'https://example.com/loaded' + assert request.error_messages == ['boom'] + assert request.handled_at == HANDLED_AT + + async def ensure_queue_is_populated( rq_client: RequestQueueClient | RequestQueueClientAsync, *, @@ -204,7 +265,7 @@ async def test_request_queue_add_and_get_request(client: ApifyClient | ApifyClie try: # Add a request - request_data: RequestDraftDict = { + request_data: RequestWithoutIdDict = { 'url': 'https://example.com/test', 'unique_key': 'test-key-1', 'method': 'GET', @@ -322,7 +383,7 @@ async def test_request_queue_batch_add_requests(client: ApifyClient | ApifyClien try: # Batch add requests - requests_to_add: list[RequestDraftDict] = [ + requests_to_add: list[RequestWithoutIdDict] = [ {'url': f'https://example.com/batch-{i}', 'unique_key': f'batch-{i}'} for i in range(10) ] batch_response = await maybe_await(rq_client.batch_add_requests(requests_to_add)) @@ -546,7 +607,7 @@ async def test_request_queue_update_request(client: ApifyClient | ApifyClientAsy try: # Add a request - request_data: RequestDraftDict = { + request_data: RequestWithoutIdDict = { 'url': 'https://example.com/original', 'unique_key': 'update-test', 'method': 'GET', @@ -578,6 +639,144 @@ async def get_added_request() -> Request | None: await maybe_await(rq_client.delete()) +async def test_request_queue_add_request_round_trips_all_fields(client: ApifyClient | ApifyClientAsync) -> None: + """Every field of a snake_cased request survives `add_request` and comes back from `get_request`.""" + rq = await maybe_await(client.request_queues().get_or_create(name=get_random_resource_name('rq'))) + assert isinstance(rq, RequestQueue) + rq_client = client.request_queue(rq.id) + + try: + request_data: RequestWithoutIdDict = { + 'unique_key': 'round-trip', + 'url': 'https://example.com/round-trip', + **ALL_REQUEST_FIELDS, + } + add_result = await maybe_await(rq_client.add_request(request_data)) + assert isinstance(add_result, RequestRegistration) + + stored = await fetch_stored_request(rq_client, add_result.request_id) + assert_all_fields_stored(stored) + finally: + await maybe_await(rq_client.delete()) + + +async def test_request_queue_batch_add_requests_round_trips_all_fields( + client: ApifyClient | ApifyClientAsync, +) -> None: + """Every field of a snake_cased request survives `batch_add_requests` and comes back from `get_request`.""" + rq = await maybe_await(client.request_queues().get_or_create(name=get_random_resource_name('rq'))) + assert isinstance(rq, RequestQueue) + rq_client = client.request_queue(rq.id) + + try: + requests_to_add: list[RequestWithoutIdDict] = [ + { + 'unique_key': 'batch-round-trip', + 'url': 'https://example.com/batch-round-trip', + **ALL_REQUEST_FIELDS, + } + ] + batch_result = await maybe_await(rq_client.batch_add_requests(requests_to_add)) + assert isinstance(batch_result, BatchAddResult) + assert len(batch_result.unprocessed_requests) == 0 + assert len(batch_result.processed_requests) == 1 + + stored = await fetch_stored_request(rq_client, batch_result.processed_requests[0].request_id) + assert_all_fields_stored(stored) + finally: + await maybe_await(rq_client.delete()) + + +async def test_request_queue_add_and_update_request_store_identical_fields( + client: ApifyClient | ApifyClientAsync, +) -> None: + """The same field dict stored through `add_request` and through `update_request` lands identically.""" + rq = await maybe_await(client.request_queues().get_or_create(name=get_random_resource_name('rq'))) + assert isinstance(rq, RequestQueue) + rq_client = client.request_queue(rq.id) + + try: + added = await maybe_await( + rq_client.add_request( + {'unique_key': 'parity-add', 'url': 'https://example.com/parity-add', **ALL_REQUEST_FIELDS} + ) + ) + assert isinstance(added, RequestRegistration) + + seeded = await maybe_await( + rq_client.add_request({'unique_key': 'parity-update', 'url': 'https://example.com/parity-update'}) + ) + assert isinstance(seeded, RequestRegistration) + updated = await maybe_await( + rq_client.update_request( + { + 'id': seeded.request_id, + 'unique_key': 'parity-update', + 'url': 'https://example.com/parity-update', + **ALL_REQUEST_FIELDS, + } + ) + ) + assert isinstance(updated, RequestRegistration) + + from_add = await fetch_stored_request(rq_client, added.request_id) + from_update = await fetch_stored_request(rq_client, seeded.request_id) + assert non_identity_fields(from_add) == non_identity_fields(from_update) + assert_all_fields_stored(from_add) + assert_all_fields_stored(from_update) + finally: + await maybe_await(rq_client.delete()) + + +async def test_request_queue_add_request_accepts_camel_cased_fields(client: ApifyClient | ApifyClientAsync) -> None: + """A camelCased request dict stores the same fields as its snake_cased equivalent.""" + rq = await maybe_await(client.request_queues().get_or_create(name=get_random_resource_name('rq'))) + assert isinstance(rq, RequestQueue) + rq_client = client.request_queue(rq.id) + + try: + camel_request: RequestWithoutIdCamelDict = { + 'uniqueKey': 'camel', + 'url': 'https://example.com/camel', + 'method': 'POST', + 'userData': {'label': 'DETAIL', 'depth': 2}, + 'noRetry': True, + 'retryCount': 3, + 'headers': {'X-Test': 'yes'}, + 'payload': '{"a": 1}', + 'loadedUrl': 'https://example.com/loaded', + 'errorMessages': ['boom'], + 'handledAt': HANDLED_AT_ISO, + } + add_result = await maybe_await(rq_client.add_request(camel_request)) + assert isinstance(add_result, RequestRegistration) + + stored = await fetch_stored_request(rq_client, add_result.request_id) + assert_all_fields_stored(stored) + finally: + await maybe_await(rq_client.delete()) + + +async def test_request_queue_add_request_rejects_undeclared_fields(client: ApifyClient | ApifyClientAsync) -> None: + """The API refuses a body key its schema does not declare, so a snake_cased field takes the whole write down.""" + rq = await maybe_await(client.request_queues().get_or_create(name=get_random_resource_name('rq'))) + assert isinstance(rq, RequestQueue) + rq_client = client.request_queue(rq.id) + + try: + # `model_validate` keeps the undeclared key as a model extra, which is serialized verbatim. + draft = RequestWithoutId.model_validate( + {'unique_key': 'undeclared', 'url': 'https://example.com/undeclared', 'undeclared_field': 'value'} + ) + with pytest.raises(ApifyApiError, match='not allowed by the schema'): + await maybe_await(rq_client.add_request(draft)) + + with pytest.raises(ApifyApiError, match='not allowed by the schema'): + await maybe_await(rq_client.batch_add_requests([draft])) + finally: + await maybe_await(rq_client.delete()) + + async def test_request_queue_collection_iterate(client: ApifyClient | ApifyClientAsync, *, is_async: bool) -> None: """Test paginated iteration over user request queues.""" created_ids: list[str] = [] @@ -612,7 +811,7 @@ async def test_request_queue_iterate_requests(client: ApifyClient | ApifyClientA # Add several requests added_urls: list[str] = [] for i in range(7): - request_draft = RequestDraft(url=f'https://example.com/page-{i}', unique_key=f'unique-{i}') + request_draft = RequestWithoutId(url=f'https://example.com/page-{i}', unique_key=f'unique-{i}') await maybe_await(rq_client.add_request(request_draft)) added_urls.append(request_draft.url) @@ -650,7 +849,7 @@ async def test_request_queue_list_requests_with_cursor(client: ApifyClient | Api try: for i in range(5): await maybe_await( - rq_client.add_request(RequestDraft(url=f'https://example.com/p-{i}', unique_key=f'u-{i}')) + rq_client.add_request(RequestWithoutId(url=f'https://example.com/p-{i}', unique_key=f'u-{i}')) ) # Wait for all 5 requests to be indexed so pagination is exercised, not truncated @@ -682,7 +881,7 @@ async def test_request_queue_list_requests_with_filter(client: ApifyClient | Api try: for i in range(3): await maybe_await( - rq_client.add_request(RequestDraft(url=f'https://example.com/f-{i}', unique_key=f'f-{i}')) + rq_client.add_request(RequestWithoutId(url=f'https://example.com/f-{i}', unique_key=f'f-{i}')) ) # Wait for all 3 requests to be indexed before filtering diff --git a/tests/unit/test_client_request_queue.py b/tests/unit/test_client_request_queue.py index 8c34b025..4e283a29 100644 --- a/tests/unit/test_client_request_queue.py +++ b/tests/unit/test_client_request_queue.py @@ -6,9 +6,11 @@ from typing import TYPE_CHECKING import pytest +from pydantic import ValidationError from werkzeug.wrappers import Response from apify_client import ApifyClient, ApifyClientAsync +from apify_client._models import RequestDraftDelete from apify_client.errors import ApifyApiError if TYPE_CHECKING: @@ -17,7 +19,7 @@ from pytest_httpserver import HTTPServer from werkzeug.wrappers import Request - from apify_client._typeddicts import RequestDraftDict + from apify_client._typeddicts import RequestDict, RequestWithoutIdDict # The Apify API limit on the payload size of a batch-add request, which the client's batching must respect. _API_MAX_PAYLOAD_SIZE_BYTES = 9 * 1024 * 1024 @@ -54,7 +56,7 @@ async def test_batch_not_processed_raises_exception_async(httpserver: HTTPServer api_public_url=server_url, ) httpserver.expect_oneshot_request(re.compile(r'.*'), method='POST').respond_with_data(status=401) - requests: list[RequestDraftDict] = [ + requests: list[RequestWithoutIdDict] = [ {'unique_key': 'http://example.com/1', 'url': 'http://example.com/1', 'method': 'GET'}, {'unique_key': 'http://example.com/2', 'url': 'http://example.com/2', 'method': 'GET'}, ] @@ -75,7 +77,7 @@ async def test_batch_processed_partially_async(httpserver: HTTPServer) -> None: httpserver.expect_oneshot_request(re.compile(r'.*'), method='POST').respond_with_data( status=200, response_data=_PARTIALLY_ADDED_BATCH_RESPONSE_CONTENT ) - requests: list[RequestDraftDict] = [ + requests: list[RequestWithoutIdDict] = [ {'unique_key': 'http://example.com/1', 'url': 'http://example.com/1', 'method': 'GET'}, {'unique_key': 'http://example.com/2', 'url': 'http://example.com/2', 'method': 'GET'}, ] @@ -97,7 +99,7 @@ def test_batch_not_processed_raises_exception_sync(httpserver: HTTPServer) -> No ) httpserver.expect_oneshot_request(re.compile(r'.*'), method='POST').respond_with_data(status=401) - requests: list[RequestDraftDict] = [ + requests: list[RequestWithoutIdDict] = [ {'unique_key': 'http://example.com/1', 'url': 'http://example.com/1', 'method': 'GET'}, {'unique_key': 'http://example.com/2', 'url': 'http://example.com/2', 'method': 'GET'}, ] @@ -107,7 +109,7 @@ def test_batch_not_processed_raises_exception_sync(httpserver: HTTPServer) -> No rq_client.batch_add_requests(requests=requests) -def _make_large_requests() -> list[RequestDraftDict]: +def _make_large_requests() -> list[RequestWithoutIdDict]: """Return 3 requests of ~4 MB each, so that all of them together exceed the 9 MB payload limit.""" return [ { @@ -119,8 +121,11 @@ def _make_large_requests() -> list[RequestDraftDict]: ] -def _payload_capturing_handler(payloads: list[bytes]) -> Callable[[Request], Response]: - """Return a handler that records each POST body and responds with an empty batch result. +def _payload_capturing_handler( + payloads: list[bytes], + response_content: str = _EMPTY_BATCH_RESPONSE_CONTENT, +) -> Callable[[Request], Response]: + """Return a handler that records each request body and responds with `response_content`. Bodies below the client's compression threshold arrive uncompressed, so the recorded payload is decompressed only when the request says it was encoded. @@ -129,7 +134,7 @@ def _payload_capturing_handler(payloads: list[bytes]) -> Callable[[Request], Res def handler(request: Request) -> Response: body = request.get_data() payloads.append(gzip.decompress(body) if request.headers.get('Content-Encoding') == 'gzip' else body) - return Response(_EMPTY_BATCH_RESPONSE_CONTENT, status=200, content_type='application/json') + return Response(response_content, status=200, content_type='application/json') return handler @@ -178,7 +183,7 @@ def test_batch_add_requests_splits_batches_by_payload_size_sync(httpserver: HTTP assert sum(len(json.loads(payload)) for payload in payloads) == 3 -def _make_oversized_and_small_requests() -> list[RequestDraftDict]: +def _make_oversized_and_small_requests() -> list[RequestWithoutIdDict]: """Return a small request plus one whose serialized size alone exceeds the 9 MB payload limit.""" return [ {'unique_key': 'small', 'url': 'http://example.com/small', 'method': 'GET'}, @@ -243,7 +248,7 @@ def test_batch_processed_partially_sync(httpserver: HTTPServer) -> None: httpserver.expect_oneshot_request(re.compile(r'.*'), method='POST').respond_with_data( status=200, response_data=_PARTIALLY_ADDED_BATCH_RESPONSE_CONTENT ) - requests: list[RequestDraftDict] = [ + requests: list[RequestWithoutIdDict] = [ {'unique_key': 'http://example.com/1', 'url': 'http://example.com/1', 'method': 'GET'}, {'unique_key': 'http://example.com/2', 'url': 'http://example.com/2', 'method': 'GET'}, ] @@ -253,3 +258,258 @@ def test_batch_processed_partially_sync(httpserver: HTTPServer) -> None: assert requests[0]['unique_key'] in {request.unique_key for request in batch_response.processed_requests} assert len(batch_response.unprocessed_requests) == 1 assert batch_response.unprocessed_requests[0].unique_key == requests[1]['unique_key'] + + +_REQUEST_REGISTRATION_RESPONSE_CONTENT = ( + '{"data": {"requestId": "YiKoxjkaS9gjGTqhF", "wasAlreadyPresent": false, "wasAlreadyHandled": false}}' +) + +_FULL_REQUEST_DICT: RequestDict = { + 'id': 'YiKoxjkaS9gjGTqhF', + 'unique_key': 'http://example.com/1', + 'url': 'http://example.com/1', + 'method': 'GET', + 'user_data': {'label': 'DETAIL'}, + 'no_retry': True, + 'retry_count': 2, + 'loaded_url': 'http://example.com/1/final', + 'headers': {'X-Test': 'value'}, + 'payload': 'body', + 'error_messages': ['boom'], + 'handled_at': '2019-06-16T10:23:31.607Z', +} + + +class Unserializable: + """A value no JSON serializer can encode without a stringification fallback.""" + + def __str__(self) -> str: + return 'unserializable' + + +async def test_add_request_matches_update_request_casing_async(httpserver: HTTPServer) -> None: + """The same snake_case request dict reaches the API camelCased on add_request just as on update_request.""" + server_url = httpserver.url_for('/').removesuffix('/') + client = ApifyClientAsync(token='placeholder_token', api_url=server_url, api_public_url=server_url) + + payloads = list[bytes]() + httpserver.expect_request(re.compile(r'.*')).respond_with_handler( + _payload_capturing_handler(payloads, _REQUEST_REGISTRATION_RESPONSE_CONTENT) + ) + rq_client = client.request_queue(request_queue_id='whatever') + + await rq_client.add_request(_FULL_REQUEST_DICT) + await rq_client.update_request(_FULL_REQUEST_DICT) + + added, updated = (json.loads(payload) for payload in payloads) + assert added == updated + assert added['userData'] == {'label': 'DETAIL'} + assert [key for key in added if '_' in key] == [] + assert added['handledAt'] == '2019-06-16T10:23:31.607000Z' + + +def test_add_request_matches_update_request_casing_sync(httpserver: HTTPServer) -> None: + """The same snake_case request dict reaches the API camelCased on add_request just as on update_request.""" + server_url = httpserver.url_for('/').removesuffix('/') + client = ApifyClient(token='placeholder_token', api_url=server_url, api_public_url=server_url) + + payloads = list[bytes]() + httpserver.expect_request(re.compile(r'.*')).respond_with_handler( + _payload_capturing_handler(payloads, _REQUEST_REGISTRATION_RESPONSE_CONTENT) + ) + rq_client = client.request_queue(request_queue_id='whatever') + + rq_client.add_request(_FULL_REQUEST_DICT) + rq_client.update_request(_FULL_REQUEST_DICT) + + added, updated = (json.loads(payload) for payload in payloads) + assert added == updated + assert added['userData'] == {'label': 'DETAIL'} + assert [key for key in added if '_' in key] == [] + assert added['handledAt'] == '2019-06-16T10:23:31.607000Z' + + +async def test_batch_add_requests_camel_cases_every_field_async(httpserver: HTTPServer) -> None: + """Every field of a snake_case request dict is camelCased in the batch-add payload.""" + server_url = httpserver.url_for('/').removesuffix('/') + client = ApifyClientAsync(token='placeholder_token', api_url=server_url, api_public_url=server_url) + + payloads = list[bytes]() + httpserver.expect_request(re.compile(r'.*'), method='POST').respond_with_handler( + _payload_capturing_handler(payloads) + ) + rq_client = client.request_queue(request_queue_id='whatever') + + await rq_client.batch_add_requests(requests=[_FULL_REQUEST_DICT]) + + (sent_request,) = json.loads(payloads[0]) + assert sent_request['userData'] == {'label': 'DETAIL'} + assert [key for key in sent_request if '_' in key] == [] + assert sent_request['handledAt'] == '2019-06-16T10:23:31.607000Z' + + +def test_batch_add_requests_camel_cases_every_field_sync(httpserver: HTTPServer) -> None: + """Every field of a snake_case request dict is camelCased in the batch-add payload.""" + server_url = httpserver.url_for('/').removesuffix('/') + client = ApifyClient(token='placeholder_token', api_url=server_url, api_public_url=server_url) + + payloads = list[bytes]() + httpserver.expect_request(re.compile(r'.*'), method='POST').respond_with_handler( + _payload_capturing_handler(payloads) + ) + rq_client = client.request_queue(request_queue_id='whatever') + + rq_client.batch_add_requests(requests=[_FULL_REQUEST_DICT]) + + (sent_request,) = json.loads(payloads[0]) + assert sent_request['userData'] == {'label': 'DETAIL'} + assert [key for key in sent_request if '_' in key] == [] + assert sent_request['handledAt'] == '2019-06-16T10:23:31.607000Z' + + +async def test_add_request_stringifies_unserializable_user_data_async(httpserver: HTTPServer) -> None: + """A `user_data` value JSON cannot represent is stringified instead of aborting the call.""" + server_url = httpserver.url_for('/').removesuffix('/') + client = ApifyClientAsync(token='placeholder_token', api_url=server_url, api_public_url=server_url) + + payloads = list[bytes]() + httpserver.expect_request(re.compile(r'.*')).respond_with_handler( + _payload_capturing_handler(payloads, _REQUEST_REGISTRATION_RESPONSE_CONTENT) + ) + rq_client = client.request_queue(request_queue_id='whatever') + + await rq_client.add_request( + {'unique_key': 'http://example.com/1', 'url': 'http://example.com/1', 'user_data': {'tag': Unserializable()}} + ) + + assert json.loads(payloads[0])['userData'] == {'tag': 'unserializable'} + + +def test_add_request_stringifies_unserializable_user_data_sync(httpserver: HTTPServer) -> None: + """A `user_data` value JSON cannot represent is stringified instead of aborting the call.""" + server_url = httpserver.url_for('/').removesuffix('/') + client = ApifyClient(token='placeholder_token', api_url=server_url, api_public_url=server_url) + + payloads = list[bytes]() + httpserver.expect_request(re.compile(r'.*')).respond_with_handler( + _payload_capturing_handler(payloads, _REQUEST_REGISTRATION_RESPONSE_CONTENT) + ) + rq_client = client.request_queue(request_queue_id='whatever') + + rq_client.add_request( + {'unique_key': 'http://example.com/1', 'url': 'http://example.com/1', 'user_data': {'tag': Unserializable()}} + ) + + assert json.loads(payloads[0])['userData'] == {'tag': 'unserializable'} + + +async def test_batch_add_requests_stringifies_unserializable_user_data_async(httpserver: HTTPServer) -> None: + """A `user_data` value JSON cannot represent is stringified instead of aborting the call.""" + server_url = httpserver.url_for('/').removesuffix('/') + client = ApifyClientAsync(token='placeholder_token', api_url=server_url, api_public_url=server_url) + + payloads = list[bytes]() + httpserver.expect_request(re.compile(r'.*'), method='POST').respond_with_handler( + _payload_capturing_handler(payloads) + ) + rq_client = client.request_queue(request_queue_id='whatever') + + await rq_client.batch_add_requests( + requests=[ + { + 'unique_key': 'http://example.com/1', + 'url': 'http://example.com/1', + 'user_data': {'tag': Unserializable()}, + } + ] + ) + + (sent_request,) = json.loads(payloads[0]) + assert sent_request['userData'] == {'tag': 'unserializable'} + + +def test_batch_add_requests_stringifies_unserializable_user_data_sync(httpserver: HTTPServer) -> None: + """A `user_data` value JSON cannot represent is stringified instead of aborting the call.""" + server_url = httpserver.url_for('/').removesuffix('/') + client = ApifyClient(token='placeholder_token', api_url=server_url, api_public_url=server_url) + + payloads = list[bytes]() + httpserver.expect_request(re.compile(r'.*'), method='POST').respond_with_handler( + _payload_capturing_handler(payloads) + ) + rq_client = client.request_queue(request_queue_id='whatever') + + rq_client.batch_add_requests( + requests=[ + { + 'unique_key': 'http://example.com/1', + 'url': 'http://example.com/1', + 'user_data': {'tag': Unserializable()}, + } + ] + ) + + (sent_request,) = json.loads(payloads[0]) + assert sent_request['userData'] == {'tag': 'unserializable'} + + +async def test_batch_delete_requests_stringifies_unserializable_extra_async(httpserver: HTTPServer) -> None: + """An extra field JSON cannot represent is stringified instead of aborting the call.""" + server_url = httpserver.url_for('/').removesuffix('/') + client = ApifyClientAsync(token='placeholder_token', api_url=server_url, api_public_url=server_url) + + payloads = list[bytes]() + httpserver.expect_request(re.compile(r'.*'), method='DELETE').respond_with_handler( + _payload_capturing_handler(payloads) + ) + rq_client = client.request_queue(request_queue_id='whatever') + + request = RequestDraftDelete.model_validate({'id': 'YiKoxjkaS9gjGTqhF', 'weird': Unserializable()}) + await rq_client.batch_delete_requests(requests=[request]) + + (sent_request,) = json.loads(payloads[0]) + assert sent_request == {'id': 'YiKoxjkaS9gjGTqhF', 'weird': 'unserializable'} + + +def test_batch_delete_requests_stringifies_unserializable_extra_sync(httpserver: HTTPServer) -> None: + """An extra field JSON cannot represent is stringified instead of aborting the call.""" + server_url = httpserver.url_for('/').removesuffix('/') + client = ApifyClient(token='placeholder_token', api_url=server_url, api_public_url=server_url) + + payloads = list[bytes]() + httpserver.expect_request(re.compile(r'.*'), method='DELETE').respond_with_handler( + _payload_capturing_handler(payloads) + ) + rq_client = client.request_queue(request_queue_id='whatever') + + request = RequestDraftDelete.model_validate({'id': 'YiKoxjkaS9gjGTqhF', 'weird': Unserializable()}) + rq_client.batch_delete_requests(requests=[request]) + + (sent_request,) = json.loads(payloads[0]) + assert sent_request == {'id': 'YiKoxjkaS9gjGTqhF', 'weird': 'unserializable'} + + +async def test_add_request_requires_unique_key_and_url_async(httpserver: HTTPServer) -> None: + """Pydantic still rejects a dict missing `unique_key`/`url`, even past a type-check bypass.""" + server_url = httpserver.url_for('/').removesuffix('/') + client = ApifyClientAsync(token='placeholder_token', api_url=server_url, api_public_url=server_url) + rq_client = client.request_queue(request_queue_id='whatever') + + with pytest.raises(ValidationError) as exc_info: + await rq_client.add_request({'method': 'GET'}) # ty: ignore[invalid-argument-type] + + assert {error['loc'][0] for error in exc_info.value.errors()} == {'uniqueKey', 'url'} + assert httpserver.log == [] + + +def test_add_request_requires_unique_key_and_url_sync(httpserver: HTTPServer) -> None: + """Pydantic still rejects a dict missing `unique_key`/`url`, even past a type-check bypass.""" + server_url = httpserver.url_for('/').removesuffix('/') + client = ApifyClient(token='placeholder_token', api_url=server_url, api_public_url=server_url) + rq_client = client.request_queue(request_queue_id='whatever') + + with pytest.raises(ValidationError) as exc_info: + rq_client.add_request({'method': 'GET'}) # ty: ignore[invalid-argument-type] + + assert {error['loc'][0] for error in exc_info.value.errors()} == {'uniqueKey', 'url'} + assert httpserver.log == [] diff --git a/tests/unit/test_url_path_encoding.py b/tests/unit/test_url_path_encoding.py index fa33d9e8..bc99319d 100644 --- a/tests/unit/test_url_path_encoding.py +++ b/tests/unit/test_url_path_encoding.py @@ -417,18 +417,16 @@ def test_degenerate_resource_id_is_rejected_before_the_request( def test_update_request_without_an_id_is_rejected_sync(*, api_url: str) -> None: """A request carrying no ID cannot address a queue record, so the update is refused before it is sent.""" client = ApifyClient(token='test_token', api_url=api_url) + request = Request(url='https://example.com', unique_key='https://example.com') with pytest.raises(ValueError, match='must have an ID'): - client.request_queue(_QUEUE_ID).update_request( - Request(url='https://example.com', unique_key='https://example.com') - ) + client.request_queue(_QUEUE_ID).update_request(request) async def test_update_request_without_an_id_is_rejected_async(*, api_url: str) -> None: """A request carrying no ID cannot address a queue record, so the update is refused before it is sent.""" client = ApifyClientAsync(token='test_token', api_url=api_url) + request = Request(url='https://example.com', unique_key='https://example.com') with pytest.raises(ValueError, match='must have an ID'): - await client.request_queue(_QUEUE_ID).update_request( - Request(url='https://example.com', unique_key='https://example.com') - ) + await client.request_queue(_QUEUE_ID).update_request(request)