feat(ops): add async support for operations

- add AsyncOperations class for async operations functionality
- add async tests for operations functionality
- add async fixtures for testing async operations

Author: tharropoulos

src/typesense/async_operations.py

@@ -0,0 +1,279 @@
+"""
+This module provides async functionality for performing various operations in the Typesense API.
+
+It contains the AsyncOperations class, which handles different API operations such as
+health checks, snapshots, and configuration changes asynchronously.
+
+Classes:
+    AsyncOperations: Manages various async operations in the Typesense API.
+
+Dependencies:
+    - typesense.types.operations:
+        Provides type definitions for operation responses and parameters.
+    - typesense.async_api_call: Provides the AsyncApiCall class for making async API requests.
+
+Note: This module uses conditional imports to support both Python 3.11+ and earlier versions.
+"""
+
+import sys
+
+from typesense.async_api_call import AsyncApiCall
+from typesense.types.operations import (
+    HealthCheckResponse,
+    LogSlowRequestsTimeParams,
+    OperationResponse,
+    SchemaChangesResponse,
+    SnapshotParameters,
+)
+
+if sys.version_info >= (3, 11):
+    import typing
+else:
+    import typing_extensions as typing
+
+
+class AsyncOperations:
+    """
+    Manages various async operations in the Typesense API.
+
+    This class provides async methods to perform different operations such as
+    health checks, snapshots, and configuration changes.
+
+    Attributes:
+        resource_path (str): The base path for operations endpoints.
+        health_path (str): The path for the health check endpoint.
+        config_path (str): The path for the configuration endpoint.
+        api_call (AsyncApiCall): The AsyncApiCall instance for making async API requests.
+    """
+
+    resource_path: typing.Final[str] = "/operations"
+    health_path: typing.Final[str] = "/health"
+    config_path: typing.Final[str] = "/config"
+    schema_changes: typing.Final[str] = "/schema_changes"
+
+    def __init__(self, api_call: AsyncApiCall):
+        """
+        Initialize the AsyncOperations instance.
+
+        Args:
+            api_call (AsyncApiCall): The AsyncApiCall instance for making async API requests.
+        """
+        self.api_call = api_call
+
+    @typing.overload
+    async def perform(
+        self,
+        operation_name: typing.Literal["schema_changes"],
+        query_params: None = None,
+    ) -> typing.List[SchemaChangesResponse]:
+        """
+        Perform a schema_changes operation.
+
+        Args:
+            operation_name (Literal["schema_changes"]): The name of the operation.
+            query_params (None, optional): Query parameters (not used for schema_changes operation).
+
+        Returns:
+            List[SchemaChangesResponse]: The response from the schema_changes operation.
+        """
+
+    @typing.overload
+    async def perform(
+        self,
+        operation_name: typing.Literal["vote"],
+        query_params: None = None,
+    ) -> OperationResponse:
+        """
+        Perform a vote operation.
+
+        Args:
+            operation_name (Literal["vote"]): The name of the operation.
+            query_params (None, optional): Query parameters (not used for vote operation).
+
+        Returns:
+            OperationResponse: The response from the vote operation.
+        """
+
+    @typing.overload
+    async def perform(
+        self,
+        operation_name: typing.Literal["db/compact"],
+        query_params: None = None,
+    ) -> OperationResponse:
+        """
+        Perform a database compaction operation.
+
+        Args:
+            operation_name (Literal["db/compact"]): The name of the operation.
+            query_params (None, optional): Query parameters (not used for db/compact operation).
+
+        Returns:
+            OperationResponse: The response from the database compaction operation.
+        """
+
+    @typing.overload
+    async def perform(
+        self,
+        operation_name: typing.Literal["cache/clear"],
+        query_params: None = None,
+    ) -> OperationResponse:
+        """
+        Perform a cache clear operation.
+
+        Args:
+            operation_name (Literal["cache/clear"]): The name of the operation.
+            query_params (None, optional):
+                Query parameters (not used for cache/clear operation).
+
+        Returns:
+            OperationResponse: The response from the cache clear operation.
+        """
+
+    @typing.overload
+    async def perform(
+        self,
+        operation_name: str,
+        query_params: typing.Union[typing.Dict[str, str], None] = None,
+    ) -> OperationResponse:
+        """
+        Perform a generic operation.
+
+        Args:
+            operation_name (str): The name of the operation.
+            query_params (Union[Dict[str, str], None], optional):
+                Query parameters for the operation.
+
+        Returns:
+            OperationResponse: The response from the operation.
+        """
+
+    @typing.overload
+    async def perform(
+        self,
+        operation_name: typing.Literal["snapshot"],
+        query_params: SnapshotParameters,
+    ) -> OperationResponse:
+        """
+        Perform a snapshot operation.
+
+        Args:
+            operation_name (Literal["snapshot"]): The name of the operation.
+            query_params (SnapshotParameters): Query parameters for the snapshot operation.
+
+        Returns:
+            OperationResponse: The response from the snapshot operation.
+        """
+
+    async def perform(
+        self,
+        operation_name: typing.Union[
+            typing.Literal[
+                "snapshot",
+                "vote",
+                "db/compact",
+                "cache/clear",
+                "schema_changes",
+            ],
+            str,
+        ],
+        query_params: typing.Union[
+            SnapshotParameters,
+            typing.Dict[str, str],
+            None,
+        ] = None,
+    ) -> OperationResponse:
+        """
+        Perform an operation on the Typesense API.
+
+        This method is the actual implementation for all the overloaded perform methods.
+
+        Args:
+            operation_name (Literal["snapshot, vote, db/compact, cache/clear, schema_changes"]):
+               The name of the operation to perform.
+            query_params (Union[SnapshotParameters, Dict[str, str], None], optional):
+               Query parameters for the operation.
+
+        Returns:
+            Union[OperationResponse, List[SchemaChangesResponse]]:
+                The response from the performed operation.
+
+        Example:
+            >>> operations = AsyncOperations(async_api_call)
+            >>> response = await operations.perform("vote")
+            >>> health = await operations.is_healthy()
+        """
+        response: OperationResponse = await self.api_call.post(
+            self._endpoint_path(operation_name),
+            params=query_params,
+            as_json=True,
+            entity_type=OperationResponse,
+        )
+        return response
+
+    async def is_healthy(self) -> bool:
+        """
+        Check if the Typesense server is healthy.
+
+        Returns:
+            bool: True if the server is healthy, False otherwise.
+
+        Example:
+            >>> operations = AsyncOperations(async_api_call)
+            >>> healthy = await operations.is_healthy()
+            >>> print(healthy)
+        """
+        call_resp: HealthCheckResponse = await self.api_call.get(
+            AsyncOperations.health_path,
+            as_json=True,
+            entity_type=HealthCheckResponse,
+        )
+        if isinstance(call_resp, typing.Dict):
+            is_ok: bool = call_resp.get("ok", False)
+        else:
+            is_ok = False
+        return is_ok
+
+    async def toggle_slow_request_log(
+        self,
+        log_slow_requests_time_params: LogSlowRequestsTimeParams,
+    ) -> typing.Dict[str, typing.Union[str, bool]]:
+        """
+        Toggle the slow request log configuration.
+
+        Args:
+            log_slow_requests_time_params (LogSlowRequestsTimeParams):
+               Parameters for configuring slow request logging.
+
+        Returns:
+            Dict[str, Union[str, bool]]: The response from the configuration change operation.
+
+        Example:
+            >>> operations = AsyncOperations(async_api_call)
+            >>> response = await operations.toggle_slow_request_log(
+            ...     {"log_slow_requests_time_ms": 100}
+            ... )
+        """
+        data_dashed = {
+            key.replace("_", "-"): dashed_value
+            for key, dashed_value in log_slow_requests_time_params.items()
+        }
+        response: typing.Dict[str, typing.Union[str, bool]] = await self.api_call.post(
+            AsyncOperations.config_path,
+            as_json=True,
+            entity_type=typing.Dict[str, typing.Union[str, bool]],
+            body=data_dashed,
+        )
+        return response
+
+    @staticmethod
+    def _endpoint_path(operation_name: str) -> str:
+        """
+        Generate the endpoint path for a given operation.
+
+        Args:
+            operation_name (str): The name of the operation.
+
+        Returns:
+            str: The full endpoint path for the operation.
+        """
+        return "/".join([AsyncOperations.resource_path, operation_name])

tests/fixtures/operation_fixtures.py

@@ -3,6 +3,8 @@
 import pytest
 
 from typesense.api_call import ApiCall
+from typesense.async_api_call import AsyncApiCall
+from typesense.async_operations import AsyncOperations
 from typesense.operations import Operations
 
 
@@ -16,3 +18,19 @@ def actual_operations_fixture(actual_api_call: ApiCall) -> Operations:
 def fake_operations_fixture(fake_api_call: ApiCall) -> Operations:
     """Return a Collection object with test values."""
     return Operations(fake_api_call)
+
+
+@pytest.fixture(scope="function", name="actual_async_operations")
+def actual_async_operations_fixture(
+    actual_async_api_call: AsyncApiCall,
+) -> AsyncOperations:
+    """Return a AsyncOperations object using a real API."""
+    return AsyncOperations(actual_async_api_call)
+
+
+@pytest.fixture(scope="function", name="fake_async_operations")
+def fake_async_operations_fixture(
+    fake_async_api_call: AsyncApiCall,
+) -> AsyncOperations:
+    """Return a AsyncOperations object with test values."""
+    return AsyncOperations(fake_async_api_call)