Extract storage related complexity from Actor to dedicated storage client

Author: Pijukatel

docs/04_upgrading/upgrading_to_v3.md

@@ -53,6 +53,28 @@ async def main():
 
 <!-- TODO -->
 
+## 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, ApifyHybridStorageClient, MemoryStorageClient
+from apify import Actor
+
+
+async def main():
+    service_locator.set_storage_client(
+        ApifyHybridStorageClient(
+            cloud_storage_client=ApifyStorageClient(access="single"),
+            local_storage_client=MemoryStorageClient()
+        )
+    )
+    async with Actor:
+        rq = await Actor.open_request_queue()
+```
+
+
 ## The default use of optimized ApifyRequestQueueClient
 
 - 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.
@@ -61,12 +83,13 @@ async def main():
 **Now (v3.0):**
 
 ```python
-from apify.storages import RequestQueue
+from crawlee import service_locator
 from apify.storage_clients import ApifyStorageClient
+from apify import Actor
 
 async def main():
    # Full client that supports multiple consumers of the Apify Request Queue
-    rq_shared = await RequestQueue.open(storage_client=ApifyStorageClient(access="shared"))
-    # Default optimized client that expects only single consumer of the Apify Request Queue
-    rq_single = await RequestQueue.open(storage_client=ApifyStorageClient())
+    service_locator.set_storage_client(ApifyStorageClient(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 @ git+https://github.com/apify/crawlee-python.git@include-storag-client-in-additional-cache-key",
     "cachetools>=5.5.0",
     "cryptography>=42.0.0",
     "impit>=0.6.1",

src/apify/_actor.py

@@ -38,7 +38,7 @@
 from apify.log import _configure_logging, logger
 from apify.storage_clients import ApifyStorageClient
 from apify.storage_clients._file_system import ApifyFileSystemStorageClient
-from apify.storages import Dataset, KeyValueStore, RequestQueue
+from apify.storage_clients._hybrid_apify._storage_client import ApifyHybridStorageClient
 
 if TYPE_CHECKING:
     import logging
@@ -48,9 +48,9 @@
     from typing_extensions import Self
 
     from crawlee.proxy_configuration import _NewUrlFunction
-    from crawlee.storage_clients import StorageClient
 
     from apify._models import Webhook
+    from apify.storages import Dataset, KeyValueStore, RequestQueue
 
 
 MainReturnType = TypeVar('MainReturnType')
@@ -131,7 +131,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 +233,49 @@ 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) -> ApifyHybridStorageClient:
+        """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()
+            # Notning was set by the user.
+            implicit_storage_client = ApifyHybridStorageClient(
+                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.'
             )
+        else:
+            return implicit_storage_client
 
-        self._local_storage_client = service_locator.get_storage_client()
-        if type(self._local_storage_client) is FileSystemStorageClient:
+        # User set something in the service locator.
+        storage_client = service_locator.get_storage_client()
+        if isinstance(storage_client, ApifyHybridStorageClient):
+            # The client was manually set to the right type in the service locator. This is the explicit way.
+            return storage_client
+
+        if isinstance(storage_client, ApifyStorageClient):
+            # The cloud storage client was manually set in the service locator.
+            return ApifyHybridStorageClient(cloud_storage_client=storage_client)
+
+        # The local storage client was manually set in the service locator
+        if type(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.'
-            )
+        return ApifyHybridStorageClient(cloud_storage_client=ApifyStorageClient(), local_storage_client=storage_client)
 
     async def init(self) -> None:
         """Initialize the Actor instance.
@@ -298,22 +301,13 @@ 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)
 
@@ -470,17 +464,7 @@ 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,
-        )
+        return await self._storage_client.open_dataset(id=id, name=name, alias=alias, force_cloud=force_cloud)
 
     async def open_key_value_store(
         self,
@@ -509,17 +493,7 @@ 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,
-        )
+        return await self._storage_client.open_key_value_store(id=id, name=name, alias=alias, force_cloud=force_cloud)
 
     async def open_request_queue(
         self,
@@ -550,17 +524,7 @@ 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,
-        )
+        return await self._storage_client.open_request_queue(id=id, name=name, alias=alias, force_cloud=force_cloud)
 
     @overload
     async def push_data(self, data: dict | list[dict]) -> None: ...

src/apify/storage_clients/__init__.py

@@ -2,8 +2,10 @@
 
 from ._apify import ApifyStorageClient
 from ._file_system import ApifyFileSystemStorageClient as FileSystemStorageClient
+from ._hybrid_apify import ApifyHybridStorageClient
 
 __all__ = [
+    'ApifyHybridStorageClient',
     'ApifyStorageClient',
     'FileSystemStorageClient',
     'MemoryStorageClient',

src/apify/storage_clients/_hybrid_apify/__init__.py

@@ -0,0 +1 @@
+from ._storage_client import ApifyHybridStorageClient