feat: Add KeyValueStoreClient(Async).get_record_public_url (#506)

Pijukatel · web-flow · commit 6417d26f90af · 2025-09-30T09:16:04.000+02:00
### Description - Add `KeyValueStoreClient.get_record_public_url`. - Add `KeyValueStoreClientAsync.get_record_public_url`. - Add tests. ### Issues Closes: #497
diff --git a/src/apify_client/clients/resource_clients/key_value_store.py b/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,36 @@ def delete_record(self, key: str) -> None:
             timeout_secs=_SMALL_TIMEOUT,
         )
 
+    def get_record_public_url(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')
+
+        metadata = self.get()
+
+        request_params = self._params()
+
+        if metadata and 'urlSigningSecretKey' in metadata:
+            request_params['signature'] = create_hmac_signature(metadata['urlSigningSecretKey'], key)
+
+        key_public_url = urlparse(self._url(f'records/{key}', public=True))
+        filtered_params = {k: v for k, v in request_params.items() if v is not None}
+
+        if filtered_params:
+            key_public_url = key_public_url._replace(query=urlencode(filtered_params))
+
+        return urlunparse(key_public_url)
+
     def create_keys_public_url(
         self,
         *,
@@ -290,7 +320,7 @@ def create_keys_public_url(
         Returns:
             The public key-value store keys URL.
         """
-        store = self.get()
+        metadata = self.get()
 
         request_params = self._params(
             limit=limit,
@@ -299,10 +329,10 @@ def create_keys_public_url(
             prefix=prefix,
         )
 
-        if store and 'urlSigningSecretKey' in store:
+        if metadata and 'urlSigningSecretKey' in metadata:
             signature = create_storage_content_signature(
-                resource_id=store['id'],
-                url_signing_secret_key=store['urlSigningSecretKey'],
+                resource_id=metadata['id'],
+                url_signing_secret_key=metadata['urlSigningSecretKey'],
                 expires_in_millis=expires_in_secs * 1000 if expires_in_secs is not None else None,
             )
             request_params['signature'] = signature
@@ -555,6 +585,36 @@ async def delete_record(self, key: str) -> None:
             timeout_secs=_SMALL_TIMEOUT,
         )
 
+    async def get_record_public_url(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')
+
+        metadata = await self.get()
+
+        request_params = self._params()
+
+        if metadata and 'urlSigningSecretKey' in metadata:
+            request_params['signature'] = create_hmac_signature(metadata['urlSigningSecretKey'], key)
+
+        key_public_url = urlparse(self._url(f'records/{key}', public=True))
+        filtered_params = {k: v for k, v in request_params.items() if v is not None}
+
+        if filtered_params:
+            key_public_url = key_public_url._replace(query=urlencode(filtered_params))
+
+        return urlunparse(key_public_url)
+
     async def create_keys_public_url(
         self,
         *,
@@ -578,7 +638,7 @@ async def create_keys_public_url(
         Returns:
             The public key-value store keys URL.
         """
-        store = await self.get()
+        metadata = await self.get()
 
         keys_public_url = urlparse(self._url('keys'))
 
@@ -589,10 +649,10 @@ async def create_keys_public_url(
             prefix=prefix,
         )
 
-        if store and 'urlSigningSecretKey' in store:
+        if metadata and 'urlSigningSecretKey' in metadata:
             signature = create_storage_content_signature(
-                resource_id=store['id'],
-                url_signing_secret_key=store['urlSigningSecretKey'],
+                resource_id=metadata['id'],
+                url_signing_secret_key=metadata['urlSigningSecretKey'],
                 expires_in_millis=expires_in_secs * 1000 if expires_in_secs is not None else None,
             )
             request_params['signature'] = signature
diff --git a/tests/integration/test_key_value_store.py b/tests/integration/test_key_value_store.py
@@ -1,39 +1,43 @@
 from __future__ import annotations
 
+import json
 from unittest import mock
 from unittest.mock import Mock
 
 import impit
+import pytest
+from apify_shared.utils import create_hmac_signature, create_storage_content_signature
 
 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"
-  }
-}"""
+MOCKED_ID = 'someID'
+
+
+def _get_mocked_api_kvs_response(signing_key: str | None = None) -> str:
+    response_data = {
+        'data': {
+            'id': MOCKED_ID,
+            '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 +77,48 @@ 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('signing_key', [None, 'custom-signing-key'])
     @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, signing_key: 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')
+        kvs = apify_client.key_value_store(MOCKED_ID)
 
         # 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=signing_key)),
+        ):
             public_url = kvs.create_keys_public_url()
+            if signing_key:
+                signature_value = create_storage_content_signature(
+                    resource_id=MOCKED_ID, url_signing_secret_key=signing_key
+                )
+                expected_signature = f'?signature={signature_value}'
+            else:
+                expected_signature = ''
             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}'
+            )
+
+    @pytest.mark.parametrize('signing_key', [None, 'custom-signing-key'])
+    @parametrized_api_urls
+    def test_record_public_url(self, api_token: str, api_url: str, api_public_url: str, signing_key: str) -> None:
+        apify_client = ApifyClient(token=api_token, api_url=api_url, api_public_url=api_public_url)
+        key = 'some_key'
+        kvs = apify_client.key_value_store(MOCKED_ID)
+
+        # Mock the API call to return predefined response
+        with mock.patch.object(
+            apify_client.http_client,
+            'call',
+            return_value=Mock(text=_get_mocked_api_kvs_response(signing_key=signing_key)),
+        ):
+            public_url = kvs.get_record_public_url(key=key)
+            expected_signature = f'?signature={create_hmac_signature(signing_key, key)}' if signing_key else ''
+            assert public_url == (
+                f'{(api_public_url or DEFAULT_API_URL).strip("/")}/v2/key-value-stores/someID/'
+                f'records/{key}{expected_signature}'
             )
 
 
@@ -130,15 +165,47 @@ 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('signing_key', [None, 'custom-signing-key'])
     @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, signing_key: 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')
+        kvs = apify_client.key_value_store(MOCKED_ID)
+        mocked_response = _get_mocked_api_kvs_response(signing_key=signing_key)
 
         # 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=mocked_response),
+        ):
             public_url = await kvs.create_keys_public_url()
+            if signing_key:
+                signature_value = create_storage_content_signature(
+                    resource_id=MOCKED_ID, url_signing_secret_key=signing_key
+                )
+                expected_signature = f'?signature={signature_value}'
+            else:
+                expected_signature = ''
+            assert public_url == (
+                f'{(api_public_url or DEFAULT_API_URL).strip("/")}/v2/key-value-stores/someID/keys{expected_signature}'
+            )
+
+    @pytest.mark.parametrize('signing_key', [None, 'custom-signing-key'])
+    @parametrized_api_urls
+    async def test_record_public_url(self, api_token: str, api_url: str, api_public_url: str, signing_key: str) -> None:
+        apify_client = ApifyClientAsync(token=api_token, api_url=api_url, api_public_url=api_public_url)
+        key = 'some_key'
+        kvs = apify_client.key_value_store(MOCKED_ID)
+
+        # Mock the API call to return predefined response
+        with mock.patch.object(
+            apify_client.http_client,
+            'call',
+            return_value=Mock(text=_get_mocked_api_kvs_response(signing_key=signing_key)),
+        ):
+            public_url = await kvs.get_record_public_url(key=key)
+            expected_signature = f'?signature={create_hmac_signature(signing_key, key)}' if signing_key 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/'
+                f'records/{key}{expected_signature}'
             )
