feat: Add specialized ApifyRequestQueue clients (#573)

### Description

- `ApifyRequestQueueClient` can be created in two access modes -
`single`, `shared`:
- `shared` - current version that supports multiple producers/consumers
and locking of requests. More Apify API calls, higher API usage -> more
expensive, slower.
- `single` - new constrained client for self-consumer and multiple
constrained producers. (Detailed constraints in the docs). Fewer Apify
API calls, lower API usage -> cheaper, faster.
 
- Most of the `ApifyRequestQueueClient` tests were moved away from
actor-based tests, so that they can be parametrized for both variants of
the `ApifyRequestQueueClients` and to make local debugging easier.

#### Usage:
RequestQueue with `shared`:
`await
RequestQueue.open(storage_client=ApifyStorageClient(request_queue_access="shared"))`
RequestQueue with default `single`:
`await RequestQueue.open(storage_client=ApifyStorageClient())`

#### Stats difference:
The full client is doing significantly more API calls and regarding the
API usage it is doing 50% more RequestQueue writes and also more
RequestQueue reads.

**Example rq related stats for crawler started with 1000 requests:**
`shared`:
API calls: 2123
API usage: {'readCount': 1000, 'writeCount': 3000, 'deleteCount': 0,
'headItemReadCount': 0, 'storageBytes': 104035}

`single`:
API calls: 1059
API usage: {'readCount': 3, 'writeCount': 2000, 'deleteCount': 0,
'headItemReadCount': 14, 'storageBytes': 103826}


### Issues

- Part of: #513

---------

Co-authored-by: Jan Buchar <[email protected]>

Author: Pijukatel

docs/04_upgrading/upgrading_to_v3.md

@@ -69,6 +69,13 @@ Some changes in the related model classes:
 ## Removed Actor.config property
 - `Actor.config` property has been removed. Use `Actor.configuration` instead.
 
+## Default storage ids in configuration changed to None
+- `Configuration.default_key_value_store_id` changed from `'default'` to `None`.
+- `Configuration.default_dataset_id` changed from `'default'` to `None`.
+- `Configuration.default_request_queue_id` changed from `'default'` to `None`.
+
+Previously using the default storage without specifying its `id` in `Configuration` would lead to using specific storage with id `'default'`. Now it will use newly created unnamed storage with `'id'` assigned by the Apify platform, consecutive calls to get the default storage will return the same storage.
+
 ## Actor initialization and ServiceLocator changes
 
 `Actor` initialization and global `service_locator` services setup is more strict and predictable.
@@ -102,20 +109,51 @@ async def main():
         )
 ```
 
-## Removed Actor.config property
-- `Actor.config` property has been removed. Use `Actor.configuration` instead.
+### Changes in storage clients
 
-## Default storage ids in configuration changed to None
-- `Configuration.default_key_value_store_id` changed from `'default'` to `None`.
-- `Configuration.default_dataset_id` changed from `'default'` to `None`.
-- `Configuration.default_request_queue_id` changed from `'default'` to `None`.
+## Explicit control over storage clients used in Actor
+- It is now possible to have full control over which storage clients are used by the `Actor`. To make development of Actors convenient, the `Actor` has two storage clients. One that is used when running on Apify platform or when opening storages with `force_cloud=True` and the other client that is used when running outside the Apify platform. The `Actor` has reasonable defaults and for the majority of use-cases there is no need to change it. However, if you need to use a different storage client, you can set it up before entering `Actor` context through `service_locator`.
+
+**Now (v3.0):**
+
+```python
+from crawlee import service_locator
+from apify.storage_clients import ApifyStorageClient, SmartApifyStorageClient, MemoryStorageClient
+from apify import Actor
+
+
+async def main():
+    service_locator.set_storage_client(
+        SmartApifyStorageClient(
+            cloud_storage_client=ApifyStorageClient(request_queue_access="single"),
+            local_storage_client=MemoryStorageClient()
+        )
+    )
+    async with Actor:
+        rq = await Actor.open_request_queue()
+```
 
-Previously using the default storage without specifying its `id` in `Configuration` would lead to using specific storage with id `'default'`. Now it will use newly created unnamed storage with `'id'` assigned by the Apify platform, consecutive calls to get the default storage will return the same storage.
 
-## Storages
+## The default use of optimized ApifyRequestQueueClient
 
-<!-- TODO -->
+- The default client for working with Apify platform based `RequestQueue` is now optimized and simplified client which does significantly lower amount of API calls, but does not support multiple consumers working on the same queue. It is cheaper and faster and is suitable for the majority of the use cases.
+- The full client is still available, but it has to be explicitly requested via `request_queue_access="shared"` argument when using the `ApifyStorageClient`.
 
-## Storage clients
+**Now (v3.0):**
+
+```python
+from crawlee import service_locator
+from apify.storage_clients import ApifyStorageClient, SmartApifyStorageClient
+from apify import Actor
 
