Enhanced Ingestion with Named Graph Restrictions

ingestion_service/producer/core/configuration.py

@@ -62,7 +62,10 @@ def load_environment(env_name="env"):
         "RABBITMQ_PASSWORD": os.getenv("RABBITMQ_PASSWORD"),
         "RABBITMQ_URL": os.getenv("RABBITMQ_URL", "localhost"),
         "RABBITMQ_PORT": os.getenv("RABBITMQ_PORT", 5672),
-        "RABBITMQ_VHOST": os.getenv("RABBITMQ_VHOST","/")
+        "RABBITMQ_VHOST": os.getenv("RABBITMQ_VHOST","/"),
+
+        #query service
+        "QUERY_SERVICE_BASE_URL": os.getenv("QUERY_SERVICE_BASE_URL", "localhost:8010")
     }
 
 

ingestion_service/producer/core/pydantic_schema.py

@@ -37,6 +37,7 @@ class Config:
 
 class InputJSONSLdchema(BaseModel):
     user: str
+    graph: str
     kg_data: Dict[Any, Any]
 
 

ingestion_service/producer/core/shared.py

@@ -21,7 +21,7 @@
 from rdflib import Graph
 import requests
 import logging
-
+from core.configuration import load_environment
 logger = logging.getLogger(__name__)
 
 # Helper function to resolve issues during the conversion from JSON-LD to Turtle representation.
@@ -121,3 +121,61 @@ def is_valid_jsonld(jsonld_str):
     except ValueError:
         return False
 
+def check_url_for_slash(url:str):
+    if not url.endswith("/"):
+        return url + "/"
+    return url
+
+def check_if_url_wellformed(url:str):
+    "We want to ensure that the name graph IRI is wellformed, i.e., starts with http or https, not www"
+    if url is None:
+        return False
+    else:
+        return True if url.startswith("http://") or  url.startswith("https://") else False
+
+
+import requests
+
+
+def named_graph_exists(named_graph_iri: str) -> dict:
+    """
+    Checks whether a named graph exists in the registered named graphs list.
+
+    Args:
+        named_graph_iri (str): The IRI of the named graph to check.
+
+    Returns:
+        dict: A dictionary indicating success or failure with a relevant message.
+    """
+
+    query_service_url = load_environment().get("QUERY_SERVICE_BASE_URL", "")
+    endpoint = f"{check_url_for_slash(query_service_url)}query/registered-named-graphs"
+
+    # Validate the named graph IRI
+    print(check_if_url_wellformed(named_graph_iri))
+    if not check_if_url_wellformed(named_graph_iri):
+        return {
+            "status": "error",
+            "message": "The graph IRI is not well-formed. It should start with 'http' or 'https'."
+        }
+
+    try:
+        response = requests.get(endpoint)
+        response.raise_for_status()  # Raise an error for bad responses (4xx, 5xx)
+
+        registered_graphs = response.json()
+        formatted_iri= check_url_for_slash(named_graph_iri)
+        if formatted_iri in registered_graphs:
+            return {
+                "status": True,
+                "formatted_iri": formatted_iri
+            }
+        return {
+                "status": False,
+                "message": f"The graph is not registered. Available graphs: {list(registered_graphs.keys())}"
+            }
+    except requests.exceptions.RequestException as e:
+        return {
+            "status": "error",
+            "message": f"Error connecting to query service: {str(e)}"
+        }

ingestion_service/producer/requirements.txt

@@ -17,4 +17,6 @@ psycopg==3.1.18
 python-jose==3.3.0
 python-multipart==0.0.9
 passlib[bcrypt]==1.7.4
-asyncpg==0.29.0
+asyncpg==0.29.0
+rdflib
+PyLD

ingestion_service/worker/core/configuration.py

@@ -62,7 +62,10 @@ def load_environment(env_name="env"):
         "RABBITMQ_URL": os.getenv("RABBITMQ_URL", "localhost"),
         "RABBITMQ_PORT": os.getenv("RABBITMQ_PORT", 5672),
         "RABBITMQ_VHOST": os.getenv("RABBITMQ_VHOST","/"),
-        "INGEST_URL": os.getenv("INGEST_URL")
+        "INGEST_URL": os.getenv("INGEST_URL"),
+        "JWT_LOGIN_EMAIL": os.getenv("JWT_LOGIN_EMAIL"),
+        "JWT_LOGIN_PASSWORD": os.getenv("JWT_LOGIN_PASSWORD"),
+        "JWT_BEARER_TOKEN_URL": os.getenv("JWT_BEARER_TOKEN_URL")
     }
 
 

ingestion_service/worker/core/main.py

@@ -9,14 +9,17 @@
 from core.rabbit_mq_listener import start_consuming
 from core.configure_logging import configure_logging
 from core.routers.worker import router as index_router
+import logging
 
+logger = logging.getLogger(__name__)
 
 async def background_task():
-    print("waiting for messages...")
-    asyncio.create_task(start_consuming())
+    logger.info("#### waiting for messages... ####")
+    loop = asyncio.get_event_loop()
+    # This will run start_consuming in a separate thread
+    await loop.run_in_executor(None, start_consuming)
 
 app = FastAPI()
-logger = logging.getLogger(__name__)
 app.add_middleware(CorrelationIdMiddleware)
 
 
@@ -27,7 +30,7 @@ async def background_task():
 async def startup_event():
     asyncio.create_task(background_task())
     configure_logging()
-    logger.info("Starting FastAPI")
+    logger.info("#### Starting FastAPI... ####")
 
 
 # log all HTTP exception when raised