feat: Implement charge() for multi-tenant Server Actors

docs/02_concepts/11_pay_per_event.mdx

@@ -9,6 +9,7 @@ import ConditionalActorChargeSource from '!!raw-loader!roa-loader!./code/11_cond
 import ChargeLimitCheckSource from '!!raw-loader!roa-loader!./code/11_charge_limit_check.py';
 import AdvancedChargingExample from '!!raw-loader!roa-loader!./code/11_advanced_charging.py';
 import ChargePerUnitSource from '!!raw-loader!roa-loader!./code/11_charge_per_unit.py';
+import MultiTenantChargeSource from '!!raw-loader!roa-loader!./code/11_multi_tenant_charge.py';
 import ApiLink from '@theme/ApiLink';
 import RunnableCodeBlock from '@site/src/components/RunnableCodeBlock';
 
@@ -38,6 +39,14 @@ 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.
 
+### Charge in multi-tenant Server Actors
+
+A multi-tenant [Server Actor](https://docs.apify.com/platform/actors/development/programming-interface/standby) run serves requests from many users. To charge the user who made a request rather than the account that started the run, pass the request's `X-Actor-Request-ID` header value as `request_id`. Multi-tenant runs reject charges without it.
+
+<RunnableCodeBlock className="language-python" language="python">
+    {MultiTenantChargeSource}
+</RunnableCodeBlock>
+
 ### 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:

docs/02_concepts/code/11_multi_tenant_charge.py

@@ -0,0 +1,33 @@
+import asyncio
+from typing import Annotated
+
+import uvicorn
+from fastapi import FastAPI, Header
+
+from apify import Actor
+
+app = FastAPI()
+
+
+@app.get('/')
+async def index(
+    x_actor_request_id: Annotated[str | None, Header()] = None,
+) -> str:
+    # The platform identifies the caller of each request with this header.
+    # Pass it along so the charge lands on the caller's account, not yours.
+    await Actor.charge(event_name='api-call', request_id=x_actor_request_id)
+    return 'done'
+
+
+async def main() -> None:
+    async with Actor:
+        config = uvicorn.Config(
+            app,
+            host='0.0.0.0',  # noqa: S104
+            port=Actor.configuration.web_server_port,
+        )
+        await uvicorn.Server(config).serve()
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

pyproject.toml

@@ -36,7 +36,7 @@ keywords = [
     "scraping",
 ]
 dependencies = [
-    "apify-client[brotli]>=3.3.0,<4.0.0",
+    "apify-client[brotli]==3.4.1b2",
     "crawlee>=1.9.1,<2.0.0,!=1.10.1",
     "cachetools>=5.5.0",
     "cryptography>=42.0.0",

src/apify/_actor.py

@@ -878,7 +878,14 @@ def get_charging_manager(self) -> ChargingManager:
         return self._charging_manager_implementation
 
     @_ensure_context
-    async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str | None = None) -> ChargeResult:
+    async def charge(
+        self,
+        event_name: str,
+        *,
+        count: int = 1,
+        idempotency_key: str | None = None,
+        request_id: 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.
@@ -893,10 +900,15 @@ async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str
                 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.
+            request_id: ID of the Server Actor request being charged, taken from the `X-Actor-Request-ID` header of
+                the incoming request. Required in multi-tenant Server Actor runs, where it maps the charge to the
+                caller's account instead of the account that started the run. Ignored elsewhere.
         """
         # charging_manager.charge() acquires charge_lock internally.
         charging_manager = self.get_charging_manager()
-        return await charging_manager.charge(event_name, count=count, idempotency_key=idempotency_key)
+        return await charging_manager.charge(
+            event_name, count=count, idempotency_key=idempotency_key, request_id=request_id
+        )
 
     @overload
     def on(

src/apify/_charging.py

@@ -88,7 +88,14 @@ 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, idempotency_key: str | None = None) -> ChargeResult:
+    async def charge(
+        self,
+        event_name: str,
+        *,
+        count: int = 1,
+        idempotency_key: str | None = None,
+        request_id: 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.
@@ -103,6 +110,9 @@ async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str
                 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.
+            request_id: ID of the Server Actor request being charged, taken from the `X-Actor-Request-ID` header of
+                the incoming request. Required in multi-tenant Server Actor runs, where it maps the charge to the
+                caller's account instead of the account that started the run. Ignored elsewhere.
         """
 
     def calculate_total_charged_amount(self) -> Decimal:
@@ -210,6 +220,9 @@ def __init__(self, configuration: Configuration, client: ApifyClientAsync) -> No
         self._actor_run_id = configuration.actor_run_id
         self._purge_charging_log_dataset = configuration.purge_on_start
         self._pricing_model: PricingModel | None = None
+        self._is_multi_tenant = False
+        """Multi-tenant Server Actor runs carry no pricing snapshot; pricing comes from the Actor and the platform
+        prices and limits each caller separately."""
 
         self._client = client
         self._charging_log_dataset: Dataset | None = None
@@ -302,10 +315,23 @@ async def __aexit__(
         self.active = False
 
     @_ensure_context
-    async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str | None = None) -> ChargeResult:
+    async def charge(
+        self,
+        event_name: str,
+        *,
+        count: int = 1,
+        idempotency_key: str | None = None,
+        request_id: str | None = None,
+    ) -> ChargeResult:
         if idempotency_key is not None and not idempotency_key.strip():
             raise ValueError('idempotency_key must not be blank')
 
+        if self._is_multi_tenant and request_id is None and not event_name.startswith('apify-'):
+            raise ValueError(
+                f"Charging for '{event_name}' in a multi-tenant Server Actor run requires `request_id` - pass the "
+                "value of the incoming request's X-Actor-Request-ID header to Actor.charge()."
+            )
+
         # 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:
@@ -379,6 +405,7 @@ async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str
                         event_name,
                         count=charged_count,
                         idempotency_key=idempotency_key,
+                        request_id=request_id,
                     )
                     logger.debug(f"Charged {charged_count} occurrence(s) of event '{event_name}'.")
                 elif event_name in self._tier_priced_events:
@@ -411,6 +438,7 @@ async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str
                         'event_price_usd': float(round(pricing_info.price, 3)),
                         'charged_count': charged_count,
                         'idempotency_key': idempotency_key,
+                        'request_id': request_id,
                         'timestamp': datetime.now(UTC).isoformat(),
                     }
                 )
@@ -533,6 +561,9 @@ async def _fetch_pricing_info(self) -> _FetchedPricingInfoDict:
             if run is None:
                 raise RuntimeError('Actor run not found')
 
+            if run.standby is not None and run.standby.deployment.startswith('MULTI_TENANT'):
+                return await self._fetch_multi_tenant_pricing_info(run.charged_event_counts or {})
+
             max_charge = run.options.max_total_charge_usd
             return _FetchedPricingInfoDict(
                 pricing_info=run.pricing_info,
@@ -551,6 +582,29 @@ async def _fetch_pricing_info(self) -> _FetchedPricingInfoDict:
             ),
         )
 
+    async def _fetch_multi_tenant_pricing_info(self, charged_event_counts: dict[str, int]) -> _FetchedPricingInfoDict:
+        actor_id = self._configuration.actor_id
+        if actor_id is None:
+            raise RuntimeError('Actor ID not found even though the Actor is running on Apify')
+
+        actor = await self._client.actor(actor_id).get()
+        if actor is None:
+            raise RuntimeError('Actor not found')
+
+        self._is_multi_tenant = True
+
+        # The history includes scheduled future changes, so the effective entry is the latest one that has started.
+        # Resolved once on purpose - the platform prices each charge from its own current pricing anyway.
+        now = datetime.now(UTC)
+        started = [info for info in actor.pricing_infos or [] if info.started_at <= now]
+        pricing_info = max(started, key=lambda info: info.started_at, default=None)
+
+        return _FetchedPricingInfoDict(
+            pricing_info=pricing_info,
+            charged_event_counts=charged_event_counts,
+            max_total_charge_usd=Decimal('inf'),
+        )
+
     def _compute_charged_count(self, event_name: str, count: int) -> int:
         """Compute how many of `count` events to charge, given the remaining budget."""
         max_chargeable = self.calculate_max_event_charge_count_within_limit(event_name)

tests/e2e/test_actor_charge.py

@@ -1,9 +1,12 @@
 from __future__ import annotations
 
+import json
+import os
 from decimal import Decimal
 from functools import partial
 from typing import TYPE_CHECKING
 
+import impit
 import pytest_asyncio
 
 from .._utils import poll_until_condition
@@ -226,3 +229,91 @@ async def test_actor_push_data_combined_budget_limit(
 
     assert run.status == 'SUCCEEDED'
     assert run.charged_event_counts == expected_counts
+
+
+async def test_actor_charge_in_multi_tenant_standby_run(
+    make_actor: MakeActorFunction,
+    apify_client_async: ApifyClientAsync,
+    apify_token: str,
+) -> None:
+    """Test that a multi-tenant Server Actor run charges with the request ID and rejects charges without it."""
+
+    async def main() -> None:
+        import json
+
+        import uvicorn
+
+        async def app(scope, receive, send):  # noqa: ANN001, ANN202, ARG001
+            headers = dict(scope['headers'])
+            if b'x-apify-container-server-readiness-probe' in headers:
+                body = b'ready'
+            else:
+                request_id = headers.get(b'x-actor-request-id', b'').decode() or None
+                if scope['query_string'] == b'omit_request_id':
+                    request_id = None
+                try:
+                    result = await Actor.charge(event_name='api-call', request_id=request_id)
+                    body = json.dumps({'charged_count': result.charged_count, 'request_id': request_id}).encode()
+                except ValueError as exc:
+                    body = json.dumps({'error': str(exc)}).encode()
+            await send({'type': 'http.response.start', 'status': 200, 'headers': []})
+            await send({'type': 'http.response.body', 'body': body})
+
+        async with Actor:
+            config = uvicorn.Config(app, host='0.0.0.0', port=Actor.configuration.web_server_port)  # noqa: S104
+            await uvicorn.Server(config).serve()
+
+    actor_client = await make_actor('ppe-multi-tenant', main_func=main)
+    await actor_client.update(
+        pricing_infos=[
+            {
+                'pricingModel': 'PAY_PER_EVENT',
+                'apifyMarginPercentage': 0.0,
+                'createdAt': '2024-01-01T00:00:00.000Z',
+                'startedAt': '2024-01-01T00:00:00.000Z',
+                'pricingPerEvent': {
+                    'actorChargeEvents': {
+                        'api-call': {
+                            'eventTitle': 'API call',
+                            'eventPriceUsd': 0.01,
+                            'eventDescription': 'One API call',
+                        },
+                    },
+                },
+            },
+        ],
+    )
+
+    actor = await actor_client.get()
+    assert actor is not None
+    auth = {'Authorization': f'Bearer {apify_token}'}
+
+    async with impit.AsyncClient(timeout=120) as http:
+        # `ActorClientAsync.update()` has no tenancy argument, so the field is set through the API directly.
+        api_url = os.getenv('APIFY_INTEGRATION_TESTS_API_URL', 'https://api.apify.com')
+        response = await http.put(
+            f'{api_url}/v2/acts/{actor.id}',
+            headers={**auth, 'Content-Type': 'application/json'},
+            content=json.dumps({'actorStandby': {'isEnabled': True, 'tenancy': 'MULTI_TENANT'}}),
+        )
+        assert response.status_code == 200, response.text
+
+        actor = await actor_client.get()
+        assert actor is not None
+        assert actor.standby_url is not None
+
+        try:
+            with_id = (await http.get(actor.standby_url, headers=auth)).json()
+            without_id = (await http.get(f'{actor.standby_url}?omit_request_id', headers=auth)).json()
+        finally:
+            runs = await actor_client.runs().list()
+            for run_item in runs.items:
+                await apify_client_async.run(run_item.id).abort()
+
+    assert with_id['charged_count'] == 1
+    assert with_id['request_id']
+    assert 'request_id' in without_id['error']
+
+    run = await _get_run(apify_client_async, runs.items[0].id)
+    assert run.standby is not None
+    assert run.standby.deployment.startswith('MULTI_TENANT')

tests/unit/actor/test_actor_charge.py

@@ -39,7 +39,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, idempotency_key=None)
+            setup.mock_charge.assert_called_once_with('event', count=1, idempotency_key=None, request_id=None)
     """
     # Mock the ApifyClientAsync
     mock_client = Mock()
@@ -81,14 +81,14 @@ async def test_actor_push_data_overcharges_one_item_when_budget_cannot_cover_one
         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, idempotency_key=None)
+        setup.mock_charge.assert_called_once_with('some-event', count=1, idempotency_key=None, request_id=None)
         setup.mock_charge.reset_mock()
 
         assert result1.charged_count == 1
 
         result = await Actor.push_data([{'hello': 'world'} for _ in range(10)], charged_event_name='another-event')
 
-        setup.mock_charge.assert_called_once_with('another-event', count=1, idempotency_key=None)
+        setup.mock_charge.assert_called_once_with('another-event', count=1, idempotency_key=None, request_id=None)
         assert result.charged_count == 1
         assert result.event_charge_limit_reached is True
 
@@ -109,7 +109,7 @@ 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, idempotency_key=None)
+        setup.mock_charge.assert_called_once_with('test-event', count=1, idempotency_key=None, request_id=None)
         assert result2.charged_count == 1
 
 
@@ -119,7 +119,7 @@ async def test_actor_charge_forwards_idempotency_key() -> None:
         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')
+        setup.mock_charge.assert_called_once_with('test-event', count=1, idempotency_key='key-1', request_id=None)
         assert result1.charged_count == 1
 
         setup.mock_charge.reset_mock()
@@ -313,7 +313,7 @@ async def test_charge_with_overdrawn_budget() -> None:
         charge_result = await Actor.charge('event', count=1)
         assert charge_result.charged_count == 1
         assert charge_result.event_charge_limit_reached is True
-        setup.mock_charge.assert_called_once_with('event', count=1, idempotency_key=None)
+        setup.mock_charge.assert_called_once_with('event', count=1, idempotency_key=None, request_id=None)
         setup.mock_charge.reset_mock()
 
         # The run is now strictly over budget, so nothing else is charged.