RQ and Apify clients (will be moved to SDK later)

Author: vdusek

docs/guides/code_examples/storages/rq_basic_example.py

@@ -12,7 +12,7 @@ async def main() -> None:
     await request_queue.add_request('https://apify.com/')
 
     # Add multiple requests as a batch.
-    await request_queue.add_requests_batched(
+    await request_queue.add_requests(
         ['https://crawlee.dev/', 'https://crawlee.dev/python/']
     )
 

docs/guides/code_examples/storages/rq_with_crawler_explicit_example.py

@@ -10,7 +10,7 @@ async def main() -> None:
     request_queue = await RequestQueue.open(name='my-request-queue')
 
     # Interact with the request queue directly, e.g. add a batch of requests.
-    await request_queue.add_requests_batched(
+    await request_queue.add_requests(
         ['https://apify.com/', 'https://crawlee.dev/']
     )
 

src/crawlee/_utils/file.py

@@ -134,11 +134,7 @@ async def export_json_to_stream(
     **kwargs: Unpack[ExportDataJsonKwargs],
 ) -> None:
     items = [item async for item in iterator]
-
-    if items:
-        json.dump(items, dst, **kwargs)
-    else:
-        logger.warning('Attempting to export an empty dataset - no file will be created')
+    json.dump(items, dst, **kwargs)
 
 
 async def export_csv_to_stream(

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -648,7 +648,7 @@ async def add_requests(
         """
         request_manager = await self.get_request_manager()
 
-        await request_manager.add_requests_batched(
+        await request_manager.add_requests(
             requests=requests,
             batch_size=batch_size,
             wait_time_between_batches=wait_time_between_batches,
@@ -976,7 +976,7 @@ async def _commit_request_handler_result(self, context: BasicCrawlingContext) ->
                 ):
                     requests.append(dst_request)
 
-            await request_manager.add_requests_batched(requests)
+            await request_manager.add_requests(requests)
 
         for push_data_call in result.push_data_calls:
             await self._push_data(**push_data_call)

src/crawlee/request_loaders/_request_loader.py

@@ -5,11 +5,13 @@
 
 from crawlee import Request
 from crawlee._utils.docs import docs_group
+from crawlee.request_loaders import RequestManagerTandem
+from crawlee.storages import RequestQueue
 
 if TYPE_CHECKING:
     from collections.abc import Sequence
 
-    from crawlee.request_loaders import RequestManager, RequestManagerTandem
+    from crawlee.request_loaders import RequestManager
     from crawlee.storage_clients.models import ProcessedRequest
 
 
@@ -56,9 +58,6 @@ async def to_tandem(self, request_manager: RequestManager | None = None) -> Requ
             request_manager: Request manager to combine the loader with.
                 If None is given, the default request queue is used.
         """
-        from crawlee.request_loaders import RequestManagerTandem
-        from crawlee.storages import RequestQueue
-
         if request_manager is None:
             request_manager = await RequestQueue.open()
 

src/crawlee/request_loaders/_request_manager.py

@@ -6,12 +6,12 @@
 
 from crawlee._utils.docs import docs_group
 from crawlee.request_loaders._request_loader import RequestLoader
+from crawlee.storage_clients.models import AddRequestsResponse, ProcessedRequest
 
 if TYPE_CHECKING:
     from collections.abc import Sequence
 
     from crawlee._request import Request
-    from crawlee.storage_clients.models import ProcessedRequest
 
 
 @docs_group('Abstract classes')
@@ -40,27 +40,35 @@ async def add_request(
             Information about the request addition to the manager.
         """
 
-    async def add_requests_batched(
+    async def add_requests(
         self,
         requests: Sequence[str | Request],
         *,
+        forefront: bool = False,
         batch_size: int = 1000,  # noqa: ARG002
         wait_time_between_batches: timedelta = timedelta(seconds=1),  # noqa: ARG002
         wait_for_all_requests_to_be_added: bool = False,  # noqa: ARG002
         wait_for_all_requests_to_be_added_timeout: timedelta | None = None,  # noqa: ARG002
-    ) -> None:
+    ) -> AddRequestsResponse:
         """Add requests to the manager in batches.
 
         Args:
             requests: Requests to enqueue.
+            forefront: If True, add requests to the beginning of the queue.
             batch_size: The number of requests to add in one batch.
             wait_time_between_batches: Time to wait between adding batches.
             wait_for_all_requests_to_be_added: If True, wait for all requests to be added before returning.
             wait_for_all_requests_to_be_added_timeout: Timeout for waiting for all requests to be added.
         """
         # Default and dumb implementation.
+        processed_requests = list[ProcessedRequest]()
         for request in requests:
-            await self.add_request(request)
+            processed_request = await self.add_request(request, forefront=forefront)
+            processed_requests.append(processed_request)
+        return AddRequestsResponse(
+            processed_requests=processed_requests,
+            unprocessed_requests=[],
+        )
 
     @abstractmethod
     async def reclaim_request(self, request: Request, *, forefront: bool = False) -> ProcessedRequest | None:

src/crawlee/request_loaders/_request_manager_tandem.py

@@ -49,7 +49,7 @@ async def add_request(self, request: str | Request, *, forefront: bool = False)
         return await self._read_write_manager.add_request(request, forefront=forefront)
 
     @override
-    async def add_requests_batched(
+    async def add_requests(
         self,
         requests: Sequence[str | Request],
         *,
@@ -58,7 +58,7 @@ async def add_requests_batched(
         wait_for_all_requests_to_be_added: bool = False,
         wait_for_all_requests_to_be_added_timeout: timedelta | None = None,
     ) -> None:
-        return await self._read_write_manager.add_requests_batched(
+        return await self._read_write_manager.add_requests(
             requests,
             batch_size=batch_size,
             wait_time_between_batches=wait_time_between_batches,

src/crawlee/storage_clients/_apify/__init__.py

@@ -0,0 +1,11 @@
+from ._dataset_client import ApifyDatasetClient
+from ._key_value_store_client import ApifyKeyValueStoreClient
+from ._request_queue_client import ApifyRequestQueueClient
+from ._storage_client import ApifyStorageClient
+
+__all__ = [
+    'ApifyDatasetClient',
+    'ApifyKeyValueStoreClient',
+    'ApifyRequestQueueClient',
+    'ApifyStorageClient',
+]

src/crawlee/storage_clients/_apify/_dataset_client.py

@@ -0,0 +1,198 @@
+from __future__ import annotations
+
+import asyncio
+from logging import getLogger
+from typing import TYPE_CHECKING, Any, ClassVar
+
+from apify_client import ApifyClientAsync
+from typing_extensions import override
+
+from crawlee.storage_clients._base import DatasetClient
+from crawlee.storage_clients.models import DatasetItemsListPage, DatasetMetadata
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+    from datetime import datetime
+
+    from apify_client.clients import DatasetClientAsync
+
+    from crawlee.configuration import Configuration
+
+logger = getLogger(__name__)
+
+
+class ApifyDatasetClient(DatasetClient):
+    """An Apify platform implementation of the dataset client."""
+
+    _cache_by_name: ClassVar[dict[str, ApifyDatasetClient]] = {}
+    """A dictionary to cache clients by their names."""
+
+    def __init__(
+        self,
+        *,
+        id: str,
+        name: str,
+        created_at: datetime,
+        accessed_at: datetime,
+        modified_at: datetime,
+        item_count: int,
+        api_client: DatasetClientAsync,
+    ) -> None:
+        """Initialize a new instance.
+
+        Preferably use the `ApifyDatasetClient.open` class method to create a new instance.
+        """
+        self._metadata = DatasetMetadata(
+            id=id,
+            name=name,
+            created_at=created_at,
+            accessed_at=accessed_at,
+            modified_at=modified_at,
+            item_count=item_count,
+        )
+
+        self._api_client = api_client
+        """The Apify dataset client for API operations."""
+
+        self._lock = asyncio.Lock()
+        """A lock to ensure that only one operation is performed at a time."""
+
+    @override
+    @property
+    def metadata(self) -> DatasetMetadata:
+        return self._metadata
+
+    @override
+    @classmethod
+    async def open(
+        cls,
+        *,
+        id: str | None,
+        name: str | None,
+        configuration: Configuration,
+    ) -> ApifyDatasetClient:
+        default_name = configuration.default_dataset_id
+        token = 'configuration.apify_token'  # TODO: use the real value
+        api_url = 'configuration.apify_api_url'  # TODO: use the real value
+
+        name = name or default_name
+
+        # Check if the client is already cached by name.
+        if name in cls._cache_by_name:
+            client = cls._cache_by_name[name]
+            await client._update_metadata()  # noqa: SLF001
+            return client
+
+        # Otherwise, create a new one.
+        apify_client_async = ApifyClientAsync(
+            token=token,
+            api_url=api_url,
+            max_retries=8,
+            min_delay_between_retries_millis=500,
+            timeout_secs=360,
+        )
+
+        apify_datasets_client = apify_client_async.datasets()
+
+        metadata = DatasetMetadata.model_validate(
+            await apify_datasets_client.get_or_create(name=id if id is not None else name),
+        )
+
+        apify_dataset_client = apify_client_async.dataset(dataset_id=metadata.id)
+
+        client = cls(
+            id=metadata.id,
+            name=metadata.name,
+            created_at=metadata.created_at,
+            accessed_at=metadata.accessed_at,
+            modified_at=metadata.modified_at,
+            item_count=metadata.item_count,
+            api_client=apify_dataset_client,
+        )
+
+        # Cache the client by name.
+        cls._cache_by_name[name] = client
+
+        return client
+
+    @override
+    async def drop(self) -> None:
+        async with self._lock:
+            await self._api_client.delete()
+
+            # Remove the client from the cache.
+            if self.metadata.name in self.__class__._cache_by_name:  # noqa: SLF001
+                del self.__class__._cache_by_name[self.metadata.name]  # noqa: SLF001
+
+    @override
+    async def push_data(self, data: list[Any] | dict[str, Any]) -> None:
+        async with self._lock:
+            await self._api_client.push_items(items=data)
+            await self._update_metadata()
+
+    @override
+    async def get_data(
+        self,
+        *,
+        offset: int = 0,
+        limit: int | None = 999_999_999_999,
+        clean: bool = False,
+        desc: bool = False,
+        fields: list[str] | None = None,
+        omit: list[str] | None = None,
+        unwind: str | None = None,
+        skip_empty: bool = False,
+        skip_hidden: bool = False,
+        flatten: list[str] | None = None,
+        view: str | None = None,
+    ) -> DatasetItemsListPage:
+        response = await self._api_client.list_items(
+            offset=offset,
+            limit=limit,
+            clean=clean,
+            desc=desc,
+            fields=fields,
+            omit=omit,
+            unwind=unwind,
+            skip_empty=skip_empty,
+            skip_hidden=skip_hidden,
+            flatten=flatten,
+            view=view,
+        )
+        result = DatasetItemsListPage.model_validate(vars(response))
+        await self._update_metadata()
+        return result
+
+    @override
+    async def iterate_items(
+        self,
+        *,
+        offset: int = 0,
+        limit: int | None = None,
+        clean: bool = False,
+        desc: bool = False,
+        fields: list[str] | None = None,
+        omit: list[str] | None = None,
+        unwind: str | None = None,
+        skip_empty: bool = False,
+        skip_hidden: bool = False,
+    ) -> AsyncIterator[dict]:
+        async for item in self._api_client.iterate_items(
+            offset=offset,
+            limit=limit,
+            clean=clean,
+            desc=desc,
+            fields=fields,
+            omit=omit,
+            unwind=unwind,
+            skip_empty=skip_empty,
+            skip_hidden=skip_hidden,
+        ):
+            yield item
+
+        await self._update_metadata()
+
+    async def _update_metadata(self) -> None:
+        """Update the dataset metadata file with current information."""
+        metadata = await self._api_client.get()
+        self._metadata = DatasetMetadata.model_validate(metadata)