update

Author: jgbradley1

backend/src/api/common.py

@@ -5,6 +5,7 @@
 import os
 import re
 
+from azure.cosmos import exceptions
 from azure.identity import DefaultAzureCredential
 from fastapi import HTTPException
 
@@ -37,6 +38,23 @@ def delete_blob_container(container_name: str):
         blob_service_client.delete_container(container_name)
 
 
+def delete_cosmos_container_item(container: str, item_id: str):
+    """
+    Delete an item from a cosmosdb container. If it does not exist, do nothing.
+    If exception is raised, the calling function should catch it.
+    """
+    azure_client_manager = AzureClientManager()
+    # cosmos_client = azure_client_manager.get_cosmos_client()
+    try:
+        azure_client_manager.get_cosmos_container_client(
+            "graphrag", container
+        ).delete_item(item_id)
+        # cosmos_client.get_database_client(database).get_container_client(container).delete_item(item_id)
+    except exceptions.CosmosResourceNotFoundError:
+        # If item does not exist, do nothing
+        pass
+
+
 def validate_index_file_exist(index_name: str, file_name: str):
     """
     Check if index exists and that the specified blob file exists.

backend/src/api/data.py

@@ -6,7 +6,6 @@
 from math import ceil
 from typing import List
 
-from azure.cosmos import exceptions
 from azure.storage.blob import ContainerClient
 from fastapi import (
     APIRouter,
@@ -17,6 +16,7 @@
 from src.api.azure_clients import AzureClientManager
 from src.api.common import (
     delete_blob_container,
+    delete_cosmos_container_item,
     sanitize_name,
     validate_blob_container_name,
 )
@@ -128,7 +128,6 @@ async def upload_files(
     Raises:
         HTTPException: If the container name is invalid or if any error occurs during the upload process.
     """
-    reporter = LoggerSingleton().get_instance()
     sanitized_storage_name = sanitize_name(storage_name)
     # ensure container name follows Azure Blob Storage naming conventions
     try:
@@ -171,7 +170,8 @@ async def upload_files(
         })
         return BaseResponse(status="File upload successful.")
     except Exception:
-        reporter.on_error("Error uploading files.", details={"files": files})
+        logger = LoggerSingleton().get_instance()
+        logger.on_error("Error uploading files.", details={"files": files})
         raise HTTPException(
             status_code=500,
             detail=f"Error uploading files to container '{storage_name}'.",
@@ -188,25 +188,16 @@ async def delete_files(storage_name: str):
     """
     Delete a specified data storage container.
     """
-    azure_client_manager = AzureClientManager()
+    # azure_client_manager = AzureClientManager()
     sanitized_storage_name = sanitize_name(storage_name)
     try:
         # delete container in Azure Storage
         delete_blob_container(sanitized_storage_name)
         # delete entry from container-store in cosmosDB
-        container_store_client = azure_client_manager.get_cosmos_container_client(
-            database_name="graphrag", container_name="container-store"
-        )
-        try:
-            container_store_client.delete_item(
-                item=sanitized_storage_name,
-                partition_key=sanitized_storage_name,
-            )
-        except exceptions.CosmosResourceNotFoundError:
-            pass
+        delete_cosmos_container_item("container-store", sanitized_storage_name)
     except Exception:
-        reporter = LoggerSingleton().get_instance()
-        reporter.on_error(
+        logger = LoggerSingleton().get_instance()
+        logger.on_error(
             f"Error deleting container {storage_name}.",
             details={"Container": storage_name},
         )

backend/src/api/graph.py

@@ -43,8 +43,8 @@ async def get_graphml_file(index_name: str):
             headers={"Content-Disposition": f"attachment; filename={graphml_filename}"},
         )
     except Exception:
-        reporter = LoggerSingleton().get_instance()
-        reporter.on_error("Could not retrieve graphml file")
+        logger = LoggerSingleton().get_instance()
+        logger.on_error("Could not retrieve graphml file")
         raise HTTPException(
             status_code=500,
             detail=f"Could not retrieve graphml file for index '{index_name}'.",

backend/src/api/index.py

@@ -170,7 +170,7 @@ async def _start_indexing_pipeline(index_name: str):
         "type": "index",
     })
 
-    reporter = LoggerSingleton().get_instance()
+    logger = LoggerSingleton().get_instance()
     pipelinejob = PipelineJob()
     pipeline_job = pipelinejob.load_item(sanitized_index_name)
     sanitized_storage_name = pipeline_job.sanitized_storage_name
