fix: address named-graph review findings

- honor enable_named_graphs flag when forwarding support

- prevent duplicate FROM/FROM NAMED clauses for same graph

- add default_graph_uri compatibility alias

- harden graph URI sanitization in prune DROP GRAPH path

- add regression tests for all fixes

Co-authored-by: Sameer6305 <sskadam6305@gmail.com>

Co-authored-by: KaifAhmad1 <kaifahmad087@gmail.com>

semantica/change_management/managers.py

@@ -22,6 +22,7 @@ License: MIT
 from abc import ABC, abstractmethod
 from datetime import datetime
 from typing import Any, Dict, List, Optional
+from urllib.parse import quote
 
 from .change_log import ChangeLogEntry
 from .version_storage import (
@@ -388,7 +389,7 @@ class TemporalVersionManager(BaseVersionManager):
                 # Clean up the actual graph if provided
                 if triplet_store and graph_uri:
                     try:
-                        safe_graph_uri = str(graph_uri).strip().strip("<>")
+                        safe_graph_uri = self._sanitize_graph_uri(graph_uri)
                         triplet_store.execute_query(
                             f"DROP SILENT GRAPH <{safe_graph_uri}>"
                         )
@@ -402,6 +403,11 @@ class TemporalVersionManager(BaseVersionManager):
             "pruned_versions": deleted_labels,
             "retained_count": len(all_versions) - len(deleted_labels)
         }
+
+    def _sanitize_graph_uri(self, graph_uri: Any) -> str:
+        """Percent-encode unsafe characters before embedding a graph URI in SPARQL."""
+        raw_uri = str(graph_uri).strip().strip("<>")
+        return quote(raw_uri, safe="/:?&=@[]!$'()*+,%-._~")
     
     # Git-like audit trails
 

semantica/triplet_store/config.py

