0.0.35

Feature: Updates to support real-time S3
Feature: Vector delete support
Feature: Neum Client included more get methods
Fix: Website and File connector response checks

Author: kevinco26

neumai/neumai/Client/NeumClient.py

@@ -42,6 +42,51 @@ def get_pipeline(self, pipeline_id:str):
         except Exception as e:
             print(f"Pipeline fetching failed. Exception - {e}")
 
+    def get_pipelines(self):
+        url = f"{self.endpoint}/pipelines/"
+
+        headers = {
+            "accept": "application/json",
+            "neum-api-key": self.api_key,
+            "content-type": "application/json"
+        }
+
+        try:
+            response = requests.get(url, headers=headers)
+            return json.loads(response.text)
+        except Exception as e:
+            print(f"Pipeline fetch failed. Exception - {e}")
+
+    def get_pipeline_runs(self, pipeline_id:str):
+        url = f"{self.endpoint}/pipelines/{pipeline_id}/runs"
+
+        headers = {
+            "accept": "application/json",
+            "neum-api-key": self.api_key,
+            "content-type": "application/json"
+        }
+
+        try:
+            response = requests.get(url, headers=headers)
+            return json.loads(response.text)
+        except Exception as e:
+            print(f"Pipeline runs fetch failed. Exception - {e}")
+
+    def get_pipeline_run(self, pipeline_id:str, pipeline_run_id:str):
+        url = f"{self.endpoint}/pipelines/{pipeline_id}/runs{pipeline_run_id}"
+
+        headers = {
+            "accept": "application/json",
+            "neum-api-key": self.api_key,
+            "content-type": "application/json"
+        }
+
+        try:
+            response = requests.get(url, headers=headers)
+            return json.loads(response.text)
+        except Exception as e:
+            print(f"Pipeline run fetch failed. Exception - {e}")
+
     def trigger_pipeline(self, pipeline_id:str, sync_type:TriggerSyncTypeEnum):
         url = f"{self.endpoint}/pipelines/{pipeline_id}/trigger"
 
