pinecone.grpc.sparse_values_factory 

 1import numbers
 2
 3from collections.abc import Mapping
 4from typing import Union, Dict
 5
 6from ..utils import convert_to_list
 7
 8from ..data import (
 9    SparseValuesTypeError,
10    SparseValuesMissingKeysError,
11    SparseValuesDictionaryExpectedError,
12)
13
14from pinecone.core.grpc.protos.vector_service_pb2 import (
15    SparseValues as GRPCSparseValues,
16)
17from pinecone import SparseValues as NonGRPCSparseValues
18
19
20class SparseValuesFactory:
21    @staticmethod
22    def build(input: Union[Dict, GRPCSparseValues, NonGRPCSparseValues]) -> GRPCSparseValues:
23        if input is None:
24            return input
25        if isinstance(input, GRPCSparseValues):
26            return input
27        if isinstance(input, NonGRPCSparseValues):
28            return GRPCSparseValues(indices=input.indices, values=input.values)
29        if not isinstance(input, Mapping):
30            raise SparseValuesDictionaryExpectedError(input)
31        if not {"indices", "values"}.issubset(input):
32            raise SparseValuesMissingKeysError(input)
33
34        indices = SparseValuesFactory._convert_to_list(input.get("indices"), int)
35        values = SparseValuesFactory._convert_to_list(input.get("values"), float)
36
37        if len(indices) != len(values):
38            raise ValueError("Sparse values indices and values must have the same length")
39
40        try:
41            return GRPCSparseValues(indices=indices, values=values)
42        except TypeError as e:
43            raise SparseValuesTypeError() from e
44
45    @staticmethod
46    def _convert_to_list(input, expected_type):
47        try:
48            converted = convert_to_list(input)
49        except TypeError as e:
50            raise SparseValuesTypeError() from e
51
52        SparseValuesFactory._validate_list_items_type(converted, expected_type)
53        return converted
54
55    @staticmethod
56    def _validate_list_items_type(input, expected_type):
57        if len(input) > 0 and not isinstance(input[0], expected_type):
58            raise SparseValuesTypeError()
class SparseValuesFactory: 
21class SparseValuesFactory:
22    @staticmethod
23    def build(input: Union[Dict, GRPCSparseValues, NonGRPCSparseValues]) -> GRPCSparseValues:
24        if input is None:
25            return input
26        if isinstance(input, GRPCSparseValues):
27            return input
28        if isinstance(input, NonGRPCSparseValues):
29            return GRPCSparseValues(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 GRPCSparseValues(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()
@staticmethod
def build( input: Union[Dict, vector_service_pb2.SparseValues, pinecone.core.openapi.data.model.sparse_values.SparseValues]) -> vector_service_pb2.SparseValues: 
22    @staticmethod
23    def build(input: Union[Dict, GRPCSparseValues, NonGRPCSparseValues]) -> GRPCSparseValues:
24        if input is None:
25            return input
26        if isinstance(input, GRPCSparseValues):
27            return input
28        if isinstance(input, NonGRPCSparseValues):
29            return GRPCSparseValues(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 GRPCSparseValues(indices=indices, values=values)
43        except TypeError as e:
44            raise SparseValuesTypeError() from e