feat: map apify-client errors to domain-level errors at API call sites

Author: vdusek

src/apify/_actor.py

@@ -32,6 +32,7 @@
 from apify._proxy_configuration import ProxyConfiguration
 from apify._utils import docs_group, docs_name, ensure_context, get_system_info, is_running_in_ipython
 from apify._webhook import to_client_representations
+from apify.errors import map_client_errors
 from apify.events import ApifyEventManager, EventManager, LocalEventManager
 from apify.log import _configure_logging, logger
 from apify.storage_clients import ApifyStorageClient, SmartApifyStorageClient
@@ -936,17 +937,18 @@ async def start(
             raise ValueError(f'Invalid timeout {timeout!r}: expected `None`, `"inherit"`, or a `timedelta`.')
 
         actor_client = client.actor(actor_id)
-        return await actor_client.start(
-            run_input=run_input,
-            content_type=content_type,
-            build=build,
-            max_total_charge_usd=max_total_charge_usd,
-            restart_on_error=restart_on_error,
-            memory_mbytes=memory_mbytes,
-            run_timeout=actor_start_timeout,
-            force_permission_level=force_permission_level,
-            webhooks=to_client_representations(webhooks),
-        )
+        with map_client_errors():
+            return await actor_client.start(
+                run_input=run_input,
+                content_type=content_type,
+                build=build,
+                max_total_charge_usd=max_total_charge_usd,
+                restart_on_error=restart_on_error,
+                memory_mbytes=memory_mbytes,
+                run_timeout=actor_start_timeout,
+                force_permission_level=force_permission_level,
+                webhooks=to_client_representations(webhooks),
+            )
 
     @_ensure_context
     async def abort(
@@ -975,10 +977,11 @@ async def abort(
         client = self.new_client(token=token) if token else self.apify_client
         run_client = client.run(run_id)
 
-        if status_message:
-            await run_client.update(status_message=status_message)
+        with map_client_errors():
+            if status_message:
+                await run_client.update(status_message=status_message)
 
-        run = await run_client.abort(gracefully=gracefully)
+            run = await run_client.abort(gracefully=gracefully)
 
         if run is None:
             raise RuntimeError(f'Failed to abort Actor run with ID "{run_id}".')
@@ -1047,19 +1050,20 @@ async def call(
             raise ValueError(f'Invalid timeout {timeout!r}: expected `None`, `"inherit"`, or a `timedelta`.')
 
         actor_client = client.actor(actor_id)
-        run = await actor_client.call(
-            run_input=run_input,
-            content_type=content_type,
-            build=build,
-            max_total_charge_usd=max_total_charge_usd,
-            restart_on_error=restart_on_error,
-            memory_mbytes=memory_mbytes,
-            run_timeout=actor_call_timeout,
-            force_permission_level=force_permission_level,
-            webhooks=to_client_representations(webhooks),
-            wait_duration=wait,
-            logger=logger,
-        )
+        with map_client_errors():
+            run = await actor_client.call(
+                run_input=run_input,
+                content_type=content_type,
+                build=build,
+                max_total_charge_usd=max_total_charge_usd,
+                restart_on_error=restart_on_error,
+                memory_mbytes=memory_mbytes,
+                run_timeout=actor_call_timeout,
+                force_permission_level=force_permission_level,
+                webhooks=to_client_representations(webhooks),
+                wait_duration=wait,
+                logger=logger,
+            )
 
         if run is None:
             raise RuntimeError(f'Failed to call Actor with ID "{actor_id}".')
@@ -1120,15 +1124,16 @@ async def call_task(
             raise ValueError(f'Invalid timeout {timeout!r}: expected `None`, `"inherit"`, or a `timedelta`.')
 
         task_client = client.task(task_id)
-        run = await task_client.call(
-            task_input=task_input,
-            build=build,
-            restart_on_error=restart_on_error,
-            memory_mbytes=memory_mbytes,
-            run_timeout=task_call_timeout,
-            webhooks=to_client_representations(webhooks),
-            wait_duration=wait,
-        )
+        with map_client_errors():
+            run = await task_client.call(
+                task_input=task_input,
+                build=build,
+                restart_on_error=restart_on_error,
+                memory_mbytes=memory_mbytes,
+                run_timeout=task_call_timeout,
+                webhooks=to_client_representations(webhooks),
+                wait_duration=wait,
+            )
 
         if run is None:
             raise RuntimeError(f'Failed to call Task with ID "{task_id}".')
@@ -1171,12 +1176,13 @@ async def metamorph(
         if not self.configuration.actor_run_id:
             raise RuntimeError('actor_run_id cannot be None when running on the Apify platform.')
 
-        await self.apify_client.run(self.configuration.actor_run_id).metamorph(
-            target_actor_id=target_actor_id,
-            run_input=run_input,
-            target_actor_build=target_actor_build,
-            content_type=content_type,
-        )
+        with map_client_errors():
+            await self.apify_client.run(self.configuration.actor_run_id).metamorph(
+                target_actor_id=target_actor_id,
+                run_input=run_input,
+                target_actor_build=target_actor_build,
+                content_type=content_type,
+            )
 
         if custom_after_sleep:
             await asyncio.sleep(custom_after_sleep.total_seconds())
@@ -1242,7 +1248,8 @@ async def safe_dispatch(listener: Any, data: Any) -> None:
             except TimeoutError:
                 self.log.warning('Pre-reboot event listeners did not finish within timeout; proceeding with reboot')
 
-            await self.apify_client.run(self.configuration.actor_run_id).reboot()
+            with map_client_errors():
+                await self.apify_client.run(self.configuration.actor_run_id).reboot()
         except BaseException:
             # Reset the flag so that a failed or cancelled reboot can be retried.
             self._is_rebooting = False
@@ -1283,17 +1290,18 @@ async def add_webhook(self, webhook: Webhook, *, idempotency_key: str | None = N
         if not self.configuration.actor_run_id:
             raise RuntimeError('actor_run_id cannot be None when running on the Apify platform.')
 
-        await self.apify_client.webhooks().create(
-            actor_run_id=self.configuration.actor_run_id,
-            event_types=webhook.event_types,
-            request_url=webhook.request_url,
-            payload_template=webhook.payload_template,
-            headers_template=webhook.headers_template,
-            ignore_ssl_errors=webhook.ignore_ssl_errors,
-            do_not_retry=webhook.do_not_retry,
-            idempotency_key=idempotency_key if idempotency_key is not None else webhook.idempotency_key,
-            is_ad_hoc=True,
-        )
+        with map_client_errors():
+            await self.apify_client.webhooks().create(
+                actor_run_id=self.configuration.actor_run_id,
+                event_types=webhook.event_types,
+                request_url=webhook.request_url,
+                payload_template=webhook.payload_template,
+                headers_template=webhook.headers_template,
+                ignore_ssl_errors=webhook.ignore_ssl_errors,
+                do_not_retry=webhook.do_not_retry,
+                idempotency_key=idempotency_key if idempotency_key is not None else webhook.idempotency_key,
+                is_ad_hoc=True,
+            )
 
     @_ensure_context
     async def set_status_message(
@@ -1321,10 +1329,11 @@ async def set_status_message(
             raise RuntimeError('actor_run_id cannot be None when running on the Apify platform.')
 
         run_client = self.apify_client.run(self.configuration.actor_run_id)
-        run = await run_client.update(
-            status_message=status_message,
-            is_status_message_terminal=is_terminal,
-        )
+        with map_client_errors():
+            run = await run_client.update(
+                status_message=status_message,
+                is_status_message_terminal=is_terminal,
+            )
 
         if run is None:
             raise RuntimeError(

src/apify/_charging.py

@@ -19,6 +19,7 @@
 from apify_client._models import PricingPerEvent as ClientPricingPerEvent
 
 from apify._utils import ReentrantLock, docs_group, ensure_context
+from apify.errors import map_client_errors
 from apify.log import logger
 from apify.storages import Dataset
 
@@ -449,7 +450,8 @@ 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)
+                    with map_client_errors():
+                        await self._client.run(self._actor_run_id).charge(event_name, count=charged_count)
                 elif event_name in self._tier_priced_events:
                     logger.warning(
                         f"Event '{event_name}' is tier-priced and is not chargeable via the pay-per-event API."
@@ -572,7 +574,8 @@ async def _fetch_pricing_info(self) -> _FetchedPricingInfoDict:
             if self._actor_run_id is None:
                 raise RuntimeError('Actor run ID not found even though the Actor is running on Apify')
 
-            run = await self._client.run(self._actor_run_id).get()
+            with map_client_errors():
+                run = await self._client.run(self._actor_run_id).get()
 
             if run is None:
                 raise RuntimeError('Actor run not found')

src/apify/errors.py

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+import contextlib
+import functools
+from typing import TYPE_CHECKING, ParamSpec, TypeVar
 
+from apify_client.errors import ApifyApiError
 from apify_client.errors import ForbiddenError as _ForbiddenError
 from apify_client.errors import InvalidRequestError as _InvalidRequestError
 from apify_client.errors import RateLimitError as _RateLimitError
@@ -11,8 +14,14 @@
 from apify._utils import docs_group
 
 if TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable, Coroutine, Iterator
+    from typing import Any
+
     from apify_client._models import Run
 
+_P = ParamSpec('_P')
+_R = TypeVar('_R')
+
 
 @docs_group('Errors')
 class ActorError(Exception):
@@ -147,6 +156,35 @@ class ActorRateLimitError(ActorError):
     retryable = True
 
 
+@contextlib.contextmanager
+def map_client_errors() -> Iterator[None]:
+    """Translate `apify_client` API errors into domain-level `ActorError`s.
+
+    Wrap any `apify_client` call with this context manager so that an `ApifyApiError` (e.g. an HTTP 401/403/429/5xx
+    response) surfaces as the matching `ActorError` subclass instead of a raw client exception. The original error
+    is preserved as the `__cause__` of the raised `ActorError`.
+    """
+    try:
+        yield
+    except ApifyApiError as error:
+        raise ActorError.from_client_error(error) from error
+
+
+def catch_client_errors(func: Callable[_P, Awaitable[_R]]) -> Callable[_P, Coroutine[Any, Any, _R]]:
+    """Decorate an async function so the `apify_client` errors it raises become domain-level `ActorError`s.
+
+    This is the method-level counterpart of `map_client_errors`, intended for thin wrappers around `apify_client`
+    calls such as the storage client operations.
+    """
+
+    @functools.wraps(func)
+    async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
+        with map_client_errors():
+            return await func(*args, **kwargs)
+
+    return wrapper
+
+
 __all__ = [
     'ActorAuthenticationError',
     'ActorChargeLimitExceededError',

src/apify/storage_clients/_apify/_dataset_client.py

@@ -12,6 +12,7 @@
 from crawlee.storage_clients.models import DatasetItemsListPage, DatasetMetadata
 
 from ._api_client_creation import create_storage_api_client
+from apify.errors import ActorError, catch_client_errors, map_client_errors
 from apify.storage_clients._ppe_dataset_mixin import DatasetClientPpeMixin
 
 if TYPE_CHECKING:
@@ -57,11 +58,12 @@ def __init__(
         """A lock to ensure that only one operation is performed at a time."""
 
     @override
+    @catch_client_errors
     async def get_metadata(self) -> DatasetMetadata:
         metadata = await self._api_client.get()
 
         if metadata is None:
-            raise ValueError('Failed to retrieve dataset metadata.')
+            raise ActorError('Failed to retrieve dataset metadata.')
 
         return DatasetMetadata(
             id=metadata.id,
@@ -73,6 +75,7 @@ async def get_metadata(self) -> DatasetMetadata:
         )
 
     @classmethod
+    @catch_client_errors
     async def open(
         cls,
         *,
@@ -132,11 +135,13 @@ async def purge(self) -> None:
         )
 
     @override
+    @catch_client_errors
     async def drop(self) -> None:
         async with self._lock:
             await self._api_client.delete()
 
     @override
+    @catch_client_errors
     async def push_data(self, data: Sequence[Mapping[str, JsonSerializable]] | Mapping[str, JsonSerializable]) -> None:
         async def payloads_generator(items: Sequence[Mapping[str, JsonSerializable]]) -> AsyncIterator[str]:
             for index, item in enumerate(items):
@@ -155,6 +160,7 @@ async def payloads_generator(items: Sequence[Mapping[str, JsonSerializable]]) ->
             await self._charge_for_items(count_items=limit)
 
     @override
+    @catch_client_errors
     async def get_data(
         self,
         *,
@@ -199,18 +205,19 @@ async def iterate_items(
         skip_empty: bool = False,
         skip_hidden: bool = False,
     ) -> AsyncIterator[dict]:
-        async for item in self._api_client.iterate_items(
-            offset=offset,
-            limit=limit,
-            clean=clean,
-            desc=desc,
-            fields=fields,
-            omit=omit,
-            unwind=unwind,
-            skip_empty=skip_empty,
-            skip_hidden=skip_hidden,
-        ):
-            yield item
+        with map_client_errors():
+            async for item in self._api_client.iterate_items(
+                offset=offset,
+                limit=limit,
+                clean=clean,
+                desc=desc,
+                fields=fields,
+                omit=omit,
+                unwind=unwind,
+                skip_empty=skip_empty,
+                skip_hidden=skip_hidden,
+            ):
+                yield item
 
     @classmethod
     async def _check_and_serialize(cls, item: Mapping[str, JsonSerializable], index: int | None = None) -> str: