feat(charging): Support the idempotency_key param in Actor.charge()

docs/02_concepts/11_pay_per_event.mdx

@@ -38,6 +38,16 @@ If you can split your work into individual units, for example scraping one page
 
 If you use the `count` parameter, always check the returned `charged_count`. It tells you how many events were charged, which may be less than what you requested.
 
+### Retry charges safely
+
+When an operation may be retried, for example after a network error, pass an `idempotency_key` to `Actor.charge()`. A repeated call under the same key isn't charged again. It reports the `charged_count` of the original call:
+
+```python
+await Actor.charge(event_name='search-result', idempotency_key=result_id)
+```
+
+Like the Apify platform, the SDK remembers a key for three minutes, so a later repeat is charged again. A repeat that names a different event is skipped with a warning and reports zero.
+
 ### Monitor charging
 
 For both custom and synthetic events, every `Actor.charge` call returns a <ApiLink to="class/ChargeResult">`ChargeResult`</ApiLink>. Inspect its fields to learn how much was charged.

src/apify/_actor.py

@@ -788,18 +788,22 @@ def get_charging_manager(self) -> ChargingManager:
         return self._charging_manager_implementation
 
     @_ensure_context
-    async def charge(self, event_name: str, *, count: int = 1) -> ChargeResult:
+    async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str | None = None) -> ChargeResult:
         """Charge for a specified number of events - sub-operations of the Actor.
 
         This is relevant only for the pay-per-event pricing model.
 
         Args:
             event_name: Name of the event to be charged for.
             count: Number of events to charge for.
+            idempotency_key: A unique key preventing a retried operation from being charged for twice. A repeat
+                under the same key within three minutes is not sent to the API and reports the `charged_count` of
+                the original call, or zero if it names a different event. Like the platform, the key is forgotten
+                afterwards.
         """
         # charging_manager.charge() acquires charge_lock internally.
         charging_manager = self.get_charging_manager()
-        return await charging_manager.charge(event_name, count=count)
+        return await charging_manager.charge(event_name, count=count, idempotency_key=idempotency_key)
 
     @overload
     def on(

src/apify/_charging.py

@@ -9,6 +9,7 @@
 from logging import getLogger
 from typing import TYPE_CHECKING, Literal, Protocol, TypedDict
 
+from cachetools import TTLCache
 from pydantic import ConfigDict
 from pydantic.alias_generators import to_camel
 
@@ -210,14 +211,18 @@ class ChargingManager(Protocol):
     charge_lock: ReentrantLock
     """Lock to synchronize charge operations. Prevents race conditions between `charge` and `push_data` calls."""
 
-    async def charge(self, event_name: str, *, count: int = 1) -> ChargeResult:
+    async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str | None = None) -> ChargeResult:
         """Charge for a specified number of events - sub-operations of the Actor.
 
         This is relevant only for the pay-per-event pricing model.
 
         Args:
             event_name: Name of the event to be charged for.
             count: Number of events to charge for.
+            idempotency_key: A unique key preventing a retried operation from being charged for twice. A repeat
+                under the same key within three minutes is not sent to the API and reports the `charged_count` of
+                the original call, or zero if it names a different event. Like the platform, the key is forgotten
+                afterwards.
         """
 
     def calculate_total_charged_amount(self) -> Decimal:
@@ -329,6 +334,9 @@ def __init__(self, configuration: Configuration, client: ApifyClientAsync) -> No
         self._charging_state: dict[str, ChargingStateItem] = {}
         self._pricing_info: dict[str, PricingInfoItem] = {}
         self._tier_priced_events: set[str] = set()
+        self._idempotent_charges: TTLCache[str, IdempotentChargeItem] = TTLCache(maxsize=float('inf'), ttl=180)
+        """Charges made with an idempotency key, by key. Keys expire after three minutes, the same as the platform's
+        own idempotency records."""
 
         self._not_ppe_warning_printed = False
         self.active = False
@@ -412,7 +420,10 @@ async def __aexit__(
         self.active = False
 
     @_ensure_context
-    async def charge(self, event_name: str, *, count: int = 1) -> ChargeResult:
+    async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str | None = None) -> ChargeResult:
+        if idempotency_key is not None and not idempotency_key.strip():
+            raise ValueError('idempotency_key must not be blank')
+
         # For runs that do not use the pay-per-event pricing model, just print a warning and return
         if self._pricing_model != 'PAY_PER_EVENT':
             if not self._not_ppe_warning_printed:
@@ -435,6 +446,25 @@ async def charge(self, event_name: str, *, count: int = 1) -> ChargeResult:
             )
 
         async with self.charge_lock():
+            # A repeat the platform would discard is resolved here too - counting it would inflate the charging state
+            # and make the run hit `max_total_charge_usd` early. The key alone decides, as on the platform.
+            if idempotency_key is not None and (previous := self._idempotent_charges.get(idempotency_key)) is not None:
+                if previous.event_name == event_name:
+                    logger.debug(f"Skipped a repeated charge of event '{event_name}' under key '{idempotency_key}'.")
+                    repeated_count = previous.charged_count
+                else:
+                    logger.warning(
+                        f"Idempotency key '{idempotency_key}' was already used to charge for event "
+                        f"'{previous.event_name}', so the charge of event '{event_name}' is skipped."
+                    )
+                    repeated_count = 0
+
+                return ChargeResult(
+                    event_charge_limit_reached=self.is_event_charge_limit_reached(event_name),
+                    charged_count=repeated_count,
+                    chargeable_within_limit=self.compute_chargeable(),
+                )
+
             # Determine the maximum amount of events that can be charged within the budget
             max_chargeable = self.calculate_max_event_charge_count_within_limit(event_name)
             charged_count = min(count, max_chargeable if max_chargeable is not None else count)
