pinecone.core.utils

  1import inspect
  2import logging
  3import re
  4import uuid
  5import warnings
  6from pathlib import Path
  7from typing import List
  8
  9import requests
 10import urllib3
 11
 12try:
 13    from pinecone.core.grpc.protos import vector_column_service_pb2
 14    from google.protobuf.struct_pb2 import Struct
 15    from google.protobuf import json_format
 16    import numpy as np
 17    import lz4.frame
 18except Exception:
 19    pass  # ignore for non-[grpc] installations
 20
 21DNS_COMPATIBLE_REGEX = re.compile("^[a-z0-9]([a-z0-9]|[-])+[a-z0-9]$")
 22
 23
 24def dump_numpy_public(np_array: "np.ndarray", compressed: bool = False) -> "vector_column_service_pb2.NdArray":
 25    """
 26    Dump numpy array to vector_column_service_pb2.NdArray
 27    """
 28    warn_deprecated(
 29        "dump_numpy_public and all numpy-related features will be removed in a future version",
 30        deprecated_in="2.2.1",
 31        removal_in="3.0.0",
 32    )
 33    protobuf_arr = vector_column_service_pb2.NdArray()
 34    protobuf_arr.dtype = str(np_array.dtype)
 35    protobuf_arr.shape.extend(np_array.shape)
 36    if compressed:
 37        protobuf_arr.buffer = lz4.frame.compress(np_array.tobytes())
 38        protobuf_arr.compressed = True
 39    else:
 40        protobuf_arr.buffer = np_array.tobytes()
 41    return protobuf_arr
 42
 43
 44def dump_strings_public(strs: List[str], compressed: bool = False) -> "vector_column_service_pb2.NdArray":
 45    warn_deprecated(
 46        "dump_strings_public and all numpy-related features will be removed in a future version",
 47        deprecated_in="2.2.1",
 48        removal_in="3.0.0",
 49    )
 50    return dump_numpy_public(np.array(strs, dtype="S"), compressed=compressed)
 51
 52
 53def get_version():
 54    return Path(__file__).parent.parent.parent.joinpath("__version__").read_text().strip()
 55
 56
 57def get_environment():
 58    return Path(__file__).parent.parent.parent.joinpath("__environment__").read_text().strip()
 59
 60
 61def validate_dns_name(name):
 62    if not DNS_COMPATIBLE_REGEX.match(name):
 63        raise ValueError(
 64            "{} is invalid - service names and node names must consist of lower case "
 65            "alphanumeric characters or '-', start with an alphabetic character, and end with an "
 66            "alphanumeric character (e.g. 'my-name', or 'abc-123')".format(name)
 67        )
 68
 69
 70def _generate_request_id() -> str:
 71    return str(uuid.uuid4())
 72
 73
 74def fix_tuple_length(t, n):
 75    """Extend tuple t to length n by adding None items at the end of the tuple. Return the new tuple."""
 76    return t + ((None,) * (n - len(t))) if len(t) < n else t
 77
 78
 79def get_user_agent():
 80    client_id = f"python-client-{get_version()}"
 81    user_agent_details = {"requests": requests.__version__, "urllib3": urllib3.__version__}
 82    user_agent = "{} ({})".format(client_id, ", ".join([f"{k}:{v}" for k, v in user_agent_details.items()]))
 83    return user_agent
 84
 85
 86def dict_to_proto_struct(d: dict) -> "Struct":
 87    if not d:
 88        d = {}
 89    s = Struct()
 90    s.update(d)
 91    return s
 92
 93
 94def proto_struct_to_dict(s: "Struct") -> dict:
 95    return json_format.MessageToDict(s)
 96
 97
 98def load_numpy_public(proto_arr: "vector_column_service_pb2.NdArray") -> "np.ndarray":
 99    """
100    Load numpy array from protobuf
101    :param proto_arr:
102    :return:
103    """
104    warn_deprecated(
105        "load_numpy_public and all numpy-related features will be removed in a future version",
106        deprecated_in="2.2.1",
107        removal_in="3.0.0",
108    )
109    if len(proto_arr.shape) == 0:
110        return np.array([])
111    if proto_arr.compressed:
112        numpy_arr = np.frombuffer(lz4.frame.decompress(proto_arr.buffer), dtype=proto_arr.dtype)
113    else:
114        numpy_arr = np.frombuffer(proto_arr.buffer, dtype=proto_arr.dtype)
115    return numpy_arr.reshape(proto_arr.shape)
116
117
118def load_strings_public(proto_arr: "vector_column_service_pb2.NdArray") -> List[str]:
119    warn_deprecated(
120        "load_strings_public and all numpy-related features will be removed in a future version",
121        deprecated_in="2.2.1",
122        removal_in="3.0.0",
123    )
124    return [str(item, "utf-8") for item in load_numpy_public(proto_arr)]
125
126
127def warn_deprecated(description: str = "", deprecated_in: str = None, removal_in: str = None):
128    message = f"DEPRECATED since v{deprecated_in} [Will be removed in v{removal_in}]: {description}"
129    warnings.warn(message, FutureWarning)
130
131
132def check_kwargs(caller, given):
133    argspec = inspect.getfullargspec(caller)
134    diff = set(given).difference(argspec.args)
135    if diff:
136        logging.exception(caller.__name__ + " had unexpected keyword argument(s): " + ", ".join(diff), exc_info=False)