Merge remote-tracking branch 'origin/master' into better-event-listener-typing

Author: janbuchar

.github/workflows/pre_release.yaml

@@ -8,6 +8,9 @@ on:
     tags-ignore:
       - "**" # Ignore all tags to prevent duplicate builds when tags are pushed.
 
+  # Or it can be triggered manually.
+  workflow_dispatch:
+
 jobs:
   release_metadata:
     if: "!startsWith(github.event.head_commit.message, 'docs') && !startsWith(github.event.head_commit.message, 'ci') && startsWith(github.repository, 'apify/')"
@@ -65,26 +68,11 @@ jobs:
     steps:
       - name: Prepare distribution
         uses: apify/workflows/prepare-pypi-distribution@main
-        with: 
+        with:
           package_name: apify
           is_prerelease: "yes"
           version_number: ${{ needs.release_metadata.outputs.version_number }}
           ref: ${{ needs.update_changelog.changelog_commitish }}
       # Publishes the package to PyPI using PyPA official GitHub action with OIDC authentication.
       - name: Publish package to PyPI
         uses: pypa/gh-action-pypi-publish@release/v1
-
-  trigger_docker_build:
-    name: Trigger Docker image build
-    needs: [release_metadata, update_changelog]
-    runs-on: ubuntu-latest
-    steps:
-      - # Trigger building the Python Docker images in apify/apify-actor-docker repo
-        name: Trigger Docker image build
-        run: |
-          gh api -X POST "/repos/apify/apify-actor-docker/dispatches" \
-            -F event_type=build-python-images \
-            -F client_payload[release_tag]=beta \
-            -F client_payload[apify_version]=${{ needs.release_metadata.outputs.version_number }}
-        env:
-          GH_TOKEN: ${{ secrets.APIFY_SERVICE_ACCOUNT_GITHUB_TOKEN }}

CHANGELOG.md

@@ -2,6 +2,19 @@
 
 All notable changes to this project will be documented in this file.
 