@@ -229,19 +229,19 @@ async def _start_indexing_pipeline(index_name: str):
     for workflow in pipeline_config.workflows:
         pipeline_job.all_workflows.append(workflow.name)
 
-    # create new reporters/callbacks just for this job
-    reporters = []
-    reporter_names = os.getenv("REPORTERS", Reporters.CONSOLE.name.upper()).split(",")
-    for reporter_name in reporter_names:
+    # create new loggers/callbacks just for this job
+    loggers = []
+    logger_names = os.getenv("REPORTERS", Reporters.CONSOLE.name.upper()).split(",")
+    for logger_name in logger_names:
         try:
-            reporters.append(Reporters[reporter_name.upper()])
+            loggers.append(Reporters[logger_name.upper()])
         except KeyError:
-            raise ValueError(f"Unknown reporter type: {reporter_name}")
+            raise ValueError(f"Unknown logger type: {logger_name}")
     workflow_callbacks = load_pipeline_logger(
         index_name=index_name,
         num_workflow_steps=len(pipeline_job.all_workflows),
         reporting_dir=sanitized_index_name,
-        reporters=reporters,
+        reporters=loggers,
     )
 
     # add pipeline job callback to the callback manager
@@ -305,7 +305,7 @@ async def _start_indexing_pipeline(index_name: str):
             details=error_details,
         )
         # log error in global index directory logs
