refactor!: Make Actor initialization stricter and more predictable

docs/04_upgrading/upgrading_to_v3.md

@@ -9,6 +9,38 @@ This page summarizes the breaking changes between Apify Python SDK v2.x and v3.0
 
 Support for Python 3.9 has been dropped. The Apify Python SDK v3.x now requires Python 3.10 or later. Make sure your environment is running a compatible version before upgrading.
 
+## Actor initialization and ServiceLocator changes
+
+`Actor` initialization and global `service_locator` services setup is more strict and predictable.
+- Services in `Actor` can't be changed after calling `Actor.init`, entering the `async with Actor` context manager or after requesting them from the `Actor`
+- Services in `Actor` can be different from services in Crawler
+
+**Now (v3.0):**
+
+```python
+from crawlee.crawlers import BasicCrawler
+from crawlee.storage_clients import MemoryStorageClient, FileSystemStorageClient
+from crawlee.configuration import Configuration
+from crawlee.events import LocalEventManager
+from apify import Actor
+
+async def main():
+
+    async with Actor():
+        # This crawler will use same services as Actor and global service_locator
+        crawler_1 = BasicCrawler()
+
+        # This crawler will use custom services
+        custom_configuration = Configuration()
+        custom_event_manager = LocalEventManager.from_config(custom_configuration)
+        custom_storage_client = MemoryStorageClient()
+        crawler_2 = BasicCrawler(
+            configuration=custom_configuration,
+            event_manager=custom_event_manager,
+            storage_client=custom_storage_client,
+        )
+```
+
 ## Storages
 
 <!-- TODO -->

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==1.0.0rc1",
+    "crawlee @ git+https://github.com/apify/crawlee-python.git@storage-clients-and-configurations-2",
     "cachetools>=5.5.0",
     "cryptography>=42.0.0",
     "impit>=0.5.3",

src/apify/_configuration.py

@@ -8,6 +8,7 @@
 from pydantic import AliasChoices, BeforeValidator, Field, model_validator
 from typing_extensions import Self, deprecated
 
+from crawlee import service_locator
 from crawlee._utils.models import timedelta_ms
 from crawlee._utils.urls import validate_http_url
 from crawlee.configuration import Configuration as CrawleeConfiguration
@@ -424,11 +425,35 @@ def disable_browser_sandbox_on_platform(self) -> Self:
     def get_global_configuration(cls) -> Configuration:
         """Retrieve the global instance of the configuration.
 
-        Mostly for the backwards compatibility. It is recommended to use the `service_locator.get_configuration()`
-        instead.
+        This method ensures that ApifyConfigration is returned, even if CrawleeConfiguration was set in the
+        service locator.
         """
-        return cls()
+        global_configuration = service_locator.get_configuration()
 
+        if isinstance(global_configuration, Configuration):
+            # If Apify configuration was already stored in service locator, return it.
+            return global_configuration
 
-# Monkey-patch the base class so that it works with the extended configuration
-CrawleeConfiguration.get_global_configuration = Configuration.get_global_configuration  # type: ignore[method-assign]
+        return cls.from_configuration(global_configuration)
+
+    @classmethod
+    def from_configuration(cls, configuration: CrawleeConfiguration) -> Configuration:
+        """Create Apify Configuration from existing Crawlee Configuration.
+
+        Args:
+            configuration: The existing Crawlee Configuration.
+
+        Returns:
+            The created Apify Configuration.
+        """
+        apify_configuration = cls()
+
+        # Ensure the returned configuration is of type Apify Configuration.
+        # Most likely crawlee configuration was already set. Create Apify configuration from it.
+        # Due to known Pydantic issue https://github.com/pydantic/pydantic/issues/9516, creating new instance of
+        # Configuration from existing one in situation where environment can have some fields set by alias is very
+        # unpredictable. Use the stable workaround.
+        for name in configuration.model_fields:
+            setattr(apify_configuration, name, getattr(configuration, name))
+
+        return apify_configuration

