API reference

class DIClient:
    """
    DIClient
    The `DIClient` class serves as a client interface for interacting with the Data Intelligence (DI) solution.
    It provides methods to retrieve and interact with various data entities such as collections, pipelines,
    and models. Additionally, it supports performing similarity searches within collections.
    This class is designed for general-purpose usage and is intended for scenarios where non-administrative
    operations are required. For administrative operations, such as managing users, permissions, or system-level
    configurations, the `DIAdminClient` class should be used instead.

    Purpose:
    --------
    The `DIClient` class simplifies the interaction with the DI solution by abstracting the underlying API calls
    and providing a Pythonic interface for common operations. It is particularly useful for developers and data
    scientists who need to work with DI collections, pipelines, and models in their workflows.

    Key Features:
    -------------
    1. Retrieve specific or all collections, pipelines and models.
    2. Perform similarity searches within a specified collection.
    3. Provides a session object for managing API interactions.

    Usage:
    ------
    - Use this class for non-administrative tasks such as querying collections, retrieving pipelines, or performing
        similarity searches.
    - Avoid using this class for administrative tasks. For such operations, use the `DIAdminClient` class.
    """

    def __init__(self, *, uri=None) -> None:
        self._session = Session(uri=uri)  # type: ignore

    @property
    def session(self) -> Session:
        """
        Property to get the session object.

        Returns:
            Session: The session object used for making API requests. This session is initialized with the provided URI.

        """
        return self._session

    def get_collection(self, *, name: str) -> V1CollectionResponse:
        """
        Retrieve a collection by its name.

        Args:
            name (str): The name of the collection to retrieve. This is a required keyword-only argument.

        Returns:
            V1CollectionResponse: The collection object corresponding to the
            specified name.

        Example usage:
            ```python
            client = DIClient(uri="https://example.com")
            collection = client.get_collection(name="example_collection")
            print(collection)
            # Output: V1CollectionResponse(
            #     name="example_collection",
            #     pipeline="rag-pipeline",
            #     buckets=["bucket-1", "bucket-2"],
            #     outputStore=None,
            #     indexingMode="HNSW"
            # )
            ```
        """
        return CollectionAPI(self.session).get_collection(name=name)

    def get_all_collections(self) -> ListCollection:
        """
        Retrieves all collections available in the system.

        Returns:
            ListCollection: A response object containing a list of
            collections available in the system.

        Example usage:
            ```python
            client = DIClient(uri="https://example.com")
            collections = client.get_all_collections()
            for collection in collections.root:
                print(collection.name)
            # Output:
            # ListCollection(
            #     root=[
            #         ListCollectionItem(id="1", name="collection1"),
            #         ListCollectionItem(id="2", name="collection2")
            #     ]
            # )
            ```
        """

        return CollectionAPI(self.session).get_collections()

    def get_pipeline(self, *, name: str) -> V1PipelineResponse:
        """
        Retrieve a pipeline by its name.
        This method fetches a pipeline object from the PipelineAPI using the provided name.

        Args:
            name (str): The name of the pipeline to retrieve. This is a required keyword-only argument.

        Returns:
            V1PipelineResponse: The response object containing details about the pipeline.

        Example usage:
            ```python
            client = DIClient(uri="https://example.com")
            pipeline = client.get_pipeline(name="example_pipeline")
            print(pipeline)
            # Output: V1PipelineResponse(
            #     name="example_pipeline",
            #     type="rag",
            #     model="example_model",
            #     customFunction=None,
            #     eventFilter={"objectSuffix": ["*.txt", "*.pdf"],
            #                  "maxObjectSize": 10485760},
            #     schema_name="example_schema",
            #     prompt=None,
            #     chunkSize=512,
            #     chunkOverlap=50
            # )
            ```

        Note:
            The schema is exposed as the ``schema_name`` attribute because ``schema`` collides with a
            reserved Pydantic attribute. It is serialized and deserialized as ``schema``.
        """
        return PipelineAPI(self.session).get_pipeline(name=name)

    def get_all_pipelines(self) -> ListPipelines:
        """
        Retrieves all pipelines available in the system.

        Returns:
            ListPipelines: A response object containing a list of
            pipelines available in the system.

        Example usage:
            ```python
            client = DIClient(uri="https://example.com")
            pipelines = client.get_all_pipelines()
            for pipeline in pipelines.root:
                print(pipeline.name)
            # Output:
            # ListPipelines(
            #     root=[
            #         ListPipeline(id="1", name="pipeline1"),
            #         ListPipeline(id="2", name="pipeline2")
            #     ]
            # )
            ```
        """
        return PipelineAPI(self.session).get_pipelines()

    def similarity_search(
        self,
        *,
        access_key: str,
        secret_key: str,
        collection_name: str,
        query: str,
        top_k: int,
        search_parameters: Union[Any, Dict[str, Any]] = None,
    ) -> Union[Any, List[Dict[str, Any]]]:
        """
        Perform a similarity search on a specified collection using the provided query.
        This method interacts with the SimilaritySearchAPI to retrieve the most relevant
        results from a collection based on the given query. The results are ranked by
        similarity, and the top `k` results are returned.

        Args:
            query (str): The search query string used to find similar items in the collection.
            collection_name (str): The name of the collection to search within.
            top_k (int): The number of top similar results to retrieve.
            access_key (str): The access key for authentication with the API.
            secret_key (str): The secret key for authentication with the API.
            search_parameters (Optional[Union[Any, Dict[str, Any]]]): Additional search parameters
                that can be passed to the API for fine-tuning the search behavior.

        Returns:
            Union[Any, List[Dict[str, Any]]]: A list of dictionaries containing the top `k` similar results, or another data type depending on the API's response.

        Example usage:
            ```python
            client = DIClient(uri="https://example.com")
            results = client.similarity_search(
                query="machine learning",
                collection_name="research_papers",
                top_k=5,
                access_key="your_access_key",
                secret_key="your_secret_key",
                search_parameters={"metric": "cosine", "ef_search": "100"}
            )
            print(results)
            # Output:
            # [
            #     {
            #         "dataChunk": "chunk1",
            #         "score": 0.9,
            #         "chunkMetadata": {
            #             "objectKey": "value",
            #             "startCharIndex": 1,
            #             "endCharIndex": 2,
            #             "bucketName": "string",
            #             "pageLabel": "string",
            #             "versionId": "string",
            #         }
            #     }
            # ]
            ```
        """
        return SimilaritySearchAPI(self.session).search(
            query=query,
            collection_name=collection_name,
            top_k=top_k,
            access_key=access_key,
            secret_key=secret_key,
            search_parameters=search_parameters,
        )

    def get_model(self, *, name: str) -> V1ModelsResponse:
        """
        Retrieve a model by its name.
        This method fetches a model object from the ModelAPI using the provided name.

        Args:
            name (str): The name of the model to retrieve. This is a required keyword-only argument.

        Returns:
            V1ModelsResponse: The response object containing details about the model.

        Example usage:
            ```python
            client = DIClient(uri="https://example.com")
            model = client.get_model(name="example_model")
            print(model)
            # Output: V1ModelsResponse(
            #     name="example_model",
            #     modelName="...",
            #     capabilities=["..."],
            #     version="...",
            #     communicationType="...", # e.g. Ollama API
            #     dimension=...,  # e.g., 768
            #     contextLength=...,  # e.g., 1024
            #     temperature=...,  # e.g., 0.7
            #     topK=...,  # e.g., 40
            #     topP=...,  # e.g., 0.9
            #     maximumTokens=...,  # e.g., 512
            #     timeout=...,  # e.g., 300
            #     language="...",  # e.g., "en-US"
            #     sampleRate=...,  # e.g., 16000
            #     automaticPunctuation=...,  # e.g., True
            #     endpoint="...",  # custom-function models only
            # )
            ```
        """
        return ModelAPI(self.session).get_model(name=name)

    def get_all_models(self) -> V1ListModelsResponse:
        """
        Retrieves all models available in the system.

        Returns:
            V1ListModelsResponse: A response object containing a list of
            models available in the system.

        Example usage:
            ```python
            client = DIClient(uri="https://example.com")
            models = client.get_all_models()
            print(models)
            # Output: V1ListModelsResponse(
            #     models=[ModelRecordSummary(id="1", name="example_model")]
            # )

            for model in models.models:
                print(model.name)
            ```
        """
        return ModelAPI(self.session).get_models()

class DIAdminClient(DIClient):
    """
    DIAdminClient
    The `DIAdminClient` class is an extension of the `DIClient` class, designed specifically for administrative operations
    within the Data Intelligence (DI) platform. This class provides methods to manage collections, pipelines, schemas,
    and their associated resources. It is intended for use cases where administrative privileges are required to perform
    operations such as creating, deleting, or modifying collections, pipelines, and schemas.

    Purpose:
    ---------
    The `DIAdminClient` is tailored for scenarios where administrative-level API interactions are necessary.
    It provides a higher level of control over the DI platform's resources, enabling administrators to configure
    and manage the system effectively.

    When to Use:
    ------------
    - Use this class when you need to perform administrative tasks such as:

        - Creating or deleting collections.

        - Assigning or unassigning buckets to/from collections.

        - Creating or deleting pipelines.

        - Creating or deleting schemas.

    - This class is specifically designed for admin APIs. For non-admin APIs, use the `DIClient` class instead.

    Initialization:
    ---------------
    The `DIAdminClient` requires authentication credentials (username and password) to establish an authenticated session
    with the DI platform. Upon initialization, it creates an authenticated session that is used for all subsequent API calls.
    --------
    """

    def __init__(self, *, uri: str, username: str, password: str) -> None:
        super().__init__(uri=uri)

        # create session with auth
        self._authenticated_session = AuthAPI.login(
            uri=uri, username=username, password=password
        )

    @property
    def authenticated_session(self) -> AuthenticatedSession:
        """
        Property to get the authenticated session object.

        Returns:
        AuthenticatedSession: The authenticated session object used for making API requests.This session is initialized with the provided URI, username, password, and token.

        """
        return self._authenticated_session

    def create_collection(
        self,
        *,
        name: str,
        pipeline: str,
        buckets: Optional[List[str]] = None,
        output_store: Optional[str] = None,
        indexing_mode: Optional[str] = None,
    ) -> V1CollectionResponse:
        """
        Creates a new collection using the specified pipeline.
        If buckets are provided, they will be associated with the collection and inline embedding generation will be triggered.
        If buckets are not provided, the collection will be created but embedding generation will be deferred until buckets are assigned to the collection.

        Args:
            name (str): The name of the collection to be created. This should be unique.
            pipeline (str): The name of the pipeline to be associated with the collection.
            buckets (Optional[List[str]], optional): A list of bucket names. Defaults to None.
            output_store (Optional[str], optional): Output bucket for transcription results. Defaults to None.
            indexing_mode (Optional[str], optional): Indexing mode for RAG collections (supported values: 'HNSW', 'GPU_CAGRA'). Auto-detected if omitted. Defaults to None.

        Returns:
            V1CollectionResponse: The created collection object.

        Example usage:
            ```python
            # Example usage of create_collection
            client = DIAdminClient(uri="https://example.com", username="admin", password="password")
            collection = client.create_collection(
                name="example_collection",
                pipeline="data_ingestion_pipeline",
                buckets=["bucket1", "bucket2"],
                output_store="output-bucket"
            )
            print(collection)
            # Sample Output:
            # V1CollectionResponse(
            #     name="example_collection",
            #     pipeline="data_ingestion_pipeline",
            #     buckets=["bucket1", "bucket2"],
            #     outputStore="output-bucket"
            # )
            ```
        """
        if buckets is None:
            buckets = []

        return CollectionAPI(session=self.authenticated_session).create_collection(
            name=name,
            buckets=buckets,
            pipeline=pipeline,
            output_store=output_store,
            indexing_mode=indexing_mode,
        )

    def delete_collection(self, *, name: str) -> V1DeleteCollectionResponse:
        """
        Deletes a collection by its name.

        Args:
            name (str): The name of the collection to be deleted.

        Returns:
            V1DeleteCollectionResponse: The response object containing details about the deleted collection.

        Example Usage:
            ```python
            client = DIAdminClient(uri="https://example.com", username="admin", password="password")
            resp = client.delete_collection(name="example_collection")
            print(resp)
            # Output:
            # V1DeleteCollectionResponse(
            #     success=True,
            #     message="Collection 'example_collection' has been deleted."
            # )
            ```
        """
        return CollectionAPI(session=self.authenticated_session).delete_collection(
            name=name
        )

    def assign_buckets_to_collection(
        self, *, collection_name: str, buckets: List[str]
    ) -> BucketUpdateResponse:
        """
        Assigns a list of buckets to a specified collection.
        This enables inline embedding generation for the specified buckets.

        Args:
            collection_name (str): The name of the collection to which the buckets
                will be assigned.
            buckets (List[str]): A list of bucket names to be assigned to the
                specified collection.

        Returns:
            BucketUpdateResponse: The response object containing details about the
            updated collection and assigned buckets.

        Example Usage:
            ```python
            # Initialize the DIAdminClient
            client = DIAdminClient(uri="http://example.com", username="admin", password="password")
            # Define the collection name and buckets
            collection_name = "my_collection"
            buckets = ["bucket1", "bucket2", "bucket3"]
            # Assign buckets to the collection
            response = client.assign_buckets_to_collection(
                collection_name=collection_name,
                buckets=buckets
            )
            print(response)
            # Output:
            # BucketUpdateResponse(
            #     success=True,
            #     message="Buckets assigned successfully to collection 'my_collection'."
            # )
            ```
        Notes:
            - This method is typically used for enabling the user buckets for intelligence using an existing collection.
        """

        return CollectionAPI(
            session=self.authenticated_session
        ).assign_buckets_to_collection(collection_name=collection_name, buckets=buckets)

    def unassign_buckets_from_collection(
        self, *, collection_name: str, buckets: List[str]
    ) -> BucketUpdateResponse:
        """
        Unassigns one or more buckets from a specified collection.

        Args:
            collection_name (str): The name of the collection from which the buckets will be unassigned.
            buckets (List[str]): A list of bucket names to be unassigned.

        Returns:
            BucketUpdateResponse: The response object containing details about the updated collection
            after the buckets have been unassigned.

        Example usage:
            ```python
            # Example usage of unassign_buckets_from_collection
            client = DIAdminClient(uri="http://example.com", username="admin", password="password")

            # Unassign multiple buckets
            response = client.unassign_buckets_from_collection(
                collection_name="example_collection",
                buckets=["bucket_1", "bucket_2", "bucket_3"]
            )
            print(response)
            # Output:
            # BucketUpdateResponse(
            #     success=True,
            #     message="Buckets unassigned successfully from collection 'example_collection'."
            # )
            ```
        """
        return CollectionAPI(
            session=self.authenticated_session
        ).unassign_buckets_from_collection(
            collection_name=collection_name, buckets=buckets
        )

    @normalize_pipeline_argument_errors
    def create_pipeline(
        self,
        *,
        name: str,
        pipeline_type: str,
        event_filter_object_suffix: List[str],
        schema: str,
        event_filter_max_object_size: Optional[int] = None,
        model: Optional[str] = None,
        custom_func: Optional[str] = None,
        prompt: Optional[str] = None,
        chunk_size: Optional[int] = None,
        chunk_overlap: Optional[int] = None,
    ) -> V1CreatePipelineResponse:
        """
        Creates a new pipeline with the specified configuration.

        Args:
            name (str): The name of the pipeline to be created.
            pipeline_type (str): The type of the pipeline ("rag", "metadata", "transcribe-metadata" or "custom-function").
            event_filter_object_suffix (List[str]): A list of file suffixes to filter events. Ex - ["*.txt", "*.pdf"]
            schema (str): The schema definition for the pipeline.
            event_filter_max_object_size Optional (int): The maximum object size for event filtering. Ex - 10485760
            model Optional (str): The model associated with the pipeline. Required for "rag" and "custom-function" pipelines.
            custom_func Optional (str): The custom function to be used in the pipeline. Required for "metadata" pipelines.
            prompt Optional (str): The prompt for transcribe pipelines (e.g., "Transcribe the image content into text.").
            chunk_size Optional (int): Chunk size for RAG pipelines.
            chunk_overlap Optional (int): Chunk overlap for RAG pipelines.

        Returns:
            V1CreatePipelineResponse: The response object containing details of the created pipeline.

        Raises:
            PipelineValidationError: If the arguments are invalid or incomplete, or if the server rejects
                the request with an HTTP 422 validation error.

        Example usage:
            ```python
            client = DIAdminClient(
                uri="http://example.com",
                username="admin",
                password="password"
            )
            pipeline_data = client.create_pipeline(
                name="example_pipeline",
                pipeline_type="rag",
                model="example_model",
                event_filter_object_suffix=["*.txt", "*.pdf"],
                event_filter_max_object_size=10485760,
                schema="example_schema"
            )
            print(pipeline_data)
            # Output: V1CreatePipelineResponse(
            #     success=true,
            #     message="Pipeline 'example_pipeline' created successfully."
            # )
            ```
        """

        return PipelineAPI(session=self.authenticated_session).create_pipeline(
            name=name,
            pipeline_type=pipeline_type,
            model=model,
            custom_func=custom_func,
            event_filter_object_suffix=event_filter_object_suffix,
            event_filter_max_object_size=event_filter_max_object_size,
            schema=schema,
            prompt=prompt,
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
        )

    def delete_pipeline(self, *, name: str) -> V1DeletePipelineResponse:
        """
        Deletes a pipeline with the specified name.

        Args:
            name (str): The name of the pipeline to be deleted.

        Returns:
            V1DeletePipelineResponse: The response object containing details about the deleted pipeline.

        Example usage:
            ```python
            # Initialize the DIAdminClient
            client = DIAdminClient(uri="http://example.com", username="admin", password="password")

            # Delete a pipeline by name
            response = client.delete_pipeline(name="example_pipeline")
            print(response)
            # Output:
            # V1DeletePipelineResponse(
            #     message="Pipeline successfully deleted",
            #     success=True,
            # )
            ```
        """
        return PipelineAPI(session=self.authenticated_session).delete_pipeline(
            name=name
        )

    def get_schema(self, *, name: str) -> V1SchemasResponse:
        """
        Retrieve a schema by its name.
        This method fetches a schema object from the SchemaAPI using the provided name.

        Args:
            name (str): The name of the schema to retrieve. This is a required keyword-only argument.

        Returns:
            V1SchemasResponse: The response object containing details about the schema.

        Example usage:
            ```python
            client = DIAdminClient(uri="https://example.com", username="admin", password="password")
            schema = client.get_schema(name="example_schema")
            print(schema)
            # Output: V1SchemasResponse(
            #     name="example_schema",
            #     type="...",
            #     schema_fields=[SchemaItem(name="id", type="varchar", nullable=False)]
            # )
            ```

        Note:
            The schema fields are exposed as the ``schema_fields`` attribute because ``schema`` collides
            with a reserved Pydantic attribute. They are serialized and deserialized as ``schema``.
        """
        return SchemaAPI(session=self.authenticated_session).get_schema(name=name)

    def get_all_schemas(self) -> V1ListSchemasResponse:
        """
        Retrieves all schemas available in the system.

        Returns:
            V1ListSchemasResponse: A response object containing a list of
            schemas available in the system.

        Example usage:
            ```python
            client = DIAdminClient(uri="https://example.com", username="admin", password="password")
            schemas = client.get_all_schemas()
            print(schemas)
            # Output: V1ListSchemasResponse(
            #     schemas=[SchemaListItem(id="1", name="example_schema")]
            # )
            ```
        """
        return SchemaAPI(session=self.authenticated_session).get_schemas()

    @deprecated(
        message="This method is deprecated and will be removed in future versions. Please use get_model() instead."
    )
    def get_embedding_model(self, *, name: str) -> V1ModelsResponse:
        """
        Retrieve an embedding model by its name.
        This method fetches an embedding model object from the ModelAPI using the provided name.

        .. Deprecated::
          This method is deprecated and will be removed in future versions. Please use `get_model` instead.

        Args:
            name (str): The name of the embedding model to retrieve. This is a required keyword-only argument.

        Returns:
            V1ModelsResponse: The response object containing details about the embedding model.

        Example usage:
            ```python
            client = DIAdminClient(uri="https://example.com", username="admin", password="password")
            model = client.get_embedding_model(name="example_model")
            print(model)
            # Output: V1ModelsResponse(
            #     name="example_model",
            #     modelName="...",
            #     capabilities=["Sentence-Similarity"],
            #     version="...",
            #     dimension=...,  # e.g., 768
            #     maximumTokens=...,  # e.g., 512
            # )
            ```
        """
        return ModelAPI(self.authenticated_session).get_model(name=name)

    @deprecated(
        message="This method is deprecated and will be removed in future versions. Please use get_all_models() instead."
    )
    def get_all_embedding_models(self) -> V1ListModelsResponse:
        """
        Retrieves all embedding models available in the system.

        .. Deprecated::
            This method is deprecated and will be removed in future versions. Please use `get_all_models` instead.

        Returns:
            V1ListModelsResponse: A response object containing a list of
            embedding models available in the system.

        Example usage:
            ```python
            client = DIAdminClient(uri="https://example.com", username="admin", password="password")
            models = client.get_all_embedding_models()
            print(models)
            # Output: V1ListModelsResponse(
            #     models=[ModelRecordSummary(id="1", name="example_embedding_model")]
            # )
            ```
        """
        try:
            model_api = ModelAPI(self.authenticated_session)
            models = model_api.get_models()
            embedding_models_list = list()
            for model in models.models:
                model_details = model_api.get_model(name=model.name)
                if any(
                    capability.lower() == ModelTags.SENTENCE_SIMILARITY.value.lower()
                    for capability in model_details.capabilities
                ):
                    embedding_models_list.append(model)
        except (UnexpectedResponse, UnexpectedStatus) as e:
            raise e

        return V1ListModelsResponse(models=embedding_models_list)

    def create_schema(
        self,
        *,
        name: str,
        schema_type: str,
        schema: List[SchemaItem],
    ) -> V1CreateSchemaResponse:
        """
        Create a new schema.

        Args:
            name (str): The name of the schema to create.
            schema_type (str): The schema type (e.g., 'custom-function').
            schema (List[SchemaItem]): List of schema fields. Each field has a
                ``name`` and ``type``, and may optionally include a ``nullable``
                flag (defaults to ``True``). Set ``nullable`` to ``False`` to
                mark a field as required: the underlying database column is created
                as ``NOT NULL`` and schema validation is strict for that field,
                while fields left as ``nullable=True`` are validated non-strictly.

        Returns:
            V1CreateSchemaResponse: Response indicating success or failure.

        Example usage:
            ```python
            schema_response = client.create_schema(
                name="yolo-detection-schema",
                schema_type="custom-function",
                schema=[{"name": "id", "type": "varchar", "nullable": False},
                        {"name": "content", "type": "varchar"},
                        {"name": "embedding", "type": "array(real)", "nullable": True},
                    ]
            )
            # "id" is required (NOT NULL, strict validation); "content" and
            # "embedding" allow nulls (nullable defaults to True when omitted).
            ```
        """
        return SchemaAPI(session=self.authenticated_session).create_schema(
            name=name,
            schema_type=schema_type,
            schema=schema,
        )

    def delete_schema(self, *, name: str) -> V1DeleteSchemaResponse:
        """
        Deletes a schema with the specified name.

        Args:
            name (str): The name of the schema to be deleted.

        Returns:
            V1DeleteSchemaResponse: The response object containing details about the deleted schema.

        Example usage:
            ```python
            # Initialize the DIAdminClient
            client = DIAdminClient(uri="http://example.com", username="admin", password="password")

            # Delete a schema by name
            response = client.delete_schema(name="example_schema")
            print(response)
            # Output:
            # V1DeleteSchemaResponse(
            #     message="Schema successfully deleted"
            #     success=True,
            # )
            ```
        """
        return SchemaAPI(session=self.authenticated_session).delete_schema(name=name)

