diff --git a/docs/02_concepts/11_pay_per_event.mdx b/docs/02_concepts/11_pay_per_event.mdx index 6137c297..408c935e 100644 --- a/docs/02_concepts/11_pay_per_event.mdx +++ b/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) (formerly Actor 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. + + + {MultiTenantChargeSource} + + ### 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: diff --git a/docs/02_concepts/code/11_multi_tenant_charge.py b/docs/02_concepts/code/11_multi_tenant_charge.py new file mode 100644 index 00000000..3993c464 --- /dev/null +++ b/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()) diff --git a/pyproject.toml b/pyproject.toml index 23f44f82..c4331ab9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,7 +36,7 @@ keywords = [ "scraping", ] dependencies = [ - "apify-client[brotli]>=3.3.0,<4.0.0", + "apify-client[brotli]>=3.5.0,<4.0.0", "crawlee>=1.9.1,<2.0.0,!=1.10.1", "cachetools>=5.5.0", "cryptography>=42.0.0", diff --git a/src/apify/_actor.py b/src/apify/_actor.py index d3db9395..dd2caf5c 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -913,7 +913,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. @@ -928,10 +935,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( diff --git a/src/apify/_charging.py b/src/apify/_charging.py index 1e6ca472..751e5340 100644 --- a/src/apify/_charging.py +++ b/src/apify/_charging.py @@ -94,7 +94,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. @@ -109,6 +116,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: @@ -216,6 +226,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 @@ -324,7 +337,14 @@ 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') @@ -342,6 +362,12 @@ async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str chargeable_within_limit=self.compute_chargeable(), ) + if self._is_multi_tenant and not request_id 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()." + ) + if count <= 0: return ChargeResult( event_charge_limit_reached=self.is_event_charge_limit_reached(event_name), @@ -404,6 +430,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("Charged %d occurrence(s) of event '%s'.", charged_count, event_name) elif event_name in self._tier_priced_events: @@ -434,6 +461,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(), } ) @@ -558,6 +586,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, @@ -576,6 +607,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) diff --git a/tests/e2e/test_actor_charge.py b/tests/e2e/test_actor_charge.py index 47bd86f4..3cab8d3c 100644 --- a/tests/e2e/test_actor_charge.py +++ b/tests/e2e/test_actor_charge.py @@ -6,6 +6,8 @@ import pytest_asyncio +from crawlee._utils.crypto import crypto_random_object_id + from .._utils import poll_until_condition from apify import Actor @@ -228,6 +230,81 @@ async def test_actor_push_data_combined_budget_limit( assert run.charged_event_counts == expected_counts +async def test_actor_charge_in_multi_tenant_standby_run( + make_actor: MakeActorFunction, + apify_client_async: ApifyClientAsync, +) -> 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 b'omit_request_id' in scope['query_string']: + 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( + # The generated name is too long for the `--` hostname label of the server URL, whose DNS + # limit is 63 characters. + name=f'sdk-mt-{crypto_random_object_id(8).lower()}', + 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', + }, + }, + }, + }, + ], + server_is_enabled=True, + server_tenancy='MULTI_TENANT', + ) + + server = actor_client.server() + try: + with_id = (await server.request()).json() + without_id = (await server.request(params={'omit_request_id': True})).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') + + @pytest_asyncio.fixture(scope='module', loop_scope='module') async def ppe_reboot_actor_build(make_actor: MakeActorFunction) -> str: async def main() -> None: diff --git a/tests/unit/actor/test_actor_charge.py b/tests/unit/actor/test_actor_charge.py index 583a18e0..6b3da7ef 100644 --- a/tests/unit/actor/test_actor_charge.py +++ b/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. diff --git a/tests/unit/actor/test_charging_manager.py b/tests/unit/actor/test_charging_manager.py index 4acf866f..2e9fcf73 100644 --- a/tests/unit/actor/test_charging_manager.py +++ b/tests/unit/actor/test_charging_manager.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +from datetime import datetime from decimal import Decimal from typing import Any from unittest.mock import AsyncMock, MagicMock @@ -522,12 +523,16 @@ async def test_charge_forwards_idempotency_key_to_client(mock_client: MagicMock) cm = ChargingManagerImplementation(config, mock_client) async with cm: await cm.charge('search', count=2, idempotency_key='key-1') - mock_client.run.return_value.charge.assert_awaited_once_with('search', count=2, idempotency_key='key-1') + mock_client.run.return_value.charge.assert_awaited_once_with( + 'search', count=2, idempotency_key='key-1', request_id=None + ) # Without a key the client gets None and generates a unique one per call, so nothing is deduplicated. mock_client.run.return_value.charge.reset_mock() await cm.charge('search', count=1) - mock_client.run.return_value.charge.assert_awaited_once_with('search', count=1, idempotency_key=None) + mock_client.run.return_value.charge.assert_awaited_once_with( + 'search', count=1, idempotency_key=None, request_id=None + ) async def test_charge_deduplicates_repeated_idempotency_key( @@ -739,7 +744,9 @@ async def test_charge_registers_the_count_limited_by_the_budget(mock_client: Mag cm = ChargingManagerImplementation(config, mock_client) async with cm: assert (await cm.charge('search', count=5, idempotency_key='key-1')).charged_count == 3 - mock_client.run.return_value.charge.assert_awaited_once_with('search', count=3, idempotency_key='key-1') + mock_client.run.return_value.charge.assert_awaited_once_with( + 'search', count=3, idempotency_key='key-1', request_id=None + ) assert (await cm.charge('search', count=5, idempotency_key='key-1')).charged_count == 3 assert cm.get_charged_event_count('search') == 3 @@ -781,11 +788,68 @@ async def test_charge_overcharges_by_one_event_when_budget_exceeded( assert cm.get_charged_event_count('search') == 4 if is_at_home: - mock_client.run.return_value.charge.assert_awaited_once_with('search', count=2, idempotency_key=None) + mock_client.run.return_value.charge.assert_awaited_once_with( + 'search', count=2, idempotency_key=None, request_id=None + ) else: mock_client.run.return_value.charge.assert_not_called() +async def test_multi_tenant_run_uses_effective_actor_pricing_and_requires_request_id(mock_client: MagicMock) -> None: + """Test that a multi-tenant Server Actor run prices from the Actor's effective pricing and requires `request_id`.""" + + def ppe(price: str, started_at: str) -> PayPerEventActorPricingInfo: + info = _make_ppe_pricing_info({'search': Decimal(price)}) + return info.model_copy(update={'started_at': datetime.fromisoformat(started_at)}) + + # The last entry is a scheduled future change, so the effective one is the latest that has already started. + actor = MagicMock( + pricing_infos=[ + ppe('1', '2020-01-01T00:00:00+00:00'), + ppe('2', '2021-01-01T00:00:00+00:00'), + ppe('3', '2999-01-01T00:00:00+00:00'), + ] + ) + run = MagicMock(standby=MagicMock(deployment='MULTI_TENANT_PAYING_USERS'), charged_event_counts={}) + charge_mock = AsyncMock() + mock_client.run = MagicMock(return_value=MagicMock(get=AsyncMock(return_value=run), charge=charge_mock)) + mock_client.actor = MagicMock(return_value=MagicMock(get=AsyncMock(return_value=actor))) + config = _make_config(is_at_home=True, actor_run_id='run-id', actor_id='actor-id') + cm = ChargingManagerImplementation(config, mock_client) + + async with cm: + pricing = cm.get_pricing_info() + assert pricing.per_event_prices == {'search': Decimal(2)} + assert pricing.max_total_charge_usd == Decimal('inf') + + for request_id in (None, ''): + with pytest.raises(ValueError, match='request_id'): + await cm.charge('search', request_id=request_id) + charge_mock.assert_not_called() + + # Synthetic `apify-*` events are exempt from the `request_id` requirement. + await cm.charge('apify-actor-start') + + result = await cm.charge('search', request_id='some-request-id') + assert result.charged_count == 1 + charge_mock.assert_awaited_once_with('search', count=1, idempotency_key=None, request_id='some-request-id') + + +async def test_multi_tenant_run_without_ppe_ignores_charge_without_request_id(mock_client: MagicMock) -> None: + """Test that a multi-tenant Server Actor run without pay-per-event pricing ignores a charge with no `request_id`.""" + run = MagicMock(standby=MagicMock(deployment='MULTI_TENANT_PAYING_USERS'), charged_event_counts={}) + charge_mock = AsyncMock() + mock_client.run = MagicMock(return_value=MagicMock(get=AsyncMock(return_value=run), charge=charge_mock)) + mock_client.actor = MagicMock(return_value=MagicMock(get=AsyncMock(return_value=MagicMock(pricing_infos=[])))) + config = _make_config(is_at_home=True, actor_run_id='run-id', actor_id='actor-id') + + cm = ChargingManagerImplementation(config, mock_client) + + async with cm: + assert (await cm.charge('search')).charged_count == 0 + charge_mock.assert_not_called() + + def _make_platform_ppe_config(**kwargs: Any) -> Configuration: """Helper to create a platform PPE Configuration with charged event counts in the env var.""" defaults: dict[str, Any] = { diff --git a/uv.lock b/uv.lock index 83eee90e..4b89e429 100644 --- a/uv.lock +++ b/uv.lock @@ -289,7 +289,7 @@ dev = [ [package.metadata] requires-dist = [ - { name = "apify-client", extras = ["brotli"], specifier = ">=3.3.0,<4.0.0" }, + { name = "apify-client", extras = ["brotli"], specifier = ">=3.5.0,<4.0.0" }, { name = "cachetools", specifier = ">=5.5.0" }, { name = "crawlee", specifier = ">=1.9.1,!=1.10.1,<2.0.0" }, { name = "cryptography", specifier = ">=42.0.0" }, @@ -329,7 +329,7 @@ dev = [ [[package]] name = "apify-client" -version = "3.4.0" +version = "3.5.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "colorama" }, @@ -338,9 +338,9 @@ dependencies = [ { name = "pydantic", extra = ["email"] }, { name = "typing-extensions" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/b7/18/914ce7f110c2ee51473753d24a5e845047f9d56db1861a48a2e113dcaf65/apify_client-3.4.0.tar.gz", hash = "sha256:bd4dac482afd02801eb9a90ab570d82ee86d7ee8d479f72723ca1aedc5411f27", size = 169880, upload-time = "2026-10-27T08:41:15.355Z" } +sdist = { url = "https://files.pythonhosted.org/packages/f3/3d/067522ecb5ba120dfc4340f99e2b9ace3ab22b313222dc939b030252907d/apify_client-3.5.0.tar.gz", hash = "sha256:82a223103b0992a6a9cfd5c4dd52abb5624cc36af692394aa7c50806242ba4cd", size = 188482, upload-time = "2026-11-09T12:01:13.271Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/0d/49/0f585a3404989627fa6fb19f6203e7af8f98bab3d25b01288588abd034b9/apify_client-3.4.0-py3-none-any.whl", hash = "sha256:a80dee5b28da98732e1ec68663e0a6c545d67481cd3eb91fe91c56391cfb6c45", size = 178712, upload-time = "2026-10-27T08:41:13.604Z" }, + { url = "https://files.pythonhosted.org/packages/20/37/0ba435b32b03d056f991d8a977383e34254b755f5a04752f3de7f8699ace/apify_client-3.5.0-py3-none-any.whl", hash = "sha256:b0d1ad9464c04d3f583eb1d9dfcff54f4bba62df13b0c415fca50ae98977439c", size = 197084, upload-time = "2026-11-09T12:01:11.401Z" }, ] [package.optional-dependencies]