Accept Pydantic models as alternatives to dicts in resource client methods

Author: vdusek

src/apify_client/_internal_models.py

@@ -1,10 +1,12 @@
-"""Internal Pydantic models that are not part of the public API and are therefore not generated."""
-
 from __future__ import annotations
 
-from pydantic import BaseModel, ConfigDict
+import json
+from base64 import b64encode
+from typing import Annotated
+
+from pydantic import BaseModel, ConfigDict, Field, RootModel
 
-from apify_client._models import ActorJobStatus  # noqa: TC001
+from apify_client._models import ActorJobStatus, WebhookCreate  # noqa: TC001
 
 
 class ActorJob(BaseModel):
@@ -27,3 +29,44 @@ class ActorJobResponse(BaseModel):
     model_config = ConfigDict(extra='allow')
 
     data: ActorJob
+
+
+class WebhookRepresentation(BaseModel):
+    """Representation of a webhook for base64-encoded API transmission.
+
+    Contains only the fields needed for the webhook payload sent via query parameters.
+    """
+
+    model_config = ConfigDict(populate_by_name=True, extra='ignore')
+
+    event_types: Annotated[list[str], Field(alias='eventTypes')]
+    request_url: Annotated[str, Field(alias='requestUrl')]
+    payload_template: Annotated[str | None, Field(alias='payloadTemplate')] = None
+    headers_template: Annotated[str | None, Field(alias='headersTemplate')] = None
+
+
+class WebhookRepresentationList(RootModel[list[WebhookRepresentation]]):
+    """List of webhook representations with base64 encoding support."""
+
+    @classmethod
+    def from_webhooks(cls, webhooks: list[WebhookCreate]) -> WebhookRepresentationList:
+        """Construct from a list of `WebhookCreate` models."""
+        representations = list[WebhookRepresentation]()
+
+        for w in webhooks:
+            webhook_dict = w.model_dump(mode='json', exclude_none=True)
+            representations.append(WebhookRepresentation.model_validate(webhook_dict))
+
+        return cls(representations)
+
+    def to_base64(self) -> str | None:
+        """Encode this list of webhook representations to a base64 string.
+
+        Returns `None` if the list is empty, so that the query parameter is omitted.
+        """
+        if not self.root:
+            return None
+
+        data = [r.model_dump(by_alias=True, exclude_none=True) for r in self.root]
+        json_string = json.dumps(data).encode(encoding='utf-8')
+        return b64encode(json_string).decode(encoding='ascii')

src/apify_client/_resource_clients/actor.py

@@ -5,6 +5,7 @@
 from pydantic import TypeAdapter
 
 from apify_client._docs import docs_group
+from apify_client._internal_models import WebhookRepresentationList
 from apify_client._models import (
     Actor,
     ActorPermissionLevel,
@@ -23,14 +24,10 @@
     RunOrigin,
     RunResponse,
     UpdateActorRequest,
+    WebhookCreate,
 )
 from apify_client._resource_clients._resource_client import ResourceClient, ResourceClientAsync
-from apify_client._utils import (
-    encode_key_value_store_record_value,
-    encode_webhook_list_to_base64,
-    response_to_dict,
-    to_seconds,
-)
+from apify_client._utils import encode_key_value_store_record_value, response_to_dict, to_seconds
 
 if TYPE_CHECKING:
     from datetime import timedelta
@@ -224,7 +221,7 @@ def start(
         timeout: timedelta | None = None,
         force_permission_level: ActorPermissionLevel | None = None,
         wait_for_finish: int | None = None,
-        webhooks: list[dict] | None = None,
+        webhooks: list[dict | WebhookCreate] | None = None,
     ) -> Run:
         """Start the Actor and immediately return the Run object.
 
@@ -261,6 +258,10 @@ def start(
         """
         run_input, content_type = encode_key_value_store_record_value(run_input, content_type)
 
+        validated_webhooks = (
+            [WebhookCreate.model_validate(w) if isinstance(w, dict) else w for w in webhooks] if webhooks else []
+        )
+
         request_params = self._build_params(
             build=build,
             maxItems=max_items,
@@ -270,7 +271,7 @@ def start(
             timeout=to_seconds(timeout, as_int=True),
             waitForFinish=wait_for_finish,
             forcePermissionLevel=force_permission_level.value if force_permission_level is not None else None,
-            webhooks=encode_webhook_list_to_base64(webhooks) if webhooks is not None else None,
+            webhooks=WebhookRepresentationList.from_webhooks(validated_webhooks).to_base64(),
         )
 
         response = self._http_client.call(
@@ -295,7 +296,7 @@ def call(
         restart_on_error: bool | None = None,
         memory_mbytes: int | None = None,
         timeout: timedelta | None = None,
-        webhooks: list[dict] | None = None,
+        webhooks: list[dict | WebhookCreate] | None = None,
         force_permission_level: ActorPermissionLevel | None = None,
         wait_duration: timedelta | None = None,
         logger: Logger | None | Literal['default'] = 'default',
@@ -689,7 +690,7 @@ async def start(
         timeout: timedelta | None = None,
         force_permission_level: ActorPermissionLevel | None = None,
         wait_for_finish: int | None = None,
-        webhooks: list[dict] | None = None,
+        webhooks: list[dict | WebhookCreate] | None = None,
     ) -> Run:
         """Start the Actor and immediately return the Run object.
 
@@ -726,6 +727,10 @@ async def start(
         """
         run_input, content_type = encode_key_value_store_record_value(run_input, content_type)
 
+        validated_webhooks = (
+            [WebhookCreate.model_validate(w) if isinstance(w, dict) else w for w in webhooks] if webhooks else []
+        )
+
         request_params = self._build_params(
             build=build,
             maxItems=max_items,
@@ -735,7 +740,7 @@ async def start(
             timeout=to_seconds(timeout, as_int=True),
             waitForFinish=wait_for_finish,
             forcePermissionLevel=force_permission_level.value if force_permission_level is not None else None,
-            webhooks=encode_webhook_list_to_base64(webhooks) if webhooks is not None else None,
+            webhooks=WebhookRepresentationList.from_webhooks(validated_webhooks).to_base64(),
         )
 
         response = await self._http_client.call(
@@ -760,7 +765,7 @@ async def call(
         restart_on_error: bool | None = None,
         memory_mbytes: int | None = None,
         timeout: timedelta | None = None,
-        webhooks: list[dict] | None = None,
+        webhooks: list[dict | WebhookCreate] | None = None,
         force_permission_level: ActorPermissionLevel | None = None,
         wait_duration: timedelta | None = None,
         logger: Logger | None | Literal['default'] = 'default',

src/apify_client/_resource_clients/request_queue.py

@@ -166,7 +166,7 @@ def list_and_lock_head(self, *, lock_duration: timedelta, limit: int | None = No
         result = response_to_dict(response)
         return HeadAndLockResponse.model_validate(result).data
 
-    def add_request(self, request: dict, *, forefront: bool | None = None) -> RequestRegistration:
+    def add_request(self, request: dict | RequestDraft, *, forefront: bool | None = None) -> RequestRegistration:
         """Add a request to the queue.
 
         https://docs.apify.com/api/v2#/reference/request-queues/request-collection/add-request
@@ -178,12 +178,15 @@ def add_request(self, request: dict, *, forefront: bool | None = None) -> Reques
         Returns:
             The added request.
         """
+        if isinstance(request, dict):
+            request = RequestDraft.model_validate(request)
+
         request_params = self._build_params(forefront=forefront, clientKey=self.client_key)
 
         response = self._http_client.call(
             url=self._build_url('requests'),
             method='POST',
-            json=request,
+            json=request.model_dump(by_alias=True, exclude_none=True),
             params=request_params,
             timeout=FAST_OPERATION_TIMEOUT,
         )
@@ -217,7 +220,7 @@ def get_request(self, request_id: str) -> Request | None:
 
         return None
 
-    def update_request(self, request: dict, *, forefront: bool | None = None) -> RequestRegistration:
+    def update_request(self, request: dict | Request, *, forefront: bool | None = None) -> RequestRegistration:
         """Update a request in the queue.
 
         https://docs.apify.com/api/v2#/reference/request-queues/request/update-request
@@ -229,14 +232,15 @@ def update_request(self, request: dict, *, forefront: bool | None = None) -> Req
         Returns:
             The updated request.
         """
-        request_id = request['id']
+        if isinstance(request, dict):
+            request = Request.model_validate(request)
 
         request_params = self._build_params(forefront=forefront, clientKey=self.client_key)
 
         response = self._http_client.call(
-            url=self._build_url(f'requests/{request_id}'),
+            url=self._build_url(f'requests/{request.id}'),
             method='PUT',
-            json=request,
+            json=request.model_dump(by_alias=True, exclude_none=True),
             params=request_params,
             timeout=STANDARD_OPERATION_TIMEOUT,
         )
@@ -315,7 +319,7 @@ def delete_request_lock(self, request_id: str, *, forefront: bool | None = None)
 
     def batch_add_requests(
         self,
-        requests: list[dict],
+        requests: list[dict | RequestDraft],
         *,
         forefront: bool = False,
         max_parallel: int = 1,
@@ -348,14 +352,19 @@ def batch_add_requests(
         if max_parallel != 1:
             raise NotImplementedError('max_parallel is only supported in async client')
 
+        requests_as_dicts = [
+            (RequestDraft.model_validate(r) if isinstance(r, dict) else r).model_dump(by_alias=True, exclude_none=True)
+            for r in requests
+        ]
+
         request_params = self._build_params(clientKey=self.client_key, forefront=forefront)
 
         # Compute the payload size limit to ensure it doesn't exceed the maximum allowed size.
         payload_size_limit_bytes = _MAX_PAYLOAD_SIZE_BYTES - math.ceil(_MAX_PAYLOAD_SIZE_BYTES * _SAFETY_BUFFER_PERCENT)
 
         # Split the requests into batches, constrained by the max payload size and max requests per batch.
         batches = constrained_batches(
-            requests,
+            requests_as_dicts,
             max_size=payload_size_limit_bytes,
             max_count=_RQ_MAX_REQUESTS_PER_BATCH,
         )
@@ -394,21 +403,26 @@ def batch_add_requests(
             )
         ).data
 
-    def batch_delete_requests(self, requests: list[dict]) -> BatchDeleteResult:
+    def batch_delete_requests(self, requests: list[dict | RequestDraft]) -> BatchDeleteResult:
         """Delete given requests from the queue.
 
         https://docs.apify.com/api/v2#/reference/request-queues/batch-request-operations/delete-requests
 
         Args:
             requests: List of the requests to delete.
         """
+        requests_as_dicts = [
+            (RequestDraft.model_validate(r) if isinstance(r, dict) else r).model_dump(by_alias=True, exclude_none=True)
+            for r in requests
+        ]
+
         request_params = self._build_params(clientKey=self.client_key)
 
         response = self._http_client.call(
             url=self._build_url('requests/batch'),
             method='DELETE',
             params=request_params,
-            json=requests,
+            json=requests_as_dicts,
             timeout=FAST_OPERATION_TIMEOUT,
         )
 
@@ -580,7 +594,7 @@ async def list_and_lock_head(self, *, lock_duration: timedelta, limit: int | Non
         result = response_to_dict(response)
         return HeadAndLockResponse.model_validate(result).data
 
-    async def add_request(self, request: dict, *, forefront: bool | None = None) -> RequestRegistration:
+    async def add_request(self, request: dict | RequestDraft, *, forefront: bool | None = None) -> RequestRegistration:
         """Add a request to the queue.
 
         https://docs.apify.com/api/v2#/reference/request-queues/request-collection/add-request
@@ -592,12 +606,15 @@ async def add_request(self, request: dict, *, forefront: bool | None = None) ->
         Returns:
             The added request.
         """
+        if isinstance(request, dict):
+            request = RequestDraft.model_validate(request)
+
         request_params = self._build_params(forefront=forefront, clientKey=self.client_key)
 
         response = await self._http_client.call(
             url=self._build_url('requests'),
             method='POST',
-            json=request,
+            json=request.model_dump(by_alias=True, exclude_none=True),
             params=request_params,
             timeout=FAST_OPERATION_TIMEOUT,
         )
@@ -629,7 +646,7 @@ async def get_request(self, request_id: str) -> Request | None:
             catch_not_found_or_throw(exc)
             return None
 
-    async def update_request(self, request: dict, *, forefront: bool | None = None) -> RequestRegistration:
+    async def update_request(self, request: dict | Request, *, forefront: bool | None = None) -> RequestRegistration:
         """Update a request in the queue.
 
         https://docs.apify.com/api/v2#/reference/request-queues/request/update-request
@@ -641,14 +658,15 @@ async def update_request(self, request: dict, *, forefront: bool | None = None)
         Returns:
             The updated request.
         """
-        request_id = request['id']
+        if isinstance(request, dict):
+            request = Request.model_validate(request)
 
         request_params = self._build_params(forefront=forefront, clientKey=self.client_key)
 
         response = await self._http_client.call(
-            url=self._build_url(f'requests/{request_id}'),
+            url=self._build_url(f'requests/{request.id}'),
             method='PUT',
-            json=request,
+            json=request.model_dump(by_alias=True, exclude_none=True),
             params=request_params,
             timeout=STANDARD_OPERATION_TIMEOUT,
         )
@@ -777,7 +795,7 @@ async def _batch_add_requests_worker(
 
     async def batch_add_requests(
         self,
-        requests: list[dict],
+        requests: list[dict | RequestDraft],
         *,
         forefront: bool = False,
         max_parallel: int = 5,
@@ -807,6 +825,11 @@ async def batch_add_requests(
         if min_delay_between_unprocessed_requests_retries:
             logger.warning('`min_delay_between_unprocessed_requests_retries` is deprecated and not used anymore.')
 
+        requests_as_dicts = [
+            (RequestDraft.model_validate(r) if isinstance(r, dict) else r).model_dump(by_alias=True, exclude_none=True)
+            for r in requests
+        ]
+
         asyncio_queue: asyncio.Queue[Iterable[dict]] = asyncio.Queue()
         request_params = self._build_params(clientKey=self.client_key, forefront=forefront)
 
@@ -815,7 +838,7 @@ async def batch_add_requests(
 
         # Split the requests into batches, constrained by the max payload size and max requests per batch.
         batches = constrained_batches(
-            requests,
+            requests_as_dicts,
             max_size=payload_size_limit_bytes,
             max_count=_RQ_MAX_REQUESTS_PER_BATCH,
         )
@@ -858,21 +881,26 @@ async def batch_add_requests(
             )
         ).data
 
-    async def batch_delete_requests(self, requests: list[dict]) -> BatchDeleteResult:
+    async def batch_delete_requests(self, requests: list[dict | RequestDraft]) -> BatchDeleteResult:
         """Delete given requests from the queue.
 
         https://docs.apify.com/api/v2#/reference/request-queues/batch-request-operations/delete-requests
 
         Args:
             requests: List of the requests to delete.
         """
+        requests_as_dicts = [
+            (RequestDraft.model_validate(r) if isinstance(r, dict) else r).model_dump(by_alias=True, exclude_none=True)
+            for r in requests
+        ]
+
         request_params = self._build_params(clientKey=self.client_key)
 
         response = await self._http_client.call(
             url=self._build_url('requests/batch'),
             method='DELETE',
             params=request_params,
-            json=requests,
+            json=requests_as_dicts,
             timeout=FAST_OPERATION_TIMEOUT,
         )
         result = response_to_dict(response)