Update existing tests, add draft of the function

Author: Pijukatel

src/apify_client/clients/resource_clients/key_value_store.py

@@ -6,7 +6,7 @@
 from typing import TYPE_CHECKING, Any
 from urllib.parse import urlencode, urlparse, urlunparse
 
-from apify_shared.utils import create_storage_content_signature
+from apify_shared.utils import create_hmac_signature, create_storage_content_signature
 
 from apify_client._utils import (
     catch_not_found_or_throw,
@@ -267,6 +267,29 @@ def delete_record(self, key: str) -> None:
             timeout_secs=_SMALL_TIMEOUT,
         )
 
+    def get_record_public_rul(self, key: str) -> str:
+        """Generate a URL that can be used to access key-value store record.
+
+        If the client has permission to access the key-value store's URL signing key, the URL will include a signature
+        to verify its authenticity.
+
+        Args:
+            key: The key for which the URL should be generated.
+
+        Returns:
+            A public URL that can be used to access the value of the given key in the KVS.
+        """
+        if self.resource_id is None:
+            raise ValueError('resource_id cannot be None when generating a public URL')
+
+        public_url = f'{self._api_public_base_url}/key-value-stores/{self.resource_id}/records/{key}'
+        metadata = self.get_metadata()
+
+        if metadata.url_signing_secret_key is not None:
+            public_url = public_url.with_query(signature=create_hmac_signature(metadata.url_signing_secret_key, key))
+
+        return str(public_url)
+
     def create_keys_public_url(
         self,
         *,

tests/integration/test_key_value_store.py

@@ -1,39 +1,40 @@
 from __future__ import annotations
 
+import json
 from unittest import mock
 from unittest.mock import Mock
 
 import impit
+import pytest
 
 from integration.integration_test_utils import parametrized_api_urls, random_resource_name
 
 from apify_client import ApifyClient, ApifyClientAsync
 from apify_client.client import DEFAULT_API_URL
 
-MOCKED_API_KVS_RESPONSE = """{
-  "data": {
-    "id": "someID",
-    "name": "name",
-    "userId": "userId",
-    "createdAt": "2025-09-11T08:48:51.806Z",
-    "modifiedAt": "2025-09-11T08:48:51.806Z",
-    "accessedAt": "2025-09-11T08:48:51.806Z",
-    "actId": null,
-    "actRunId": null,
-    "schema": null,
-    "stats": {
-      "readCount": 0,
-      "writeCount": 0,
-      "deleteCount": 0,
-      "listCount": 0,
-      "storageBytes": 0
-    },
-    "consoleUrl": "https://console.apify.com/storage/key-value-stores/someID",
-    "keysPublicUrl": "https://api.apify.com/v2/key-value-stores/someID/keys",
-    "generalAccess": "FOLLOW_USER_SETTING",
-    "urlSigningSecretKey": "urlSigningSecretKey"
-  }
-}"""
+
+def _get_mocked_api_kvs_response(signing_key: str | None = None) -> str:
+    response_data = {
+        'data': {
+            'id': 'someID',
+            'name': 'name',
+            'userId': 'userId',
+            'createdAt': '2025-09-11T08:48:51.806Z',
+            'modifiedAt': '2025-09-11T08:48:51.806Z',
+            'accessedAt': '2025-09-11T08:48:51.806Z',
+            'actId': None,
+            'actRunId': None,
+            'schema': None,
+            'stats': {'readCount': 0, 'writeCount': 0, 'deleteCount': 0, 'listCount': 0, 'storageBytes': 0},
+            'consoleUrl': 'https://console.apify.com/storage/key-value-stores/someID',
+            'keysPublicUrl': 'https://api.apify.com/v2/key-value-stores/someID/keys',
+            'generalAccess': 'FOLLOW_USER_SETTING',
+        }
+    }
+    if signing_key:
+        response_data['data']['urlSigningSecretKey'] = signing_key
+
+    return json.dumps(response_data)
 
 
 class TestKeyValueStoreSync:
@@ -73,17 +74,22 @@ def test_key_value_store_should_create_public_keys_non_expiring_url(self, apify_
         store.delete()
         assert apify_client.key_value_store(created_store['id']).get() is None
 
+    @pytest.mark.parametrize('signature', [None, 'custom-signature'])
     @parametrized_api_urls
-    def test_public_url(self, api_token: str, api_url: str, api_public_url: str) -> None:
+    def test_public_url(self, api_token: str, api_url: str, api_public_url: str, signature: str) -> None:
         apify_client = ApifyClient(token=api_token, api_url=api_url, api_public_url=api_public_url)
         kvs = apify_client.key_value_store('someID')
 
         # Mock the API call to return predefined response
-        with mock.patch.object(apify_client.http_client, 'call', return_value=Mock(text=MOCKED_API_KVS_RESPONSE)):
+        with mock.patch.object(
+            apify_client.http_client,
+            'call',
+            return_value=Mock(text=_get_mocked_api_kvs_response(signing_key=signature)),
+        ):
             public_url = kvs.create_keys_public_url()
+            expected_signature = f'?signature={public_url.split("signature=")[1]}' if signature else ''
             assert public_url == (
-                f'{(api_public_url or DEFAULT_API_URL).strip("/")}/v2/key-value-stores/'
-                f'someID/keys?signature={public_url.split("signature=")[1]}'
+                f'{(api_public_url or DEFAULT_API_URL).strip("/")}/v2/key-value-stores/someID/keys{expected_signature}'
             )
 
 
@@ -130,15 +136,20 @@ async def test_key_value_store_should_create_public_keys_non_expiring_url(
         await store.delete()
         assert await apify_client_async.key_value_store(created_store['id']).get() is None
 
+    @pytest.mark.parametrize('signature', [None, 'custom-signature'])
     @parametrized_api_urls
-    async def test_public_url(self, api_token: str, api_url: str, api_public_url: str) -> None:
+    async def test_public_url(self, api_token: str, api_url: str, api_public_url: str, signature: str) -> None:
         apify_client = ApifyClientAsync(token=api_token, api_url=api_url, api_public_url=api_public_url)
         kvs = apify_client.key_value_store('someID')
 
         # Mock the API call to return predefined response
-        with mock.patch.object(apify_client.http_client, 'call', return_value=Mock(text=MOCKED_API_KVS_RESPONSE)):
+        with mock.patch.object(
+            apify_client.http_client,
+            'call',
+            return_value=Mock(text=_get_mocked_api_kvs_response(signing_key=signature)),
+        ):
             public_url = await kvs.create_keys_public_url()
+            expected_signature = f'?signature={public_url.split("signature=")[1]}' if signature else ''
             assert public_url == (
-                f'{(api_public_url or DEFAULT_API_URL).strip("/")}/v2/key-value-stores/'
-                f'someID/keys?signature={public_url.split("signature=")[1]}'
+                f'{(api_public_url or DEFAULT_API_URL).strip("/")}/v2/key-value-stores/someID/keys{expected_signature}'
             )