@@ -110,6 +110,7 @@ class TripletStoreConfig:
         env_mappings = {
             "TRIPLET_STORE_DEFAULT_STORE": "default_store",
             "TRIPLET_STORE_DEFAULT_GRAPH": "default_graph",
+            "TRIPLET_STORE_DEFAULT_GRAPH_URI": "default_graph_uri",
             "TRIPLET_STORE_DEFAULT_NAMED_GRAPHS": "default_graphs",
             "TRIPLET_STORE_BATCH_SIZE": "batch_size",
             "TRIPLET_STORE_ENABLE_CACHING": "enable_caching",
@@ -170,6 +171,7 @@ class TripletStoreConfig:
         defaults = {
             "default_store": None,
             "default_graph": None,
+            "default_graph_uri": None,
             "default_graphs": [],
             "batch_size": 1000,
             "enable_caching": True,

semantica/triplet_store/query_engine.py

@@ -237,7 +237,11 @@ class QueryEngine:
         if not query:
             return ""
 
-        resolved_graph = graph or self.config.get("default_graph")
+        resolved_graph = (
+            graph
+            or self.config.get("default_graph")
+            or self.config.get("default_graph_uri")
+        )
         resolved_graphs = graphs
         if resolved_graphs is None:
             resolved_graphs = self.config.get("default_graphs")
@@ -246,7 +250,7 @@ class QueryEngine:
             resolved_graphs = [resolved_graphs]
         resolved_graphs = [g for g in (resolved_graphs or []) if g]
 
-        if resolved_graph and resolved_graph not in resolved_graphs:
+        if resolved_graph and resolved_graph in resolved_graphs:
             # Preserve graph as default dataset while avoiding duplicate URIs in FROM NAMED.
             resolved_graphs = [g for g in resolved_graphs if g != resolved_graph]
 

semantica/triplet_store/triplet_store.py

@@ -419,9 +419,11 @@ class TripletStore:
         if graphs is not None:
             options["graphs"] = graphs
 
+        enable_named_graphs = self.config.get("enable_named_graphs", True)
         options.setdefault(
             "supports_named_graphs",
-            self.backend_type in self.NAMED_GRAPH_CAPABLE_BACKENDS,
+            enable_named_graphs
+            and self.backend_type in self.NAMED_GRAPH_CAPABLE_BACKENDS,
         )
 
         return self.query_engine.execute_query(query, self._store_backend, **options)

tests/change_management/test_managers.py

@@ -7,6 +7,7 @@ knowledge graphs and ontologies with comprehensive change tracking.
 
 import os
 import tempfile
+from unittest.mock import MagicMock
 import pytest
 from semantica.change_management import (
     TemporalVersionManager,
@@ -179,6 +180,41 @@ class TestTemporalVersionManager:
         assert len(versions) == 1
         assert versions[0]["entity_count"] == 2
         assert versions[0]["relationship_count"] == 1
+
+    def test_prune_versions_sanitizes_graph_uri_in_drop_query(self):
+        """Ensure DROP GRAPH query uses sanitized URI encoding for unsafe characters."""
+        manager = TemporalVersionManager()
+        triplet_store = MagicMock()
+
+        manager.storage.save(
+            {
+                "label": "old-v1",
+                "timestamp": "2024-01-01T00:00:00",
+                "author": "test@example.com",
+                "description": "old",
+                "checksum": "x",
+                "entities": [],
+                "relationships": [],
+                "graph_uri": "http://example.org/graph> } ; DROP ALL ; #",
+            }
+        )
+        manager.storage.save(
+            {
+                "label": "new-v2",
+                "timestamp": "2025-01-01T00:00:00",
+                "author": "test@example.com",
+                "description": "new",
+                "checksum": "y",
+                "entities": [],
+                "relationships": [],
+                "graph_uri": "http://example.org/graph/new",
+            }
+        )
+
+        manager.prune_versions(keep_last_n=1, triplet_store=triplet_store)
+
+        query = triplet_store.execute_query.call_args[0][0]
+        assert "DROP SILENT GRAPH <http://example.org/graph%3E%20%7D%20%3B%20DROP%20ALL%20%3B%20%23>" == query
     
     def test_get_version(self):
         """Test retrieving specific version."""

tests/triplet_store/test_triplet_store.py

@@ -184,6 +184,25 @@ class TestTripletStore(unittest.TestCase):
             supports_named_graphs=True,
         )
 
+    @patch('semantica.triplet_store.blazegraph_store.BlazegraphStore')
+    def test_execute_query_respects_enable_named_graphs_flag(self, mock_blazegraph_store):
+        mock_backend_instance = MagicMock()
+        mock_blazegraph_store.return_value = mock_backend_instance
+
+        store = TripletStore(backend="blazegraph", enable_named_graphs=False)
+        store.query_engine = MagicMock()
+        store.query_engine.execute_query.return_value = QueryEngine()
+
+        query = "SELECT ?s WHERE { ?s ?p ?o }"
+        store.execute_query(query, graph="http://example.org/graph/default")
+
+        store.query_engine.execute_query.assert_called_once_with(
+            query,
+            store._store_backend,
+            graph="http://example.org/graph/default",
+            supports_named_graphs=False,
+        )
+
     def test_query_engine_injects_from_before_where(self):
         engine = QueryEngine(enable_optimization=False, enable_caching=False)
         query = "SELECT ?s ?p ?o WHERE { ?s ?p ?o }"
@@ -246,6 +265,32 @@ class TestTripletStore(unittest.TestCase):
         self.assertNotEqual(graph_a_result.bindings, graph_b_result.bindings)
         self.assertEqual(len(default_result.bindings), 2)
 
+    def test_query_engine_avoids_duplicate_dataset_clauses_for_same_graph(self):
+        engine = QueryEngine(enable_optimization=False, enable_caching=False)
+        query = "SELECT ?s WHERE { GRAPH ?g { ?s ?p ?o } }"
+
+        prepared = engine.prepare_query(
+            query,
+            graph="http://example.org/graph/a",
+            graphs=["http://example.org/graph/a", "http://example.org/graph/b"],
+        )
+
+        self.assertEqual(prepared.count("FROM <http://example.org/graph/a>"), 1)
+        self.assertEqual(prepared.count("FROM NAMED <http://example.org/graph/a>"), 0)
+        self.assertIn("FROM NAMED <http://example.org/graph/b>", prepared)
+
+    def test_query_engine_uses_default_graph_uri_alias(self):
+        engine = QueryEngine(
+            enable_optimization=False,
+            enable_caching=False,
+            default_graph_uri="http://example.org/graph/default",
+        )
+        query = "SELECT ?s WHERE { ?s ?p ?o }"
+
+        prepared = engine.prepare_query(query)
+
+        self.assertIn("FROM <http://example.org/graph/default>", prepared)
+
     def test_query_engine_fallback_when_named_graphs_unsupported(self):
         engine = QueryEngine(enable_optimization=False, enable_caching=False)
         query = "SELECT ?s WHERE { ?s ?p ?o }"