Upsert df - support non async, add reponse

1. Return the proper result type (UpsertResponse) with the total upserted number.
2. For gRPC - support both async and non-async modes

Author: igiloh-pinecone

pinecone/core/grpc/index_grpc.py

@@ -439,12 +439,21 @@ def upsert_dataframe(self,
         if not isinstance(df, pd.DataFrame):
             raise ValueError(f"Only pandas dataframes are supported. Found: {type(df)}")
 
-        async_results = [
-            self.upsert(vectors=chunk, namespace=namespace, async_req=True)
-            for chunk in tqdm(self._iter_dataframe(df, batch_size=batch_size),
-                              total=len(df) // batch_size, disable=not show_progress)
-        ]
-        res = [async_result.result() for async_result in async_results]
+        pbar = tqdm(total=len(df), disable=not show_progress, desc="sending upsert requests")
+        results = []
+        for chunk in self._iter_dataframe(df, batch_size=batch_size):
+            res = self.upsert(vectors=chunk, namespace=namespace, async_req=use_async_requests)
+            pbar.update(len(chunk))
+            results.append(res)
+
+        if use_async_requests:
+            results = [async_result.result() for async_result in tqdm(results, desc="collecting async responses")]
+
+        upserted_count = 0
+        for res in results:
+            upserted_count += res.upserted_count
+
+        return UpsertResponse(upserted_count=upserted_count)
 
     @staticmethod
     def _iter_dataframe(df, batch_size):

pinecone/index.py

@@ -214,12 +214,16 @@ def upsert_dataframe(self,
         if not isinstance(df, pd.DataFrame):
             raise ValueError(f"Only pandas dataframes are supported. Found: {type(df)}")
 
+        upserted_count = 0
         pbar = tqdm(total=len(df), disable=not show_progress)
         for i in range(0, len(df), batch_size):
             batch = df.iloc[i:i + batch_size].to_dict(orient="records")
-            self.upsert(batch, namespace=namespace)
+            res = self.upsert(batch, namespace=namespace)
+            upserted_count += res.upserted_count
             pbar.update(len(batch))
 
+        return UpsertResponse(upserted_count=upserted_count)
+
     @validate_and_convert_errors
     def delete(self,
                ids: Optional[List[str]] = None,