mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-29 04:26:20 +00:00
feat(#319): SKOS Vocabulary Module — namespace helpers, store helpers, OntologyEngine APIs, tests, docs
Extends the existing ontology and triplet-store stack with first-class SKOS support without adding any new top-level packages. ### semantica/ontology/namespace_manager.py - `get_skos_uri(local_name)` — build full skos:core# URI from local name - `build_concept_scheme_uri(name)` — slug a human name into a stable ConceptScheme URI anchored at the configured base URI ### semantica/triplet_store/triplet_store.py - `add_skos_concept(concept_uri, scheme_uri, pref_label, ...)` — asserts ConceptScheme + Concept triples, prefLabel, altLabel, broader, narrower, related, definition, notation via existing `add_triplets()` API - `get_skos_concepts(scheme_uri=None)` — SPARQL SELECT via `execute_query()`, collapses multi-valued bindings into concept dicts ### semantica/ontology/engine.py - `list_vocabularies()` — list all skos:ConceptScheme instances - `list_concepts(scheme_uri)` — list concepts in a scheme with alt labels - `search_concepts(query, scheme_uri=None)` — case-insensitive substring search over prefLabel + altLabel; sanitises user input against SPARQL injection ### tests - `TestSKOSOntologyEngine` (14 tests) in test_ontology_comprehensive.py - `TestSKOSTripletStore` (6 tests) in test_triplet_store.py - All 1162 existing + new tests pass, 0 failures ### docs/reference/ontology.md - New "SKOS Vocabulary Management" section: data-model table, import examples (add_skos_concept + rdflib bulk), list/search API, NamespaceManager helpers Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
KaifAhmad1
co-authored by
Claude Sonnet 4.6
Claude Sonnet 4.6
parent
6651386074
commit
71d037f581
@@ -335,6 +335,114 @@ else:
|
||||
|
||||
---
|
||||
|
||||
## SKOS Vocabulary Management
|
||||
|
||||
Semantica supports [SKOS (Simple Knowledge Organization System)](https://www.w3.org/TR/skos-reference/) vocabularies as first-class semantic assets. SKOS triples are stored in the existing RDF triplet store and queried through the `OntologyEngine` — no additional packages are required.
|
||||
|
||||
### Concepts and data model
|
||||
|
||||
| SKOS element | RDF type / predicate |
|
||||
|---|---|
|
||||
| ConceptScheme | `skos:ConceptScheme` |
|
||||
| Concept | `skos:Concept` |
|
||||
| Preferred label | `skos:prefLabel` |
|
||||
| Alternative label | `skos:altLabel` |
|
||||
| Broader concept | `skos:broader` |
|
||||
| Narrower concept | `skos:narrower` |
|
||||
| Related concept | `skos:related` |
|
||||
| Human definition | `skos:definition` |
|
||||
| Notation / code | `skos:notation` |
|
||||
|
||||
### Importing a SKOS vocabulary
|
||||
|
||||
Use `TripletStore.add_skos_concept()` to load individual concepts. The method automatically asserts the parent `skos:ConceptScheme` triple the first time any concept for that scheme is added.
|
||||
|
||||
```python
|
||||
from semantica.triplet_store import TripletStore
|
||||
|
||||
store = TripletStore(backend="blazegraph", endpoint="http://localhost:9999/blazegraph")
|
||||
|
||||
SCHEME = "https://vocab.example.org/colours"
|
||||
|
||||
store.add_skos_concept(
|
||||
concept_uri="https://vocab.example.org/colours/red",
|
||||
scheme_uri=SCHEME,
|
||||
pref_label="Red",
|
||||
alt_labels=["Crimson", "Rouge"],
|
||||
broader=["https://vocab.example.org/colours/warm"],
|
||||
definition="The colour at the long-wavelength end of the visible spectrum.",
|
||||
notation="RED",
|
||||
)
|
||||
|
||||
store.add_skos_concept(
|
||||
concept_uri="https://vocab.example.org/colours/blue",
|
||||
scheme_uri=SCHEME,
|
||||
pref_label="Blue",
|
||||
alt_labels=["Azure", "Cerulean"],
|
||||
)
|
||||
```
|
||||
|
||||
For bulk ingestion of an existing SKOS/Turtle file use `TripletStore.add_triplets()` after parsing the file with [rdflib](https://rdflib.readthedocs.io/):
|
||||
|
||||
```python
|
||||
import rdflib
|
||||
from semantica.semantic_extract.triplet_extractor import Triplet
|
||||
|
||||
g = rdflib.Graph()
|
||||
g.parse("my_vocabulary.ttl", format="turtle")
|
||||
|
||||
triplets = [
|
||||
Triplet(subject=str(s), predicate=str(p), object=str(o))
|
||||
for s, p, o in g
|
||||
]
|
||||
store.add_triplets(triplets)
|
||||
```
|
||||
|
||||
### Listing and searching concepts
|
||||
|
||||
Once a vocabulary is loaded, use `OntologyEngine` to browse and search it:
|
||||
|
||||
```python
|
||||
from semantica.ontology import OntologyEngine
|
||||
|
||||
engine = OntologyEngine(store=store)
|
||||
|
||||
# 1. List all ConceptSchemes in the store
|
||||
vocabularies = engine.list_vocabularies()
|
||||
# [{"uri": "https://vocab.example.org/colours", "label": "Colours"}, ...]
|
||||
|
||||
# 2. List every concept in a specific scheme
|
||||
concepts = engine.list_concepts("https://vocab.example.org/colours")
|
||||
# [{"uri": "...", "pref_label": "Red", "alt_labels": ["Crimson", "Rouge"]}, ...]
|
||||
|
||||
# 3. Case-insensitive substring search across prefLabel and altLabel
|
||||
results = engine.search_concepts("crimson")
|
||||
# [{"uri": "https://vocab.example.org/colours/red", "label": "Crimson"}]
|
||||
|
||||
# 4. Restrict search to one scheme
|
||||
results = engine.search_concepts("azure", scheme_uri="https://vocab.example.org/colours")
|
||||
```
|
||||
|
||||
### Building SKOS URIs with NamespaceManager
|
||||
|
||||
`NamespaceManager` provides helpers for constructing well-formed SKOS IRIs:
|
||||
|
||||
```python
|
||||
from semantica.ontology import NamespaceManager
|
||||
|
||||
nm = NamespaceManager(base_uri="https://vocab.example.org/")
|
||||
|
||||
# Full SKOS predicate URI
|
||||
nm.get_skos_uri("prefLabel")
|
||||
# "http://www.w3.org/2004/02/skos/core#prefLabel"
|
||||
|
||||
# Slug-based ConceptScheme URI anchored at the base
|
||||
nm.build_concept_scheme_uri("ISO 3166 Countries")
|
||||
# "https://vocab.example.org/vocab/iso-3166-countries"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Best Practices
|
||||
|
||||
1. **Reuse Standard Ontologies**: Don't reinvent `Person` or `Organization`; import FOAF or Schema.org using `ReuseManager`.
|
||||
|
||||
@@ -385,6 +385,194 @@ class OntologyEngine:
|
||||
self.progress.stop_tracking(tracking_id, status="failed", message=str(e))
|
||||
raise
|
||||
|
||||
# ── SKOS Vocabulary Management ────────────────────────────────────────────
|
||||
|
||||
_SKOS = "http://www.w3.org/2004/02/skos/core#"
|
||||
_RDF_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"
|
||||
|
||||
def list_vocabularies(self, **options) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
List all SKOS ConceptSchemes stored in the triplet store.
|
||||
|
||||
Returns:
|
||||
List of dicts with keys ``uri`` and ``label`` (may be empty string
|
||||
when no ``skos:prefLabel`` is present).
|
||||
|
||||
Raises:
|
||||
ProcessingError: If no store is configured or the query fails.
|
||||
"""
|
||||
if not self.store:
|
||||
raise ProcessingError("TripletStore instance not configured in OntologyEngine.")
|
||||
|
||||
SKOS = self._SKOS
|
||||
RDF_TYPE = self._RDF_TYPE
|
||||
|
||||
query = f"""
|
||||
SELECT DISTINCT ?scheme ?label WHERE {{
|
||||
?scheme <{RDF_TYPE}> <{SKOS}ConceptScheme> .
|
||||
OPTIONAL {{ ?scheme <{SKOS}prefLabel> ?label }}
|
||||
}}
|
||||
"""
|
||||
tracking_id = self.progress.start_tracking(
|
||||
module="ontology", submodule="OntologyEngine", message="Listing SKOS vocabularies"
|
||||
)
|
||||
try:
|
||||
result = self.store.execute_query(query, **options)
|
||||
vocabs = []
|
||||
if hasattr(result, "bindings"):
|
||||
seen: set = set()
|
||||
for b in result.bindings:
|
||||
def _v(key):
|
||||
val = b.get(key)
|
||||
return (val.get("value") if isinstance(val, dict) else val) if val else None
|
||||
uri = _v("scheme")
|
||||
if uri and uri not in seen:
|
||||
seen.add(uri)
|
||||
vocabs.append({"uri": uri, "label": _v("label") or ""})
|
||||
self.progress.stop_tracking(tracking_id, status="completed",
|
||||
message=f"Found {len(vocabs)} vocabularies")
|
||||
return vocabs
|
||||
except Exception as e:
|
||||
self.progress.stop_tracking(tracking_id, status="failed", message=str(e))
|
||||
raise ProcessingError(f"list_vocabularies failed: {e}")
|
||||
|
||||
def list_concepts(self, scheme_uri: str, **options) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
List all SKOS concepts that belong to the given ConceptScheme.
|
||||
|
||||
Args:
|
||||
scheme_uri: Full URI of the ``skos:ConceptScheme`` to inspect.
|
||||
|
||||
Returns:
|
||||
List of dicts with keys ``uri``, ``pref_label``, and
|
||||
``alt_labels`` (list, may be empty).
|
||||
|
||||
Raises:
|
||||
ProcessingError: If no store is configured or the query fails.
|
||||
"""
|
||||
if not self.store:
|
||||
raise ProcessingError("TripletStore instance not configured in OntologyEngine.")
|
||||
|
||||
SKOS = self._SKOS
|
||||
RDF_TYPE = self._RDF_TYPE
|
||||
safe_scheme = self._sanitize_uri(scheme_uri)
|
||||
|
||||
query = f"""
|
||||
SELECT DISTINCT ?concept ?prefLabel ?altLabel WHERE {{
|
||||
?concept <{RDF_TYPE}> <{SKOS}Concept> .
|
||||
?concept <{SKOS}inScheme> <{safe_scheme}> .
|
||||
OPTIONAL {{ ?concept <{SKOS}prefLabel> ?prefLabel }}
|
||||
OPTIONAL {{ ?concept <{SKOS}altLabel> ?altLabel }}
|
||||
}}
|
||||
"""
|
||||
tracking_id = self.progress.start_tracking(
|
||||
module="ontology", submodule="OntologyEngine",
|
||||
message=f"Listing concepts in {scheme_uri}"
|
||||
)
|
||||
try:
|
||||
result = self.store.execute_query(query, **options)
|
||||
concepts: Dict[str, Dict[str, Any]] = {}
|
||||
if hasattr(result, "bindings"):
|
||||
for b in result.bindings:
|
||||
def _v(key):
|
||||
val = b.get(key)
|
||||
return (val.get("value") if isinstance(val, dict) else val) if val else None
|
||||
uri = _v("concept")
|
||||
if not uri:
|
||||
continue
|
||||
if uri not in concepts:
|
||||
concepts[uri] = {"uri": uri, "pref_label": _v("prefLabel") or "", "alt_labels": []}
|
||||
if not concepts[uri]["pref_label"] and _v("prefLabel"):
|
||||
concepts[uri]["pref_label"] = _v("prefLabel")
|
||||
lbl = _v("altLabel")
|
||||
if lbl and lbl not in concepts[uri]["alt_labels"]:
|
||||
concepts[uri]["alt_labels"].append(lbl)
|
||||
self.progress.stop_tracking(tracking_id, status="completed",
|
||||
message=f"Found {len(concepts)} concepts")
|
||||
return list(concepts.values())
|
||||
except Exception as e:
|
||||
self.progress.stop_tracking(tracking_id, status="failed", message=str(e))
|
||||
raise ProcessingError(f"list_concepts failed: {e}")
|
||||
|
||||
def search_concepts(
|
||||
self,
|
||||
query: str,
|
||||
scheme_uri: Optional[str] = None,
|
||||
**options,
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Search SKOS concepts by matching ``skos:prefLabel`` or ``skos:altLabel``.
|
||||
|
||||
The search is case-insensitive substring matching performed at the
|
||||
SPARQL level via ``CONTAINS(LCASE(…))``.
|
||||
|
||||
Args:
|
||||
query: Substring to search for.
|
||||
scheme_uri: When given, restrict results to this ConceptScheme.
|
||||
|
||||
Returns:
|
||||
List of dicts with keys ``uri`` and ``label`` (the matching label).
|
||||
|
||||
Raises:
|
||||
ProcessingError: If no store is configured or the query fails.
|
||||
"""
|
||||
if not self.store:
|
||||
raise ProcessingError("TripletStore instance not configured in OntologyEngine.")
|
||||
|
||||
SKOS = self._SKOS
|
||||
RDF_TYPE = self._RDF_TYPE
|
||||
|
||||
# Sanitize user query for embedding in a SPARQL string literal
|
||||
safe_query = (
|
||||
query
|
||||
.replace("\\", "\\\\")
|
||||
.replace('"', '\\"')
|
||||
.replace("\n", " ")
|
||||
.replace("\r", " ")
|
||||
)
|
||||
|
||||
scheme_filter = ""
|
||||
if scheme_uri:
|
||||
safe_scheme = self._sanitize_uri(scheme_uri)
|
||||
scheme_filter = f"?concept <{SKOS}inScheme> <{safe_scheme}> ."
|
||||
|
||||
sparql = f"""
|
||||
SELECT DISTINCT ?concept ?label WHERE {{
|
||||
?concept <{RDF_TYPE}> <{SKOS}Concept> .
|
||||
{scheme_filter}
|
||||
{{
|
||||
?concept <{SKOS}prefLabel> ?label
|
||||
}} UNION {{
|
||||
?concept <{SKOS}altLabel> ?label
|
||||
}}
|
||||
FILTER(CONTAINS(LCASE(STR(?label)), LCASE("{safe_query}")))
|
||||
}}
|
||||
"""
|
||||
tracking_id = self.progress.start_tracking(
|
||||
module="ontology", submodule="OntologyEngine",
|
||||
message=f"Searching SKOS concepts: '{query}'"
|
||||
)
|
||||
try:
|
||||
result = self.store.execute_query(sparql, **options)
|
||||
matches = []
|
||||
seen: set = set()
|
||||
if hasattr(result, "bindings"):
|
||||
for b in result.bindings:
|
||||
def _v(key):
|
||||
val = b.get(key)
|
||||
return (val.get("value") if isinstance(val, dict) else val) if val else None
|
||||
uri = _v("concept")
|
||||
lbl = _v("label")
|
||||
if uri and uri not in seen:
|
||||
seen.add(uri)
|
||||
matches.append({"uri": uri, "label": lbl or ""})
|
||||
self.progress.stop_tracking(tracking_id, status="completed",
|
||||
message=f"Found {len(matches)} matches")
|
||||
return matches
|
||||
except Exception as e:
|
||||
self.progress.stop_tracking(tracking_id, status="failed", message=str(e))
|
||||
raise ProcessingError(f"search_concepts failed: {e}")
|
||||
|
||||
# ── Ontology Evaluation / Validation ─────────────────────────────────────
|
||||
|
||||
def evaluate(self, ontology: Dict[str, Any], **options):
|
||||
|
||||
@@ -207,6 +207,35 @@ class NamespaceManager:
|
||||
"""
|
||||
return dict(self.namespaces)
|
||||
|
||||
def get_skos_uri(self, local_name: str) -> str:
|
||||
"""
|
||||
Build a full SKOS URI from a local name.
|
||||
|
||||
Args:
|
||||
local_name: SKOS local term (e.g. ``"Concept"``, ``"prefLabel"``)
|
||||
|
||||
Returns:
|
||||
Full SKOS URI string
|
||||
"""
|
||||
skos_ns = self.namespaces["skos"]
|
||||
return f"{skos_ns}{local_name}"
|
||||
|
||||
def build_concept_scheme_uri(self, name: str) -> str:
|
||||
"""
|
||||
Build a ConceptScheme URI anchored at the current base URI.
|
||||
|
||||
The scheme name is slugified (spaces → hyphens, lower-cased) so that
|
||||
``"My Vocabulary"`` becomes ``<base>/vocab/my-vocabulary>``.
|
||||
|
||||
Args:
|
||||
name: Human-readable vocabulary name
|
||||
|
||||
Returns:
|
||||
ConceptScheme URI string
|
||||
"""
|
||||
slug = re.sub(r"[^a-zA-Z0-9]+", "-", name).strip("-").lower()
|
||||
return urljoin(self.get_base_uri(), f"vocab/{slug}")
|
||||
|
||||
def get_alignment_predicates(self) -> Dict[str, str]:
|
||||
"""
|
||||
Get standard alignment predicates for ontology mapping.
|
||||
|
||||
@@ -422,6 +422,157 @@ class TripletStore:
|
||||
|
||||
return True
|
||||
|
||||
# ── SKOS helpers ─────────────────────────────────────────────────────────
|
||||
|
||||
_SKOS = "http://www.w3.org/2004/02/skos/core#"
|
||||
_RDF_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"
|
||||
|
||||
def add_skos_concept(
|
||||
self,
|
||||
concept_uri: str,
|
||||
scheme_uri: str,
|
||||
pref_label: str,
|
||||
alt_labels: Optional[List[str]] = None,
|
||||
broader: Optional[List[str]] = None,
|
||||
narrower: Optional[List[str]] = None,
|
||||
related: Optional[List[str]] = None,
|
||||
definition: Optional[str] = None,
|
||||
notation: Optional[str] = None,
|
||||
**options,
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Add a SKOS concept (and its scheme if not already present) to the store.
|
||||
|
||||
Core triples added:
|
||||
|
||||
* ``concept_uri rdf:type skos:Concept``
|
||||
* ``concept_uri skos:inScheme scheme_uri``
|
||||
* ``concept_uri skos:prefLabel pref_label``
|
||||
* ``scheme_uri rdf:type skos:ConceptScheme`` (auto-created)
|
||||
* Optional: altLabel, broader, narrower, related, definition, notation
|
||||
|
||||
Args:
|
||||
concept_uri: Full URI for the concept.
|
||||
scheme_uri: Full URI for the parent ConceptScheme.
|
||||
pref_label: Preferred label string.
|
||||
alt_labels: Optional list of alternative label strings.
|
||||
broader: Optional list of broader concept URIs.
|
||||
narrower: Optional list of narrower concept URIs.
|
||||
related: Optional list of related concept URIs.
|
||||
definition: Optional human-readable definition string.
|
||||
notation: Optional notation / code string.
|
||||
**options: Forwarded to :meth:`add_triplets`.
|
||||
|
||||
Returns:
|
||||
:meth:`add_triplets` status dict.
|
||||
"""
|
||||
SKOS = self._SKOS
|
||||
RDF_TYPE = self._RDF_TYPE
|
||||
|
||||
triplets: List[Triplet] = [
|
||||
# Scheme declaration
|
||||
Triplet(scheme_uri, RDF_TYPE, f"{SKOS}ConceptScheme"),
|
||||
# Concept core
|
||||
Triplet(concept_uri, RDF_TYPE, f"{SKOS}Concept"),
|
||||
Triplet(concept_uri, f"{SKOS}inScheme", scheme_uri),
|
||||
Triplet(concept_uri, f"{SKOS}prefLabel", pref_label),
|
||||
]
|
||||
|
||||
for lbl in (alt_labels or []):
|
||||
triplets.append(Triplet(concept_uri, f"{SKOS}altLabel", lbl))
|
||||
for uri in (broader or []):
|
||||
triplets.append(Triplet(concept_uri, f"{SKOS}broader", uri))
|
||||
for uri in (narrower or []):
|
||||
triplets.append(Triplet(concept_uri, f"{SKOS}narrower", uri))
|
||||
for uri in (related or []):
|
||||
triplets.append(Triplet(concept_uri, f"{SKOS}related", uri))
|
||||
if definition:
|
||||
triplets.append(Triplet(concept_uri, f"{SKOS}definition", definition))
|
||||
if notation:
|
||||
triplets.append(Triplet(concept_uri, f"{SKOS}notation", notation))
|
||||
|
||||
return self.add_triplets(triplets, **options)
|
||||
|
||||
def get_skos_concepts(
|
||||
self, scheme_uri: Optional[str] = None, **options
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Retrieve SKOS concepts from the store as plain dicts.
|
||||
|
||||
Each returned dict has at minimum ``uri`` and ``pref_label``; optional
|
||||
keys ``alt_labels``, ``broader``, ``narrower``, and ``related`` are
|
||||
populated when available.
|
||||
|
||||
Args:
|
||||
scheme_uri: When given, only concepts ``skos:inScheme`` this URI
|
||||
are returned. When omitted all concepts are returned.
|
||||
**options: Forwarded to :meth:`execute_query`.
|
||||
|
||||
Returns:
|
||||
List of concept dicts.
|
||||
"""
|
||||
SKOS = self._SKOS
|
||||
RDF_TYPE = self._RDF_TYPE
|
||||
|
||||
scheme_filter = (
|
||||
f"?concept <{SKOS}inScheme> <{self.query_engine._sanitize_uri(scheme_uri)}> ."
|
||||
if scheme_uri
|
||||
else ""
|
||||
)
|
||||
|
||||
query = f"""
|
||||
SELECT DISTINCT ?concept ?prefLabel ?altLabel ?broader ?narrower ?related
|
||||
WHERE {{
|
||||
?concept <{RDF_TYPE}> <{SKOS}Concept> .
|
||||
{scheme_filter}
|
||||
OPTIONAL {{ ?concept <{SKOS}prefLabel> ?prefLabel }}
|
||||
OPTIONAL {{ ?concept <{SKOS}altLabel> ?altLabel }}
|
||||
OPTIONAL {{ ?concept <{SKOS}broader> ?broader }}
|
||||
OPTIONAL {{ ?concept <{SKOS}narrower> ?narrower }}
|
||||
OPTIONAL {{ ?concept <{SKOS}related> ?related }}
|
||||
}}
|
||||
"""
|
||||
|
||||
try:
|
||||
result = self.execute_query(query, **options)
|
||||
except Exception as e:
|
||||
self.logger.error(f"get_skos_concepts query failed: {e}")
|
||||
raise ProcessingError(f"Failed to retrieve SKOS concepts: {e}")
|
||||
|
||||
# Collapse multi-valued properties per concept URI
|
||||
concepts: Dict[str, Dict[str, Any]] = {}
|
||||
for b in result.bindings:
|
||||
def _val(key: str) -> Optional[str]:
|
||||
v = b.get(key)
|
||||
return (v.get("value") if isinstance(v, dict) else v) if v else None
|
||||
|
||||
uri = _val("concept")
|
||||
if not uri:
|
||||
continue
|
||||
if uri not in concepts:
|
||||
concepts[uri] = {
|
||||
"uri": uri,
|
||||
"pref_label": _val("prefLabel") or "",
|
||||
"alt_labels": [],
|
||||
"broader": [],
|
||||
"narrower": [],
|
||||
"related": [],
|
||||
}
|
||||
entry = concepts[uri]
|
||||
if not entry["pref_label"] and _val("prefLabel"):
|
||||
entry["pref_label"] = _val("prefLabel")
|
||||
for multi_key, sparql_key in [
|
||||
("alt_labels", "altLabel"),
|
||||
("broader", "broader"),
|
||||
("narrower", "narrower"),
|
||||
("related", "related"),
|
||||
]:
|
||||
v = _val(sparql_key)
|
||||
if v and v not in entry[multi_key]:
|
||||
entry[multi_key].append(v)
|
||||
|
||||
return list(concepts.values())
|
||||
|
||||
def get_stats(self) -> Dict[str, Any]:
|
||||
"""Get store statistics."""
|
||||
if hasattr(self._store_backend, "get_stats"):
|
||||
|
||||
@@ -431,5 +431,199 @@ class TestSHACLGeneration(unittest.TestCase):
|
||||
self.assertEqual(gen.quality_tier, "standard")
|
||||
|
||||
|
||||
class TestSKOSOntologyEngine(unittest.TestCase):
|
||||
"""Tests for SKOS vocabulary management APIs in OntologyEngine."""
|
||||
|
||||
def setUp(self):
|
||||
self.mock_logger = MagicMock()
|
||||
self.mock_tracker = MagicMock()
|
||||
self.mock_tracker.start_tracking.return_value = "track_id"
|
||||
|
||||
patchers = [
|
||||
patch('semantica.ontology.engine.get_logger', return_value=self.mock_logger),
|
||||
patch('semantica.ontology.engine.get_progress_tracker', return_value=self.mock_tracker),
|
||||
patch('semantica.ontology.ontology_generator.get_logger', return_value=self.mock_logger),
|
||||
patch('semantica.ontology.ontology_generator.get_progress_tracker', return_value=self.mock_tracker),
|
||||
patch('semantica.ontology.class_inferrer.get_logger', return_value=self.mock_logger),
|
||||
patch('semantica.ontology.class_inferrer.get_progress_tracker', return_value=self.mock_tracker),
|
||||
patch('semantica.ontology.property_generator.get_logger', return_value=self.mock_logger),
|
||||
patch('semantica.ontology.property_generator.get_progress_tracker', return_value=self.mock_tracker),
|
||||
patch('semantica.ontology.owl_generator.get_logger', return_value=self.mock_logger),
|
||||
patch('semantica.ontology.owl_generator.get_progress_tracker', return_value=self.mock_tracker),
|
||||
patch('semantica.ontology.ontology_evaluator.get_logger', return_value=self.mock_logger),
|
||||
patch('semantica.ontology.ontology_evaluator.get_progress_tracker', return_value=self.mock_tracker),
|
||||
patch('semantica.ontology.ontology_validator.get_logger', return_value=self.mock_logger),
|
||||
patch('semantica.ontology.llm_generator.get_logger', return_value=self.mock_logger),
|
||||
patch('semantica.ontology.llm_generator.get_progress_tracker', return_value=self.mock_tracker),
|
||||
patch('semantica.change_management.ontology_version_manager.get_logger', return_value=self.mock_logger),
|
||||
patch('semantica.change_management.ontology_version_manager.get_progress_tracker', return_value=self.mock_tracker),
|
||||
]
|
||||
self.patchers = patchers
|
||||
for p in self.patchers:
|
||||
p.start()
|
||||
|
||||
# Mock store with a controllable execute_query
|
||||
self.mock_store = MagicMock()
|
||||
from semantica.ontology.engine import OntologyEngine
|
||||
self.engine = OntologyEngine(store=self.mock_store)
|
||||
|
||||
def tearDown(self):
|
||||
for p in self.patchers:
|
||||
p.stop()
|
||||
|
||||
def _make_result(self, bindings):
|
||||
"""Build a fake QueryResult-like object."""
|
||||
result = MagicMock()
|
||||
result.bindings = bindings
|
||||
return result
|
||||
|
||||
# --- NamespaceManager SKOS helpers ---
|
||||
|
||||
def test_get_skos_uri(self):
|
||||
from semantica.ontology.namespace_manager import NamespaceManager
|
||||
nm = NamespaceManager()
|
||||
self.assertEqual(
|
||||
nm.get_skos_uri("Concept"),
|
||||
"http://www.w3.org/2004/02/skos/core#Concept",
|
||||
)
|
||||
self.assertEqual(
|
||||
nm.get_skos_uri("prefLabel"),
|
||||
"http://www.w3.org/2004/02/skos/core#prefLabel",
|
||||
)
|
||||
|
||||
def test_build_concept_scheme_uri(self):
|
||||
from semantica.ontology.namespace_manager import NamespaceManager
|
||||
nm = NamespaceManager(base_uri="https://example.org/onto/")
|
||||
uri = nm.build_concept_scheme_uri("My Vocabulary")
|
||||
self.assertIn("my-vocabulary", uri)
|
||||
self.assertTrue(uri.startswith("https://example.org/onto/"))
|
||||
|
||||
def test_build_concept_scheme_uri_special_chars(self):
|
||||
from semantica.ontology.namespace_manager import NamespaceManager
|
||||
nm = NamespaceManager()
|
||||
uri = nm.build_concept_scheme_uri("ISO 3166 Countries")
|
||||
self.assertIn("iso-3166-countries", uri)
|
||||
|
||||
# --- list_vocabularies ---
|
||||
|
||||
def test_list_vocabularies_returns_schemes(self):
|
||||
self.mock_store.execute_query.return_value = self._make_result([
|
||||
{"scheme": {"value": "http://example.org/vocab/colours"},
|
||||
"label": {"value": "Colours"}},
|
||||
{"scheme": {"value": "http://example.org/vocab/sizes"},
|
||||
"label": None},
|
||||
])
|
||||
vocabs = self.engine.list_vocabularies()
|
||||
self.assertEqual(len(vocabs), 2)
|
||||
uris = [v["uri"] for v in vocabs]
|
||||
self.assertIn("http://example.org/vocab/colours", uris)
|
||||
self.assertIn("http://example.org/vocab/sizes", uris)
|
||||
colours = next(v for v in vocabs if "colours" in v["uri"])
|
||||
self.assertEqual(colours["label"], "Colours")
|
||||
|
||||
def test_list_vocabularies_deduplicates(self):
|
||||
# Same scheme URI appearing twice (multi-valued label rows)
|
||||
self.mock_store.execute_query.return_value = self._make_result([
|
||||
{"scheme": {"value": "http://example.org/vocab/colours"},
|
||||
"label": {"value": "Colours"}},
|
||||
{"scheme": {"value": "http://example.org/vocab/colours"},
|
||||
"label": {"value": "Colors"}},
|
||||
])
|
||||
vocabs = self.engine.list_vocabularies()
|
||||
self.assertEqual(len(vocabs), 1)
|
||||
|
||||
def test_list_vocabularies_no_store_raises(self):
|
||||
from semantica.utils.exceptions import ProcessingError
|
||||
from semantica.ontology.engine import OntologyEngine
|
||||
engine_no_store = OntologyEngine()
|
||||
with self.assertRaises(ProcessingError):
|
||||
engine_no_store.list_vocabularies()
|
||||
|
||||
# --- list_concepts ---
|
||||
|
||||
def test_list_concepts_returns_concepts(self):
|
||||
self.mock_store.execute_query.return_value = self._make_result([
|
||||
{"concept": {"value": "http://example.org/concept/red"},
|
||||
"prefLabel": {"value": "Red"},
|
||||
"altLabel": {"value": "Crimson"}},
|
||||
{"concept": {"value": "http://example.org/concept/red"},
|
||||
"prefLabel": {"value": "Red"},
|
||||
"altLabel": {"value": "Rouge"}},
|
||||
{"concept": {"value": "http://example.org/concept/blue"},
|
||||
"prefLabel": {"value": "Blue"},
|
||||
"altLabel": None},
|
||||
])
|
||||
concepts = self.engine.list_concepts("http://example.org/vocab/colours")
|
||||
self.assertEqual(len(concepts), 2)
|
||||
red = next(c for c in concepts if "red" in c["uri"])
|
||||
self.assertEqual(red["pref_label"], "Red")
|
||||
self.assertIn("Crimson", red["alt_labels"])
|
||||
self.assertIn("Rouge", red["alt_labels"])
|
||||
blue = next(c for c in concepts if "blue" in c["uri"])
|
||||
self.assertEqual(blue["alt_labels"], [])
|
||||
|
||||
def test_list_concepts_no_store_raises(self):
|
||||
from semantica.utils.exceptions import ProcessingError
|
||||
from semantica.ontology.engine import OntologyEngine
|
||||
engine_no_store = OntologyEngine()
|
||||
with self.assertRaises(ProcessingError):
|
||||
engine_no_store.list_concepts("http://example.org/vocab/colours")
|
||||
|
||||
# --- search_concepts ---
|
||||
|
||||
def test_search_concepts_returns_matches(self):
|
||||
self.mock_store.execute_query.return_value = self._make_result([
|
||||
{"concept": {"value": "http://example.org/concept/red"},
|
||||
"label": {"value": "Red"}},
|
||||
{"concept": {"value": "http://example.org/concept/infrared"},
|
||||
"label": {"value": "Infrared"}},
|
||||
])
|
||||
results = self.engine.search_concepts("red")
|
||||
self.assertEqual(len(results), 2)
|
||||
uris = [r["uri"] for r in results]
|
||||
self.assertIn("http://example.org/concept/red", uris)
|
||||
self.assertIn("http://example.org/concept/infrared", uris)
|
||||
|
||||
def test_search_concepts_with_scheme_filter(self):
|
||||
self.mock_store.execute_query.return_value = self._make_result([
|
||||
{"concept": {"value": "http://example.org/concept/red"},
|
||||
"label": {"value": "Red"}},
|
||||
])
|
||||
results = self.engine.search_concepts("red", scheme_uri="http://example.org/vocab/colours")
|
||||
self.assertEqual(len(results), 1)
|
||||
# Scheme URI should appear in the SPARQL issued to the store
|
||||
issued_sparql = self.mock_store.execute_query.call_args[0][0]
|
||||
self.assertIn("http://example.org/vocab/colours", issued_sparql)
|
||||
|
||||
def test_search_concepts_empty_result(self):
|
||||
self.mock_store.execute_query.return_value = self._make_result([])
|
||||
results = self.engine.search_concepts("zzznomatch")
|
||||
self.assertEqual(results, [])
|
||||
|
||||
def test_search_concepts_no_store_raises(self):
|
||||
from semantica.utils.exceptions import ProcessingError
|
||||
from semantica.ontology.engine import OntologyEngine
|
||||
engine_no_store = OntologyEngine()
|
||||
with self.assertRaises(ProcessingError):
|
||||
engine_no_store.search_concepts("red")
|
||||
|
||||
def test_search_concepts_sanitizes_query(self):
|
||||
"""Ensure user input containing SPARQL-special chars doesn't break the query."""
|
||||
self.mock_store.execute_query.return_value = self._make_result([])
|
||||
# Should not raise
|
||||
self.engine.search_concepts('red" } MALICIOUS { ?x ?y ?z')
|
||||
|
||||
def test_search_concepts_deduplicates(self):
|
||||
# Same concept URI matched by both prefLabel and altLabel
|
||||
self.mock_store.execute_query.return_value = self._make_result([
|
||||
{"concept": {"value": "http://example.org/concept/red"},
|
||||
"label": {"value": "Red"}},
|
||||
{"concept": {"value": "http://example.org/concept/red"},
|
||||
"label": {"value": "Reddish"}},
|
||||
])
|
||||
results = self.engine.search_concepts("red")
|
||||
self.assertEqual(len(results), 1)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
@@ -162,3 +162,183 @@ class TestTripletStore(unittest.TestCase):
|
||||
self.assertIn("http://aligned.org/2", sparql_query)
|
||||
self.assertIn("VALUES ?subject", sparql_query)
|
||||
mock_backend.execute_sparql.assert_called_once()
|
||||
|
||||
|
||||
class TestSKOSTripletStore(unittest.TestCase):
|
||||
"""Tests for SKOS helper methods on TripletStore."""
|
||||
|
||||
_SKOS = "http://www.w3.org/2004/02/skos/core#"
|
||||
_RDF_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"
|
||||
|
||||
def setUp(self):
|
||||
self.mock_logger = MagicMock()
|
||||
self.mock_tracker = MagicMock()
|
||||
|
||||
self.logger_patcher = patch(
|
||||
'semantica.triplet_store.triplet_store.get_logger', return_value=self.mock_logger
|
||||
)
|
||||
self.tracker_patcher = patch(
|
||||
'semantica.triplet_store.triplet_store.get_progress_tracker', return_value=self.mock_tracker
|
||||
)
|
||||
self.logger_patcher.start()
|
||||
self.tracker_patcher.start()
|
||||
|
||||
def tearDown(self):
|
||||
self.logger_patcher.stop()
|
||||
self.tracker_patcher.stop()
|
||||
|
||||
def _make_store(self, mock_blazegraph):
|
||||
"""Return a TripletStore backed by a MagicMock BlazegraphStore."""
|
||||
mock_backend = MagicMock()
|
||||
mock_blazegraph.return_value = mock_backend
|
||||
store = TripletStore(backend="blazegraph")
|
||||
# Provide a fast no-op bulk loader
|
||||
mock_loader = MagicMock()
|
||||
mock_progress = MagicMock()
|
||||
mock_progress.metadata = {"success": True}
|
||||
mock_progress.total_triplets = 0
|
||||
mock_progress.loaded_triplets = 0
|
||||
mock_progress.failed_triplets = 0
|
||||
mock_progress.total_batches = 0
|
||||
mock_loader.load_triplets.return_value = mock_progress
|
||||
store.bulk_loader = mock_loader
|
||||
return store, mock_backend, mock_loader
|
||||
|
||||
# --- add_skos_concept ---
|
||||
|
||||
@patch('semantica.triplet_store.blazegraph_store.BlazegraphStore')
|
||||
def test_add_skos_concept_core_triples(self, mock_bg):
|
||||
"""add_skos_concept must produce ConceptScheme + Concept + inScheme + prefLabel triples."""
|
||||
store, _, mock_loader = self._make_store(mock_bg)
|
||||
|
||||
store.add_skos_concept(
|
||||
concept_uri="http://example.org/concept/red",
|
||||
scheme_uri="http://example.org/vocab/colours",
|
||||
pref_label="Red",
|
||||
)
|
||||
|
||||
mock_loader.load_triplets.assert_called_once()
|
||||
triplets = mock_loader.load_triplets.call_args[0][0]
|
||||
subjects_predicates = {(t.subject, t.predicate) for t in triplets}
|
||||
|
||||
SKOS = self._SKOS
|
||||
RDF_TYPE = self._RDF_TYPE
|
||||
|
||||
self.assertIn(("http://example.org/vocab/colours", RDF_TYPE), subjects_predicates)
|
||||
self.assertIn(("http://example.org/concept/red", RDF_TYPE), subjects_predicates)
|
||||
self.assertIn(("http://example.org/concept/red", f"{SKOS}inScheme"), subjects_predicates)
|
||||
self.assertIn(("http://example.org/concept/red", f"{SKOS}prefLabel"), subjects_predicates)
|
||||
|
||||
@patch('semantica.triplet_store.blazegraph_store.BlazegraphStore')
|
||||
def test_add_skos_concept_optional_fields(self, mock_bg):
|
||||
"""Optional fields produce extra triples."""
|
||||
store, _, mock_loader = self._make_store(mock_bg)
|
||||
|
||||
store.add_skos_concept(
|
||||
concept_uri="http://example.org/concept/red",
|
||||
scheme_uri="http://example.org/vocab/colours",
|
||||
pref_label="Red",
|
||||
alt_labels=["Crimson", "Rouge"],
|
||||
broader=["http://example.org/concept/colour"],
|
||||
definition="The colour red.",
|
||||
notation="RED",
|
||||
)
|
||||
|
||||
triplets = mock_loader.load_triplets.call_args[0][0]
|
||||
predicates = [t.predicate for t in triplets]
|
||||
SKOS = self._SKOS
|
||||
|
||||
self.assertIn(f"{SKOS}altLabel", predicates)
|
||||
self.assertEqual(predicates.count(f"{SKOS}altLabel"), 2)
|
||||
self.assertIn(f"{SKOS}broader", predicates)
|
||||
self.assertIn(f"{SKOS}definition", predicates)
|
||||
self.assertIn(f"{SKOS}notation", predicates)
|
||||
|
||||
@patch('semantica.triplet_store.blazegraph_store.BlazegraphStore')
|
||||
def test_add_skos_concept_scheme_triple_always_included(self, mock_bg):
|
||||
"""ConceptScheme rdf:type triple is always included even without optional args."""
|
||||
store, _, mock_loader = self._make_store(mock_bg)
|
||||
|
||||
store.add_skos_concept(
|
||||
concept_uri="http://example.org/concept/blue",
|
||||
scheme_uri="http://example.org/vocab/colours",
|
||||
pref_label="Blue",
|
||||
)
|
||||
|
||||
triplets = mock_loader.load_triplets.call_args[0][0]
|
||||
scheme_types = [
|
||||
t for t in triplets
|
||||
if t.subject == "http://example.org/vocab/colours"
|
||||
and t.predicate == self._RDF_TYPE
|
||||
and t.object == f"{self._SKOS}ConceptScheme"
|
||||
]
|
||||
self.assertEqual(len(scheme_types), 1)
|
||||
|
||||
# --- get_skos_concepts ---
|
||||
|
||||
@patch('semantica.triplet_store.blazegraph_store.BlazegraphStore')
|
||||
def test_get_skos_concepts_all(self, mock_bg):
|
||||
"""get_skos_concepts returns all concepts when no scheme_uri given."""
|
||||
store, mock_backend, _ = self._make_store(mock_bg)
|
||||
|
||||
from semantica.triplet_store.query_engine import QueryResult
|
||||
mock_result = QueryResult(
|
||||
bindings=[
|
||||
{"concept": {"value": "http://example.org/concept/red"},
|
||||
"prefLabel": {"value": "Red"},
|
||||
"altLabel": {"value": "Crimson"},
|
||||
"broader": None, "narrower": None, "related": None},
|
||||
{"concept": {"value": "http://example.org/concept/red"},
|
||||
"prefLabel": {"value": "Red"},
|
||||
"altLabel": {"value": "Rouge"},
|
||||
"broader": None, "narrower": None, "related": None},
|
||||
{"concept": {"value": "http://example.org/concept/blue"},
|
||||
"prefLabel": {"value": "Blue"},
|
||||
"altLabel": None,
|
||||
"broader": None, "narrower": None, "related": None},
|
||||
],
|
||||
variables=["concept", "prefLabel", "altLabel"],
|
||||
)
|
||||
mock_backend.execute_sparql.return_value = {
|
||||
"bindings": mock_result.bindings,
|
||||
"variables": mock_result.variables,
|
||||
"metadata": {},
|
||||
}
|
||||
|
||||
# Patch query_engine.execute_query to return mock_result directly
|
||||
store.query_engine.execute_query = MagicMock(return_value=mock_result)
|
||||
|
||||
concepts = store.get_skos_concepts()
|
||||
self.assertEqual(len(concepts), 2)
|
||||
|
||||
red = next(c for c in concepts if "red" in c["uri"])
|
||||
self.assertEqual(red["pref_label"], "Red")
|
||||
self.assertIn("Crimson", red["alt_labels"])
|
||||
self.assertIn("Rouge", red["alt_labels"])
|
||||
|
||||
blue = next(c for c in concepts if "blue" in c["uri"])
|
||||
self.assertEqual(blue["alt_labels"], [])
|
||||
|
||||
@patch('semantica.triplet_store.blazegraph_store.BlazegraphStore')
|
||||
def test_get_skos_concepts_scheme_filter_in_query(self, mock_bg):
|
||||
"""When scheme_uri is given the scheme URI appears in the issued SPARQL."""
|
||||
store, _, _ = self._make_store(mock_bg)
|
||||
|
||||
from semantica.triplet_store.query_engine import QueryResult
|
||||
empty_result = QueryResult(bindings=[], variables=[])
|
||||
store.query_engine.execute_query = MagicMock(return_value=empty_result)
|
||||
|
||||
store.get_skos_concepts(scheme_uri="http://example.org/vocab/colours")
|
||||
|
||||
issued_sparql = store.query_engine.execute_query.call_args[0][0]
|
||||
self.assertIn("http://example.org/vocab/colours", issued_sparql)
|
||||
|
||||
@patch('semantica.triplet_store.blazegraph_store.BlazegraphStore')
|
||||
def test_get_skos_concepts_empty_store(self, mock_bg):
|
||||
"""Returns empty list when no concepts exist."""
|
||||
store, _, _ = self._make_store(mock_bg)
|
||||
from semantica.triplet_store.query_engine import QueryResult
|
||||
store.query_engine.execute_query = MagicMock(
|
||||
return_value=QueryResult(bindings=[], variables=[])
|
||||
)
|
||||
self.assertEqual(store.get_skos_concepts(), [])
|
||||
|
||||
Reference in New Issue
Block a user