Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 10 additions & 6 deletions src/apify/storage_clients/_apify/_request_queue_shared_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

from cachetools import LRUCache

from apify_client._models import RequestResource, RequestWithoutId
from crawlee.storage_clients.models import AddRequestsResponse, ProcessedRequest, RequestQueueMetadata

from ._models import ApifyRequestQueueMetadata, CachedRequest, RequestQueueHead
Expand Down Expand Up @@ -162,14 +163,16 @@ async def add_batch_of_requests(
self._requests_being_added[request_id] = loop.create_future()

if new_requests:
# Prepare requests for API by converting to dictionaries.
requests_dict = [request.model_dump(by_alias=True) for request in new_requests]
# Prepare requests for API by converting to the client model.
api_requests = [
RequestWithoutId.model_validate(request.model_dump(by_alias=True)) for request in new_requests
]

committed_request_ids: set[str] = set()
try:
# Send requests to API.
batch_response = await self._api_client.batch_add_requests(
requests=requests_dict,
requests=api_requests,
forefront=forefront,
)

Expand Down Expand Up @@ -593,10 +596,11 @@ async def _update_request(
Returns:
The updated request
"""
request_dict = request.model_dump(by_alias=True)
request_dict['id'] = unique_key_to_request_id(request.unique_key)
api_request = RequestResource.model_validate(
{**request.model_dump(by_alias=True), 'id': unique_key_to_request_id(request.unique_key)}
)
response = await self._api_client.update_request(
request=request_dict,
request=api_request,
forefront=forefront,
)

Expand Down
16 changes: 10 additions & 6 deletions src/apify/storage_clients/_apify/_request_queue_single_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

from cachetools import LRUCache

from apify_client._models import RequestResource, RequestWithoutId
from crawlee.storage_clients.models import AddRequestsResponse, ProcessedRequest, RequestQueueMetadata

from ._utils import (
Expand Down Expand Up @@ -173,13 +174,15 @@ async def add_batch_of_requests(
self._requests_being_added[request_id] = loop.create_future()

if new_requests:
# Prepare requests for API by converting to dictionaries.
requests_dict = [request.model_dump(by_alias=True) for request in new_requests]
# Prepare requests for API by converting to the client model.
api_requests = [
RequestWithoutId.model_validate(request.model_dump(by_alias=True)) for request in new_requests
]

committed_request_ids: set[str] = set()
try:
# Send requests to API.
batch_response = await self._api_client.batch_add_requests(requests=requests_dict, forefront=forefront)
batch_response = await self._api_client.batch_add_requests(requests=api_requests, forefront=forefront)
batch_response_dict = batch_response.model_dump(by_alias=True)
api_response = AddRequestsResponse.model_validate(batch_response_dict)

Expand Down Expand Up @@ -424,10 +427,11 @@ async def _update_request(
Returns:
The updated request
"""
request_dict = request.model_dump(by_alias=True)
request_dict['id'] = unique_key_to_request_id(request.unique_key)
api_request = RequestResource.model_validate(
{**request.model_dump(by_alias=True), 'id': unique_key_to_request_id(request.unique_key)}
)
response = await self._api_client.update_request(
request=request_dict,
request=api_request,
forefront=forefront,
)

Expand Down
18 changes: 11 additions & 7 deletions tests/integration/test_request_queue.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@

import pytest

from apify_client._models import BatchAddResult, UnprocessedRequest
from apify_client._models import BatchAddResult, RequestWithoutId, UnprocessedRequest
from crawlee import service_locator
from crawlee.crawlers import BasicCrawler

Expand Down Expand Up @@ -963,7 +963,11 @@ async def test_request_queue_metadata_another_client(
"""Test metadata tracking. The delayed metadata should be reliable even when changed by another client."""
rq = request_queue_apify
api_client = apify_client_async.request_queue(request_queue_id=rq.id, client_key=None)
await api_client.add_request(Request.from_url('http://example.com/1').model_dump(by_alias=True, exclude={'id'}))
await api_client.add_request(
RequestWithoutId.model_validate(
Request.from_url('http://example.com/1').model_dump(by_alias=True, exclude={'id'})
)
)

# Poll until the API has propagated the metadata change.
metadata = await poll_until_condition(
Expand Down Expand Up @@ -1201,7 +1205,7 @@ async def test_pre_existing_request_with_user_data(

# Add request by a different producer
rq_client = apify_client_async.request_queue(request_queue_id=rq.id)
await rq_client.add_request(req.model_dump(by_alias=True))
await rq_client.add_request(RequestWithoutId.model_validate(req.model_dump(by_alias=True)))

# Fetch the request by the client under test. It was added by a different producer, so single-mode immediate
# consistency (which only covers the client's own writes) does not apply: poll until the write propagates,
Expand Down Expand Up @@ -1287,13 +1291,13 @@ async def test_request_queue_deduplication_unprocessed_requests(
assert stats_before is not None
assert stats_before.write_count is not None

def return_unprocessed_requests(requests: list[dict], *_: Any, **__: Any) -> BatchAddResult:
def return_unprocessed_requests(requests: list[RequestWithoutId], *_: Any, **__: Any) -> BatchAddResult:
"""Simulate API returning unprocessed requests."""
unprocessed_requests = [
UnprocessedRequest.model_construct(
url=request['url'],
unique_key=request['uniqueKey'],
method=request['method'],
url=request.url,
unique_key=request.unique_key,
method=request.method,
)
for request in requests
]
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -379,7 +379,7 @@ async def test_partial_unprocessed_commits_only_accepted_requests(access: str) -
api_client.batch_add_requests.assert_awaited_once()
assert api_client.batch_add_requests.await_args is not None
resent = api_client.batch_add_requests.await_args.kwargs['requests']
assert [request['uniqueKey'] for request in resent] == [rejected.unique_key]
assert [request.unique_key for request in resent] == [rejected.unique_key]


@pytest.mark.parametrize(
Expand Down
Loading
Loading