feat(api-keys): add async support for key operations

tharropoulos · tharropoulos · commit f5e27fd0dd99 · 2025-12-09T11:05:54.000+02:00
- add AsyncKey class for async individual key operations
- add AsyncKeys class for async keys collection operations
- add async tests for key and keys functionality
- add async fixtures for testing async key operations
diff --git a/src/typesense/async_key.py b/src/typesense/async_key.py
@@ -0,0 +1,80 @@
+"""
+This module provides async functionality for managing individual API keys in Typesense.
+
+It contains the AsyncKey class, which allows for retrieving and deleting
+API keys asynchronously.
+
+Classes:
+    AsyncKey: Manages async operations on a single API key in the Typesense API.
+
+Dependencies:
+    - typesense.async_api_call: Provides the AsyncApiCall class for making async API requests.
+    - typesense.types.key: Provides ApiKeyDeleteSchema and ApiKeySchema types.
+
+Note: This module uses conditional imports to support both Python 3.11+ and earlier versions.
+"""
+
+from typesense.async_api_call import AsyncApiCall
+from typesense.types.key import ApiKeyDeleteSchema, ApiKeySchema
+
+
+class AsyncKey:
+    """
+    Manages async operations on a single API key in the Typesense API.
+
+    This class provides async methods to retrieve and delete an API key.
+
+    Attributes:
+        key_id (int): The ID of the API key.
+        api_call (AsyncApiCall): The AsyncApiCall instance for making async API requests.
+    """
+
+    def __init__(self, api_call: AsyncApiCall, key_id: int) -> None:
+        """
+        Initialize the AsyncKey instance.
+
+        Args:
+            api_call (AsyncApiCall): The AsyncApiCall instance for making async API requests.
+            key_id (int): The ID of the API key.
+        """
+        self.key_id = key_id
+        self.api_call = api_call
+
+    async def retrieve(self) -> ApiKeySchema:
+        """
+        Retrieve this specific API key.
+
+        Returns:
+            ApiKeySchema: The schema containing the API key details.
+        """
+        response: ApiKeySchema = await self.api_call.get(
+            self._endpoint_path,
+            as_json=True,
+            entity_type=ApiKeySchema,
+        )
+        return response
+
+    async def delete(self) -> ApiKeyDeleteSchema:
+        """
+        Delete this specific API key.
+
+        Returns:
+            ApiKeyDeleteSchema: The schema containing the deletion response.
+        """
+        response: ApiKeyDeleteSchema = await self.api_call.delete(
+            self._endpoint_path,
+            entity_type=ApiKeyDeleteSchema,
+        )
+        return response
+
+    @property
+    def _endpoint_path(self) -> str:
+        """
+        Construct the API endpoint path for this specific API key.
+
+        Returns:
+            str: The constructed endpoint path.
+        """
+        from typesense.async_keys import AsyncKeys
+
+        return "/".join([AsyncKeys.resource_path, str(self.key_id)])
diff --git a/src/typesense/async_keys.py b/src/typesense/async_keys.py
@@ -0,0 +1,170 @@
+"""
+This module provides async functionality for managing API keys in Typesense.
+
+It contains the AsyncKeys class, which allows for creating, retrieving, and
+generating scoped search keys asynchronously.
+
+Classes:
+    AsyncKeys: Manages API keys in the Typesense API (async).
+
+Dependencies:
+    - typesense.async_api_call: Provides the AsyncApiCall class for making async API requests.
+    - typesense.async_key: Provides the AsyncKey class for individual API key operations.
+    - typesense.types.document: Provides GenerateScopedSearchKeyParams type.
+    - typesense.types.key: Provides various API key schema types.
+
+Note: This module uses conditional imports to support both Python 3.11+ and earlier versions.
+"""
+
+import base64
+import hashlib
+import hmac
+import json
+import sys
+
+from typesense.async_api_call import AsyncApiCall
+from typesense.async_key import AsyncKey
+from typesense.types.document import GenerateScopedSearchKeyParams
+from typesense.types.key import (
+    ApiKeyCreateResponseSchema,
+    ApiKeyCreateSchema,
+    ApiKeyRetrieveSchema,
+    ApiKeySchema,
+)
+
+if sys.version_info >= (3, 11):
+    import typing
+else:
+    import typing_extensions as typing
+
+
+class AsyncKeys:
+    """
+    Manages API keys in the Typesense API (async).
+
+    This class provides async methods to create, retrieve, and generate scoped search keys.
+
+    Attributes:
+        resource_path (str): The API endpoint path for key operations.
+        api_call (AsyncApiCall): The AsyncApiCall instance for making async API requests.
+        keys (Dict[int, AsyncKey]): A dictionary of AsyncKey instances, keyed by key ID.
+    """
+
+    resource_path: typing.Final[str] = "/keys"
+
+    def __init__(self, api_call: AsyncApiCall) -> None:
+        """
+        Initialize the AsyncKeys instance.
+
+        Args:
+            api_call (AsyncApiCall): The AsyncApiCall instance for making async API requests.
+        """
+        self.api_call = api_call
+        self.keys: typing.Dict[int, AsyncKey] = {}
+
+    def __getitem__(self, key_id: int) -> AsyncKey:
+        """
+        Get or create an AsyncKey instance for a given key ID.
+
+        This method allows accessing API keys using dictionary-like syntax.
+        If the AsyncKey instance doesn't exist, it creates a new one.
+
+        Args:
+            key_id (int): The ID of the API key.
+
+        Returns:
+            AsyncKey: The AsyncKey instance for the specified key ID.
+
+        Example:
+            >>> keys = AsyncKeys(async_api_call)
+            >>> key = keys[1]
+        """
+        if not self.keys.get(key_id):
+            self.keys[key_id] = AsyncKey(self.api_call, key_id)
+        return self.keys[key_id]
+
+    async def create(self, schema: ApiKeyCreateSchema) -> ApiKeyCreateResponseSchema:
+        """
+        Create a new API key.
+
+        Args:
+            schema (ApiKeyCreateSchema): The schema for creating the API key.
+
+        Returns:
+            ApiKeyCreateResponseSchema: The created API key.
+
+        Example:
+            >>> keys = AsyncKeys(async_api_call)
+            >>> key = await keys.create(
+            ...     {
+            ...         "actions": ["documents:search"],
+            ...         "collections": ["companies"],
+            ...         "description": "Search-only key",
+            ...     }
+            ... )
+        """
+        response: ApiKeySchema = await self.api_call.post(
+            AsyncKeys.resource_path,
+            as_json=True,
+            body=schema,
+            entity_type=ApiKeySchema,
+        )
+        return response
+
+    def generate_scoped_search_key(
+        self,
+        search_key: str,
+        key_parameters: GenerateScopedSearchKeyParams,
+    ) -> bytes:
+        """
+        Generate a scoped search key.
+
+        Note: This is a synchronous method as it performs local computation
+        and does not make any API calls. Only a key generated with the
+        `documents:search` action will be accepted by the server.
+
+        Args:
+            search_key (str): The search key to use as a base.
+            key_parameters (GenerateScopedSearchKeyParams): Parameters for the scoped key.
+
+        Returns:
+            bytes: The generated scoped search key.
+
+        Example:
+            >>> keys = AsyncKeys(async_api_call)
+            >>> scoped_key = keys.generate_scoped_search_key(
+            ...     "KmacipDKNqAM3YiigXfw5pZvNOrPQUba",
+            ...     {"q": "search query", "collection": "companies"},
+            ... )
+        """
+        params_str = json.dumps(key_parameters)
+        digest = base64.b64encode(
+            hmac.new(
+                search_key.encode("utf-8"),
+                params_str.encode("utf-8"),
+                digestmod=hashlib.sha256,
+            ).digest(),
+        )
+        key_prefix = search_key[:4]
+        raw_scoped_key = f"{digest.decode('utf-8')}{key_prefix}{params_str}"
+        return base64.b64encode(raw_scoped_key.encode("utf-8"))
+
+    async def retrieve(self) -> ApiKeyRetrieveSchema:
+        """
+        Retrieve all API keys.
+
+        Returns:
+            ApiKeyRetrieveSchema: The schema containing all API keys.
+
+        Example:
+            >>> keys = AsyncKeys(async_api_call)
+            >>> all_keys = await keys.retrieve()
+            >>> for key in all_keys["keys"]:
+            ...     print(key["id"])
+        """
+        response: ApiKeyRetrieveSchema = await self.api_call.get(
+            AsyncKeys.resource_path,
+            entity_type=ApiKeyRetrieveSchema,
+            as_json=True,
+        )
+        return response
diff --git a/tests/fixtures/key_fixtures.py b/tests/fixtures/key_fixtures.py
@@ -4,6 +4,9 @@
 import requests
 
 from typesense.api_call import ApiCall
