11
11
from google .protobuf import json_format
12
12
from grpc ._channel import _InactiveRpcError , _MultiThreadedRendezvous
13
13
from tqdm import tqdm
14
+ import json
14
15
15
16
from pinecone import FetchResponse , QueryResponse , ScoredVector , SingleQueryResults , DescribeIndexStatsResponse
16
17
from pinecone .config import Config
@@ -81,9 +82,38 @@ def __init__(self, index_name: str, channel=None, grpc_config: GRPCClientConfig
81
82
"client-version" : CLIENT_VERSION
82
83
}
83
84
self ._endpoint_override = _endpoint_override
85
+
86
+ self .method_config = json .dumps (
87
+ {
88
+ "methodConfig" : [
89
+ {
90
+ "name" : [{"service" : "VectorService.Upsert" }],
91
+ "retryPolicy" : {
92
+ "maxAttempts" : 5 ,
93
+ "initialBackoff" : "0.1s" ,
94
+ "maxBackoff" : "1s" ,
95
+ "backoffMultiplier" : 2 ,
96
+ "retryableStatusCodes" : ["UNAVAILABLE" ],
97
+ },
98
+ },
99
+ {
100
+ "name" : [{"service" : "VectorService" }],
101
+ "retryPolicy" : {
102
+ "maxAttempts" : 5 ,
103
+ "initialBackoff" : "0.1s" ,
104
+ "maxBackoff" : "1s" ,
105
+ "backoffMultiplier" : 2 ,
106
+ "retryableStatusCodes" : ["UNAVAILABLE" ],
107
+ },
108
+ }
109
+ ]
110
+ }
111
+ )
112
+
84
113
self ._channel = channel or self ._gen_channel ()
85
114
self .stub = self .stub_class (self ._channel )
86
115
116
+
87
117
@property
88
118
@abstractmethod
89
119
def stub_class (self ):
@@ -97,7 +127,9 @@ def _gen_channel(self, options=None):
97
127
target = self ._endpoint ()
98
128
default_options = {
99
129
"grpc.max_send_message_length" : MAX_MSG_SIZE ,
100
- "grpc.max_receive_message_length" : MAX_MSG_SIZE
130
+ "grpc.max_receive_message_length" : MAX_MSG_SIZE ,
131
+ "grpc.service_config" : self .method_config ,
132
+ "grpc.enable_retries" : True
101
133
}
102
134
if self .grpc_client_config .secure :
103
135
default_options ['grpc.ssl_target_name_override' ] = target .split (':' )[0 ]
@@ -111,8 +143,8 @@ def _gen_channel(self, options=None):
111
143
root_cas = open (certifi .where (), "rb" ).read ()
112
144
tls = grpc .ssl_channel_credentials (root_certificates = root_cas )
113
145
channel = grpc .secure_channel (target , tls , options = _options )
114
- interceptor = RetryOnRpcErrorClientInterceptor ( self . retry_config )
115
- return grpc . intercept_channel ( channel , interceptor )
146
+
147
+ return channel
116
148
117
149
@property
118
150
def channel (self ):
@@ -243,7 +275,7 @@ def add_done_callback(self, fun):
243
275
244
276
def result (self , timeout = None ):
245
277
try :
246
- self ._delegate .result (timeout = timeout )
278
+ return self ._delegate .result (timeout = timeout )
247
279
except _MultiThreadedRendezvous as e :
248
280
raise PineconeException (e ._state .debug_error_string ) from e
249
281
0 commit comments