src/apify/storage_clients/_apify/_storage_client.py

@@ -9,72 +9,71 @@
 from ._dataset_client import ApifyDatasetClient
 from ._key_value_store_client import ApifyKeyValueStoreClient
 from ._request_queue_client import ApifyRequestQueueClient
+from apify._configuration import Configuration as ApifyConfiguration
 from apify._utils import docs_group
 
 if TYPE_CHECKING:
-    from crawlee.configuration import Configuration
+    from collections.abc import Hashable
+
+    from crawlee.configuration import Configuration as CrawleeConfiguration
 
 
 @docs_group('Storage clients')
 class ApifyStorageClient(StorageClient):
     """Apify storage client."""
 
+    # This class breaches Liskov Substitution Principle. It requires specialized Configuration compared to its parent.
+    _lsp_violation_error_message_template = (
+        'Expected "configuration" to be an instance of "apify.Configuration", but got {} instead.'
+    )
+
+    @override
+    def get_additional_cache_key(self, configuration: CrawleeConfiguration) -> Hashable:
+        if isinstance(configuration, ApifyConfiguration):
+            return f'{configuration.api_base_url},{configuration.token}'
+        raise TypeError(self._lsp_violation_error_message_template.format(type(configuration).__name__))
+
     @override
     async def create_dataset_client(
         self,
         *,
         id: str | None = None,
         name: str | None = None,
-        configuration: Configuration | None = None,
+        alias: str | None = None,
+        configuration: CrawleeConfiguration | None = None,
     ) -> ApifyDatasetClient:
-        # Import here to avoid circular imports.
-        from apify import Configuration as ApifyConfiguration  # noqa: PLC0415
-
         configuration = configuration or ApifyConfiguration.get_global_configuration()
         if isinstance(configuration, ApifyConfiguration):
             return await ApifyDatasetClient.open(id=id, name=name, configuration=configuration)
 
-        raise TypeError(
-            f'Expected "configuration" to be an instance of "apify.Configuration", '
-            f'but got {type(configuration).__name__} instead.'
-        )
+        raise TypeError(self._lsp_violation_error_message_template.format(type(configuration).__name__))
 
     @override
     async def create_kvs_client(
         self,
         *,
         id: str | None = None,
         name: str | None = None,
-        configuration: Configuration | None = None,
+        alias: str | None = None,
+        configuration: CrawleeConfiguration | None = None,
     ) -> ApifyKeyValueStoreClient:
-        # Import here to avoid circular imports.
-        from apify import Configuration as ApifyConfiguration  # noqa: PLC0415
-
         configuration = configuration or ApifyConfiguration.get_global_configuration()
         if isinstance(configuration, ApifyConfiguration):
             return await ApifyKeyValueStoreClient.open(id=id, name=name, configuration=configuration)
 
-        raise TypeError(
-            f'Expected "configuration" to be an instance of "apify.Configuration", '
-            f'but got {type(configuration).__name__} instead.'
-        )
+        raise TypeError(self._lsp_violation_error_message_template.format(type(configuration).__name__))
 
     @override
     async def create_rq_client(
         self,
         *,
         id: str | None = None,
         name: str | None = None,
-        configuration: Configuration | None = None,
+        alias: str | None = None,
+        configuration: CrawleeConfiguration | None = None,
     ) -> ApifyRequestQueueClient:
-        # Import here to avoid circular imports.
-        from apify import Configuration as ApifyConfiguration  # noqa: PLC0415
-
         configuration = configuration or ApifyConfiguration.get_global_configuration()
         if isinstance(configuration, ApifyConfiguration):
             return await ApifyRequestQueueClient.open(id=id, name=name, configuration=configuration)
 
