Merge remote-tracking branch 'origin/master' into unique-key

Author: Pijukatel

CHANGELOG.md

@@ -5,10 +5,19 @@ All notable changes to this project will be documented in this file.
 <!-- git-cliff-unreleased-start -->
 ## 2.7.1 - **not yet released**
 
+### 🚀 Features
+
+- Add deduplication to `add_batch_of_requests` ([#534](https://github.com/apify/apify-sdk-python/pull/534)) ([dd03c4d](https://github.com/apify/apify-sdk-python/commit/dd03c4d446f611492adf35f1b5738648ee5a66f7)) by [@Pijukatel](https://github.com/Pijukatel), closes [#514](https://github.com/apify/apify-sdk-python/issues/514)
+
 ### 🐛 Bug Fixes
 
 - Restrict apify-shared and apify-client versions ([#523](https://github.com/apify/apify-sdk-python/pull/523)) ([b3ae5a9](https://github.com/apify/apify-sdk-python/commit/b3ae5a972a65454a4998eda59c9fcc3f6b7e8579)) by [@vdusek](https://github.com/vdusek)
 - Expose `APIFY_USER_IS_PAYING` env var to the configuration ([#507](https://github.com/apify/apify-sdk-python/pull/507)) ([0801e54](https://github.com/apify/apify-sdk-python/commit/0801e54887317c1280cc6828ecd3f2cc53287e76)) by [@stepskop](https://github.com/stepskop)
+- Resolve DeprecationWarning in ApifyEventManager ([#555](https://github.com/apify/apify-sdk-python/pull/555)) ([0c5111d](https://github.com/apify/apify-sdk-python/commit/0c5111dafe19796ec1fb9652a44c031bed9758df)) by [@vdusek](https://github.com/vdusek), closes [#343](https://github.com/apify/apify-sdk-python/issues/343)
+
+### Chore
+
+- [**breaking**] Update apify-client and apify-shared to v2.0 ([#548](https://github.com/apify/apify-sdk-python/pull/548)) ([8ba084d](https://github.com/apify/apify-sdk-python/commit/8ba084ded6cd018111343f2219260b481c8d4e35)) by [@vdusek](https://github.com/vdusek)
 
 ### Refactor
 
@@ -64,6 +73,54 @@ All notable changes to this project will be documented in this file.
 - Tagline overlap ([#501](https://github.com/apify/apify-sdk-python/pull/501)) ([bae8340](https://github.com/apify/apify-sdk-python/commit/bae8340c46fea756ea35ea4d591da84c09d478e2)) by [@katzino](https://github.com/katzino)
 
 
+## [2.7.0](https://github.com/apify/apify-sdk-python/releases/tag/v2.7.0) (2025-07-14)
+
+### 🚀 Features
+
+- **crypto:** Decrypt secret objects ([#482](https://github.com/apify/apify-sdk-python/pull/482)) ([ce9daf7](https://github.com/apify/apify-sdk-python/commit/ce9daf7381212b8dc194e8a643e5ca0dedbc0078)) by [@MFori](https://github.com/MFori)
+
+### 🐛 Bug Fixes
+
+- Sync `@docusaurus` theme version [internal] ([#500](https://github.com/apify/apify-sdk-python/pull/500)) ([a7485e7](https://github.com/apify/apify-sdk-python/commit/a7485e7d2276fde464ce862573d5b95e7d4d836a)) by [@katzino](https://github.com/katzino)
+- Tagline overlap ([#501](https://github.com/apify/apify-sdk-python/pull/501)) ([bae8340](https://github.com/apify/apify-sdk-python/commit/bae8340c46fea756ea35ea4d591da84c09d478e2)) by [@katzino](https://github.com/katzino)
+
+
+## [2.7.0](https://github.com/apify/apify-sdk-python/releases/tag/v2.7.0) (2025-07-14)
+
+### 🚀 Features
+
+- **crypto:** Decrypt secret objects ([#482](https://github.com/apify/apify-sdk-python/pull/482)) ([ce9daf7](https://github.com/apify/apify-sdk-python/commit/ce9daf7381212b8dc194e8a643e5ca0dedbc0078)) by [@MFori](https://github.com/MFori)
+
+### 🐛 Bug Fixes
+
+- Sync `@docusaurus` theme version [internal] ([#500](https://github.com/apify/apify-sdk-python/pull/500)) ([a7485e7](https://github.com/apify/apify-sdk-python/commit/a7485e7d2276fde464ce862573d5b95e7d4d836a)) by [@katzino](https://github.com/katzino)
+- Tagline overlap ([#501](https://github.com/apify/apify-sdk-python/pull/501)) ([bae8340](https://github.com/apify/apify-sdk-python/commit/bae8340c46fea756ea35ea4d591da84c09d478e2)) by [@katzino](https://github.com/katzino)
+
+
+## [2.7.0](https://github.com/apify/apify-sdk-python/releases/tag/v2.7.0) (2025-07-14)
+
+### 🚀 Features
+
+- **crypto:** Decrypt secret objects ([#482](https://github.com/apify/apify-sdk-python/pull/482)) ([ce9daf7](https://github.com/apify/apify-sdk-python/commit/ce9daf7381212b8dc194e8a643e5ca0dedbc0078)) by [@MFori](https://github.com/MFori)
+
+### 🐛 Bug Fixes
+
+- Sync `@docusaurus` theme version [internal] ([#500](https://github.com/apify/apify-sdk-python/pull/500)) ([a7485e7](https://github.com/apify/apify-sdk-python/commit/a7485e7d2276fde464ce862573d5b95e7d4d836a)) by [@katzino](https://github.com/katzino)
+- Tagline overlap ([#501](https://github.com/apify/apify-sdk-python/pull/501)) ([bae8340](https://github.com/apify/apify-sdk-python/commit/bae8340c46fea756ea35ea4d591da84c09d478e2)) by [@katzino](https://github.com/katzino)
+
+
+## [2.7.0](https://github.com/apify/apify-sdk-python/releases/tag/v2.7.0) (2025-07-14)
+
+### 🚀 Features
+
+- **crypto:** Decrypt secret objects ([#482](https://github.com/apify/apify-sdk-python/pull/482)) ([ce9daf7](https://github.com/apify/apify-sdk-python/commit/ce9daf7381212b8dc194e8a643e5ca0dedbc0078)) by [@MFori](https://github.com/MFori)
+
+### 🐛 Bug Fixes
+
+- Sync `@docusaurus` theme version [internal] ([#500](https://github.com/apify/apify-sdk-python/pull/500)) ([a7485e7](https://github.com/apify/apify-sdk-python/commit/a7485e7d2276fde464ce862573d5b95e7d4d836a)) by [@katzino](https://github.com/katzino)
+- Tagline overlap ([#501](https://github.com/apify/apify-sdk-python/pull/501)) ([bae8340](https://github.com/apify/apify-sdk-python/commit/bae8340c46fea756ea35ea4d591da84c09d478e2)) by [@katzino](https://github.com/katzino)
+
+
 ## [2.7.3](https://github.com/apify/apify-sdk-python/releases/tag/v2.7.3) (2025-08-11)
 
 ### 🐛 Bug Fixes

pyproject.toml

@@ -65,22 +65,22 @@ scrapy = ["scrapy>=2.11.0"]
 dev = [
     "build~=1.3.0",
     "crawlee[parsel]",
-    "dycw-pytest-only>=2.1.1",
-    "griffe~=1.12.1",
+    "dycw-pytest-only~=2.1.0",
+    "griffe~=1.12.0",
     "mypy~=1.17.0",
     "pre-commit~=4.3.0",
     "pydoc-markdown~=4.8.0",
     "pytest-asyncio~=1.1.0",
     "pytest-cov~=6.2.0",
-    "pytest-httpserver>=1.1.3",
-    "pytest-timeout>=2.4.0",
+    "pytest-httpserver~=1.1.0",
+    "pytest-timeout~=2.4.0",
     "pytest-xdist~=3.8.0",
     "pytest~=8.4.0",
     "ruff~=0.12.0",
     "setuptools", # setuptools are used by pytest but not explicitly required
-    "types-cachetools>=6.0.0.20250525",
+    "types-cachetools~=6.0.0.20250525",
     "uvicorn[standard]",
-    "werkzeug~=3.1.3", # Werkzeug is used by httpserver
+    "werkzeug~=3.1.0", # Werkzeug is used by httpserver
     "yarl~=1.20.0", # yarl is used by crawlee
 ]
 

src/apify/_actor.py

@@ -130,7 +130,7 @@ def __init__(
         # Set the event manager based on whether the Actor is running on the platform or locally.
         self._event_manager = (
             ApifyEventManager(
-                config=self._configuration,
+                configuration=self._configuration,
                 persist_state_interval=self._configuration.persist_state_interval,
             )
             if self.is_at_home()

src/apify/events/_apify_event_manager.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 from typing import TYPE_CHECKING, Annotated
 
 import websockets.asyncio.client
@@ -29,39 +30,51 @@
 
 @docs_group('Event managers')
 class ApifyEventManager(EventManager):
-    """A class for managing Actor events.
+    """Event manager for the Apify platform.
 
-    You shouldn't use this class directly,
-    but instead use it via the `Actor.on()` and `Actor.off()` methods.
-    """
+    This class extends Crawlee's `EventManager` to provide Apify-specific functionality, including websocket
+    connectivity to the Apify platform for receiving platform events.
+
+    The event manager handles:
+    - Registration and emission of events and their listeners.
+    - Websocket connection to Apify platform events.
+    - Processing and validation of platform messages.
+    - Automatic event forwarding from the platform to local event listeners.
 
-    _platform_events_websocket: websockets.asyncio.client.ClientConnection | None = None
-    _process_platform_messages_task: asyncio.Task | None = None
-    _send_system_info_interval_task: asyncio.Task | None = None
-    _connected_to_platform_websocket: asyncio.Future = asyncio.Future()
+    This class should not be used directly. Use the `Actor.on` and `Actor.off` methods to interact
+    with the event system.
+    """
 
-    def __init__(self, config: Configuration, **kwargs: Unpack[EventManagerOptions]) -> None:
-        """Create an instance of the EventManager.
+    def __init__(self, configuration: Configuration, **kwargs: Unpack[EventManagerOptions]) -> None:
+        """Initialize a new instance.
 
         Args:
-            config: The Actor configuration to be used in this event manager.
-            kwargs: Event manager options - forwarded to the base class
+            configuration: The Actor configuration for the event manager.
+            **kwargs: Additional event manager options passed to the parent class.
         """
         super().__init__(**kwargs)
 
-        self._config = config
-        self._listener_tasks = set()
-        self._connected_to_platform_websocket = asyncio.Future[bool]()
+        self._configuration = configuration
+        """The Actor configuration for the event manager."""
+
+        self._platform_events_websocket: websockets.asyncio.client.ClientConnection | None = None
+        """WebSocket connection to the platform events."""
+
+        self._process_platform_messages_task: asyncio.Task | None = None
+        """Task for processing messages from the platform websocket."""
+
+        self._connected_to_platform_websocket: asyncio.Future[bool] | None = None
+        """Future that resolves when the connection to the platform websocket is established."""
 
     @override
     async def __aenter__(self) -> Self:
         await super().__aenter__()
         self._connected_to_platform_websocket = asyncio.Future()
 
         # Run tasks but don't await them
-        if self._config.actor_events_ws_url:
+        if self._configuration.actor_events_ws_url:
             self._process_platform_messages_task = asyncio.create_task(
-                self._process_platform_messages(self._config.actor_events_ws_url)
+                self._process_platform_messages(self._configuration.actor_events_ws_url)
             )
             is_connected = await self._connected_to_platform_websocket
             if not is_connected:
@@ -81,16 +94,19 @@ async def __aexit__(
         if self._platform_events_websocket:
             await self._platform_events_websocket.close()
 
-        if self._process_platform_messages_task:
-            await self._process_platform_messages_task
+        if self._process_platform_messages_task and not self._process_platform_messages_task.done():
+            self._process_platform_messages_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._process_platform_messages_task
 
         await super().__aexit__(exc_type, exc_value, exc_traceback)
 
     async def _process_platform_messages(self, ws_url: str) -> None:
         try:
             async with websockets.asyncio.client.connect(ws_url) as websocket:
                 self._platform_events_websocket = websocket
-                self._connected_to_platform_websocket.set_result(True)
+                if self._connected_to_platform_websocket is not None:
+                    self._connected_to_platform_websocket.set_result(True)
 
                 async for message in websocket:
                     try:
@@ -110,7 +126,7 @@ async def _process_platform_messages(self, ws_url: str) -> None:
                             event=parsed_message.name,
                             event_data=parsed_message.data
                             if not isinstance(parsed_message.data, SystemInfoEventData)
-                            else parsed_message.data.to_crawlee_format(self._config.dedicated_cpus or 1),
+                            else parsed_message.data.to_crawlee_format(self._configuration.dedicated_cpus or 1),
                         )
 
                         if parsed_message.name == Event.MIGRATING:
@@ -120,4 +136,5 @@ async def _process_platform_messages(self, ws_url: str) -> None:
                         logger.exception('Cannot parse Actor event', extra={'message': message})
         except Exception:
             logger.exception('Error in websocket connection')
-            self._connected_to_platform_websocket.set_result(False)
+            if self._connected_to_platform_websocket is not None:
+                self._connected_to_platform_websocket.set_result(False)

src/apify/storage_clients/_apify/_request_queue_client.py

@@ -228,20 +228,77 @@ async def add_batch_of_requests(
         Returns:
             Response containing information about the added requests.
         """
-        # Prepare requests for API by converting to dictionaries.
-        requests_dict = [
-            request.model_dump(
-                by_alias=True,
-                exclude={'id'},  # Exclude ID fields from requests since the API doesn't accept them.
+        # Do not try to add previously added requests to avoid pointless expensive calls to API
+
+        new_requests: list[Request] = []
+        already_present_requests: list[ProcessedRequest] = []
+
+        for request in requests:
+            if self._requests_cache.get(request.id):
+                # We are not sure if it was already handled at this point, and it is not worth calling API for it.
+                # It could have been handled by another client in the meantime, so cached information about
+                # `request.was_already_handled` is not reliable.
+                already_present_requests.append(
+                    ProcessedRequest.model_validate(
+                        {
+                            'id': request.id,
+                            'uniqueKey': request.unique_key,
+                            'wasAlreadyPresent': True,
+                            'wasAlreadyHandled': request.was_already_handled,
+                        }
+                    )
+                )
+
+            else:
+                # Add new request to the cache.
+                processed_request = ProcessedRequest.model_validate(
+                    {
+                        'id': request.id,
+                        'uniqueKey': request.unique_key,
+                        'wasAlreadyPresent': True,
+                        'wasAlreadyHandled': request.was_already_handled,
+                    }
+                )
+                self._cache_request(
+                    unique_key_to_request_id(request.unique_key),
+                    processed_request,
+                )
+                new_requests.append(request)
+
+        if new_requests:
+            # Prepare requests for API by converting to dictionaries.
+            requests_dict = [
+                request.model_dump(
+                    by_alias=True,
+                    exclude={'id'},  # Exclude ID fields from requests since the API doesn't accept them.
+                )
+                for request in new_requests
+            ]
+
+            # Send requests to API.
+            api_response = AddRequestsResponse.model_validate(
+                await self._api_client.batch_add_requests(requests=requests_dict, forefront=forefront)
             )
-            for request in requests
-        ]
 
-        # Send requests to API.
-        response = await self._api_client.batch_add_requests(requests=requests_dict, forefront=forefront)
+            # Add the locally known already present processed requests based on the local cache.
+            api_response.processed_requests.extend(already_present_requests)
+
+            # Remove unprocessed requests from the cache
+            for unprocessed_request in api_response.unprocessed_requests:
+                self._requests_cache.pop(unique_key_to_request_id(unprocessed_request.unique_key), None)
+
+        else:
+            api_response = AddRequestsResponse.model_validate(
+                {'unprocessedRequests': [], 'processedRequests': already_present_requests}
+            )
+
+        logger.debug(
+            f'Tried to add new requests: {len(new_requests)}, '
+            f'succeeded to add new requests: {len(api_response.processed_requests) - len(already_present_requests)}, '
+            f'skipped already present requests: {len(already_present_requests)}'
+        )
 
         # Update assumed total count for newly added requests.
-        api_response = AddRequestsResponse.model_validate(response)
         new_request_count = 0
         for processed_request in api_response.processed_requests:
             if not processed_request.was_already_present and not processed_request.was_already_handled: