chore: generate sync version of the client

Author: tharropoulos

src/typesense/sync/alias.py

@@ -0,0 +1,80 @@
+"""
+This module provides async functionality for managing individual aliases in Typesense.
+
+It contains the Alias class, which allows for retrieving and deleting
+aliases asynchronously.
+
+Classes:
+    Alias: Manages async operations on a single alias in the Typesense API.
+
+Dependencies:
+    - typesense.async_api_call: Provides the ApiCall class for making async API requests.
+    - typesense.types.alias: Provides AliasSchema type.
+
+Note: This module uses conditional imports to support both Python 3.11+ and earlier versions.
+"""
+
+from .api_call import ApiCall
+from typesense.types.alias import AliasSchema
+
+
+class Alias:
+    """
+    Manages async operations on a single alias in the Typesense API.
+
+    This class provides async methods to retrieve and delete an alias.
+
+    Attributes:
+        api_call (ApiCall): The ApiCall instance for making async API requests.
+        name (str): The name of the alias.
+    """
+
+    def __init__(self, api_call: ApiCall, name: str):
+        """
+        Initialize the Alias instance.
+
+        Args:
+            api_call (ApiCall): The ApiCall instance for making async API requests.
+            name (str): The name of the alias.
+        """
+        self.api_call = api_call
+        self.name = name
+
+    def retrieve(self) -> AliasSchema:
+        """
+        Retrieve this specific alias.
+
+        Returns:
+            AliasSchema: The schema containing the alias details.
+        """
+        response: AliasSchema = self.api_call.get(
+            self._endpoint_path,
+            entity_type=AliasSchema,
+            as_json=True,
+        )
+        return response
+
+    def delete(self) -> AliasSchema:
+        """
+        Delete this specific alias.
+
+        Returns:
+            AliasSchema: The schema containing the deletion response.
+        """
+        response: AliasSchema = self.api_call.delete(
+            self._endpoint_path,
+            entity_type=AliasSchema,
+        )
+        return response
+
+    @property
+    def _endpoint_path(self) -> str:
+        """
+        Construct the API endpoint path for this specific alias.
+
+        Returns:
+            str: The constructed endpoint path.
+        """
+        from .aliases import Aliases
+
+        return "/".join([Aliases.resource_path, self.name])

src/typesense/sync/aliases.py

@@ -0,0 +1,129 @@
+"""
+This module provides async functionality for managing aliases in Typesense.
+
+It contains the Aliases class, which allows for creating, updating, retrieving, and
+accessing individual aliases asynchronously.
+
+Classes:
+    Aliases: Manages aliases in the Typesense API (async).
+
+Dependencies:
+    - typesense.async_api_call: Provides the ApiCall class for making async API requests.
+    - typesense.async_alias: Provides the Alias class for individual alias operations.
+    - typesense.types.alias: Provides AliasCreateSchema, AliasSchema, and AliasesResponseSchema types.
+
+Note: This module uses conditional imports to support both Python 3.11+ and earlier versions.
+"""
+
+import sys
+
+from .api_call import ApiCall
+from .alias import Alias
+from typesense.types.alias import AliasCreateSchema, AliasSchema, AliasesResponseSchema
+
+if sys.version_info >= (3, 11):
+    import typing
+else:
+    import typing_extensions as typing
+
+
+class Aliases:
+    """
+    Manages aliases in the Typesense API (async).
+
+    This class provides async methods to create, update, retrieve, and access individual aliases.
+
+    Attributes:
+        resource_path (str): The API endpoint path for alias operations.
+        api_call (ApiCall): The ApiCall instance for making async API requests.
+        aliases (Dict[str, Alias]): A dictionary of Alias instances, keyed by alias name.
+    """
+
+    resource_path: typing.Final[str] = "/aliases"
+
+    def __init__(self, api_call: ApiCall):
+        """
+        Initialize the Aliases instance.
+
+        Args:
+            api_call (ApiCall): The ApiCall instance for making async API requests.
+        """
+        self.api_call = api_call
+        self.aliases: typing.Dict[str, Alias] = {}
+
+    def __getitem__(self, name: str) -> Alias:
+        """
+        Get or create an Alias instance for a given alias name.
+
+        This method allows accessing aliases using dictionary-like syntax.
+        If the Alias instance doesn't exist, it creates a new one.
+
+        Args:
+            name (str): The name of the alias.
+
+        Returns:
+            Alias: The Alias instance for the specified alias name.
+
+        Example:
+            >>> aliases = Aliases(async_api_call)
+            >>> company_alias = aliases["company_alias"]
+        """
+        if not self.aliases.get(name):
+            self.aliases[name] = Alias(self.api_call, name)
+        return self.aliases[name]
+
+    def upsert(self, name: str, mapping: AliasCreateSchema) -> AliasSchema:
+        """
+        Create or update an alias.
+
+        Args:
+            name (str): The name of the alias.
+            mapping (AliasCreateSchema): The schema for creating or updating the alias.
+
+        Returns:
+            AliasSchema: The created or updated alias.
+
+        Example:
+            >>> aliases = Aliases(async_api_call)
+            >>> alias = await aliases.upsert(
+            ...     "company_alias", {"collection_name": "companies"}
+            ... )
+        """
+        response: AliasSchema = self.api_call.put(
+            self._endpoint_path(name),
+            body=mapping,
+            entity_type=AliasSchema,
+        )
+        return response
+
+    def retrieve(self) -> AliasesResponseSchema:
+        """
+        Retrieve all aliases.
+
+        Returns:
+            AliasesResponseSchema: The schema containing all aliases.
+
+        Example:
+            >>> aliases = Aliases(async_api_call)
+            >>> all_aliases = await aliases.retrieve()
+            >>> for alias in all_aliases["aliases"]:
+            ...     print(alias["name"])
+        """
+        response: AliasesResponseSchema = self.api_call.get(
+            Aliases.resource_path,
+            as_json=True,
+            entity_type=AliasesResponseSchema,
+        )
+        return response
+
+    def _endpoint_path(self, alias_name: str) -> str:
+        """
+        Construct the API endpoint path for alias operations.
+
+        Args:
+            alias_name (str): The name of the alias.
+
+        Returns:
+            str: The constructed endpoint path.
+        """
+        return "/".join([Aliases.resource_path, alias_name])

