pinecone.data.sparse_values_factory

 1from collections.abc import Mapping
 2from typing import Union, Optional
 3
 4from ..utils import convert_to_list
 5
 6from .errors import (
 7    SparseValuesTypeError,
 8    SparseValuesMissingKeysError,
 9    SparseValuesDictionaryExpectedError,
10)
11
12from .dataclasses import SparseValues
13from .types import SparseVectorTypedDict
14from pinecone.core.openapi.db_data.models import SparseValues as OpenApiSparseValues
15
16
17class SparseValuesFactory:
18    """SparseValuesFactory is used to convert various types of user input into SparseValues objects used in generated request code."""
19
20    @staticmethod
21    def build(
22        input: Optional[Union[SparseValues, OpenApiSparseValues, SparseVectorTypedDict]],
23    ) -> Optional[OpenApiSparseValues]:
24        if input is None:
25            return input
26        if isinstance(input, OpenApiSparseValues):
27            return input
28        if isinstance(input, SparseValues):
29            return OpenApiSparseValues(indices=input.indices, values=input.values)
30        if not isinstance(input, Mapping):
31            raise SparseValuesDictionaryExpectedError(input)
32        if not {"indices", "values"}.issubset(input):
33            raise SparseValuesMissingKeysError(input)
34
35        indices = SparseValuesFactory._convert_to_list(input.get("indices"), int)
36        values = SparseValuesFactory._convert_to_list(input.get("values"), float)
37
38        if len(indices) != len(values):
39            raise ValueError("Sparse values indices and values must have the same length")
40
41        try:
42            return OpenApiSparseValues(indices=indices, values=values)
43        except TypeError as e:
44            raise SparseValuesTypeError() from e
45
46    @staticmethod
47    def _convert_to_list(input, expected_type):
48        try:
49            converted = convert_to_list(input)
50        except TypeError as e:
51            raise SparseValuesTypeError() from e
52
53        SparseValuesFactory._validate_list_items_type(converted, expected_type)
54        return converted
55
56    @staticmethod
57    def _validate_list_items_type(input, expected_type):
58        if len(input) > 0 and not isinstance(input[0], expected_type):
59            raise SparseValuesTypeError()
class SparseValuesFactory:
18class SparseValuesFactory:
19    """SparseValuesFactory is used to convert various types of user input into SparseValues objects used in generated request code."""
20
21    @staticmethod
22    def build(
23        input: Optional[Union[SparseValues, OpenApiSparseValues, SparseVectorTypedDict]],
24    ) -> Optional[OpenApiSparseValues]:
25        if input is None:
26            return input
27        if isinstance(input, OpenApiSparseValues):
28            return input
29        if isinstance(input, SparseValues):
30            return OpenApiSparseValues(indices=input.indices, values=input.values)
31        if not isinstance(input, Mapping):
32            raise SparseValuesDictionaryExpectedError(input)
33        if not {"indices", "values"}.issubset(input):
34            raise SparseValuesMissingKeysError(input)
35
36        indices = SparseValuesFactory._convert_to_list(input.get("indices"), int)
37        values = SparseValuesFactory._convert_to_list(input.get("values"), float)
38
39        if len(indices) != len(values):
40            raise ValueError("Sparse values indices and values must have the same length")
41
42        try:
43            return OpenApiSparseValues(indices=indices, values=values)
44        except TypeError as e:
45            raise SparseValuesTypeError() from e
46
47    @staticmethod
48    def _convert_to_list(input, expected_type):
49        try:
50            converted = convert_to_list(input)
51        except TypeError as e:
52            raise SparseValuesTypeError() from e
53
54        SparseValuesFactory._validate_list_items_type(converted, expected_type)
55        return converted
56
57    @staticmethod
58    def _validate_list_items_type(input, expected_type):
59        if len(input) > 0 and not isinstance(input[0], expected_type):
60            raise SparseValuesTypeError()

SparseValuesFactory is used to convert various types of user input into SparseValues objects used in generated request code.

@staticmethod
def build( input: Union[pinecone.data.dataclasses.sparse_values.SparseValues, pinecone.core.openapi.db_data.model.sparse_values.SparseValues, pinecone.data.types.sparse_vector_typed_dict.SparseVectorTypedDict, NoneType]) -> Optional[pinecone.core.openapi.db_data.model.sparse_values.SparseValues]:
21    @staticmethod
22    def build(
23        input: Optional[Union[SparseValues, OpenApiSparseValues, SparseVectorTypedDict]],
24    ) -> Optional[OpenApiSparseValues]:
25        if input is None:
26            return input
27        if isinstance(input, OpenApiSparseValues):
28            return input
29        if isinstance(input, SparseValues):
30            return OpenApiSparseValues(indices=input.indices, values=input.values)
31        if not isinstance(input, Mapping):
32            raise SparseValuesDictionaryExpectedError(input)
33        if not {"indices", "values"}.issubset(input):
34            raise SparseValuesMissingKeysError(input)
35
36        indices = SparseValuesFactory._convert_to_list(input.get("indices"), int)
37        values = SparseValuesFactory._convert_to_list(input.get("values"), float)
38
39        if len(indices) != len(values):
40            raise ValueError("Sparse values indices and values must have the same length")
41
42        try:
43            return OpenApiSparseValues(indices=indices, values=values)
44        except TypeError as e:
45            raise SparseValuesTypeError() from e