-        raise TypeError(
-            f'Expected "configuration" to be an instance of "apify.Configuration", '
-            f'but got {type(configuration).__name__} instead.'
-        )
+        raise TypeError(self._lsp_violation_error_message_template.format(type(configuration).__name__))

src/apify/storage_clients/_file_system/_storage_client.py

@@ -27,9 +27,12 @@ async def create_kvs_client(
         *,
         id: str | None = None,
         name: str | None = None,
+        alias: str | None = None,
         configuration: Configuration | None = None,
     ) -> FileSystemKeyValueStoreClient:
         configuration = configuration or Configuration.get_global_configuration()
-        client = await ApifyFileSystemKeyValueStoreClient.open(id=id, name=name, configuration=configuration)
+        client = await ApifyFileSystemKeyValueStoreClient.open(
+            id=id, name=name, alias=alias, configuration=configuration
+        )
         await self._purge_if_needed(client, configuration)
         return client

tests/integration/actor_source_base/requirements.txt

@@ -1,4 +1,4 @@
 # The test fixture will put the Apify SDK wheel path on the next line
 APIFY_SDK_WHEEL_PLACEHOLDER
 uvicorn[standard]
-crawlee[parsel]==1.0.0rc1
+crawlee[parsel] @ git+https://github.com/apify/crawlee-python.git@storage-clients-and-configurations-2

tests/integration/conftest.py

@@ -58,18 +58,10 @@ def _prepare_test_env() -> None:
         service_locator._configuration = None
         service_locator._event_manager = None
         service_locator._storage_client = None
-        service_locator._storage_instance_manager = None
-
-        # Reset the retrieval flags.
-        service_locator._configuration_was_retrieved = False
-        service_locator._event_manager_was_retrieved = False
-        service_locator._storage_client_was_retrieved = False
+        service_locator.storage_instance_manager.clear_cache()
 
         # Verify that the test environment was set up correctly.
         assert os.environ.get(ApifyEnvVars.LOCAL_STORAGE_DIR) == str(tmp_path)
-        assert service_locator._configuration_was_retrieved is False
-        assert service_locator._storage_client_was_retrieved is False
-        assert service_locator._event_manager_was_retrieved is False
 
     return _prepare_test_env
 

tests/integration/test_actor_migration.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .conftest import MakeActorFunction, RunActorFunction
+
+
+async def test_migration_through_reboot(make_actor: MakeActorFunction, run_actor: RunActorFunction) -> None:
+    """Test that actor works as expected after migration through testing behavior after reboot.
+
+    Handle two requests. Migrate in between the two requests."""
+
+    async def main() -> None:
+        from crawlee._types import BasicCrawlingContext, ConcurrencySettings
+        from crawlee.crawlers import BasicCrawler
+
+        from apify import Actor
+
+        async with Actor:
+            crawler = BasicCrawler(concurrency_settings=ConcurrencySettings(max_concurrency=1))
+            requests = ['https://example.com/1', 'https://example.com/2']
+
+            run = await Actor.apify_client.run(Actor.config.actor_run_id or '').get()
+            assert run
+            first_run = run.get('stats', {}).get('rebootCount', 0) == 0
+            Actor.log.warning(run)
+
+            @crawler.router.default_handler
+            async def default_handler(context: BasicCrawlingContext) -> None:
+                context.log.info(f'Processing {context.request.url} ...')
+
+                # Simulate migration through reboot
+                if context.request.url == requests[1] and first_run:
+                    context.log.info(f'Reclaiming {context.request.url} ...')
+                    rq = await crawler.get_request_manager()
+                    await rq.reclaim_request(context.request)
+                    await Actor.reboot()
+
+            await crawler.run(requests)
+
+            # Each time one request is finished.
+            assert crawler.statistics.state.requests_finished == 1
+
+    actor = await make_actor(label='migration', main_func=main)
+    run_result = await run_actor(actor)
+
+    assert run_result.status == 'SUCCEEDED'

tests/unit/actor/test_actor_lifecycle.py