@@ -75,7 +120,7 @@ def search_pipeline(self, pipeline_id:str, query:str, num_of_results:int = 3, tr
             response = requests.post(url, json=payload, headers=headers)
             return json.loads(response.text)
         except Exception as e:
-            print(f"Pipeline trigger failed. Exception - {e}")
+            print(f"Pipeline search failed. Exception - {e}")
 
     def search_file(self, pipeline_id:str, file_id:str, query:str, num_of_results:int = 3, track:bool = False, requested_by:str = None):
         url = f"{self.endpoint}/pipelines/{pipeline_id}/files/search?file_id={file_id}"
@@ -95,7 +140,7 @@ def search_file(self, pipeline_id:str, file_id:str, query:str, num_of_results:in
             response = requests.post(url, json=payload, headers=headers)
             return json.loads(response.text)
         except Exception as e:
-            print(f"Pipeline trigger failed. Exception - {e}")
+            print(f"File search failed. Exception - {e}")
 
     def get_files(self, pipeline_id:str):
         url = f"{self.endpoint}/pipelines/{pipeline_id}/files"
@@ -110,7 +155,7 @@ def get_files(self, pipeline_id:str):
             response = requests.get(url, headers=headers)
             return json.loads(response.text)
         except Exception as e:
-            print(f"Pipeline trigger failed. Exception - {e}")
+            print(f"Files fetch failed. Exception - {e}")
 
     def get_file(self, pipeline_id:str, file_id:str):
         url = f"{self.endpoint}/pipelines/{pipeline_id}/files?file_id={file_id}"
@@ -125,7 +170,7 @@ def get_file(self, pipeline_id:str, file_id:str):
             response = requests.get(url, headers=headers)
             return json.loads(response.text)
         except Exception as e:
-            print(f"Pipeline trigger failed. Exception - {e}")
+            print(f"File fetch failed. Exception - {e}")
 
     def get_retrievals_by_file_id(self, pipeline_id:str, file_id:str):
         url = f"{self.endpoint}/retrievals/{pipeline_id}/files?file_id={file_id}"
@@ -140,7 +185,7 @@ def get_retrievals_by_file_id(self, pipeline_id:str, file_id:str):
             response = requests.get(url, headers=headers)
             return json.loads(response.text)
         except Exception as e:
-            print(f"Pipeline trigger failed. Exception - {e}")
+            print(f"Retrievals fetch failed. Exception - {e}")
 
     def get_retrievals_by_pipeline_id(self, pipeline_id:str):
         url = f"{self.endpoint}/retrievals/{pipeline_id}"
@@ -155,7 +200,7 @@ def get_retrievals_by_pipeline_id(self, pipeline_id:str):
             response = requests.get(url, headers=headers)
             return json.loads(response.text)
         except Exception as e:
-            print(f"Pipeline trigger failed. Exception - {e}")
+            print(f"Retrievals fetch failed. Exception - {e}")
 
     def provide_retrieval_feedback(self, pipeline_id:str, retrieval_id:str, status:str):
         url = f"{self.endpoint}/retrievals/{pipeline_id}/{retrieval_id}"
@@ -173,4 +218,4 @@ def provide_retrieval_feedback(self, pipeline_id:str, retrieval_id:str, status:s
             response = requests.post(url, json=payload, headers=headers)
             return json.loads(response.text)
         except Exception as e:
-            print(f"Pipeline trigger failed. Exception - {e}")
+            print(f"Retrieval feedback failed. Exception - {e}")

neumai/neumai/DataConnectors/FileConnector.py

@@ -73,6 +73,8 @@ def connect_and_download(self, cloudFile:CloudFile) -> Generator[LocalFile, None
 
         headers = {"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.182 Safari/537.36"}
         response = requests.get(cloudFile.file_identifier, headers=headers)
+        if not response.ok:
+            raise NeumFileException(f"File can't be accessed. Please make sure it is publicly available.")     
 
         # Parse the URL to get the path
         path = urlparse(cloudFile.file_identifier).path
@@ -89,6 +91,8 @@ def config_validation(self) -> bool:
         import requests
         try:
             response = requests.get(self.url)
+            if not response.ok:
+                raise NeumFileException(f"File can't be accessed. Please make sure it is publicly available.")     
         except Exception as e:
             raise NeumFileException(f"Connection to file failed, check url. See Exception: {e}")     
         # Check for metadata

neumai/neumai/DataConnectors/WebsiteConnector.py

@@ -74,6 +74,8 @@ def connect_and_list_delta(self) -> Generator[CloudFile, None, None]:
     def connect_and_download(self, cloudFile:CloudFile) -> Generator[LocalFile, None, None]:
             headers = {"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.182 Safari/537.36"}
             response = requests.get(cloudFile.file_identifier, headers=headers)
+            if not response.ok:
+                raise WebsiteConnectionException(f"File can't be accessed. Please make sure it is publicly available.")     
             # Parse the HTML content
             soup = BeautifulSoup(response.content, 'html.parser')
             # Find the <body> element and extract its HTML content
@@ -94,7 +96,9 @@ def config_validation(self) -> bool:
 
         # Check to see that site exists
         try:
-            requests.get(self.url)
+            response = requests.get(self.url)
+            if not response.ok:
+                raise WebsiteConnectionException(f"File can't be accessed. Please make sure it is publicly available.")     
         except Exception as e:
             raise WebsiteConnectionException(f"Connection to website failed, check url. See Exception: {e}")      
         return True 

neumai/neumai/Pipelines/TriggerSyncTypeEnum.py

@@ -3,6 +3,7 @@
 class TriggerSyncTypeEnum(str, Enum):
     delta = "delta"
     full = "full"
+    event_based="event_based"
 
     def as_trigger_sync_type(trigger_sync_type: str):
         if trigger_sync_type == None or trigger_sync_type == "":

neumai/neumai/SinkConnectors/PineconeSink.py

@@ -64,6 +64,20 @@ def validation(self) -> bool:
             raise PineconeConnectionException(f"Pinecone connection couldn't be initialized. See exception: {e}")
         return True 
 
+    def delete_vectors_with_file_id(self, file_id: str) -> bool:
+        api_key =  self.api_key
+        environment = self.environment
+        index = self.index
+        namespace = self.namespace
+        if environment == "gcp-starter":
+            raise Exception("Pinecone does not support deleting vectors by metadata in the gcp starter environment")
+        pinecone.init(      
+            api_key=api_key,      
+            environment=environment)    
+        index = pinecone.Index(index)
+        index.delete(filter={"_file_entry_id": {"$eq": file_id}}, namespace=namespace)
+        return True
+
     def store(self, vectors_to_store:List[NeumVector]) -> int:
         api_key =  self.api_key
         environment = self.environment

neumai/neumai/SinkConnectors/QdrantSink.py

@@ -59,6 +59,9 @@ def validation(self) -> bool:
         )
         return True 
 
+    def delete_vectors_with_file_id(self, file_id: str) -> bool:
+        raise Exception("Qdrant doesn't have support to delete vectors by metadata")
+    
     def store(self, vectors_to_store:List[NeumVector]) -> int:
         url = self.url
         api_key = self.api_key

neumai/neumai/SinkConnectors/SingleStoreSink.py

@@ -62,6 +62,13 @@ def validation(self) -> bool:
             raise SinglestoreConnectionException(f"There was a problem connecting to Singlestore. See Exception: {e}")
         return True 
 
+    def delete_vectors_with_file_id(self, file_id: str) -> bool:
+        with s2.connect(self.url) as conn:
+            with conn.cursor() as cur:
+                delete_query = f"""DELETE FROM {self.table} WHERE _file_entry_id='{file_id}';"""
+                cur.execute(delete_query)
+        return True
+    
     def store(self, vectors_to_store:List[NeumVector]) -> int:
         batch_size = self.batch_size
         url = self.url

neumai/neumai/SinkConnectors/SinkConnector.py

@@ -35,6 +35,10 @@ def store(self, vectors_to_store:List[NeumVector]) -> int:
     def search(self, vector:List[float], number_of_results:int, filter:dict={}) -> List[NeumSearchResult]:
         """Search vectors for a given service"""
 
+    @abstractmethod
+    def delete_vectors_with_file_id(self, file_id:str ) -> bool:
+        """Deletes vectors for a specific file id"""
+    
     @abstractmethod
     def info(self) -> NeumSinkInfo:
         """Get information about what is stores in the sink"""

neumai/neumai/SinkConnectors/SupabaseSink.py

@@ -52,6 +52,19 @@ def validation(self) -> bool:
             raise SupabaseConnectionException(f"Supabase connection couldn't be initialized. See exception: {e}")
         return True 
 
+    def delete_vectors_with_file_id(self, file_id: str) -> bool:
+        database_connection = self.database_connection
+        vx = vecs.create_client(database_connection)
+        try:
+            collection_name = self.collection_name
+            db = vx.get_collection(name=collection_name)
+            db.delete(filters={"_file_entry_id": {"$eq": file_id}})
+        except Exception as e:
+            raise Exception(f"Supabase deletion failed. Exception {e}")
+        finally:
+            vx.disconnect()
+        return True
+    
     def store(self, vectors_to_store:List[NeumVector]) -> int:
         database_connection = self.database_connection
         vx = vecs.create_client(database_connection)

neumai/neumai/SinkConnectors/WeaviateSink.py

@@ -101,6 +101,26 @@ def _check_batch_result(self, results: Optional[List[dict[str, any]]], partial_f
                         partial_failure['latest_failure'] = result["result"]["errors"]["error"]
                         partial_failure['number_of_failures'] += 1
 
+    def delete_vectors_with_file_id(self, file_id: str) -> bool:
+        api_key = self.api_key
+        url = self.url
+        # Weaviate requires first letter to be capitalized
+        class_name = self.class_name
+        class_name = _capitalize_first_letter(class_name)
+        client = weaviate.Client(
+            url=url,
+            auth_client_secret=weaviate.AuthApiKey(api_key=api_key),
+        )
+        client.batch.delete_objects(
+            class_name=class_name,
+            where={
+                "path": ["_file_entry_id"],
+                "operator": "Equal",
+                "valueText": file_id
+            },
+        )
+        return True
+    
     def store(self, vectors_to_store:List[NeumVector]) -> Tuple[List, dict]:
         url = self.url
         num_workers = self.num_workers