+## [2.2.0](https://github.com/apify/apify-sdk-python/releases/tag/v2.2.0) (2025-01-10)
+
+### 🚀 Features
+
+- Add new config variables to `Actor.config` ([#351](https://github.com/apify/apify-sdk-python/pull/351)) ([7b6478c](https://github.com/apify/apify-sdk-python/commit/7b6478c3fc239b454f733fbd98348dab7b3a1766)) by [@fnesveda](https://github.com/fnesveda)
+- Upgrade to Crawlee v0.5 ([#355](https://github.com/apify/apify-sdk-python/pull/355)) ([826f4db](https://github.com/apify/apify-sdk-python/commit/826f4dbcc8cfd693d97e40c17faf91d225d7ffaf)) by [@vdusek](https://github.com/vdusek)
+
+### 🐛 Bug Fixes
+
+- Better error message when attempting to use force_cloud without an Apify token ([#356](https://github.com/apify/apify-sdk-python/pull/356)) ([33245ce](https://github.com/apify/apify-sdk-python/commit/33245ceddb1fa0ed39548181fb57fb3e6b98f954)) by [@janbuchar](https://github.com/janbuchar)
+- Allow calling `Actor.reboot()` from migrating handler, align reboot behavior with JS SDK ([#361](https://github.com/apify/apify-sdk-python/pull/361)) ([7ba0221](https://github.com/apify/apify-sdk-python/commit/7ba022121fe7b65470fec901295f74cebce72610)) by [@fnesveda](https://github.com/fnesveda)
+
+
 ## [2.1.0](https://github.com/apify/apify-sdk-python/releases/tag/v2.1.0) (2024-12-03)
 
 ### 🚀 Features

docs/03-concepts/04-actor-events.mdx

@@ -40,6 +40,8 @@ During its runtime, the Actor receives Actor events sent by the Apify platform o
                 {' '}to another worker server soon.</p>
                 You can use it to persist the state of the Actor so that once it is executed again on the new server,
                 it doesn't have to start over from the beginning.
+                Once you have persisted the state of your Actor, you can call <a href="../../reference/class/Actor#reboot"><code>Actor.reboot()</code></a>
+                to reboot the Actor and trigger the migration immediately, to speed up the process.
             </td>
         </tr>
         <tr>

poetry.lock

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "apify"
-version = "2.1.0"
+version = "2.2.0"
 description = "Apify SDK for Python"
 authors = ["Apify Technologies s.r.o. <[email protected]>"]
 license = "Apache-2.0"
@@ -45,32 +45,33 @@ keywords = [
 python = "^3.9"
 apify-client = ">=1.8.1"
 apify-shared = ">=1.1.2"
-crawlee = { git = "https://github.com/apify/crawlee-python.git", branch = "improve-event-types" }
+crawlee = "~0.5.1"
 cryptography = ">=42.0.0"
-# TODO: relax the upper bound once the issue is resolved:
-# https://github.com/apify/apify-sdk-python/issues/348
-httpx = "~0.27.0"
+httpx = ">=0.27.0"
 lazy-object-proxy = ">=1.10.0"
+more_itertools = ">=10.2.0"
 scrapy = { version = ">=2.11.0", optional = true }
 typing-extensions = ">=4.1.0"
+# TODO: Relax the upper bound once the issue is resolved:
+# https://github.com/apify/apify-sdk-python/issues/325
 websockets = ">=10.0 <14.0.0"
 
 [tool.poetry.group.dev.dependencies]
 build = "~1.2.0"
 filelock = "~3.16.0"
 griffe = "~1.5.0"
-mypy = "~1.13.0"
+mypy = "~1.14.0"
 pre-commit = "~4.0.0"
 pydoc-markdown = "~4.8.0"
 pytest = "~8.3.0"
-pytest-asyncio = "~0.24.0"
+pytest-asyncio = "~0.25.0"
 pytest-cov = "~6.0.0"
 pytest-only = "~2.1.0"
 pytest-timeout = "~2.3.0"
 pytest-xdist = "~3.6.0"
-respx = "~0.21.0"
-ruff = "~0.8.0"
-setuptools = "~75.6.0"     # setuptools are used by pytest but not explicitly required
+respx = "~0.22.0"
+ruff = "~0.9.0"
+setuptools = "~75.8.0"     # setuptools are used by pytest but not explicitly required
 
 [tool.poetry.extras]
 scrapy = ["scrapy"]

pyproject.toml

@@ -7,12 +7,13 @@
 from typing import TYPE_CHECKING, Any, Callable, Literal, TypeVar, cast, overload
 
 from lazy_object_proxy import Proxy
+from more_itertools import flatten
 from pydantic import AliasChoices
 
 from apify_client import ApifyClientAsync
 from apify_shared.consts import ActorEnvVars, ActorExitCodes, ApifyEnvVars
 from apify_shared.utils import ignore_docs, maybe_extract_enum_member_value
-from crawlee import service_container
+from crawlee import service_locator
 from crawlee.events import (
     Event,
     EventAbortingData,
@@ -41,6 +42,7 @@
     from typing_extensions import Self
 
     from crawlee.proxy_configuration import _NewUrlFunction
+    from crawlee.storage_clients import BaseStorageClient
 
     from apify._models import Webhook
 
@@ -56,6 +58,7 @@ class _ActorType:
     _apify_client: ApifyClientAsync
     _configuration: Configuration
     _is_exiting = False
+    _is_rebooting = False
 
     def __init__(
         self,
@@ -77,17 +80,22 @@ def __init__(
         self._configure_logging = configure_logging
         self._apify_client = self.new_client()
 
-        self._event_manager: EventManager
-        if self._configuration.is_at_home:
-            self._event_manager = PlatformEventManager(
+        # Create an instance of the cloud storage client, the local storage client is obtained
+        # from the service locator.
+        self._cloud_storage_client = ApifyStorageClient.from_config(config=self._configuration)
+
+        # Set the event manager based on whether the Actor is running on the platform or locally.
+        self._event_manager = (
+            PlatformEventManager(
                 config=self._configuration,
                 persist_state_interval=self._configuration.persist_state_interval,
             )
-        else:
-            self._event_manager = LocalEventManager(
+            if self.is_at_home()
+            else LocalEventManager(
                 system_info_interval=self._configuration.system_info_interval,
                 persist_state_interval=self._configuration.persist_state_interval,
             )
+        )
 
         self._is_initialized = False
 
@@ -100,9 +108,6 @@ async def __aenter__(self) -> Self:
         When you exit the `async with` block, the `Actor.exit()` method is called, and if any exception happens while
         executing the block code, the `Actor.fail` method is called.
         """
-        if self._configure_logging:
-            _configure_logging(self._configuration)
-
         await self.init()
         return self
 
@@ -162,10 +167,25 @@ def log(self) -> logging.Logger:
         """The logging.Logger instance the Actor uses."""
         return logger
 
+    @property
+    def _local_storage_client(self) -> BaseStorageClient:
+        """The local storage client the Actor instance uses."""
+        return service_locator.get_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.config.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.'
+            )
+
     async def init(self) -> None:
         """Initialize the Actor instance.
 
@@ -180,18 +200,19 @@ async def init(self) -> None:
         if self._is_initialized:
             raise RuntimeError('The Actor was already initialized!')
 
-        if self._configuration.token:
-            service_container.set_cloud_storage_client(ApifyStorageClient(configuration=self._configuration))
+        self._is_exiting = False
+        self._was_final_persist_state_emitted = False
 
-        if self._configuration.is_at_home:
-            service_container.set_default_storage_client_type('cloud')
-        else:
-            service_container.set_default_storage_client_type('local')
+        # 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)
 
-        service_container.set_event_manager(self._event_manager)
+        service_locator.set_event_manager(self.event_manager)
+        service_locator.set_configuration(self.configuration)
 
-        self._is_exiting = False
-        self._was_final_persist_state_emitted = False
+        # 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())
@@ -241,7 +262,6 @@ async def finalize() -> None:
                 await self._event_manager.wait_for_all_listeners_to_complete(timeout=event_listeners_timeout)
 
             await self._event_manager.__aexit__(None, None, None)
-            cast(dict, service_container._services).clear()  # noqa: SLF001
 
         await asyncio.wait_for(finalize(), cleanup_timeout.total_seconds())
         self._is_initialized = False
@@ -343,12 +363,15 @@ 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._local_storage_client
 
         return await Dataset.open(
             id=id,
             name=name,
             configuration=self._configuration,
-            storage_client=service_container.get_storage_client(client_type='cloud' if force_cloud else None),
+            storage_client=storage_client,
         )
 
     async def open_key_value_store(
@@ -375,12 +398,14 @@ 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._local_storage_client
 
         return await KeyValueStore.open(
             id=id,
             name=name,
             configuration=self._configuration,
-            storage_client=service_container.get_storage_client(client_type='cloud' if force_cloud else None),
+            storage_client=storage_client,
         )
 
     async def open_request_queue(
@@ -409,12 +434,15 @@ 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._local_storage_client
 
         return await RequestQueue.open(
             id=id,
             name=name,
             configuration=self._configuration,
-            storage_client=service_container.get_storage_client(client_type='cloud' if force_cloud else None),
+            storage_client=storage_client,
         )
 
     async def push_data(self, data: dict | list[dict]) -> None:
@@ -870,12 +898,32 @@ async def reboot(
             self.log.error('Actor.reboot() is only supported when running on the Apify platform.')
             return
 
+        if self._is_rebooting:
+            self.log.debug('Actor is already rebooting, skipping the additional reboot call.')
+            return
+
+        self._is_rebooting = True
+
         if not custom_after_sleep:
             custom_after_sleep = self._configuration.metamorph_after_sleep
 
-        self._event_manager.emit(event=Event.PERSIST_STATE, event_data=EventPersistStateData(is_migrating=True))
+        # Call all the listeners for the PERSIST_STATE and MIGRATING events, and wait for them to finish.
+        # PERSIST_STATE listeners are called to allow the Actor to persist its state before the reboot.
+        # MIGRATING listeners are called to allow the Actor to gracefully stop in-progress tasks before the reboot.
+        # Typically, crawlers are listening for the MIIGRATING event to stop processing new requests.
+        # We can't just emit the events and wait for all listeners to finish,
+        # because this method might be called from an event listener itself, and we would deadlock.
+        persist_state_listeners = flatten(
+            (self._event_manager._listeners_to_wrappers[Event.PERSIST_STATE] or {}).values()  # noqa: SLF001
+        )
+        migrating_listeners = flatten(
+            (self._event_manager._listeners_to_wrappers[Event.MIGRATING] or {}).values()  # noqa: SLF001
+        )
 
-        await self._event_manager.__aexit__(None, None, None)
+        await asyncio.gather(
+            *[listener(EventPersistStateData(is_migrating=True)) for listener in persist_state_listeners],
+            *[listener(EventMigratingData()) for listener in migrating_listeners],
+        )
 
         if not self._configuration.actor_run_id:
             raise RuntimeError('actor_run_id cannot be None when running on the Apify platform.')
@@ -972,7 +1020,7 @@ async def create_proxy_configuration(
         password: str | None = None,
         groups: list[str] | None = None,
         country_code: str | None = None,
-        proxy_urls: list[str] | None = None,
+        proxy_urls: list[str | None] | None = None,
         new_url_function: _NewUrlFunction | None = None,
     ) -> ProxyConfiguration | None:
         """Create a ProxyConfiguration object with the passed proxy configuration.