class PipelineValidationError(ValueError):
    """Exception raised when pipeline creation input is invalid."""

    def __init__(
        self,
        *,
        errors: List[Dict[str, Any]],
        source: Literal["client", "server"],
        status_code: Optional[int] = None,
        raw_response: Optional[bytes] = None,
    ):
        self.errors = errors
        self.source = source
        self.status_code = status_code
        self.raw_response = raw_response

        super().__init__(self._message())

    def _message(self) -> str:
        details = []
        for error in self.errors:
            rendered_error = self._format_error(error)
            if rendered_error:
                details.append(rendered_error)

        prefix = f"Pipeline validation failed ({self.source})"
        return (
            f"{prefix}: {'; '.join(details)}"
            if details
            else f"{prefix}: no error details"
        )

    @staticmethod
    def _format_error(error: Dict[str, Any]) -> str:
        message = error.get("msg")
        if not isinstance(message, str):
            return ""

        location = error.get("loc")
        if isinstance(location, (list, tuple)) and location:
            return f"{'.'.join(str(part) for part in location)}: {message}"

        status = error.get("status")
        if isinstance(status, str) and status:
            return f"{status}: {message}"

        return message

class NotImplementedException(Exception):
    """Exception raised for methods that are not implemented."""

    def __init__(self, message="This method is not implemented."):
        super().__init__(message)

class HTTPUnauthorizedException(Exception):
    """Exception raised for HTTP 401 Unauthorized errors."""

    def __init__(
        self,
        message="HTTP 401 Unauthorized: Access is denied due to invalid credentials.",
    ):
        super().__init__(message)

class SimilaritySearchFailureException(Exception):
    """Exception raised when a similarity search operation fails."""

    def __init__(self, message="Similarity search operation failed."):
        super().__init__(message)

class UnexpectedStatus(Exception):
    """Raised by api functions when the response status an unexpected status"""

    def __init__(self, status_code: int, content: bytes):
        self.status_code = status_code
        self.content = content

        super().__init__(
            f"Unexpected status code: {status_code}\n\nResponse content:\n{content.decode(errors='ignore')}"
        )

class UnexpectedResponse(Exception):
    """Exception raised when an API response is not as expected."""

    def __init__(self, status_code: int, response: bytes):
        self.status_code = status_code
        self.response = response

        super().__init__(
            f"Unexpected response: {status_code}\n\nResponse content:\n{response.decode(errors='ignore')}"
        )