Coverage for lode / reader / modules.py: 93%
90 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-10-05 16:02 +0000
1# modules.py - RDF graph enrichment modules
2"""Enrichment modules: load external documents and add them to the graph of the
3documented ontology.
5Three modules, activated by the Loader (one at a time; partial_import takes precedence):
7 imported direct owl:imports (depth 1)
8 closure transitive closure of owl:imports (depth N)
9 partial_import only the definitions of the external terms mentioned, taken from
10 their "home" ontology (1-hop dereferencing)
12Two ways of loading an external document, shared by the modules:
14 formal import declared with owl:imports; its own owl:imports are followed
15 (_load_formal_import, used by imported and closure)
16 informal import a document reached by dereferencing a mentioned IRI; loaded alone,
17 its imports are not followed (_load_informal_import, used by
18 partial_import)
20Provenance. The graph received is a ProvenanceGraph: one named graph per source
21document, named after the ontology IRI declared in the document (owl:Ontology or
22skos:ConceptScheme), falling back to the URL it was loaded from. Modules never build
23contexts by hand: every document enters through ProvenanceGraph.add_document, so each
24triple stays in the graph of the document that asserts it.
26Every external document is loaded with a Loader without modules: no implicit
27recursion, and the same security checks as the main document.
29Contract: every apply_* modifies the graph in place and returns (graph, visited),
30where visited is the set of sources loaded successfully.
31"""
32from rdflib import Graph, OWL, URIRef, BNode, RDF, RDFS, XSD
33from urllib.parse import urldefrag
34from typing import Optional
36from lode.helpers.rdf import iri_namespace, iri_stem
37from lode.reader.provenance_graph import ProvenanceGraph
39# Core vocabularies: never treated as external terms to dereference.
40_SKIP_NS = (str(RDF), str(RDFS), str(OWL), str(XSD))
42# ----------------------------------------------------------
43# DOCUMENT LOADERS: formal / informal imports
44# ----------------------------------------------------------
46def _load_formal_import(graph: ProvenanceGraph, source: str, depth: int,
47 max_depth: Optional[int], visited: set) -> None:
48 """Load a formally imported document (owl:imports) into its own named graph and
49 follow its owl:imports, depth-first, down to max_depth (None = no limit).
51 `visited` prevents cycles and documents imported by several ontologies: each one
52 is loaded only once. An unreachable document does not stop loading: it is
53 reported and skipped, together with its imports."""
54 if source in visited:
55 return
56 if max_depth is not None and depth > max_depth:
57 return
59 visited.add(source)
61 # Reuse the Loader for the whole load (content negotiation, formats, security).
62 # Modules are not propagated: each imported ontology is loaded as-is, and the
63 # recursion over imports is handled by this function.
64 from lode.reader.loader import Loader # local import: loader imports modules
65 try:
66 imported_loader = Loader(source)
67 except Exception:
68 print(f" [formal import] could not load {source}")
69 return
71 imported_graph = imported_loader.get_graph()
72 print(f" [formal import] {len(imported_graph)} triples from {source}")
73 # Context = ontology IRI declared in the document (fallback: source)
74 graph.add_document(imported_graph, source)
76 for nested_uri in list(imported_graph.objects(None, OWL.imports)):
77 _load_formal_import(graph, str(nested_uri), depth + 1, max_depth, visited)
80def _load_informal_import(source: str, target: ProvenanceGraph, visited: set, failed: set,
81 homes: dict) -> None:
82 """Load a single document reached by dereferencing an IRI (informal import) into
83 its own named graph of `target`. Its owl:imports are not followed.
85 Records the outcome: success -> visited + homes[source] (name of its named graph);
86 error -> failed (never retried)."""
87 if source in visited or source in failed:
88 return
89 from lode.reader.loader import Loader # local import: loader imports modules
90 try:
91 g = Loader(source).get_graph()
92 except Exception as e:
93 failed.add(source)
94 print(f" [informal import] could not load {source}: {type(e).__name__}: {e}")
95 return
96 visited.add(source)
97 print(f" [informal import] {len(g)} triples from {source}")
98 homes[source] = target.add_document(g, source)
100# ----------------------------------------------------------
101# IMPORTED / CLOSURE MODULES (formal imports)
102# ----------------------------------------------------------
104def _expand_formal_imports(graph: ProvenanceGraph, max_depth: Optional[int]):
105 """Load every owl:imports of the graph, down to max_depth (None = no limit)."""
106 visited: set = set()
107 for uri in list(graph.objects(None, OWL.imports)):
108 _load_formal_import(graph, str(uri), depth=1, max_depth=max_depth, visited=visited)
109 return graph, visited
112def apply_imported(graph: ProvenanceGraph):
113 """Add the directly imported ontologies (depth 1)."""
114 return _expand_formal_imports(graph, max_depth=1)
117def apply_closure(graph: ProvenanceGraph):
118 """Add the full transitive closure of owl:imports (depth N, max_depth=None)."""
119 return _expand_formal_imports(graph, max_depth=None)
121# ----------------------------------------------------------
122# PARTIAL IMPORT MODULE (informal imports)
123# ----------------------------------------------------------
124#
125# Boundary: an external term receives
126# - what the documented ontology says about it (with priority, per predicate), and
127# - the triples of its "home" graph, i.e. of the ontology that owns it.
128# Redefinitions found in third-party ontologies (formally imported ones included) are
129# NOT copied: they belong to the import closure, available through imported/closure.
130# If the home cannot be reached, the term receives no external definition.
132def _namespace(term) -> str:
133 """Namespace of an IRI, without the trailing separator:
134 'http://ex.org/onto#T' -> 'http://ex.org/onto', 'http://ex.org/voc/t' -> 'http://ex.org/voc'."""
135 s = str(term)
136 return iri_stem(iri_namespace(term))
139def _home(term, homes: dict):
140 """Home graph of the term, or None. Criteria, in order:
142 1. a loaded ontology whose IRI matches the term's namespace;
143 2. a document downloaded from a URL in the same namespace (vocabularies without
144 owl:Ontology, e.g. DC Terms: a single download for all its terms; or BIBO,
145 whose namespace .../bibo/status matches the downloaded URL);
146 3. the document obtained by dereferencing the term itself (ontology IRI that
147 differs from the namespace, e.g. a versioned IRI or a purl redirect).
149 `homes` maps every successfully downloaded source to the name of its named graph."""
150 ns = _namespace(term)
151 for ctx in homes.values():
152 if iri_stem(ctx) == ns:
153 return ctx
154 for source, ctx in homes.items():
155 if iri_stem(source) == ns or _namespace(source) == ns:
156 return ctx
157 return homes.get(urldefrag(str(term))[0])
160def _namespace_failed(term, failed: set) -> bool:
161 """True if a URL in the same namespace has already failed: avoids retrying, term
162 by term, a vocabulary that answers with HTML or does not answer at all."""
163 ns = _namespace(term)
164 return any(iri_stem(f) == ns or _namespace(f) == ns for f in failed)
167def apply_partial_import(graph: ProvenanceGraph):
168 """Dereference (1 hop) the external IRIs mentioned and add ONLY their definition,
169 taken from their home ontology (not the redefinitions in other ontologies)."""
170 # 1. External terms: IRIs mentioned in the graph, excluding the import IRIs, the
171 # core vocabularies and the ontology's own terms.
172 import_iris = set(graph.objects(None, OWL.imports))
173 own = tuple(iri_stem(o) for o in graph.subjects(RDF.type, OWL.Ontology))
174 mentioned = {t for triple in graph for t in triple if isinstance(t, URIRef)} - import_iris
175 external = {t for t in mentioned
176 if not str(t).startswith(_SKIP_NS) and not (own and str(t).startswith(own))}
178 # 2. Load into a separate graph (one named graph per document). The owl:imports
179 # documents are loaded first, as informal imports (no recursion): they are often
180 # the home of the mentioned terms, so they are not downloaded again. Then one
181 # attempt per namespace not covered yet.
182 fetched = ProvenanceGraph()
183 visited, failed, homes = set(), set(), {} # homes: source -> loaded ontology IRI
184 for iri in import_iris:
185 _load_informal_import(str(iri), fetched, visited, failed, homes)
186 for term in external:
187 if _home(term, homes) is None and not _namespace_failed(term, failed):
188 _load_informal_import(urldefrag(str(term))[0], fetched, visited, failed, homes)
190 # 3. Headers of external ontologies (owl:Ontology) are not terms to document.
191 external -= set(fetched.subjects(RDF.type, OWL.Ontology))
193 # 4. Priority to the documented ontology: predicates it already asserts on an
194 # external term (e.g. rdfs:label) are not copied again from the home.
195 base_pairs = {(s, p) for s, p, _ in graph if s in external}
197 # 5. Copy from the home graph only, into its own named graph.
198 seen: set = set()
199 for term in external:
200 home = _home(term, homes)
201 if home is not None: # home unreachable: no external definition
202 _copy_description(fetched, term, graph, seen, skip=base_pairs, ctx=home)
203 return graph, visited
206def _copy_description(src: ProvenanceGraph, node, dst: ProvenanceGraph, seen: set,
207 skip=frozenset(), ctx=None) -> None:
208 """Copy into `dst` the triples of `src` whose subject is `node`, in the same named graph.
210 ctx restricts the copy to a single graph (the home); None = all graphs.
211 skip (subject, predicate) pairs already populated in the base: not overwritten.
212 Blank nodes (restrictions, lists) are copied in full and stay in their parent's
213 context; `seen` prevents cycles and repeated copies."""
214 if node in seen:
215 return
216 seen.add(node)
217 for _, p, o, c in src.quads((node, None, None, ctx)):
218 if (node, p) in skip:
219 continue
220 dst.graph(c).add((node, p, o))
221 if isinstance(o, BNode):
222 _copy_description(src, o, dst, seen, ctx=c)