@@ -6,6 +6,7 @@
 import sys
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any, cast
+from unittest import mock
 from unittest.mock import AsyncMock, Mock
 
 import pytest
@@ -179,25 +180,24 @@ async def handler(websocket: websockets.asyncio.server.ServerConnection) -> None
             }
         )
 
-        monkeypatch.setattr(Actor._charging_manager, '_client', mock_run_client)
-
-        async with Actor:
-            Actor.on(Event.PERSIST_STATE, log_persist_state)
-            await asyncio.sleep(2)
-
-            for socket in ws_server.connections:
-                await socket.send(
-                    json.dumps(
-                        {
-                            'name': 'migrating',
-                            'data': {
-                                'isMigrating': True,
-                            },
-                        }
+        with mock.patch.object(Actor, 'new_client', return_value=mock_run_client):
+            async with Actor:
+                Actor.on(Event.PERSIST_STATE, log_persist_state)
+                await asyncio.sleep(2)
+
+                for socket in ws_server.connections:
+                    await socket.send(
+                        json.dumps(
+                            {
+                                'name': 'migrating',
+                                'data': {
+                                    'isMigrating': True,
+                                },
+                            }
+                        )
                     )
-                )
 
-            await asyncio.sleep(1)
+                await asyncio.sleep(1)
 
     assert len(persist_state_events_data) >= 3
 

tests/unit/actor/test_apify_storage.py

@@ -0,0 +1,54 @@
+from unittest import mock
+from unittest.mock import AsyncMock
+
+import pytest
+
+from crawlee.storages import Dataset, KeyValueStore, RequestQueue
+from crawlee.storages._base import Storage
+
+from apify import Configuration
+from apify.storage_clients import ApifyStorageClient
+from apify.storage_clients._apify import ApifyDatasetClient, ApifyKeyValueStoreClient, ApifyRequestQueueClient
+
+
+@pytest.mark.parametrize(
+    ('storage', '_storage_client'),
+    [
+        (Dataset, ApifyDatasetClient),
+        (KeyValueStore, ApifyKeyValueStoreClient),
+        (RequestQueue, ApifyRequestQueueClient),
+    ],
+)
+async def test_get_additional_cache_key(
+    storage: Storage, _storage_client: ApifyDatasetClient | ApifyKeyValueStoreClient | ApifyRequestQueueClient
+) -> None:
+    """Test that Storages based on `ApifyStorageClient` include `token` and `api_base_url` in additional cache key."""
+    storage_names = iter(['1', '2', '3', '1', '3'])
+
+    apify_storage_client = ApifyStorageClient()
+
+    config_1 = Configuration(token='a')
+    config_2 = Configuration(token='b')
+    config_3 = Configuration(token='a', api_base_url='https://super_custom_api.com')
+
+    config_4 = Configuration(token='a')
+    config_5 = Configuration(token='a', api_base_url='https://super_custom_api.com')
+
+    mocked_open = AsyncMock(spec=_storage_client.open)
+    mocked_open.get_metadata = AsyncMock(storage_names)
+
+    with mock.patch.object(_storage_client, 'open', mocked_open):
+        storage_1 = await storage.open(storage_client=apify_storage_client, configuration=config_1)
+        storage_2 = await storage.open(storage_client=apify_storage_client, configuration=config_2)
+        storage_3 = await storage.open(storage_client=apify_storage_client, configuration=config_3)
+        storage_4 = await storage.open(storage_client=apify_storage_client, configuration=config_4)
+        storage_5 = await storage.open(storage_client=apify_storage_client, configuration=config_5)
+
+        # Different configuration results in different storage clients.
+        assert storage_1 is not storage_2
+        assert storage_1 is not storage_3
+        assert storage_2 is not storage_3
+
+        # Equivalent configuration results in same storage clients.
+        assert storage_1 is storage_4
+        assert storage_3 is storage_5