pinecone.data.vector_factory

 1import numbers
 2
 3from collections.abc import Iterable, Mapping
 4from typing import Union, Tuple, Dict
 5
 6from ..utils import fix_tuple_length, convert_to_list
 7from ..utils.constants import REQUIRED_VECTOR_FIELDS, OPTIONAL_VECTOR_FIELDS
 8from .sparse_vector_factory import SparseValuesFactory
 9
10from pinecone.core.openapi.data.models import Vector, SparseValues
11
12from .errors import (
13    VectorDictionaryMissingKeysError,
14    VectorDictionaryExcessKeysError,
15    VectorTupleLengthError,
16    MetadataDictionaryExpectedError,
17)
18
19
20class VectorFactory:
21    @staticmethod
22    def build(item: Union[Vector, Tuple, Dict], check_type: bool = True) -> Vector:
23        if isinstance(item, Vector):
24            return item
25        elif isinstance(item, tuple):
26            return VectorFactory._tuple_to_vector(item, check_type)
27        elif isinstance(item, Mapping):
28            return VectorFactory._dict_to_vector(item, check_type)
29        else:
30            raise ValueError(f"Invalid vector value passed: cannot interpret type {type(item)}")
31
32    @staticmethod
33    def _tuple_to_vector(item, check_type: bool) -> Vector:
34        if len(item) < 2 or len(item) > 3:
35            raise VectorTupleLengthError(item)
36        id, values, metadata = fix_tuple_length(item, 3)
37        if isinstance(values, SparseValues):
38            raise ValueError(
39                "Sparse values are not supported in tuples. Please use either dicts or Vector objects as inputs."
40            )
41        else:
42            return Vector(
43                id=id,
44                values=convert_to_list(values),
45                metadata=metadata or {},
46                _check_type=check_type,
47            )
48
49    @staticmethod
50    def _dict_to_vector(item, check_type: bool) -> Vector:
51        item_keys = set(item.keys())
52        if not item_keys.issuperset(REQUIRED_VECTOR_FIELDS):
53            raise VectorDictionaryMissingKeysError(item)
54
55        excessive_keys = item_keys - (REQUIRED_VECTOR_FIELDS | OPTIONAL_VECTOR_FIELDS)
56        if len(excessive_keys) > 0:
57            raise VectorDictionaryExcessKeysError(item)
58
59        values = item.get("values")
60        if "values" in item:
61            item["values"] = convert_to_list(values)
62
63        sparse_values = item.get("sparse_values")
64        if sparse_values is None:
65            item.pop("sparse_values", None)
66        else:
67            item["sparse_values"] = SparseValuesFactory.build(sparse_values)
68
69        metadata = item.get("metadata")
70        if metadata and not isinstance(metadata, Mapping):
71            raise MetadataDictionaryExpectedError(item)
72
73        try:
74            return Vector(**item, _check_type=check_type)
75        except TypeError as e:
76            if not isinstance(item["values"], Iterable) or not isinstance(
77                item["values"].__iter__().__next__(), numbers.Real
78            ):
79                raise TypeError(f"Column `values` is expected to be a list of floats")
80            raise e