From 04e83336a5b42319cfce7e7eb6f68576707d5c5a Mon Sep 17 00:00:00 2001 From: Jan Buchar Date: Thu, 29 Oct 2026 17:29:56 +0100 Subject: [PATCH] feat: Implement charge() for multi-tenant Server Actors --- docs/02_concepts/11_pay_per_event.mdx | 9 ++ .../code/11_multi_tenant_charge.py | 33 +++++++ pyproject.toml | 2 +- src/apify/_actor.py | 16 +++- src/apify/_charging.py | 58 +++++++++++- tests/e2e/test_actor_charge.py | 91 +++++++++++++++++++ tests/unit/actor/test_actor_charge.py | 12 +-- tests/unit/actor/test_charging_manager.py | 53 ++++++++++- uv.lock | 8 +- 9 files changed, 263 insertions(+), 19 deletions(-) create mode 100644 docs/02_concepts/code/11_multi_tenant_charge.py diff --git a/docs/02_concepts/11_pay_per_event.mdx b/docs/02_concepts/11_pay_per_event.mdx index 6137c2978..69c5ff3a2 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) 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 000000000..3993c4642 --- /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 45696db0a..9bbc0cc4e 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.4.1b2", "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 7bb01019f..e60dbf14b 100644 --- a/src/apify/_actor.py +++ b/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( diff --git a/src/apify/_charging.py b/src/apify/_charging.py index fc404ffbf..2ec5401b1 100644 --- a/src/apify/_charging.py +++ b/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) diff --git a/tests/e2e/test_actor_charge.py b/tests/e2e/test_actor_charge.py index dc09602a9..daf6fde19 100644 --- a/tests/e2e/test_actor_charge.py +++ b/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') diff --git a/tests/unit/actor/test_actor_charge.py b/tests/unit/actor/test_actor_charge.py index 583a18e0b..6b3da7ef9 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 ae50b041d..203209852 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 @@ -519,12 +520,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( @@ -736,7 +741,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 @@ -778,6 +785,44 @@ 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') + + with pytest.raises(ValueError, match='request_id'): + await cm.charge('search') + charge_mock.assert_not_called() + + 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') diff --git a/uv.lock b/uv.lock index 003c13b02..0968a5060 100644 --- a/uv.lock +++ b/uv.lock @@ -268,7 +268,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.4.1b2" }, { name = "cachetools", specifier = ">=5.5.0" }, { name = "crawlee", specifier = ">=1.9.1,!=1.10.1,<2.0.0" }, { name = "cryptography", specifier = ">=42.0.0" }, @@ -309,7 +309,7 @@ dev = [ [[package]] name = "apify-client" -version = "3.4.0" +version = "3.4.1b2" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "colorama" }, @@ -318,9 +318,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/e8/59/2dec029a069e8c6808ba4bb669c14ea6777101532c91c95bf1c56c176085/apify_client-3.4.1b2.tar.gz", hash = "sha256:5c55b687418815366a5e87025d956305c59efcccb353512bd09407fee15eace0", size = 177467, upload-time = "2026-10-29T13:42:32.938Z" } 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/4b/ee/c0081eba3521ee038cf0519c48a2ca3b5a8898b26b4e163c248eb4e12a61/apify_client-3.4.1b2-py3-none-any.whl", hash = "sha256:20bc48e268eae9f0031239ec60317226361066827c12c3740503f101427b220f", size = 186317, upload-time = "2026-10-29T13:42:31.331Z" }, ] [package.optional-dependencies]