+from typesense.async_api_call import AsyncApiCall
+from typesense.async_key import AsyncKey
+from typesense.async_keys import AsyncKeys
 from typesense.key import Key
 from typesense.keys import Keys
 
@@ -61,3 +64,21 @@ def fake_keys_fixture(fake_api_call: ApiCall) -> Keys:
 def fake_key_fixture(fake_api_call: ApiCall) -> Key:
     """Return a Key object with test values."""
     return Key(fake_api_call, 1)
+
+
+@pytest.fixture(scope="function", name="actual_async_keys")
+def actual_async_keys_fixture(actual_async_api_call: AsyncApiCall) -> AsyncKeys:
+    """Return a AsyncKeys object using a real API."""
+    return AsyncKeys(actual_async_api_call)
+
+
+@pytest.fixture(scope="function", name="fake_async_keys")
+def fake_async_keys_fixture(fake_async_api_call: AsyncApiCall) -> AsyncKeys:
+    """Return a AsyncKeys object with test values."""
+    return AsyncKeys(fake_async_api_call)
+
+
+@pytest.fixture(scope="function", name="fake_async_key")
+def fake_async_key_fixture(fake_async_api_call: AsyncApiCall) -> AsyncKey:
+    """Return a AsyncKey object with test values."""
+    return AsyncKey(fake_async_api_call, 1)
diff --git a/tests/key_test.py b/tests/key_test.py
@@ -9,6 +9,9 @@
     assert_to_contain_object,
 )
 from typesense.api_call import ApiCall