src/typesense/sync/analytics.py

@@ -0,0 +1,14 @@
+"""Client for Typesense Analytics module (async)."""
+
+from .analytics_events import AnalyticsEvents
+from .analytics_rules import AnalyticsRules
+from .api_call import ApiCall
+
+
+class Analytics:
+    """Client for v30 Analytics endpoints (async)."""
+
+    def __init__(self, api_call: ApiCall) -> None:
+        self.api_call = api_call
+        self.rules = AnalyticsRules(api_call)
+        self.events = AnalyticsEvents(api_call)

src/typesense/sync/analytics_events.py

@@ -0,0 +1,71 @@
+"""Client for Analytics events and status operations (async)."""
+
+import sys
+
+if sys.version_info >= (3, 11):
+    import typing
+else:
+    import typing_extensions as typing
+
+from .api_call import ApiCall
+from typesense.types.analytics import (
+    AnalyticsEvent as AnalyticsEventSchema,
+    AnalyticsEventCreateResponse,
+    AnalyticsEventsResponse,
+    AnalyticsStatus,
+)
+
+
+class AnalyticsEvents:
+    events_path: typing.Final[str] = "/analytics/events"
+    flush_path: typing.Final[str] = "/analytics/flush"
+    status_path: typing.Final[str] = "/analytics/status"
+
+    def __init__(self, api_call: ApiCall) -> None:
+        self.api_call = api_call
+
+    def create(self, event: AnalyticsEventSchema) -> AnalyticsEventCreateResponse:
+        response: AnalyticsEventCreateResponse = self.api_call.post(
+            AnalyticsEvents.events_path,
+            body=event,
+            as_json=True,
+            entity_type=AnalyticsEventCreateResponse,
+        )
+        return response
+
+    def retrieve(
+        self,
+        *,
+        user_id: str,
+        name: str,
+        n: int,
+    ) -> AnalyticsEventsResponse:
+        params: typing.Dict[str, typing.Union[str, int]] = {
+            "user_id": user_id,
+            "name": name,
+            "n": n,
+        }
+        response: AnalyticsEventsResponse = self.api_call.get(
+            AnalyticsEvents.events_path,
+            params=params,
+            as_json=True,
+            entity_type=AnalyticsEventsResponse,
+        )
+        return response
+
+    def flush(self) -> AnalyticsEventCreateResponse:
+        response: AnalyticsEventCreateResponse = self.api_call.post(
+            AnalyticsEvents.flush_path,
+            body={},
+            as_json=True,
+            entity_type=AnalyticsEventCreateResponse,
+        )
+        return response
+
+    def status(self) -> AnalyticsStatus:
+        response: AnalyticsStatus = self.api_call.get(
+            AnalyticsEvents.status_path,
+            as_json=True,
+            entity_type=AnalyticsStatus,
+        )
+        return response

src/typesense/sync/analytics_rule.py

@@ -0,0 +1,31 @@
+"""Per-rule client for Analytics rules operations (async)."""
+
+from .api_call import ApiCall
+from typesense.types.analytics import AnalyticsRuleSchema
+
+
+class AnalyticsRule:
+    def __init__(self, api_call: ApiCall, rule_name: str) -> None:
+        self.api_call = api_call
+        self.rule_name = rule_name
+
+    @property
+    def _endpoint_path(self) -> str:
+        from .analytics_rules import AnalyticsRules
+
+        return "/".join([AnalyticsRules.resource_path, self.rule_name])
+
+    def retrieve(self) -> AnalyticsRuleSchema:
+        response: AnalyticsRuleSchema = self.api_call.get(
+            self._endpoint_path,
+            as_json=True,
+            entity_type=AnalyticsRuleSchema,
+        )
+        return response
+
+    def delete(self) -> AnalyticsRuleSchema:
+        response: AnalyticsRuleSchema = self.api_call.delete(
+            self._endpoint_path,
+            entity_type=AnalyticsRuleSchema,
+        )
+        return response