-        reporter.on_error(
+        logger.on_error(
             message=f"Indexing pipeline failed for index '{index_name}'.",
             cause=e,
             stack=traceback.format_exc(),
@@ -337,8 +337,8 @@ async def get_all_indexes():
             if item["type"] == "index":
                 items.append(item["human_readable_name"])
     except Exception:
-        reporter = LoggerSingleton().get_instance()
-        reporter.on_error("Error retrieving index names")
+        logger = LoggerSingleton().get_instance()
+        logger.on_error("Error retrieving index names")
     return IndexNameList(index_name=items)
 
 
@@ -363,13 +363,13 @@ def _delete_k8s_job(job_name: str, namespace: str) -> None:
     # function should only work when running in AKS
     if not os.getenv("KUBERNETES_SERVICE_HOST"):
         return None
-    reporter = LoggerSingleton().get_instance()
+    logger = LoggerSingleton().get_instance()
     kubernetes_config.load_incluster_config()
     try:
         batch_v1 = kubernetes_client.BatchV1Api()
         batch_v1.delete_namespaced_job(name=job_name, namespace=namespace)
     except Exception:
-        reporter.on_error(
+        logger.on_error(
             message=f"Error deleting k8s job {job_name}.",
             details={"container": job_name},
         )
@@ -380,7 +380,7 @@ def _delete_k8s_job(job_name: str, namespace: str) -> None:
         if job_pod:
             core_v1.delete_namespaced_pod(job_pod, namespace=namespace)
     except Exception:
-        reporter.on_error(
+        logger.on_error(
             message=f"Error deleting k8s pod for job {job_name}.",
             details={"container": job_name},
         )
@@ -442,8 +442,8 @@ async def delete_index(index_name: str):
             index_client.delete_index(ai_search_index_name)
 
     except Exception:
-        reporter = LoggerSingleton().get_instance()
-        reporter.on_error(
+        logger = LoggerSingleton().get_instance()
+        logger.on_error(
             message=f"Error encountered while deleting all data for index {index_name}.",
             stack=None,
             details={"container": index_name},

backend/src/api/index_configuration.py

@@ -69,11 +69,11 @@ async def generate_prompts(storage_name: str, limit: int = 5):
             output=f"{temp_dir}/prompts",
         )
     except Exception as e:
-        reporter = LoggerSingleton().get_instance()
+        logger = LoggerSingleton().get_instance()
         error_details = {
             "storage_name": storage_name,
         }
-        reporter.on_error(
+        logger.on_error(
             message="Auto-prompt generation failed.",
             cause=e,
             stack=traceback.format_exc(),

backend/src/api/query.py

@@ -197,8 +197,8 @@ async def global_query(request: GraphRequest):
 
         return GraphResponse(result=result[0], context_data=context_data)
     except Exception as e:
-        reporter = LoggerSingleton().get_instance()
-        reporter.on_error(
+        logger = LoggerSingleton().get_instance()
+        logger.on_error(
             message="Could not perform global search.",
             cause=e,
             stack=traceback.format_exc(),

backend/src/api/query_streaming.py

@@ -186,8 +186,8 @@ async def global_search_streaming(request: GraphRequest):
             media_type="application/json",
         )
     except Exception as e:
-        reporter = LoggerSingleton().get_instance()
-        reporter.on_error(
+        logger = LoggerSingleton().get_instance()
+        logger.on_error(
             message="Error encountered while streaming global search response",
             cause=e,
             stack=traceback.format_exc(),
@@ -429,8 +429,8 @@ async def local_search_streaming(request: GraphRequest):
             media_type="application/json",
         )
     except Exception as e:
-        reporter = LoggerSingleton().get_instance()
-        reporter.on_error(
+        logger = LoggerSingleton().get_instance()
+        logger.on_error(
             message="Error encountered while streaming local search response",
             cause=e,
             stack=traceback.format_exc(),

backend/src/api/source.py

@@ -55,8 +55,8 @@ async def get_report_info(index_name: str, report_id: str):
         ][0]
         return ReportResponse(text=report_content)
     except Exception:
-        reporter = LoggerSingleton().get_instance()
-        reporter.on_error("Could not get report.")
+        logger = LoggerSingleton().get_instance()
+        logger.on_error("Could not get report.")
         raise HTTPException(
             status_code=500,
             detail=f"Error retrieving report '{report_id}' from index '{index_name}'.",
@@ -97,8 +97,8 @@ async def get_chunk_info(index_name: str, text_unit_id: str):
             text=row["chunk"].values[0], source_document=row["source_doc"].values[0]
         )
     except Exception:
-        reporter = LoggerSingleton().get_instance()
-        reporter.on_error("Could not get text chunk.")
+        logger = LoggerSingleton().get_instance()
+        logger.on_error("Could not get text chunk.")
         raise HTTPException(
             status_code=500,
             detail=f"Error retrieving text chunk '{text_unit_id}' from index '{index_name}'.",
@@ -127,8 +127,8 @@ async def get_entity_info(index_name: str, entity_id: int):
             text_units=row["text_unit_ids"].values[0].tolist(),
         )
     except Exception:
-        reporter = LoggerSingleton().get_instance()
-        reporter.on_error("Could not get entity")
+        logger = LoggerSingleton().get_instance()
+        logger.on_error("Could not get entity")
         raise HTTPException(
             status_code=500,
             detail=f"Error retrieving entity '{entity_id}' from index '{index_name}'.",
@@ -172,8 +172,8 @@ async def get_claim_info(index_name: str, claim_id: int):
             document_ids=row["document_ids"].values[0].tolist(),
         )
     except Exception:
-        reporter = LoggerSingleton().get_instance()
-        reporter.on_error("Could not get claim.")
+        logger = LoggerSingleton().get_instance()
+        logger.on_error("Could not get claim.")
         raise HTTPException(
             status_code=500,
             detail=f"Error retrieving claim '{claim_id}' from index '{index_name}'.",
@@ -218,8 +218,8 @@ async def get_relationship_info(index_name: str, relationship_id: int):
             ],  # extract text_unit_ids from a list of panda series
         )
     except Exception:
-        reporter = LoggerSingleton().get_instance()
-        reporter.on_error("Could not get relationship.")
+        logger = LoggerSingleton().get_instance()
+        logger.on_error("Could not get relationship.")
         raise HTTPException(
             status_code=500,
             detail=f"Error retrieving relationship '{relationship_id}' from index '{index_name}'.",

backend/tests/conftest.py

@@ -76,7 +76,7 @@ def container_with_graphml_file(blob_service_client, cosmos_client):
     yield container_name
     # cleanup
     blob_service_client.delete_container(sanitized_name)
-    container_store_client.delete_item(sanitized_name, sanitized_name)
+    # container_store_client.delete_item(sanitized_name, sanitized_name)
 
 
 @pytest.fixture(scope="session")
@@ -120,7 +120,7 @@ def container_with_index_files(blob_service_client, cosmos_client):
     yield container_name
     # cleanup
     blob_service_client.delete_container(sanitized_name)
-    container_store_client.delete_item(sanitized_name, sanitized_name)
+    # container_store_client.delete_item(sanitized_name, sanitized_name)
 
 
 @pytest.fixture(scope="session")