Source code for alibabacloud_oss_v2.vectors.client

# pylint: disable=line-too-long
"""Client used to interact with **Alibaba Cloud Object Storage Service (OSS)**."""
import copy
from .._client import _SyncClientImpl
from ..config import Config
from ..types import OperationInput, OperationOutput
from ..signer.vectors_v4 import VectorsSignerV4
from .. import utils
from .. import validation

from . import models
from . import operations
from . import endpoints
from .paginator import (
    ListVectorBucketsPaginator,
    ListVectorIndexesPaginator,
    ListVectorsPaginator
)


[docs] class Client: """Vectors Client """ def __init__(self, config: Config, **kwargs) -> None: """Initialize Vectors Client Args: config (Config): _description_ """ _config = copy.copy(config) self._resolve_vectors_endpoint(_config) self._build_vectors_user_agent(_config) self._client = _SyncClientImpl(_config, **kwargs) self._client._options.signer = VectorsSignerV4(account_id=config.account_id) self._client._options.endpoint_provider = endpoints.VectorsEndpointProvider( account_id=config.account_id, endpoint=self._client._options.endpoint ) def __repr__(self) -> str: return "<OssVectorsClient>" def _resolve_vectors_endpoint(self, config: Config) -> None: """vectors endpoint""" if config.endpoint is not None: return if not validation.is_valid_region(config.region): return if bool(config.use_internal_endpoint): etype = "internal" else: etype = "default" config.endpoint = endpoints.from_region(config.region, etype) def _build_vectors_user_agent(self, config: Config) -> str: if config.user_agent: return f'{utils.get_vector_user_agent()}/{config.user_agent}' return utils.get_vector_user_agent()
[docs] def invoke_operation(self, op_input: OperationInput, **kwargs ) -> OperationOutput: """invoke operation Args: op_input (OperationInput): _description_ Returns: OperationOutput: _description_ """ return self._client.invoke_operation(op_input, **kwargs)
# bucket
[docs] def put_vector_bucket(self, request: models.PutVectorBucketRequest, **kwargs ) -> models.PutVectorBucketResult: """ Creates a bucket. Args: request (PutVectorBucketRequest): Request parameters for PutVectorBucket operation. Returns: PutVectorBucketResult: Response result for PutBucket operation. """ return operations.put_vector_bucket(self._client, request, **kwargs)
[docs] def get_vector_bucket(self, request: models.GetVectorBucketRequest, **kwargs ) -> models.GetVectorBucketResult: """ GetBucketInfo Queries information about a bucket. Args: request (GetVectorBucketRequest): Request parameters for GetVectorBucket operation. Returns: GetVectorBucketResult: Response result for GetVectorBucket operation. """ return operations.get_vector_bucket(self._client, request, **kwargs)
[docs] def delete_vector_bucket(self, request: models.DeleteVectorBucketRequest, **kwargs ) -> models.DeleteVectorBucketResult: """ Deletes a bucket. Args: request (DeleteVectorBucketRequest): Request parameters for DeleteVectorBucket operation. Returns: DeleteVectorBucketResult: Response result for DeleteVectorBucket operation. """ return operations.delete_vector_bucket(self._client, request, **kwargs)
[docs] def list_vector_buckets(self, request: models.ListVectorBucketsRequest, **kwargs ) -> models.ListVectorBucketsResult: """ Lists all buckets that belong to your Alibaba Cloud account. Args: request (ListVectorBucketsRequest): Request parameters for ListBuckets operation. Returns: ListVectorBucketsResult: Response result for ListVectorBuckets operation. """ return operations.list_vector_buckets(self._client, request, **kwargs)
# bucket policy
[docs] def put_bucket_policy(self, request: models.PutBucketPolicyRequest, **kwargs ) -> models.PutBucketPolicyResult: """ Configures a policy for a bucket. Args: request (PutBucketPolicyRequest): Request parameters for PutBucketPolicy operation. Returns: PutBucketPolicyResult: Response result for PutBucketPolicy operation. """ return operations.put_bucket_policy(self._client, request, **kwargs)
[docs] def get_bucket_policy(self, request: models.GetBucketPolicyRequest, **kwargs ) -> models.GetBucketPolicyResult: """ Queries the policies configured for a bucket. Args: request (GetBucketPolicyRequest): Request parameters for GetBucketPolicy operation. Returns: GetBucketPolicyResult: Response result for GetBucketPolicy operation. """ return operations.get_bucket_policy(self._client, request, **kwargs)
[docs] def delete_bucket_policy(self, request: models.DeleteBucketPolicyRequest, **kwargs ) -> models.DeleteBucketPolicyResult: """ Deletes a policy for a bucket. Args: request (DeleteBucketPolicyRequest): Request parameters for DeleteBucketPolicy operation. Returns: DeleteBucketPolicyResult: Response result for DeleteBucketPolicy operation. """ return operations.delete_bucket_policy(self._client, request, **kwargs)
# index
[docs] def put_vector_index(self, request: models.PutVectorIndexRequest, **kwargs) -> models.PutVectorIndexResult: """ Create or update a vector index. Args: request (PutVectorIndexRequest): The request for the PutVectorIndex operation. Returns: PutVectorIndexResult: The result for the PutVectorIndex operation. """ return operations.put_vector_index(self._client, request, **kwargs)
[docs] def get_vector_index(self, request: models.GetVectorIndexRequest, **kwargs) -> models.GetVectorIndexResult: """ Get information about a specific vector index. Args: request (GetVectorIndexRequest): The request for the GetVectorIndex operation. Returns: GetVectorIndexResult: The result for the GetVectorIndex operation. """ return operations.get_vector_index(self._client, request, **kwargs)
[docs] def list_vector_indexes(self, request: models.ListVectorIndexesRequest, **kwargs) -> models.ListVectorIndexesResult: """ List vector indexes in a bucket. Args: request (ListVectorIndexesRequest): The request for the ListVectorIndexes operation. Returns: ListVectorIndexesResult: The result for the ListVectorIndexes operation. """ return operations.list_vector_indexes(self._client, request, **kwargs)
[docs] def delete_vector_index(self, request: models.DeleteVectorIndexRequest, **kwargs) -> models.DeleteVectorIndexResult: """ Delete a vector index. Args: request (DeleteVectorIndexRequest): The request for the DeleteVectorIndex operation. Returns: DeleteVectorIndexResult: The result for the DeleteVectorIndex operation. """ return operations.delete_vector_index(self._client, request, **kwargs)
# vector
[docs] def put_vectors(self, request: models.PutVectorsRequest, **kwargs) -> models.PutVectorsResult: """ Put vectors into an index. Args: request (PutVectorsRequest): The request for the PutVectors operation. Returns: PutVectorsResult: The result for the PutVectors operation. """ return operations.put_vectors(self._client, request, **kwargs)
[docs] def get_vectors(self, request: models.GetVectorsRequest, **kwargs) -> models.GetVectorsResult: """ Get vectors from an index. Args: request (GetVectorsRequest): The request for the GetVectors operation. Returns: GetVectorsResult: The result for the GetVectors operation. """ return operations.get_vectors(self._client, request, **kwargs)
[docs] def list_vectors(self, request: models.ListVectorsRequest, **kwargs) -> models.ListVectorsResult: """ List vectors in an index. Args: request (ListVectorsRequest): The request for the ListVectors operation. Returns: ListVectorsResult: The result for the ListVectors operation. """ return operations.list_vectors(self._client, request, **kwargs)
[docs] def delete_vectors(self, request: models.DeleteVectorsRequest, **kwargs) -> models.DeleteVectorsResult: """ Delete vectors from an index. Args: request (DeleteVectorsRequest): The request for the DeleteVectors operation. Returns: DeleteVectorsResult: The result for the DeleteVectors operation. """ return operations.delete_vectors(self._client, request, **kwargs)
[docs] def query_vectors(self, request: models.QueryVectorsRequest, **kwargs) -> models.QueryVectorsResult: """ Query vectors in an index. Args: request (QueryVectorsRequest): The request for the QueryVectors operation. Returns: QueryVectorsResult: The result for the QueryVectors operation. """ return operations.query_vectors(self._client, request, **kwargs)
# bucket logging
[docs] def put_bucket_logging(self, request: models.PutBucketLoggingRequest, **kwargs ) -> models.PutBucketLoggingResult: """ Enables logging for a bucket. Args: request (PutBucketLoggingRequest): Request parameters for PutBucketLogging operation. Returns: PutBucketLoggingResult: Response result for PutBucketLogging operation. """ return operations.put_bucket_logging(self._client, request, **kwargs)
[docs] def get_bucket_logging(self, request: models.GetBucketLoggingRequest, **kwargs ) -> models.GetBucketLoggingResult: """ Queries the logging configurations of a bucket. Args: request (GetBucketLoggingRequest): Request parameters for GetBucketLogging operation. Returns: GetBucketLoggingResult: Response result for GetBucketLogging operation. """ return operations.get_bucket_logging(self._client, request, **kwargs)
[docs] def delete_bucket_logging(self, request: models.DeleteBucketLoggingRequest, **kwargs ) -> models.DeleteBucketLoggingResult: """ Disables logging for a bucket. Args: request (DeleteBucketLoggingRequest): Request parameters for DeleteBucketLogging operation. Returns: DeleteBucketLoggingResult: Response result for DeleteBucketLogging operation. """ return operations.delete_bucket_logging(self._client, request, **kwargs)
# paginator
[docs] def list_vector_buckets_paginator(self, **kwargs) -> ListVectorBucketsPaginator: """Creates a paginator for ListVectorBuckets Returns: ListVectorBucketsPaginator: a paginator for ListVectorBuckets """ return ListVectorBucketsPaginator(self, **kwargs)
[docs] def list_vector_indexes_paginator(self, **kwargs) -> ListVectorIndexesPaginator: """Creates a paginator for ListVectorIndex Returns: ListVectorIndexesPaginator: a paginator for ListVectorIndex """ return ListVectorIndexesPaginator(self, **kwargs)
[docs] def list_vectors_paginator(self, **kwargs) -> ListVectorsPaginator: """Creates a paginator for ListVectors Returns: ListVectorsPaginator: a paginator for ListVectors """ return ListVectorsPaginator(self, **kwargs)