refactor(operations): format operations class based on linting rules

Author: tharropoulos

src/typesense/operations.py

@@ -1,3 +1,21 @@
+"""
+This module provides functionality for performing various operations in the Typesense API.
+
+It contains the Operations class, which handles different API operations such as
+health checks, snapshots, and configuration changes.
+
+Classes:
+    Operations: Manages various operations in the Typesense API.
+
+Dependencies:
+    - typesense.types.operations:
+        Provides type definitions for operation responses and parameters.
+    - typesense.api_call: Provides the ApiCall class for making API requests.
+    - typesense.configuration: Provides the Configuration class.
+
+Note: This module uses conditional imports to support both Python 3.11+ and earlier versions.
+"""
+
 import sys
 
 from typesense.types.operations import (
@@ -12,67 +30,149 @@
 else:
     import typing_extensions as typing
 
-from typesense import configuration
 from typesense.api_call import ApiCall
-from typesense.configuration import Configuration
 
 
-class Operations(object):
-    RESOURCE_PATH = "/operations"
-    HEALTH_PATH = "/health"
-    CONFIG_PATH = "/config"
+class Operations:
+    """
+    Manages various operations in the Typesense API.
+
+    This class provides methods to perform different operations such as
+    health checks, snapshots, and configuration changes.
+
+    Attributes:
+        resource_path (str): The base path for operations endpoints.
+        healht_path (str): The path for the health check endpoint.
+        config_path (str): The path for the configuration endpoint.
+        api_call (ApiCall): The ApiCall instance for making API requests.
+    """
+
+    resource_path: typing.Final[str] = "/operations"
+    healht_path: typing.Final[str] = "/health"
+    config_path: typing.Final[str] = "/config"
 
     def __init__(self, api_call: ApiCall):
-        self.api_call = api_call
+        """
+        Initialize the Operations instance.
 
-    @staticmethod
-    def _endpoint_path(operation_name: str) -> str:
-        return "{0}/{1}".format(Operations.RESOURCE_PATH, operation_name)
+        Args:
+            api_call (ApiCall): The ApiCall instance for making API requests.
+        """
+        self.api_call = api_call
 
     @typing.overload
     def perform(
         self,
         operation_name: typing.Literal["vote"],
         query_params: None = None,
-    ) -> OperationResponse: ...
+    ) -> 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
     def perform(
         self,
         operation_name: typing.Literal["db/compact"],
         query_params: None = None,
-    ) -> OperationResponse: ...
+    ) -> 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
     def perform(
         self,
         operation_name: typing.Literal["cache/clear"],
         query_params: None = None,
-    ) -> OperationResponse: ...
+    ) -> 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
     def perform(
         self,
         operation_name: str,
         query_params: typing.Union[typing.Dict[str, str], None] = None,
-    ) -> OperationResponse: ...
+    ) -> 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
     def perform(
         self,
         operation_name: typing.Literal["snapshot"],
         query_params: SnapshotParameters,
-    ) -> OperationResponse: ...
+    ) -> 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.
+        """
 
     def perform(
         self,
         operation_name: typing.Union[
-            typing.Literal["snapshot, vote, db/compact, cache/clear"], str
+            typing.Literal["snapshot, vote, db/compact, cache/clear"],
+            str,
         ],
         query_params: typing.Union[
-            SnapshotParameters, typing.Dict[str, str], None
+            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"]):
+               The name of the operation to perform.
+            query_params (Union[SnapshotParameters, None], optional):
+               Query parameters for the operation.
+
+        Returns:
+            OperationResponse: The response from the performed operation.
+        """
         response: OperationResponse = self.api_call.post(
             self._endpoint_path(operation_name),
             params=query_params,
@@ -82,8 +182,16 @@ def perform(
         return response
 
     def is_healthy(self) -> bool:
+        """
+        Check if the Typesense server is healthy.
+
+        Returns:
+            bool: True if the server is healthy, False otherwise.
+        """
         call_resp = self.api_call.get(
-            Operations.HEALTH_PATH, as_json=True, entity_type=HealthCheckResponse
+            Operations.healht_path,
+            as_json=True,
+            entity_type=HealthCheckResponse,
         )
         if isinstance(call_resp, typing.Dict):
             is_ok: bool = call_resp.get("ok", False)
@@ -92,13 +200,40 @@ def is_healthy(self) -> bool:
         return is_ok
 
     def toggle_slow_request_log(
-        self, data: LogSlowRequestsTimeParams
+        self,
+        log_slow_requests_time_params: LogSlowRequestsTimeParams,
     ) -> typing.Dict[str, typing.Union[str, bool]]:
-        data_dashed = {key.replace("_", "-"): value for key, value in data.items()}
+        """
+        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.
+        """
+        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]] = self.api_call.post(
-            Operations.CONFIG_PATH,
+            Operations.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([Operations.resource_path, operation_name])