Retry processing

backend/score.py

@@ -137,7 +137,8 @@ async def extract_knowledge_graph_from_file(
     allowedNodes=Form(None),
     allowedRelationship=Form(None),
     language=Form(None),
-    access_token=Form(None)
+    access_token=Form(None),
+    retry_condition=Form(None)
 ):
     """
     Calls 'extract_graph_from_file' in a new thread to create Neo4jGraph from a
@@ -161,30 +162,30 @@ async def extract_knowledge_graph_from_file(
             merged_file_path = os.path.join(MERGED_DIR,file_name)
             logging.info(f'File path:{merged_file_path}')
             result = await asyncio.to_thread(
-                extract_graph_from_file_local_file, uri, userName, password, database, model, merged_file_path, file_name, allowedNodes, allowedRelationship)
+                extract_graph_from_file_local_file, uri, userName, password, database, model, merged_file_path, file_name, allowedNodes, allowedRelationship, retry_condition)
 
         elif source_type == 's3 bucket' and source_url:
             result = await asyncio.to_thread(
-                extract_graph_from_file_s3, uri, userName, password, database, model, source_url, aws_access_key_id, aws_secret_access_key, allowedNodes, allowedRelationship)
+                extract_graph_from_file_s3, uri, userName, password, database, model, source_url, aws_access_key_id, aws_secret_access_key, file_name, allowedNodes, allowedRelationship, retry_condition)
 
         elif source_type == 'web-url':
             result = await asyncio.to_thread(
-                extract_graph_from_web_page, uri, userName, password, database, model, source_url, allowedNodes, allowedRelationship)
+                extract_graph_from_web_page, uri, userName, password, database, model, source_url, file_name, allowedNodes, allowedRelationship, retry_condition)
 
         elif source_type == 'youtube' and source_url:
             result = await asyncio.to_thread(
-                extract_graph_from_file_youtube, uri, userName, password, database, model, source_url, allowedNodes, allowedRelationship)
+                extract_graph_from_file_youtube, uri, userName, password, database, model, source_url, file_name, allowedNodes, allowedRelationship, retry_condition)
 
         elif source_type == 'Wikipedia' and wiki_query:
             result = await asyncio.to_thread(
-                extract_graph_from_file_Wikipedia, uri, userName, password, database, model, wiki_query, max_sources, language, allowedNodes, allowedRelationship)
+                extract_graph_from_file_Wikipedia, uri, userName, password, database, model, wiki_query, language, file_name, allowedNodes, allowedRelationship, retry_condition)
 
         elif source_type == 'gcs bucket' and gcs_bucket_name:
             result = await asyncio.to_thread(
-                extract_graph_from_file_gcs, uri, userName, password, database, model, gcs_project_id, gcs_bucket_name, gcs_bucket_folder, gcs_blob_filename, access_token, allowedNodes, allowedRelationship)
+                extract_graph_from_file_gcs, uri, userName, password, database, model, gcs_project_id, gcs_bucket_name, gcs_bucket_folder, gcs_blob_filename, access_token, file_name, allowedNodes, allowedRelationship, retry_condition)
         else:
             return create_api_response('Failed',message='source_type is other than accepted source')
-        
+
         if result is not None:
             result['db_url'] = uri
             result['api_name'] = 'extract'
@@ -626,6 +627,22 @@ async def merge_duplicate_nodes(uri=Form(), userName=Form(), password=Form(), da
         return create_api_response(job_status, message=message, error=error_message)
     finally:
         gc.collect()
+
+@app.post("/retry_processing")
+async def retry_processing(uri=Form(), userName=Form(), password=Form(), database=Form(), file_name=Form(), retry_condition=Form()):
+    try:
+        graph = create_graph_database_connection(uri, userName, password, database)
+        await asyncio.to_thread(set_status_retry, graph,file_name,retry_condition)
+        #set_status_retry(graph,file_name,retry_condition)
+        return create_api_response('Success',message=f"Status set to Reprocess for filename : {file_name}")
+    except Exception as e:
+        job_status = "Failed"
+        message="Unable to set status to Retry"
+        error_message = str(e)
+        logging.exception(f'{error_message}')
+        return create_api_response(job_status, message=message, error=error_message)
+    finally:
+        gc.collect()        
 
 if __name__ == "__main__":
     uvicorn.run(app)

backend/src/entities/source_node.py

@@ -24,3 +24,4 @@ class sourceNode:
     is_cancelled:bool=None
     processed_chunk:int=None
     access_token:str=None
+    retry_condition:str=None

backend/src/graphDB_dataAccess.py

@@ -71,10 +71,10 @@ def update_source_node(self, obj_source_node:sourceNode):
             if obj_source_node.processing_time is not None and obj_source_node.processing_time != 0:
                 params['processingTime'] = round(obj_source_node.processing_time.total_seconds(),2)
 
-            if obj_source_node.node_count is not None and obj_source_node.node_count != 0:
+            if obj_source_node.node_count is not None :
                 params['nodeCount'] = obj_source_node.node_count
 
-            if obj_source_node.relationship_count is not None and obj_source_node.relationship_count != 0:
+            if obj_source_node.relationship_count is not None :
                 params['relationshipCount'] = obj_source_node.relationship_count
 
             if obj_source_node.model is not None and obj_source_node.model != '':
@@ -86,11 +86,14 @@ def update_source_node(self, obj_source_node:sourceNode):
             if obj_source_node.total_chunks is not None and obj_source_node.total_chunks != 0:
                 params['total_chunks'] = obj_source_node.total_chunks
 
-            if obj_source_node.is_cancelled is not None and obj_source_node.is_cancelled != False:
+            if obj_source_node.is_cancelled is not None:
                 params['is_cancelled'] = obj_source_node.is_cancelled
 
-            if obj_source_node.processed_chunk is not None and obj_source_node.processed_chunk != 0:
+            if obj_source_node.processed_chunk is not None :
                 params['processed_chunk'] = obj_source_node.processed_chunk
+
+            if obj_source_node.retry_condition is not None :
+                params['retry_condition'] = obj_source_node.retry_condition    
 
             param= {"props":params}