0.0.36 (#37)

Feature:  client updates for retrieval gets

Feature: Add processing time to object

Fix: fix for pipeline run

Fix: fix for weaviate class name

Author: ddematheu

neumai/neumai/Client/NeumClient.py

@@ -202,6 +202,51 @@ def get_retrievals_by_pipeline_id(self, pipeline_id:str):
         except Exception as e:
             print(f"Retrievals fetch failed. Exception - {e}")
 
+    def get_retrievals_by_pipeline_id_user_id(self, pipeline_id:str, user_id:str):
+        url = f"{self.endpoint}/retrievals/{pipeline_id}/user/{user_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"Retrievals fetch failed. Exception - {e}")
+    
+    def get_retrievals_by_user_id(self, user_id:str):
+        url = f"{self.endpoint}/retrievals/user/{user_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"Retrievals fetch failed. Exception - {e}")
+
+    def get_retrievals_by_file_id_user_id(self, pipeline_id:str, file_id:str, user_id:str):
+        url = f"{self.endpoint}/retrievals/{pipeline_id}/files?file_id={file_id}&user_id={user_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"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}"
 

neumai/neumai/Pipelines/PipelineRun.py

@@ -58,6 +58,7 @@ class PipelineRun(BaseModel):
     last_updated: Optional[float] = None
     number_of_documents: Optional[int] = None
     finished_distributing: bool = False
+    processing_time: Optional[float] = None
 
     def set_id(self, id: str):
         self.id = id
@@ -76,6 +77,9 @@ def set_pipeline_id(self, pipeline_id: str):
 
     def set_created(self, created: float):
         self.created = created
+
+    def set_processing_time(self, processing_time: float):
+        self.processing_time = processing_time
 
     def as_pipeline_run(dct:dict):
         if dct == None:
@@ -91,5 +95,6 @@ def as_pipeline_run(dct:dict):
             task_details=PipelineRunTaskDetails.as_pipeline_run_task_details(dct.get("task_details", None)),
             last_updated=dct.get("last_updated", None),
             number_of_documents=dct.get("number_of_documents",None),
-            finished_distributing=dct.get("finished_distributing",None) # we could do something like in distributed tasks where we store the state of the DAG in the pipeline run object.. for now just doing finished_distributing
+            finished_distributing=dct.get("finished_distributing",None),# we could do something like in distributed tasks where we store the state of the DAG in the pipeline run object.. for now just doing finished_distributing
+            processing_time=dct.get("processing_time", None)
         )

neumai/neumai/SinkConnectors/WeaviateSink.py

@@ -105,7 +105,7 @@ 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 = self.class_name.replace("-","_")
         class_name = _capitalize_first_letter(class_name)
         client = weaviate.Client(
             url=url,
@@ -128,7 +128,8 @@ def store(self, vectors_to_store:List[NeumVector]) -> Tuple[List, dict]:
         batch_size = self.batch_size
         is_dynamic_batch = self.is_dynamic_batch
         batch_connection_error_retries = self.batch_connection_error_retries
-        class_name = self.class_name
+        class_name = self.class_name.replace("-","_")
+        class_name = _capitalize_first_letter(class_name)
         partial_failure = {'did_fail': False, 'latest_failure': None, 'number_of_failures': 0}
 
         if 'https' not in url:
@@ -174,7 +175,7 @@ def search(self, vector: List[float], number_of_results: int, filter:dict={}) ->
         api_key = self.api_key
         url = self.url
         # Weaviate requires first letter to be capitalized
-        class_name = self.class_name
+        class_name = self.class_name.replace("-","_")
         class_name = _capitalize_first_letter(class_name)
         client = weaviate.Client(
             url=url,
@@ -215,7 +216,7 @@ def info(self) -> NeumSinkInfo:
         api_key = self.api_key
         url = self.url
 
-        class_name = self.class_name
+        class_name = self.class_name.replace("-","_")
         class_name = _capitalize_first_letter(class_name)
         client = weaviate.Client(
             url=url,

neumai/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "neumai"
-version = "0.0.35"
+version = "0.0.36"
 description = "Package containing connectors for Neum AI."
 authors = ["David de Matheu <david@tryneum.com>"]
 license = "MIT"