+from typesense.async_api_call import AsyncApiCall
+from typesense.async_key import AsyncKey
+from typesense.async_keys import AsyncKeys
 from typesense.key import Key
 from typesense.keys import Keys
 
@@ -30,6 +33,23 @@ def test_init(fake_api_call: ApiCall) -> None:
     assert key._endpoint_path == "/keys/3"  # noqa: WPS437
 
 
+def test_init_async(fake_async_api_call: AsyncApiCall) -> None:
+    """Test that the AsyncKey object is initialized correctly."""
+    key = AsyncKey(fake_async_api_call, 3)
+
+    assert key.key_id == 3
+    assert_match_object(key.api_call, fake_async_api_call)
+    assert_object_lists_match(
+        key.api_call.node_manager.nodes,
+        fake_async_api_call.node_manager.nodes,
+    )
+    assert_match_object(
+        key.api_call.config.nearest_node,
+        fake_async_api_call.config.nearest_node,
+    )
+    assert key._endpoint_path == "/keys/3"  # noqa: WPS437
+
+
 def test_actual_retrieve(
     actual_keys: Keys,
     delete_all_keys: None,
@@ -60,3 +80,35 @@ def test_actual_delete(
     response = actual_keys[create_key_id].delete()
 
     assert response == {"id": create_key_id}
+
+
+async def test_actual_retrieve_async(
+    actual_async_keys: AsyncKeys,
+    delete_all_keys: None,
+    delete_all: None,
+    create_key_id: int,
+) -> None:
+    """Test that the AsyncKey object can retrieve an key from Typesense Server."""
+    response = await actual_async_keys[create_key_id].retrieve()
+
+    assert_to_contain_object(
+        response,
+        {
+            "actions": ["documents:search"],
+            "collections": ["companies"],
+            "description": "Search-only key",
+            "id": create_key_id,
+        },
+    )
+
+
+async def test_actual_delete_async(
+    actual_async_keys: AsyncKeys,
+    delete_all_keys: None,
+    delete_all: None,
+    create_key_id: int,
+) -> None:
+    """Test that the AsyncKey object can delete an key from Typesense Server."""
+    response = await actual_async_keys[create_key_id].delete()
+
+    assert response == {"id": create_key_id}
diff --git a/tests/keys_test.py b/tests/keys_test.py