-<!-- TODO -->
+
+async def main():
+    # Full client that supports multiple consumers of the Apify Request Queue
+    service_locator.set_storage_client(
+        SmartApifyStorageClient(
+            cloud_storage_client=ApifyStorageClient(request_queue_access="shared"),
+        )
+    )
+    async with Actor:
+        rq = await Actor.open_request_queue()
+```

pyproject.toml

@@ -36,7 +36,7 @@ keywords = [
 dependencies = [
     "apify-client>=2.0.0,<3.0.0",
     "apify-shared>=2.0.0,<3.0.0",
-    "crawlee==0.6.13b42",
+    "crawlee==0.6.13b46",
     "cachetools>=5.5.0",
     "cryptography>=42.0.0",
     "impit>=0.6.1",

src/apify/_actor.py

@@ -25,7 +25,6 @@
     EventPersistStateData,
     EventSystemInfoData,
 )
-from crawlee.storage_clients import FileSystemStorageClient
 
 from apify._charging import ChargeResult, ChargingManager, ChargingManagerImplementation
 from apify._configuration import Configuration
@@ -38,6 +37,7 @@
 from apify.log import _configure_logging, logger
 from apify.storage_clients import ApifyStorageClient
 from apify.storage_clients._file_system import ApifyFileSystemStorageClient
+from apify.storage_clients._smart_apify._storage_client import SmartApifyStorageClient
 from apify.storages import Dataset, KeyValueStore, RequestQueue
 
 if TYPE_CHECKING:
@@ -48,7 +48,6 @@
     from typing_extensions import Self
 
     from crawlee.proxy_configuration import _NewUrlFunction
-    from crawlee.storage_clients import StorageClient
 
     from apify._models import Webhook
 
@@ -131,7 +130,6 @@ def __init__(
         self._configuration = configuration
         self._configure_logging = configure_logging
         self._apify_client: ApifyClientAsync | None = None
-        self._local_storage_client: StorageClient | None = None
 
         self._is_initialized = False
 
@@ -234,45 +232,42 @@ def log(self) -> logging.Logger:
         """The logging.Logger instance the Actor uses."""
         return logger
 
-    def _get_local_storage_client(self) -> StorageClient:
-        """Get the local storage client the Actor instance uses."""
-        if self._local_storage_client:
-            return self._local_storage_client
+    def _raise_if_not_initialized(self) -> None:
+        if not self._is_initialized:
+            raise RuntimeError('The Actor was not initialized!')
+
+    @cached_property
+    def _storage_client(self) -> SmartApifyStorageClient:
+        """Storage client used by the actor.
 
+        Depending on the initialization of the service locator the client can be created in different ways.
+        """
         try:
-            # Set implicit default local storage client, unless local storage client was already set.
-            implicit_storage_client = ApifyFileSystemStorageClient()
+            # Nothing was set by the user.
+            implicit_storage_client = SmartApifyStorageClient(
+                local_storage_client=ApifyFileSystemStorageClient(), cloud_storage_client=ApifyStorageClient()
+            )
             service_locator.set_storage_client(implicit_storage_client)
-            self._local_storage_client = implicit_storage_client
         except ServiceConflictError:
             self.log.debug(
                 'Storage client in service locator was set explicitly before Actor.init was called.'
                 'Using the existing storage client as implicit storage client for the Actor.'
             )
-
-        self._local_storage_client = service_locator.get_storage_client()
-        if type(self._local_storage_client) is FileSystemStorageClient:
-            self.log.warning(
-                f'Using {FileSystemStorageClient.__module__}.{FileSystemStorageClient.__name__} in Actor context is not'
-                f' recommended and can lead to problems with reading the input file. Use '
-                f'`apify.storage_clients.FileSystemStorageClient` instead.'
-            )
-
-        return self._local_storage_client
-
-    def _raise_if_not_initialized(self) -> None:
-        if not self._is_initialized:
-            raise RuntimeError('The Actor was not initialized!')
-
-    def _raise_if_cloud_requested_but_not_configured(self, *, force_cloud: bool) -> None:
-        if not force_cloud:
-            return
-
-        if not self.is_at_home() and self.configuration.token is None:
-            raise RuntimeError(
-                'In order to use the Apify cloud storage from your computer, '
-                'you need to provide an Apify token using the APIFY_TOKEN environment variable.'
-            )
+        else:
+            return implicit_storage_client
+
+        # User set something in the service locator.
+        explicit_storage_client = service_locator.get_storage_client()
+        if isinstance(explicit_storage_client, SmartApifyStorageClient):
+            # The client was manually set to the right type in the service locator. This is the explicit way.
+            return explicit_storage_client
+
+        raise RuntimeError(
+            'The storage client in the service locator has to be instance of SmartApifyStorageClient. If you want to '
+            'set the storage client manually you have to call '
+            '`service_locator.set_storage_client(SmartApifyStorageClient(...))` before entering Actor context or '
+            'awaiting `Actor.init`.'
+        )
 
     async def init(self) -> None:
         """Initialize the Actor instance.
@@ -285,6 +280,7 @@ async def init(self) -> None:
         This method should be called immediately before performing any additional Actor actions, and it should be
         called only once.
         """
+        self.log.info('Initializing Actor...')
         if self._configuration:
             # Set explicitly the configuration in the service locator
             service_locator.set_configuration(self.configuration)
@@ -298,30 +294,20 @@ async def init(self) -> None:
         if _ActorType._is_any_instance_initialized:
             self.log.warning('Repeated Actor initialization detected - this is non-standard usage, proceed with care')
 
-        # Create an instance of the cloud storage client, the local storage client is obtained
-        # from the service locator
-        self._cloud_storage_client = ApifyStorageClient()
-
         # Make sure that the currently initialized instance is also available through the global `Actor` proxy
         cast('Proxy', Actor).__wrapped__ = self
 
         self._is_exiting = False
         self._was_final_persist_state_emitted = False
 
-        # If the Actor is running on the Apify platform, we set the cloud storage client.
-        if self.is_at_home():
-            service_locator.set_storage_client(self._cloud_storage_client)
-            self._local_storage_client = self._cloud_storage_client
-        else:
-            self._get_local_storage_client()
+        self.log.debug(f'Storage client set to {self._storage_client}')
 
         service_locator.set_event_manager(self.event_manager)
 
         # The logging configuration has to be called after all service_locator set methods.
         if self._configure_logging:
             _configure_logging()
 
-        self.log.info('Initializing Actor...')
         self.log.info('System info', extra=get_system_info())
 
         await self.event_manager.__aenter__()
@@ -470,16 +456,11 @@ async def open_dataset(
             An instance of the `Dataset` class for the given ID or name.
         """
         self._raise_if_not_initialized()
-        self._raise_if_cloud_requested_but_not_configured(force_cloud=force_cloud)
-
-        storage_client = self._cloud_storage_client if force_cloud else self._get_local_storage_client()
-
         return await Dataset.open(
             id=id,
-            alias=alias,
             name=name,
-            configuration=self.configuration,
-            storage_client=storage_client,
+            alias=alias,
+            storage_client=self._storage_client.get_suitable_storage_client(force_cloud=force_cloud),
         )
 
     async def open_key_value_store(
@@ -509,16 +490,11 @@ async def open_key_value_store(
             An instance of the `KeyValueStore` class for the given ID or name.
         """
         self._raise_if_not_initialized()
-        self._raise_if_cloud_requested_but_not_configured(force_cloud=force_cloud)
-
-        storage_client = self._cloud_storage_client if force_cloud else self._get_local_storage_client()
-
         return await KeyValueStore.open(
             id=id,
-            alias=alias,
             name=name,
-            configuration=self.configuration,
-            storage_client=storage_client,
+            alias=alias,
+            storage_client=self._storage_client.get_suitable_storage_client(force_cloud=force_cloud),
         )
 
     async def open_request_queue(
@@ -550,16 +526,11 @@ async def open_request_queue(
             An instance of the `RequestQueue` class for the given ID or name.
         """
         self._raise_if_not_initialized()
-        self._raise_if_cloud_requested_but_not_configured(force_cloud=force_cloud)
-
-        storage_client = self._cloud_storage_client if force_cloud else self._get_local_storage_client()
-
         return await RequestQueue.open(
             id=id,
-            alias=alias,
             name=name,
-            configuration=self.configuration,
-            storage_client=storage_client,
+            alias=alias,
+            storage_client=self._storage_client.get_suitable_storage_client(force_cloud=force_cloud),
         )
 
     @overload

src/apify/storage_clients/__init__.py

@@ -2,9 +2,11 @@
 
 from ._apify import ApifyStorageClient
 from ._file_system import ApifyFileSystemStorageClient as FileSystemStorageClient
+from ._smart_apify import SmartApifyStorageClient
 
 __all__ = [
     'ApifyStorageClient',
     'FileSystemStorageClient',
     'MemoryStorageClient',
+    'SmartApifyStorageClient',
 ]