@@ -455,11 +485,6 @@ async def charge(self, event_name: str, *, count: int = 1) -> ChargeResult:
                 ),
             )
 
-            # Update the charging state
-            self._charging_state.setdefault(event_name, ChargingStateItem(0, Decimal()))
-            self._charging_state[event_name].charge_count += charged_count
-            self._charging_state[event_name].total_charged_amount += charged_count * pricing_info.price
-
             # If running on the platform, call the charge endpoint
             if self._is_at_home:
                 if self._actor_run_id is None:
@@ -470,7 +495,11 @@ async def charge(self, event_name: str, *, count: int = 1) -> ChargeResult:
                     # the platform handles them automatically based on dataset writes.
                     pass
                 elif event_name in self._pricing_info:
-                    await self._client.run(self._actor_run_id).charge(event_name, count=charged_count)
+                    await self._client.run(self._actor_run_id).charge(
+                        event_name,
+                        count=charged_count,
+                        idempotency_key=idempotency_key,
+                    )
                     logger.debug(f"Charged {charged_count} occurrence(s) of event '{event_name}'.")
                 elif event_name in self._tier_priced_events:
                     logger.warning(
@@ -479,6 +508,20 @@ async def charge(self, event_name: str, *, count: int = 1) -> ChargeResult:
                 else:
                     logger.warning(f"Attempting to charge for an unknown event '{event_name}'")
 
+            # Count the charge only after the API call succeeds, so a failed call retried under the same key is
+            # counted once.
+            self._charging_state.setdefault(event_name, ChargingStateItem(0, Decimal()))
+            self._charging_state[event_name].charge_count += charged_count
+            self._charging_state[event_name].total_charged_amount += charged_count * pricing_info.price
+
+            # Remember the key for every charge that was counted, including events the API never receives, such as
+            # synthetic and tier-priced ones - those are counted locally and a repeat would count them twice.
+            if idempotency_key is not None:
+                self._idempotent_charges[idempotency_key] = IdempotentChargeItem(
+                    event_name=event_name,
+                    charged_count=charged_count,
+                )
+
             # Log the charged operation (if enabled)
             if self._charging_log_dataset:
                 await self._charging_log_dataset.push_data(
@@ -487,6 +530,7 @@ async def charge(self, event_name: str, *, count: int = 1) -> ChargeResult:
                         'event_title': pricing_info.title,
                         'event_price_usd': float(round(pricing_info.price, 3)),
                         'charged_count': charged_count,
+                        'idempotency_key': idempotency_key,
                         'timestamp': datetime.now(UTC).isoformat(),
                     }
                 )
@@ -636,6 +680,12 @@ class PricingInfoItem:
     title: str
 
 
+@dataclass(frozen=True)
+class IdempotentChargeItem:
+    event_name: str
+    charged_count: int
+
+
 class _FetchedPricingInfoDict(TypedDict):
     pricing_info: ActorPricingInfoModel | None
     charged_event_counts: dict[str, int]

tests/unit/actor/test_actor_charge.py

@@ -37,7 +37,7 @@ async def setup_mocked_charging(
             setup.charging_mgr._pricing_info['event'] = PricingInfoItem(Decimal('1.0'), 'Event')
 
             result = await Actor.charge('event', count=1)
-            setup.mock_charge.assert_called_once_with('event', count=1)
+            setup.mock_charge.assert_called_once_with('event', count=1, idempotency_key=None)
     """
     # Mock the ApifyClientAsync
     mock_client = Mock()
@@ -82,7 +82,7 @@ async def test_actor_charge_push_data_with_no_remaining_budget() -> None:
         result1 = await Actor.charge('some-event', count=1)  # Costs $1, leaving $0.5
 
         # Verify the first charge call was made correctly
-        setup.mock_charge.assert_called_once_with('some-event', count=1)
+        setup.mock_charge.assert_called_once_with('some-event', count=1, idempotency_key=None)
         setup.mock_charge.reset_mock()
 
         assert result1.charged_count == 1
@@ -117,10 +117,27 @@ async def test_actor_charge_api_call_verification() -> None:
 
         # Call charge with count=1 - this SHOULD call the API
         result2 = await Actor.charge('test-event', count=1)
-        setup.mock_charge.assert_called_once_with('test-event', count=1)
+        setup.mock_charge.assert_called_once_with('test-event', count=1, idempotency_key=None)
         assert result2.charged_count == 1
 
 
+async def test_actor_charge_forwards_idempotency_key() -> None:
+    """Verify that Actor.charge passes the idempotency key down to the API and deduplicates repeats."""
+    async with setup_mocked_charging(
+        Configuration(max_total_charge_usd=Decimal('10.0'), test_pay_per_event=True), {'test-event': Decimal('1.0')}
+    ) as setup:
+        result1 = await Actor.charge('test-event', count=1, idempotency_key='key-1')
+        setup.mock_charge.assert_called_once_with('test-event', count=1, idempotency_key='key-1')
+        assert result1.charged_count == 1
+
+        setup.mock_charge.reset_mock()
+
+        result2 = await Actor.charge('test-event', count=1, idempotency_key='key-1')
+        setup.mock_charge.assert_not_called()
+        assert result2.charged_count == 1
+        assert setup.charging_mgr.get_charged_event_count('test-event') == 1
+
+
 async def test_max_event_charge_count_within_limit_tolerates_overdraw() -> None:
     """Test that calculate_max_event_charge_count_within_limit does not return nonsensical (e.g., negative) values when
     the total number of charged events overdraws the max_total_charge_usd limit."""