Skip to content

pynmms.rdf.backends

pynmms.rdf.backends

Graph backends: where the stored graph G and its closure live.

A backend answers membership and pattern queries against the asserted graph and against its closure under the backend's regime (the same graph when no regime is configured). Proof search mostly asks for membership of a handful of triples per node and hands conjunctions to :meth:join; a backend is iterated only when a whole side must be walked (a :class:~pynmms.rdf.view.GraphView read as a set, a procedural rule scanning its trigger predicates during materialisation).

GraphBackend

Bases: Protocol

What a store must provide to serve as the base of a :class:GraphView.

Source code in src/pynmms/rdf/backends/__init__.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@runtime_checkable
class GraphBackend(Protocol):
    """What a store must provide to serve as the base of a :class:`GraphView`."""

    @property
    def generation(self) -> int:
        """Bumped whenever the asserted graph or the closure changes."""
        ...

    @property
    def regime(self) -> Regime | None:
        """The entailment regime whose closure this backend maintains, if any."""
        ...

    @property
    def resolver(self) -> Resolver:
        """Prefix resolver for parsing and display."""
        ...

    def size(self) -> int:
        """Number of asserted triples."""
        ...

    def contains(self, t: Triple) -> bool:
        """Is *t* asserted?"""
        ...

    def triples(self, pattern: Pattern) -> Iterator[Triple]:
        """Asserted triples matching *pattern* (``None`` = wildcard)."""
        ...

    def closure_contains(self, t: Triple) -> bool:
        """Is *t* in cl_R(G)? (Equals ``contains`` when there is no regime.)"""
        ...

    def closure_triples(self, pattern: Pattern) -> Iterator[Triple]:
        """Closure triples matching *pattern*."""
        ...

    def is_inconsistent(self) -> bool:
        """Does the regime derive ⊥ from G alone?"""
        ...

    def join(self, patterns: list[Any], bindings: dict[Any, Node]) -> Iterator[dict[Any, Node]]:
        """All extensions of *bindings* making every pattern a closure triple.

        Patterns contain :class:`~pynmms.rdf.rules.Var` terms. One round trip
        on a remote store; the in-memory backend joins locally.
        """
        ...

generation property

Bumped whenever the asserted graph or the closure changes.

regime property

The entailment regime whose closure this backend maintains, if any.

resolver property

Prefix resolver for parsing and display.

closure_contains(t)

Is t in cl_R(G)? (Equals contains when there is no regime.)

Source code in src/pynmms/rdf/backends/__init__.py 
58
59
60
def closure_contains(self, t: Triple) -> bool:
    """Is *t* in cl_R(G)? (Equals ``contains`` when there is no regime.)"""
    ...

closure_triples(pattern)

Closure triples matching pattern.

Source code in src/pynmms/rdf/backends/__init__.py 
62
63
64
def closure_triples(self, pattern: Pattern) -> Iterator[Triple]:
    """Closure triples matching *pattern*."""
    ...

contains(t)

Is t asserted?

Source code in src/pynmms/rdf/backends/__init__.py 
50
51
52
def contains(self, t: Triple) -> bool:
    """Is *t* asserted?"""
    ...

is_inconsistent()

Does the regime derive ⊥ from G alone?

Source code in src/pynmms/rdf/backends/__init__.py 
66
67
68
def is_inconsistent(self) -> bool:
    """Does the regime derive ⊥ from G alone?"""
    ...

join(patterns, bindings)

All extensions of bindings making every pattern a closure triple.

Patterns contain :class:~pynmms.rdf.rules.Var terms. One round trip on a remote store; the in-memory backend joins locally.

Source code in src/pynmms/rdf/backends/__init__.py 
70
71
72
73
74
75
76
def join(self, patterns: list[Any], bindings: dict[Any, Node]) -> Iterator[dict[Any, Node]]:
    """All extensions of *bindings* making every pattern a closure triple.

    Patterns contain :class:`~pynmms.rdf.rules.Var` terms. One round trip
    on a remote store; the in-memory backend joins locally.
    """
    ...

size()

Number of asserted triples.

Source code in src/pynmms/rdf/backends/__init__.py 
46
47
48
def size(self) -> int:
    """Number of asserted triples."""
    ...

triples(pattern)

Asserted triples matching pattern (None = wildcard).

Source code in src/pynmms/rdf/backends/__init__.py 
54
55
56
def triples(self, pattern: Pattern) -> Iterator[Triple]:
    """Asserted triples matching *pattern* (``None`` = wildcard)."""
    ...

MemoryBackend

rdflib Graph as the store, with the regime closure kept in a second graph.

Parameters:

Name Type Description Default
graph Graph | None

Existing graph to wrap (copied by reference, not duplicated).

 None
regime Regime | None

Entailment regime to materialise; None for no closure.

 None
skolemize bool

Replace blank nodes by Skolem IRIs on load (sound in the antecedent, lem:skolem of the paper).

 True
Source code in src/pynmms/rdf/backends/memory.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
class MemoryBackend:
    """rdflib ``Graph`` as the store, with the regime closure kept in a second graph.

    Parameters:
        graph: Existing graph to wrap (copied by reference, not duplicated).
        regime: Entailment regime to materialise; ``None`` for no closure.
        skolemize: Replace blank nodes by Skolem IRIs on load (sound in the
            antecedent, ``lem:skolem`` of the paper).
    """

    def __init__(
        self,
        graph: Graph | None = None,
        *,
        regime: Regime | None = None,
        skolemize: bool = True,
    ) -> None:
        self._graph = graph if graph is not None else Graph()
        self._regime = regime
        self._skolemize = skolemize
        self._closure: Graph | None = None
        self._inconsistent = False
        self._generation = 0
        self._resolver = Resolver(self._graph)
        if self._skolemize and len(self._graph):
            self._graph = _skolemized(self._graph)
            self._resolver = Resolver(self._graph)
        self._materialize()

    # --- Protocol ---

    @property
    def generation(self) -> int:
        return self._generation

    @property
    def regime(self) -> Regime | None:
        return self._regime

    @property
    def resolver(self) -> Resolver:
        return self._resolver

    @property
    def graph(self) -> Graph:
        return self._graph

    @property
    def closure(self) -> Graph:
        """cl_R(G) (the asserted graph itself when there is no regime)."""
        return self._closure if self._closure is not None else self._graph

    def size(self) -> int:
        return len(self._graph)

    def contains(self, t: Triple) -> bool:
        return t in self._graph

    def triples(self, pattern: Pattern) -> Iterator[Triple]:
        return self._graph.triples(pattern)  # type: ignore[return-value]

    def closure_contains(self, t: Triple) -> bool:
        return t in self.closure

    def closure_triples(self, pattern: Pattern) -> Iterator[Triple]:
        return self.closure.triples(pattern)  # type: ignore[return-value]

    def is_inconsistent(self) -> bool:
        return self._inconsistent

    def join(self, patterns: list[Any], bindings: dict[Any, Node]) -> Iterator[dict[Any, Node]]:
        """Nested-loop join over the closure graph (no batching to be had locally)."""
        from pynmms.rdf.closure import join_patterns

        return join_patterns(patterns, bindings, self.closure_triples)

    def closure_size(self) -> int:
        return len(self.closure)

    # --- Loading and mutation ---

    def load(self, source: str | Path, format: str | None = None) -> int:
        """Parse a file into the graph, Skolemize if configured, re-materialise.

        Returns the number of triples loaded.
        """
        before = len(self._graph)
        t0 = time.perf_counter()
        incoming = Graph()
        incoming.parse(str(source), format=format)
        for prefix, ns in incoming.namespaces():
            self._graph.namespace_manager.bind(prefix, ns, replace=True)
        bnodes = len({n for t in incoming for n in t if isinstance(n, BNode)})
        if self._skolemize and bnodes:
            incoming = _skolemized(incoming)
        for t in incoming:
            self._graph.add(t)
        loaded = len(self._graph) - before
        logger.info(
            "Loaded %d triples from %s (%d blank nodes%s) in %.1f ms",
            loaded, source, bnodes, ", Skolemized" if self._skolemize and bnodes else "",
            (time.perf_counter() - t0) * 1000,
        )
        self._resolver = Resolver(self._graph)
        self._materialize()
        return loaded

    def graphs_of(self, t: Triple) -> list[URIRef]:
        """Named-graph provenance is not kept by the in-memory backend."""
        return []

    def add(self, triples: Iterable[Triple], *, source: URIRef | None = None) -> int:
        """Assert triples and extend the closure incrementally (*source* is not kept)."""
        added = [t for t in triples if t not in self._graph]
        for t in added:
            self._graph.add(t)
        if added:
            if self._closure is not None and self._regime is not None:
                from pynmms.rdf.closure import ClosureEngine

                engine = ClosureEngine(self._regime)
                new, bottom = engine.extend(
                    added, self.closure.triples, lambda t: t in self.closure
                )
                for t in new.triples:
                    self._closure.add(t)
                self._inconsistent = self._inconsistent or bottom
            self._generation += 1
            logger.debug("Added %d triples; generation %d", len(added), self._generation)
        return len(added)

    def _materialize(self) -> None:
        self._generation += 1
        if self._regime is None or (not self._regime.rules and not self._regime.axioms):
            self._closure = None
            self._inconsistent = False
            return
        from pynmms.rdf.closure import ClosureEngine

        t0 = time.perf_counter()
        closed, bottom = ClosureEngine(self._regime).close(iter(self._graph))
        closure = Graph()
        for t in closed:
            closure.add(t)
        self._closure = closure
        self._inconsistent = bottom
        logger.info(
            "Materialised %s closure: %d asserted -> %d triples%s in %.1f ms",
            self._regime.name, len(self._graph), len(closure),
            " (INCONSISTENT)" if bottom else "", (time.perf_counter() - t0) * 1000,
        )

    def __repr__(self) -> str:
        return f"MemoryBackend({len(self._graph)} triples, regime={self._regime})"

closure property

cl_R(G) (the asserted graph itself when there is no regime).

add(triples, *, source=None)

Assert triples and extend the closure incrementally (source is not kept).

Source code in src/pynmms/rdf/backends/memory.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def add(self, triples: Iterable[Triple], *, source: URIRef | None = None) -> int:
    """Assert triples and extend the closure incrementally (*source* is not kept)."""
    added = [t for t in triples if t not in self._graph]
    for t in added:
        self._graph.add(t)
    if added:
        if self._closure is not None and self._regime is not None:
            from pynmms.rdf.closure import ClosureEngine

            engine = ClosureEngine(self._regime)
            new, bottom = engine.extend(
                added, self.closure.triples, lambda t: t in self.closure
            )
            for t in new.triples:
                self._closure.add(t)
            self._inconsistent = self._inconsistent or bottom
        self._generation += 1
        logger.debug("Added %d triples; generation %d", len(added), self._generation)
    return len(added)

graphs_of(t)

Named-graph provenance is not kept by the in-memory backend.

Source code in src/pynmms/rdf/backends/memory.py 
143
144
145
def graphs_of(self, t: Triple) -> list[URIRef]:
    """Named-graph provenance is not kept by the in-memory backend."""
    return []

join(patterns, bindings)

Nested-loop join over the closure graph (no batching to be had locally).

Source code in src/pynmms/rdf/backends/memory.py 
106
107
108
109
110
def join(self, patterns: list[Any], bindings: dict[Any, Node]) -> Iterator[dict[Any, Node]]:
    """Nested-loop join over the closure graph (no batching to be had locally)."""
    from pynmms.rdf.closure import join_patterns

    return join_patterns(patterns, bindings, self.closure_triples)

load(source, format=None)

Parse a file into the graph, Skolemize if configured, re-materialise.

Returns the number of triples loaded.

Source code in src/pynmms/rdf/backends/memory.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def load(self, source: str | Path, format: str | None = None) -> int:
    """Parse a file into the graph, Skolemize if configured, re-materialise.

    Returns the number of triples loaded.
    """
    before = len(self._graph)
    t0 = time.perf_counter()
    incoming = Graph()
    incoming.parse(str(source), format=format)
    for prefix, ns in incoming.namespaces():
        self._graph.namespace_manager.bind(prefix, ns, replace=True)
    bnodes = len({n for t in incoming for n in t if isinstance(n, BNode)})
    if self._skolemize and bnodes:
        incoming = _skolemized(incoming)
    for t in incoming:
        self._graph.add(t)
    loaded = len(self._graph) - before
    logger.info(
        "Loaded %d triples from %s (%d blank nodes%s) in %.1f ms",
        loaded, source, bnodes, ", Skolemized" if self._skolemize and bnodes else "",
        (time.perf_counter() - t0) * 1000,
    )
    self._resolver = Resolver(self._graph)
    self._materialize()
    return loaded

OxigraphBackend

A :class:~pynmms.rdf.backends.GraphBackend over an embedded Oxigraph store.

Parameters:

Name Type Description Default
path str | Path | None

Directory of an on-disk store (created if absent); None for an in-memory store.

 None
regime Regime | None

The entailment regime to materialise. None means the closure is the asserted graph.

 None
skolemize bool

Replace blank nodes by Skolem IRIs on load (lem:skolem).

 True
prefixes dict[str, str] | None

Prefix bindings for parsing and display.

 None
materialize bool

Materialise on construction when the store's recorded regime differs from regime (default). Pass False to load several files first and call :meth:materialize once.

 True
in_memory bool | None

For an on-disk store, compute the closure in a temporary in-memory store and bulk-load the result (fast, but about 600 bytes of memory per closure triple) rather than by rule updates against the disk store (RocksDB-write-bound, no memory cost). None chooses in memory below :data:IN_MEMORY_LIMIT asserted triples.

 None
Source code in src/pynmms/rdf/backends/oxigraph.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
class OxigraphBackend:
    """A :class:`~pynmms.rdf.backends.GraphBackend` over an embedded Oxigraph store.

    Args:
        path: Directory of an on-disk store (created if absent); ``None`` for
            an in-memory store.
        regime: The entailment regime to materialise. ``None`` means the
            closure is the asserted graph.
        skolemize: Replace blank nodes by Skolem IRIs on load (``lem:skolem``).
        prefixes: Prefix bindings for parsing and display.
        materialize: Materialise on construction when the store's recorded
            regime differs from *regime* (default). Pass ``False`` to load
            several files first and call :meth:`materialize` once.
        in_memory: For an on-disk store, compute the closure in a temporary
            in-memory store and bulk-load the result (fast, but about 600
            bytes of memory per closure triple) rather than by rule updates
            against the disk store (RocksDB-write-bound, no memory cost).
            ``None`` chooses in memory below :data:`IN_MEMORY_LIMIT`
            asserted triples.
    """

    def __init__(
        self,
        path: str | Path | None = None,
        *,
        regime: Regime | None = None,
        skolemize: bool = True,
        prefixes: dict[str, str] | None = None,
        materialize: bool = True,
        in_memory: bool | None = None,
    ) -> None:
        self._ox = _ox()
        self.in_memory = in_memory
        ox = self._ox
        self._store = ox.Store(str(path)) if path is not None else ox.Store()
        self._path = Path(path) if path is not None else None
        self._regime = regime
        self._skolemize = skolemize
        self._generation = 0
        self._inconsistent = False
        self._resolver = Resolver(Graph())
        for pfx, ns in (prefixes or {}).items():
            self._resolver.bind(pfx, ns)
        self._default = ox.DefaultGraph()
        self._asserted = ox.NamedNode(ASSERTED) if self._materialising else self._default
        self._meta = ox.NamedNode(META)
        self.round_trips = 0
        if self._materialising:
            if self._recorded_regime() == regime.name:  # type: ignore[union-attr]
                self._inconsistent = self._recorded_inconsistent()
                logger.info("Oxigraph store %s: closure for regime %s already materialised%s",
                            self._path, regime.name,  # type: ignore[union-attr]
                            " (INCONSISTENT)" if self._inconsistent else "")
            elif materialize:
                self.materialize()

    # --- Protocol ---
    @property
    def generation(self) -> int:
        return self._generation

    @property
    def regime(self) -> Regime | None:
        return self._regime

    @property
    def resolver(self) -> Resolver:
        return self._resolver

    @property
    def store(self) -> Any:
        """The underlying ``pyoxigraph.Store``."""
        return self._store

    @property
    def path(self) -> Path | None:
        return self._path

    @property
    def materialised(self) -> bool:
        """Does the store hold ``cl_R(G)`` for this regime? (Always true without one.)"""
        if not self._materialising:
            return True
        assert self._regime is not None
        return self._recorded_regime() == self._regime.name

    @property
    def _materialising(self) -> bool:
        return self._regime is not None and bool(self._regime.rules or self._regime.axioms)

    def size(self) -> int:
        return self._count(self._asserted)

    def closure_size(self) -> int:
        return self._count(self._default)

    def contains(self, t: Triple) -> bool:
        return self._has(t, self._asserted)

    def triples(self, pattern: Pattern) -> Iterator[Triple]:
        return self._match(pattern, self._asserted)

    def closure_contains(self, t: Triple) -> bool:
        return self._has(t, self._default)

    def closure_triples(self, pattern: Pattern) -> Iterator[Triple]:
        return self._match(pattern, self._default)

    def is_inconsistent(self) -> bool:
        return self._inconsistent

    def join(self, patterns: list[Any], bindings: dict[Any, Node]) -> Iterator[dict[Any, Node]]:
        """One ``SELECT`` over the closure for the whole conjunction (an ``ASK``
        when every term is bound)."""
        from pynmms.rdf.sparql_rules import build_select

        bgp, query, free, names = build_select(patterns, bindings)
        if not free:
            yield from ([dict(bindings)] if bool(self._query(f"ASK {{ {bgp} }}")) else [])
            return
        for row in self._query(query):
            b = dict(bindings)
            for v in free:
                val = row[names[v][1:]]
                if val is None:
                    break
                b[v] = from_ox(val, self._ox)
            else:
                yield b

    # --- Loading and mutation ---
    def load(self, source: str | Path, format: str | None = None, *,
             materialize: bool = True) -> int:
        """Load an RDF file into the store, Skolemize, and (re)materialise.

        Parsing is Oxigraph's own; *format* is a MIME type or guessed from
        the extension. Turtle ``@prefix`` lines are bound in the resolver.
        Returns the number of asserted triples added. With
        ``materialize=False`` the closure is left stale until
        :meth:`materialize` is called.
        """
        ox = self._ox
        path = Path(source)
        mime = format or _FORMATS.get(path.suffix.lower())
        if mime is None:
            raise ValueError(f"Cannot guess the RDF format of {source}")
        before = self.size()
        t0 = time.perf_counter()
        with open(path, "rb") as fh:
            self._store.bulk_load(fh, format=ox.RdfFormat.from_media_type(mime),
                                  to_graph=self._asserted)
        if self._skolemize:
            self._skolemize_graph()
        for pfx, ns in _file_prefixes(path).items():
            self._resolver.bind(pfx, ns)
        added = self.size() - before
        logger.info("Loaded %d triples from %s in %.1f ms", added, source,
                    (time.perf_counter() - t0) * 1000)
        if self._materialising and materialize:
            self.materialize()
        else:
            self._generation += 1
        return added

    def dump(self, destination: str | Path, format: str | None = None) -> None:
        """Write the asserted graph to a file (format from the extension)."""
        ox = self._ox
        path = Path(destination)
        mime = format or _FORMATS.get(path.suffix.lower())
        if mime is None:
            raise ValueError(f"Cannot guess the RDF format of {destination}")
        with open(path, "wb") as fh:
            self._store.dump(fh, format=ox.RdfFormat.from_media_type(mime),
                             from_graph=self._asserted)

    def materialize(self) -> None:
        """Recompute ``cl_R(G)`` in the store from the asserted graph.

        An in-memory store runs the rules in place. An on-disk store does so
        too when ``in_memory=False`` was given, or when it was left to choose
        and holds more than :data:`IN_MEMORY_LIMIT` asserted triples;
        otherwise it computes the closure in a temporary in-memory store and
        bulk-loads the result, since RocksDB writes made the rule updates
        five times slower than in memory, slower than the Python engine at
        two million closure triples.
        """
        if not self._materialising:
            self._generation += 1
            return
        if self._path is None:
            scratch_in_memory = False
        elif self.in_memory is not None:
            scratch_in_memory = self.in_memory
        else:
            scratch_in_memory = self.size() <= IN_MEMORY_LIMIT
        if not scratch_in_memory:
            self._reset_closure()
            self._materialize()
            return
        import tempfile

        ox = self._ox
        disk = self._store
        t0 = time.perf_counter()
        with tempfile.TemporaryDirectory() as tmp:
            dump = Path(tmp) / "graph.nt"
            disk.dump(str(dump), format=ox.RdfFormat.N_TRIPLES, from_graph=self._asserted)
            scratch = ox.Store()
            scratch.bulk_load(path=str(dump), format=ox.RdfFormat.N_TRIPLES)
            self._store = scratch
            try:
                self._asserted, asserted_on_disk = self._default, self._asserted
                self._materialize(asserted=self._count(self._default), record=False)
            finally:
                self._store, self._asserted = disk, asserted_on_disk
            scratch.dump(str(dump), format=ox.RdfFormat.N_TRIPLES, from_graph=self._default)
            del scratch
            self._update("DROP SILENT DEFAULT")
            disk.bulk_load(path=str(dump), format=ox.RdfFormat.N_TRIPLES)
        self._record_meta()
        logger.info("Closure written to %s in %.1f ms total", self._path,
                    (time.perf_counter() - t0) * 1000)

    def load_graph(self, graph: Graph, *, source: URIRef | None = None) -> int:
        """Load an rdflib graph (its namespace bindings included).

        With *source*, the triples are also recorded in that named graph, so
        :meth:`graphs_of` reports them as coming from it (provenance).
        """
        for pfx, ns in graph.namespaces():
            self._resolver.bind(pfx, str(ns))
        triples = [skolemize_triple(t) if self._skolemize else t for t in graph]
        before = self.size()
        self._insert(triples, self._asserted)
        if source is not None:
            self._insert(triples, self._ox.NamedNode(str(source)))
        self.materialize()
        return self.size() - before

    def graphs_of(self, t: Triple) -> list[URIRef]:
        """The named graphs (sources) holding *t*, the backend's own graphs excluded."""
        ox = self._ox
        s, p, o = t
        if isinstance(s, Literal):
            return []
        out: list[URIRef] = []
        for q in self._store.quads_for_pattern(to_ox(s, ox), to_ox(p, ox), to_ox(o, ox), None):
            g = q.graph_name
            if isinstance(g, ox.NamedNode) and g.value not in (ASSERTED, META):
                out.append(URIRef(g.value))
        return sorted(out)

    def add(self, triples: Iterable[Triple], *, source: URIRef | None = None) -> int:
        """Assert triples and extend the closure incrementally in process.

        The new triples are closed by :meth:`ClosureEngine.extend` with the
        store as the join partner, and the result is written to the closure
        graph; the store is never re-materialised. With *source*, they are
        also recorded in that named graph.
        """
        triples = list(triples)
        added = [t for t in triples if not self.contains(t)]
        if source is not None and triples:
            self._insert(triples, self._ox.NamedNode(str(source)))
        if not added:
            return 0
        self._insert(added, self._asserted)
        if self._materialising:
            from pynmms.rdf.closure import ClosureEngine

            assert self._regime is not None
            new, bottom = ClosureEngine(self._regime).extend(
                added, self.closure_triples, self.closure_contains, store_join=self.join
            )
            self._insert(list(new.triples), self._default)
            if bottom and not self._inconsistent:
                self._inconsistent = True
                self._record_meta()
        else:
            self._insert(added, self._default) if self._asserted is not self._default else None
        self._generation += 1
        logger.debug("Added %d triples; generation %d", len(added), self._generation)
        return len(added)

    # --- Materialisation in the store ---
    def _materialize(self, asserted: int | None = None, *, record: bool = True) -> None:
        """Run the regime to a fixpoint inside the store.

        Pattern rules run as SPARQL updates; guarded and procedural rules run
        in process against the store between rounds, until neither adds a
        triple. ⊥ rules are ASKs (``prop:incoherence``). With ``record=False``
        the meta graph is left to the caller (the scratch store of
        :meth:`materialize` is discarded, so recording there is wasted).
        """
        from pynmms.rdf.sparql_rules import partition

        assert self._regime is not None
        rules = partition(self._regime, skip=UNSTORABLE_RULES)
        t0 = time.perf_counter()
        self._insert(list(self._regime.axioms), self._default)
        rounds = 0
        while True:
            rounds += self._sparql_fixpoint(rules.updates)
            if not self._in_process_round(rules.in_process):
                break
        bottom = any(bool(self._query(q)) for q in rules.asks) or self._in_process_bottom(
            rules.in_process
        )
        self._inconsistent = bottom
        self._generation += 1
        if record:
            self._record_meta()
        logger.info(
            "Materialised %s closure in the store: %d asserted -> %d triples%s in %d round(s), "
            "%.1f ms (%d store-side rules, %d in process%s)",
            self._regime.name, self.size() if asserted is None else asserted, self.closure_size(),
            " (INCONSISTENT)" if bottom else "", rounds, (time.perf_counter() - t0) * 1000,
            rules.store_side, len(rules.in_process),
            f", skipped {', '.join(rules.skipped)}" if rules.skipped else "",
        )

    def _sparql_fixpoint(self, updates: tuple[str, ...]) -> int:
        rounds = 0
        while updates:
            before = self.closure_size()
            for u in updates:
                self._update(u)
            rounds += 1
            if self.closure_size() == before:
                break
        return rounds

    def _in_process_round(self, rules: tuple[Any, ...]) -> bool:
        """One pass of the guarded and procedural rules over the closure graph."""
        from pynmms.rdf.rules import ProceduralRule, Rule

        new: list[Triple] = []
        for rule in rules:
            if isinstance(rule, ProceduralRule):
                for concl in self._fire_procedural(rule):
                    if concl is not None and not self.closure_contains(concl):
                        new.append(concl)
            elif isinstance(rule, Rule) and rule.conclusion is not None:
                for b in self.join(list(rule.premises), {}):
                    if rule.guard is None or rule.guard(b):
                        bound: Triple = tuple(b.get(t, t) for t in rule.conclusion)  # type: ignore[assignment]
                        if not isinstance(bound[0], Literal) and not self.closure_contains(bound):
                            new.append(bound)
        if not new:
            return False
        self._insert(new, self._default)
        return True

    def _in_process_bottom(self, rules: tuple[Any, ...]) -> bool:
        from pynmms.rdf.rules import ProceduralRule, Rule

        for rule in rules:
            if isinstance(rule, ProceduralRule):
                if any(c is None for c in self._fire_procedural(rule)):
                    return True
            elif isinstance(rule, Rule) and rule.conclusion is None:
                for b in self.join(list(rule.premises), {}):
                    if rule.guard is None or rule.guard(b):
                        return True
        return False

    def _fire_procedural(self, rule: Any) -> Iterator[Triple | None]:
        triggers = rule.triggers or (None,)
        seen: set[Triple] = set()
        for p in triggers:
            for t in self.closure_triples((None, p, None)):
                if t in seen:
                    continue
                seen.add(t)
                yield from rule.fire(t, self.closure_triples)

    # --- Store plumbing ---
    def _reset_closure(self) -> None:
        """Make the default graph a copy of the asserted graph."""
        if self._asserted is self._default:
            return
        self._update(f"DROP SILENT DEFAULT; ADD SILENT <{ASSERTED}> TO DEFAULT")

    def _skolemize_graph(self) -> None:
        ox = self._ox
        rows = list(self._query(
            f"SELECT ?s ?p ?o WHERE {{ GRAPH <{ASSERTED}> {{ ?s ?p ?o }} "
            f"FILTER(isBlank(?s) || isBlank(?o)) }}"
        )) if self._asserted is not self._default else list(self._query(
            "SELECT ?s ?p ?o WHERE { ?s ?p ?o FILTER(isBlank(?s) || isBlank(?o)) }"
        ))
        if not rows:
            return
        old = [ox.Quad(r["s"], r["p"], r["o"], self._asserted) for r in rows]
        new = [skolemize_triple((from_ox(r["s"], ox), from_ox(r["p"], ox), from_ox(r["o"], ox)))
               for r in rows]
        for q in old:
            self._store.remove(q)
        self._insert(new, self._asserted)
        logger.info("Skolemized %d triple(s) with blank nodes", len(rows))

    def _insert(self, triples: list[Triple], graph: Any) -> None:
        ox = self._ox
        quads = []
        for s, p, o in triples:
            if isinstance(s, Literal):
                continue  # generalized triple; not storable
            quads.append(ox.Quad(to_ox(s, ox), to_ox(p, ox), to_ox(o, ox), graph))
        if quads:
            self._store.extend(quads)

    def _has(self, t: Triple, graph: Any) -> bool:
        ox = self._ox
        s, p, o = t
        if isinstance(s, Literal):
            return False
        self.round_trips += 1
        return ox.Quad(to_ox(s, ox), to_ox(p, ox), to_ox(o, ox), graph) in self._store

    def _match(self, pattern: Pattern, graph: Any) -> Iterator[Triple]:
        ox = self._ox
        s, p, o = pattern
        if isinstance(s, Literal):
            return
        self.round_trips += 1
        for q in self._store.quads_for_pattern(
            None if s is None else to_ox(s, ox), None if p is None else to_ox(p, ox),
            None if o is None else to_ox(o, ox), graph,
        ):
            yield (from_ox(q.subject, ox), from_ox(q.predicate, ox), from_ox(q.object, ox))

    def _count(self, graph: Any) -> int:
        clause = ("{ ?s ?p ?o }" if graph is self._default
                  else f"{{ GRAPH <{ASSERTED}> {{ ?s ?p ?o }} }}")
        row = next(iter(self._query(f"SELECT (COUNT(*) AS ?n) WHERE {clause}")))
        return int(row["n"].value)

    def _query(self, q: str) -> Any:
        self.round_trips += 1
        return self._store.query(q)

    def _update(self, u: str) -> None:
        self.round_trips += 1
        self._store.update(u)

    def _record_meta(self) -> None:
        assert self._regime is not None
        self._update(
            f"DROP SILENT GRAPH <{META}>; INSERT DATA {{ GRAPH <{META}> {{ "
            f"<{_STORE}> <{_REGIME}> {Literal(self._regime.name).n3()} ; "
            f"<{_INCONSISTENT}> {Literal(self._inconsistent).n3()} }} }}"
        )

    def _recorded_regime(self) -> str | None:
        q = f"SELECT ?r WHERE {{ GRAPH <{META}> {{ <{_STORE}> <{_REGIME}> ?r }} }}"
        for row in self._query(q):
            return str(row["r"].value)
        return None

    def _recorded_inconsistent(self) -> bool:
        for row in self._query(
            f"SELECT ?i WHERE {{ GRAPH <{META}> {{ <{_STORE}> <{_INCONSISTENT}> ?i }} }}"
        ):
            return str(row["i"].value) == "true"
        return False

    def close(self) -> None:
        """Release the store (an on-disk store holds a lock while open)."""
        self._store = None

    def __enter__(self) -> OxigraphBackend:
        return self

    def __exit__(self, *exc: object) -> None:
        self.close()

    def __repr__(self) -> str:
        where = str(self._path) if self._path else "memory"
        return f"OxigraphBackend({where}, {self.size()} triples, regime={self._regime})"

materialised property

Does the store hold cl_R(G) for this regime? (Always true without one.)

store property

The underlying pyoxigraph.Store.

add(triples, *, source=None)

Assert triples and extend the closure incrementally in process.

The new triples are closed by :meth:ClosureEngine.extend with the store as the join partner, and the result is written to the closure graph; the store is never re-materialised. With source, they are also recorded in that named graph.

Source code in src/pynmms/rdf/backends/oxigraph.py 
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
def add(self, triples: Iterable[Triple], *, source: URIRef | None = None) -> int:
    """Assert triples and extend the closure incrementally in process.

    The new triples are closed by :meth:`ClosureEngine.extend` with the
    store as the join partner, and the result is written to the closure
    graph; the store is never re-materialised. With *source*, they are
    also recorded in that named graph.
    """
    triples = list(triples)
    added = [t for t in triples if not self.contains(t)]
    if source is not None and triples:
        self._insert(triples, self._ox.NamedNode(str(source)))
    if not added:
        return 0
    self._insert(added, self._asserted)
    if self._materialising:
        from pynmms.rdf.closure import ClosureEngine

        assert self._regime is not None
        new, bottom = ClosureEngine(self._regime).extend(
            added, self.closure_triples, self.closure_contains, store_join=self.join
        )
        self._insert(list(new.triples), self._default)
        if bottom and not self._inconsistent:
            self._inconsistent = True
            self._record_meta()
    else:
        self._insert(added, self._default) if self._asserted is not self._default else None
    self._generation += 1
    logger.debug("Added %d triples; generation %d", len(added), self._generation)
    return len(added)

close()

Release the store (an on-disk store holds a lock while open).

Source code in src/pynmms/rdf/backends/oxigraph.py 
601
602
603
def close(self) -> None:
    """Release the store (an on-disk store holds a lock while open)."""
    self._store = None

dump(destination, format=None)

Write the asserted graph to a file (format from the extension).

Source code in src/pynmms/rdf/backends/oxigraph.py 
299
300
301
302
303
304
305
306
307
308
def dump(self, destination: str | Path, format: str | None = None) -> None:
    """Write the asserted graph to a file (format from the extension)."""
    ox = self._ox
    path = Path(destination)
    mime = format or _FORMATS.get(path.suffix.lower())
    if mime is None:
        raise ValueError(f"Cannot guess the RDF format of {destination}")
    with open(path, "wb") as fh:
        self._store.dump(fh, format=ox.RdfFormat.from_media_type(mime),
                         from_graph=self._asserted)

graphs_of(t)

The named graphs (sources) holding t, the backend's own graphs excluded.

Source code in src/pynmms/rdf/backends/oxigraph.py 
374
375
376
377
378
379
380
381
382
383
384
385
def graphs_of(self, t: Triple) -> list[URIRef]:
    """The named graphs (sources) holding *t*, the backend's own graphs excluded."""
    ox = self._ox
    s, p, o = t
    if isinstance(s, Literal):
        return []
    out: list[URIRef] = []
    for q in self._store.quads_for_pattern(to_ox(s, ox), to_ox(p, ox), to_ox(o, ox), None):
        g = q.graph_name
        if isinstance(g, ox.NamedNode) and g.value not in (ASSERTED, META):
            out.append(URIRef(g.value))
    return sorted(out)

join(patterns, bindings)

One SELECT over the closure for the whole conjunction (an ASK when every term is bound).

Source code in src/pynmms/rdf/backends/oxigraph.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
def join(self, patterns: list[Any], bindings: dict[Any, Node]) -> Iterator[dict[Any, Node]]:
    """One ``SELECT`` over the closure for the whole conjunction (an ``ASK``
    when every term is bound)."""
    from pynmms.rdf.sparql_rules import build_select

    bgp, query, free, names = build_select(patterns, bindings)
    if not free:
        yield from ([dict(bindings)] if bool(self._query(f"ASK {{ {bgp} }}")) else [])
        return
    for row in self._query(query):
        b = dict(bindings)
        for v in free:
            val = row[names[v][1:]]
            if val is None:
                break
            b[v] = from_ox(val, self._ox)
        else:
            yield b

load(source, format=None, *, materialize=True)

Load an RDF file into the store, Skolemize, and (re)materialise.

Parsing is Oxigraph's own; format is a MIME type or guessed from the extension. Turtle @prefix lines are bound in the resolver. Returns the number of asserted triples added. With materialize=False the closure is left stale until :meth:materialize is called.

Source code in src/pynmms/rdf/backends/oxigraph.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def load(self, source: str | Path, format: str | None = None, *,
         materialize: bool = True) -> int:
    """Load an RDF file into the store, Skolemize, and (re)materialise.

    Parsing is Oxigraph's own; *format* is a MIME type or guessed from
    the extension. Turtle ``@prefix`` lines are bound in the resolver.
    Returns the number of asserted triples added. With
    ``materialize=False`` the closure is left stale until
    :meth:`materialize` is called.
    """
    ox = self._ox
    path = Path(source)
    mime = format or _FORMATS.get(path.suffix.lower())
    if mime is None:
        raise ValueError(f"Cannot guess the RDF format of {source}")
    before = self.size()
    t0 = time.perf_counter()
    with open(path, "rb") as fh:
        self._store.bulk_load(fh, format=ox.RdfFormat.from_media_type(mime),
                              to_graph=self._asserted)
    if self._skolemize:
        self._skolemize_graph()
    for pfx, ns in _file_prefixes(path).items():
        self._resolver.bind(pfx, ns)
    added = self.size() - before
    logger.info("Loaded %d triples from %s in %.1f ms", added, source,
                (time.perf_counter() - t0) * 1000)
    if self._materialising and materialize:
        self.materialize()
    else:
        self._generation += 1
    return added

load_graph(graph, *, source=None)

Load an rdflib graph (its namespace bindings included).

With source, the triples are also recorded in that named graph, so :meth:graphs_of reports them as coming from it (provenance).

Source code in src/pynmms/rdf/backends/oxigraph.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def load_graph(self, graph: Graph, *, source: URIRef | None = None) -> int:
    """Load an rdflib graph (its namespace bindings included).

    With *source*, the triples are also recorded in that named graph, so
    :meth:`graphs_of` reports them as coming from it (provenance).
    """
    for pfx, ns in graph.namespaces():
        self._resolver.bind(pfx, str(ns))
    triples = [skolemize_triple(t) if self._skolemize else t for t in graph]
    before = self.size()
    self._insert(triples, self._asserted)
    if source is not None:
        self._insert(triples, self._ox.NamedNode(str(source)))
    self.materialize()
    return self.size() - before

materialize()

Recompute cl_R(G) in the store from the asserted graph.

An in-memory store runs the rules in place. An on-disk store does so too when in_memory=False was given, or when it was left to choose and holds more than :data:IN_MEMORY_LIMIT asserted triples; otherwise it computes the closure in a temporary in-memory store and bulk-loads the result, since RocksDB writes made the rule updates five times slower than in memory, slower than the Python engine at two million closure triples.

Source code in src/pynmms/rdf/backends/oxigraph.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
def materialize(self) -> None:
    """Recompute ``cl_R(G)`` in the store from the asserted graph.

    An in-memory store runs the rules in place. An on-disk store does so
    too when ``in_memory=False`` was given, or when it was left to choose
    and holds more than :data:`IN_MEMORY_LIMIT` asserted triples;
    otherwise it computes the closure in a temporary in-memory store and
    bulk-loads the result, since RocksDB writes made the rule updates
    five times slower than in memory, slower than the Python engine at
    two million closure triples.
    """
    if not self._materialising:
        self._generation += 1
        return
    if self._path is None:
        scratch_in_memory = False
    elif self.in_memory is not None:
        scratch_in_memory = self.in_memory
    else:
        scratch_in_memory = self.size() <= IN_MEMORY_LIMIT
    if not scratch_in_memory:
        self._reset_closure()
        self._materialize()
        return
    import tempfile

    ox = self._ox
    disk = self._store
    t0 = time.perf_counter()
    with tempfile.TemporaryDirectory() as tmp:
        dump = Path(tmp) / "graph.nt"
        disk.dump(str(dump), format=ox.RdfFormat.N_TRIPLES, from_graph=self._asserted)
        scratch = ox.Store()
        scratch.bulk_load(path=str(dump), format=ox.RdfFormat.N_TRIPLES)
        self._store = scratch
        try:
            self._asserted, asserted_on_disk = self._default, self._asserted
            self._materialize(asserted=self._count(self._default), record=False)
        finally:
            self._store, self._asserted = disk, asserted_on_disk
        scratch.dump(str(dump), format=ox.RdfFormat.N_TRIPLES, from_graph=self._default)
        del scratch
        self._update("DROP SILENT DEFAULT")
        disk.bulk_load(path=str(dump), format=ox.RdfFormat.N_TRIPLES)
    self._record_meta()
    logger.info("Closure written to %s in %.1f ms total", self._path,
                (time.perf_counter() - t0) * 1000)

SPARQLBackend

A remote (or local server) graph reached over SPARQL.

Parameters:

Name Type Description Default
endpoint str

Query endpoint URL.

 required
regime Regime | None

The regime the endpoint materialises (declared, not enforced).

 None
probe tuple[Triple, ...] | None

Optional triples that the endpoint must entail under the declared regime; raises ValueError otherwise.

 None
update_endpoint str | None

SPARQL UPDATE endpoint for :meth:add.

 None
prefixes dict[str, str] | None

{prefix: namespace} for parsing and display; endpoints do not expose their prefix declarations to clients.

 None
Source code in src/pynmms/rdf/backends/sparql.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
class SPARQLBackend:
    """A remote (or local server) graph reached over SPARQL.

    Parameters:
        endpoint: Query endpoint URL.
        regime: The regime the endpoint materialises (declared, not enforced).
        probe: Optional triples that the endpoint must entail under the
            declared regime; raises ``ValueError`` otherwise.
        update_endpoint: SPARQL UPDATE endpoint for :meth:`add`.
        prefixes: ``{prefix: namespace}`` for parsing and display; endpoints
            do not expose their prefix declarations to clients.
    """

    def __init__(
        self,
        endpoint: str,
        *,
        regime: Regime | None = None,
        probe: tuple[Triple, ...] | None = None,
        update_endpoint: str | None = None,
        prefixes: dict[str, str] | None = None,
    ) -> None:
        self.endpoint = endpoint
        self.update_endpoint = update_endpoint
        self._regime = regime
        self._store: SPARQLStore
        if update_endpoint is not None:
            self._store = SPARQLUpdateStore(endpoint, update_endpoint)
        else:
            self._store = SPARQLStore(endpoint)
        # The default graph: a blank-node identifier would be rejected by
        # SPARQL UPDATE, and the endpoint's prefixes are not visible to the
        # client, so they are declared here.
        self._graph = Graph(store=self._store, identifier=DATASET_DEFAULT_GRAPH_ID)
        self._resolver = Resolver(self._graph)
        for prefix, ns in (prefixes or {}).items():
            self._resolver.bind(prefix, ns)
        self._generation = 1
        self._calls = 0
        self._latency = 0.0
        # Per-generation memo: proof search asks the same memberships and
        # joins many times (every firing checks its conclusions; every query
        # re-derives the same schema-level triples). Cleared on bump().
        self._size: int | None = None
        self._contains_memo: dict[Triple, bool] = {}
        self._join_memo: dict[tuple[tuple[Any, ...], tuple[tuple[Any, Any], ...]], list] = {}
        self.memo_limit = 100_000
        if probe is not None:
            for t in probe:
                if not self.closure_contains(t):
                    raise ValueError(
                        f"Endpoint {endpoint} does not entail probe triple {t}; "
                        f"declared regime {regime} is not materialised"
                    )

    @property
    def generation(self) -> int:
        return self._generation

    @property
    def regime(self) -> Regime | None:
        return self._regime

    @property
    def resolver(self) -> Resolver:
        return self._resolver

    @property
    def stats(self) -> dict[str, float]:
        """Round trips and cumulative latency, for post-run analysis."""
        return {"calls": self._calls, "latency_ms": self._latency * 1000}

    def bump(self) -> None:
        """Signal that the remote graph changed (invalidates views and caches)."""
        self._generation += 1
        self._size = None
        self._contains_memo.clear()
        self._join_memo.clear()

    def _timed(self, what: str):  # type: ignore[no-untyped-def]
        backend = self

        class _T:
            def __enter__(self) -> None:
                self.t0 = time.perf_counter()

            def __exit__(self, *exc: object) -> None:
                dt = time.perf_counter() - self.t0
                backend._calls += 1
                backend._latency += dt
                logger.debug("SPARQL %s: %.1f ms", what, dt * 1000)

        return _T()

    def size(self) -> int:
        if self._size is None:
            with self._timed("count"):
                rows = list(self._graph.query("SELECT (COUNT(*) AS ?n) WHERE { ?s ?p ?o }"))
            self._size = int(str(rows[0][0]))  # type: ignore[index]
        return self._size

    def contains(self, t: Triple) -> bool:
        hit = self._contains_memo.get(t)
        if hit is not None:
            return hit
        with self._timed("ask"):
            found = t in self._graph
        if len(self._contains_memo) < self.memo_limit:
            self._contains_memo[t] = found
        return found

    def triples(self, pattern: Pattern) -> Iterator[Triple]:
        with self._timed("triples"):
            return iter(list(self._graph.triples(pattern)))  # type: ignore[arg-type]

    # The endpoint's materialisation is the closure.
    closure_contains = contains
    closure_triples = triples

    def is_inconsistent(self) -> bool:
        return False

    def join(self, patterns: list[Any], bindings: dict[Any, Node]) -> Iterator[dict[Any, Node]]:
        """One ``SELECT`` for the whole conjunction, bound terms inlined."""
        from pynmms.rdf.sparql_rules import build_select

        _bgp, query, free, _names = build_select(patterns, bindings)
        key = (tuple(patterns), tuple(sorted(bindings.items(), key=lambda kv: str(kv[0]))))
        rows = self._join_memo.get(key)
        if rows is None:
            with self._timed(f"join {len(patterns)}"):
                rows = list(self._graph.query(query))
            if len(self._join_memo) < self.memo_limit:
                self._join_memo[key] = rows
        out: list[dict[Any, Node]] = []
        for row in rows:
            b = dict(bindings)
            for v, val in zip(free, row):  # type: ignore[arg-type]
                b[v] = val
            out.append(b)
        return iter(out)

    BULK_CHUNK = 500

    def add(self, triples: Iterable[Triple]) -> int:
        """Insert triples through the update endpoint in chunked ``INSERT DATA`` updates.

        The store is expected to maintain its own materialisation; the
        generation is bumped once per call so views and caches invalidate.
        """
        if self.update_endpoint is None:
            raise ValueError("SPARQLBackend.add needs an update_endpoint")
        batch = list(triples)
        if not batch:
            return 0
        for t in batch:
            if any(isinstance(n, BNode) for n in t):
                raise ValueError("SPARQLBackend.add cannot insert blank nodes; Skolemize first")
        for start in range(0, len(batch), self.BULK_CHUNK):
            chunk = batch[start:start + self.BULK_CHUNK]
            data = " ".join(f"{s.n3()} {p.n3()} {o.n3()} ." for s, p, o in chunk)
            with self._timed(f"insert {len(chunk)}"):
                # Call the store directly: Graph.update() would wrap the update
                # in the graph's identifier, which single-graph endpoints reject.
                self._store.update(f"INSERT DATA {{ {data} }}")
        self.bump()
        logger.info("Inserted %d triples into %s in %d update(s); generation %d",
                    len(batch), self.update_endpoint,
                    -(-len(batch) // self.BULK_CHUNK), self._generation)
        return len(batch)

    def __repr__(self) -> str:
        return f"SPARQLBackend({self.endpoint!r}, regime={self._regime})"

stats property

Round trips and cumulative latency, for post-run analysis.

add(triples)

Insert triples through the update endpoint in chunked INSERT DATA updates.

The store is expected to maintain its own materialisation; the generation is bumped once per call so views and caches invalidate.

Source code in src/pynmms/rdf/backends/sparql.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def add(self, triples: Iterable[Triple]) -> int:
    """Insert triples through the update endpoint in chunked ``INSERT DATA`` updates.

    The store is expected to maintain its own materialisation; the
    generation is bumped once per call so views and caches invalidate.
    """
    if self.update_endpoint is None:
        raise ValueError("SPARQLBackend.add needs an update_endpoint")
    batch = list(triples)
    if not batch:
        return 0
    for t in batch:
        if any(isinstance(n, BNode) for n in t):
            raise ValueError("SPARQLBackend.add cannot insert blank nodes; Skolemize first")
    for start in range(0, len(batch), self.BULK_CHUNK):
        chunk = batch[start:start + self.BULK_CHUNK]
        data = " ".join(f"{s.n3()} {p.n3()} {o.n3()} ." for s, p, o in chunk)
        with self._timed(f"insert {len(chunk)}"):
            # Call the store directly: Graph.update() would wrap the update
            # in the graph's identifier, which single-graph endpoints reject.
            self._store.update(f"INSERT DATA {{ {data} }}")
    self.bump()
    logger.info("Inserted %d triples into %s in %d update(s); generation %d",
                len(batch), self.update_endpoint,
                -(-len(batch) // self.BULK_CHUNK), self._generation)
    return len(batch)

bump()

Signal that the remote graph changed (invalidates views and caches).

Source code in src/pynmms/rdf/backends/sparql.py 
107
108
109
110
111
112
def bump(self) -> None:
    """Signal that the remote graph changed (invalidates views and caches)."""
    self._generation += 1
    self._size = None
    self._contains_memo.clear()
    self._join_memo.clear()

join(patterns, bindings)

One SELECT for the whole conjunction, bound terms inlined.

Source code in src/pynmms/rdf/backends/sparql.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def join(self, patterns: list[Any], bindings: dict[Any, Node]) -> Iterator[dict[Any, Node]]:
    """One ``SELECT`` for the whole conjunction, bound terms inlined."""
    from pynmms.rdf.sparql_rules import build_select

    _bgp, query, free, _names = build_select(patterns, bindings)
    key = (tuple(patterns), tuple(sorted(bindings.items(), key=lambda kv: str(kv[0]))))
    rows = self._join_memo.get(key)
    if rows is None:
        with self._timed(f"join {len(patterns)}"):
            rows = list(self._graph.query(query))
        if len(self._join_memo) < self.memo_limit:
            self._join_memo[key] = rows
    out: list[dict[Any, Node]] = []
    for row in rows:
        b = dict(bindings)
        for v, val in zip(free, row):  # type: ignore[arg-type]
            b[v] = val
        out.append(b)
    return iter(out)

pynmms.rdf.backends.oxigraph

pynmms.rdf.backends.oxigraph

Oxigraph backend: an embedded store that materialises the regime itself.

Oxigraph <https://github.com/oxigraph/oxigraph>_ is a Rust triplestore with a SPARQL 1.1 engine, usable in process through pyoxigraph either in memory or on disk (RocksDB). This backend keeps the asserted graph in the named graph <urn:pynmms:asserted> and cl_R(G) in the default graph, and computes the closure inside the store: every pattern rule of the regime runs as a SPARQL INSERT ... WHERE update (:mod:pynmms.rdf.sparql_rules), iterated to a fixpoint, with the rules a store cannot run (guarded rules, :class:~pynmms.rdf.rules.ProceduralRule\s) evaluated in process against the store between rounds. Membership, pattern, and join queries are then microseconds, and a store on disk keeps its closure across sessions: reopening a path whose recorded regime matches skips materialisation.

add() extends the closure semi-naively through :meth:~pynmms.rdf.closure.ClosureEngine.extend with the store as the join partner, so a TELL of a few triples costs milliseconds at any size.

Rules whose conclusions are generalized triples with a literal subject (rdfs1, rdfD1) cannot be stored and are skipped, so literal typing is absent from the materialised closure; the in-process extras closure still derives them for a query's own triples.

Requires pip install pyoxigraph (the oxigraph extra).

OxigraphBackend

A :class:~pynmms.rdf.backends.GraphBackend over an embedded Oxigraph store.

Parameters:

Name Type Description Default
path str | Path | None

Directory of an on-disk store (created if absent); None for an in-memory store.

 None
regime Regime | None

The entailment regime to materialise. None means the closure is the asserted graph.

 None
skolemize bool

Replace blank nodes by Skolem IRIs on load (lem:skolem).

 True
prefixes dict[str, str] | None

Prefix bindings for parsing and display.

 None
materialize bool

Materialise on construction when the store's recorded regime differs from regime (default). Pass False to load several files first and call :meth:materialize once.

 True
in_memory bool | None

For an on-disk store, compute the closure in a temporary in-memory store and bulk-load the result (fast, but about 600 bytes of memory per closure triple) rather than by rule updates against the disk store (RocksDB-write-bound, no memory cost). None chooses in memory below :data:IN_MEMORY_LIMIT asserted triples.

 None
Source code in src/pynmms/rdf/backends/oxigraph.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
class OxigraphBackend:
    """A :class:`~pynmms.rdf.backends.GraphBackend` over an embedded Oxigraph store.

    Args:
        path: Directory of an on-disk store (created if absent); ``None`` for
            an in-memory store.
        regime: The entailment regime to materialise. ``None`` means the
            closure is the asserted graph.
        skolemize: Replace blank nodes by Skolem IRIs on load (``lem:skolem``).
        prefixes: Prefix bindings for parsing and display.
        materialize: Materialise on construction when the store's recorded
            regime differs from *regime* (default). Pass ``False`` to load
            several files first and call :meth:`materialize` once.
        in_memory: For an on-disk store, compute the closure in a temporary
            in-memory store and bulk-load the result (fast, but about 600
            bytes of memory per closure triple) rather than by rule updates
            against the disk store (RocksDB-write-bound, no memory cost).
            ``None`` chooses in memory below :data:`IN_MEMORY_LIMIT`
            asserted triples.
    """

    def __init__(
        self,
        path: str | Path | None = None,
        *,
        regime: Regime | None = None,
        skolemize: bool = True,
        prefixes: dict[str, str] | None = None,
        materialize: bool = True,
        in_memory: bool | None = None,
    ) -> None:
        self._ox = _ox()
        self.in_memory = in_memory
        ox = self._ox
        self._store = ox.Store(str(path)) if path is not None else ox.Store()
        self._path = Path(path) if path is not None else None
        self._regime = regime
        self._skolemize = skolemize
        self._generation = 0
        self._inconsistent = False
        self._resolver = Resolver(Graph())
        for pfx, ns in (prefixes or {}).items():
            self._resolver.bind(pfx, ns)
        self._default = ox.DefaultGraph()
        self._asserted = ox.NamedNode(ASSERTED) if self._materialising else self._default
        self._meta = ox.NamedNode(META)
        self.round_trips = 0
        if self._materialising:
            if self._recorded_regime() == regime.name:  # type: ignore[union-attr]
                self._inconsistent = self._recorded_inconsistent()
                logger.info("Oxigraph store %s: closure for regime %s already materialised%s",
                            self._path, regime.name,  # type: ignore[union-attr]
                            " (INCONSISTENT)" if self._inconsistent else "")
            elif materialize:
                self.materialize()

    # --- Protocol ---
    @property
    def generation(self) -> int:
        return self._generation

    @property
    def regime(self) -> Regime | None:
        return self._regime

    @property
    def resolver(self) -> Resolver:
        return self._resolver

    @property
    def store(self) -> Any:
        """The underlying ``pyoxigraph.Store``."""
        return self._store

    @property
    def path(self) -> Path | None:
        return self._path

    @property
    def materialised(self) -> bool:
        """Does the store hold ``cl_R(G)`` for this regime? (Always true without one.)"""
        if not self._materialising:
            return True
        assert self._regime is not None
        return self._recorded_regime() == self._regime.name

    @property
    def _materialising(self) -> bool:
        return self._regime is not None and bool(self._regime.rules or self._regime.axioms)

    def size(self) -> int:
        return self._count(self._asserted)

    def closure_size(self) -> int:
        return self._count(self._default)

    def contains(self, t: Triple) -> bool:
        return self._has(t, self._asserted)

    def triples(self, pattern: Pattern) -> Iterator[Triple]:
        return self._match(pattern, self._asserted)

    def closure_contains(self, t: Triple) -> bool:
        return self._has(t, self._default)

    def closure_triples(self, pattern: Pattern) -> Iterator[Triple]:
        return self._match(pattern, self._default)

    def is_inconsistent(self) -> bool:
        return self._inconsistent

    def join(self, patterns: list[Any], bindings: dict[Any, Node]) -> Iterator[dict[Any, Node]]:
        """One ``SELECT`` over the closure for the whole conjunction (an ``ASK``
        when every term is bound)."""
        from pynmms.rdf.sparql_rules import build_select

        bgp, query, free, names = build_select(patterns, bindings)
        if not free:
            yield from ([dict(bindings)] if bool(self._query(f"ASK {{ {bgp} }}")) else [])
            return
        for row in self._query(query):
            b = dict(bindings)
            for v in free:
                val = row[names[v][1:]]
                if val is None:
                    break
                b[v] = from_ox(val, self._ox)
            else:
                yield b

    # --- Loading and mutation ---
    def load(self, source: str | Path, format: str | None = None, *,
             materialize: bool = True) -> int:
        """Load an RDF file into the store, Skolemize, and (re)materialise.

        Parsing is Oxigraph's own; *format* is a MIME type or guessed from
        the extension. Turtle ``@prefix`` lines are bound in the resolver.
        Returns the number of asserted triples added. With
        ``materialize=False`` the closure is left stale until
        :meth:`materialize` is called.
        """
        ox = self._ox
        path = Path(source)
        mime = format or _FORMATS.get(path.suffix.lower())
        if mime is None:
            raise ValueError(f"Cannot guess the RDF format of {source}")
        before = self.size()
        t0 = time.perf_counter()
        with open(path, "rb") as fh:
            self._store.bulk_load(fh, format=ox.RdfFormat.from_media_type(mime),
                                  to_graph=self._asserted)
        if self._skolemize:
            self._skolemize_graph()
        for pfx, ns in _file_prefixes(path).items():
            self._resolver.bind(pfx, ns)
        added = self.size() - before
        logger.info("Loaded %d triples from %s in %.1f ms", added, source,
                    (time.perf_counter() - t0) * 1000)
        if self._materialising and materialize:
            self.materialize()
        else:
            self._generation += 1
        return added

    def dump(self, destination: str | Path, format: str | None = None) -> None:
        """Write the asserted graph to a file (format from the extension)."""
        ox = self._ox
        path = Path(destination)
        mime = format or _FORMATS.get(path.suffix.lower())
        if mime is None:
            raise ValueError(f"Cannot guess the RDF format of {destination}")
        with open(path, "wb") as fh:
            self._store.dump(fh, format=ox.RdfFormat.from_media_type(mime),
                             from_graph=self._asserted)

    def materialize(self) -> None:
        """Recompute ``cl_R(G)`` in the store from the asserted graph.

        An in-memory store runs the rules in place. An on-disk store does so
        too when ``in_memory=False`` was given, or when it was left to choose
        and holds more than :data:`IN_MEMORY_LIMIT` asserted triples;
        otherwise it computes the closure in a temporary in-memory store and
        bulk-loads the result, since RocksDB writes made the rule updates
        five times slower than in memory, slower than the Python engine at
        two million closure triples.
        """
        if not self._materialising:
            self._generation += 1
            return
        if self._path is None:
            scratch_in_memory = False
        elif self.in_memory is not None:
            scratch_in_memory = self.in_memory
        else:
            scratch_in_memory = self.size() <= IN_MEMORY_LIMIT
        if not scratch_in_memory:
            self._reset_closure()
            self._materialize()
            return
        import tempfile

        ox = self._ox
        disk = self._store
        t0 = time.perf_counter()
        with tempfile.TemporaryDirectory() as tmp:
            dump = Path(tmp) / "graph.nt"
            disk.dump(str(dump), format=ox.RdfFormat.N_TRIPLES, from_graph=self._asserted)
            scratch = ox.Store()
            scratch.bulk_load(path=str(dump), format=ox.RdfFormat.N_TRIPLES)
            self._store = scratch
            try:
                self._asserted, asserted_on_disk = self._default, self._asserted
                self._materialize(asserted=self._count(self._default), record=False)
            finally:
                self._store, self._asserted = disk, asserted_on_disk
            scratch.dump(str(dump), format=ox.RdfFormat.N_TRIPLES, from_graph=self._default)
            del scratch
            self._update("DROP SILENT DEFAULT")
            disk.bulk_load(path=str(dump), format=ox.RdfFormat.N_TRIPLES)
        self._record_meta()
        logger.info("Closure written to %s in %.1f ms total", self._path,
                    (time.perf_counter() - t0) * 1000)

    def load_graph(self, graph: Graph, *, source: URIRef | None = None) -> int:
        """Load an rdflib graph (its namespace bindings included).

        With *source*, the triples are also recorded in that named graph, so
        :meth:`graphs_of` reports them as coming from it (provenance).
        """
        for pfx, ns in graph.namespaces():
            self._resolver.bind(pfx, str(ns))
        triples = [skolemize_triple(t) if self._skolemize else t for t in graph]
        before = self.size()
        self._insert(triples, self._asserted)
        if source is not None:
            self._insert(triples, self._ox.NamedNode(str(source)))
        self.materialize()
        return self.size() - before

    def graphs_of(self, t: Triple) -> list[URIRef]:
        """The named graphs (sources) holding *t*, the backend's own graphs excluded."""
        ox = self._ox
        s, p, o = t
        if isinstance(s, Literal):
            return []
        out: list[URIRef] = []
        for q in self._store.quads_for_pattern(to_ox(s, ox), to_ox(p, ox), to_ox(o, ox), None):
            g = q.graph_name
            if isinstance(g, ox.NamedNode) and g.value not in (ASSERTED, META):
                out.append(URIRef(g.value))
        return sorted(out)

    def add(self, triples: Iterable[Triple], *, source: URIRef | None = None) -> int:
        """Assert triples and extend the closure incrementally in process.

        The new triples are closed by :meth:`ClosureEngine.extend` with the
        store as the join partner, and the result is written to the closure
        graph; the store is never re-materialised. With *source*, they are
        also recorded in that named graph.
        """
        triples = list(triples)
        added = [t for t in triples if not self.contains(t)]
        if source is not None and triples:
            self._insert(triples, self._ox.NamedNode(str(source)))
        if not added:
            return 0
        self._insert(added, self._asserted)
        if self._materialising:
            from pynmms.rdf.closure import ClosureEngine

            assert self._regime is not None
            new, bottom = ClosureEngine(self._regime).extend(
                added, self.closure_triples, self.closure_contains, store_join=self.join
            )
            self._insert(list(new.triples), self._default)
            if bottom and not self._inconsistent:
                self._inconsistent = True
                self._record_meta()
        else:
            self._insert(added, self._default) if self._asserted is not self._default else None
        self._generation += 1
        logger.debug("Added %d triples; generation %d", len(added), self._generation)
        return len(added)

    # --- Materialisation in the store ---
    def _materialize(self, asserted: int | None = None, *, record: bool = True) -> None:
        """Run the regime to a fixpoint inside the store.

        Pattern rules run as SPARQL updates; guarded and procedural rules run
        in process against the store between rounds, until neither adds a
        triple. ⊥ rules are ASKs (``prop:incoherence``). With ``record=False``
        the meta graph is left to the caller (the scratch store of
        :meth:`materialize` is discarded, so recording there is wasted).
        """
        from pynmms.rdf.sparql_rules import partition

        assert self._regime is not None
        rules = partition(self._regime, skip=UNSTORABLE_RULES)
        t0 = time.perf_counter()
        self._insert(list(self._regime.axioms), self._default)
        rounds = 0
        while True:
            rounds += self._sparql_fixpoint(rules.updates)
            if not self._in_process_round(rules.in_process):
                break
        bottom = any(bool(self._query(q)) for q in rules.asks) or self._in_process_bottom(
            rules.in_process
        )
        self._inconsistent = bottom
        self._generation += 1
        if record:
            self._record_meta()
        logger.info(
            "Materialised %s closure in the store: %d asserted -> %d triples%s in %d round(s), "
            "%.1f ms (%d store-side rules, %d in process%s)",
            self._regime.name, self.size() if asserted is None else asserted, self.closure_size(),
            " (INCONSISTENT)" if bottom else "", rounds, (time.perf_counter() - t0) * 1000,
            rules.store_side, len(rules.in_process),
            f", skipped {', '.join(rules.skipped)}" if rules.skipped else "",
        )

    def _sparql_fixpoint(self, updates: tuple[str, ...]) -> int:
        rounds = 0
        while updates:
            before = self.closure_size()
            for u in updates:
                self._update(u)
            rounds += 1
            if self.closure_size() == before:
                break
        return rounds

    def _in_process_round(self, rules: tuple[Any, ...]) -> bool:
        """One pass of the guarded and procedural rules over the closure graph."""
        from pynmms.rdf.rules import ProceduralRule, Rule

        new: list[Triple] = []
        for rule in rules:
            if isinstance(rule, ProceduralRule):
                for concl in self._fire_procedural(rule):
                    if concl is not None and not self.closure_contains(concl):
                        new.append(concl)
            elif isinstance(rule, Rule) and rule.conclusion is not None:
                for b in self.join(list(rule.premises), {}):
                    if rule.guard is None or rule.guard(b):
                        bound: Triple = tuple(b.get(t, t) for t in rule.conclusion)  # type: ignore[assignment]
                        if not isinstance(bound[0], Literal) and not self.closure_contains(bound):
                            new.append(bound)
        if not new:
            return False
        self._insert(new, self._default)
        return True

    def _in_process_bottom(self, rules: tuple[Any, ...]) -> bool:
        from pynmms.rdf.rules import ProceduralRule, Rule

        for rule in rules:
            if isinstance(rule, ProceduralRule):
                if any(c is None for c in self._fire_procedural(rule)):
                    return True
            elif isinstance(rule, Rule) and rule.conclusion is None:
                for b in self.join(list(rule.premises), {}):
                    if rule.guard is None or rule.guard(b):
                        return True
        return False

    def _fire_procedural(self, rule: Any) -> Iterator[Triple | None]:
        triggers = rule.triggers or (None,)
        seen: set[Triple] = set()
        for p in triggers:
            for t in self.closure_triples((None, p, None)):
                if t in seen:
                    continue
                seen.add(t)
                yield from rule.fire(t, self.closure_triples)

    # --- Store plumbing ---
    def _reset_closure(self) -> None:
        """Make the default graph a copy of the asserted graph."""
        if self._asserted is self._default:
            return
        self._update(f"DROP SILENT DEFAULT; ADD SILENT <{ASSERTED}> TO DEFAULT")

    def _skolemize_graph(self) -> None:
        ox = self._ox
        rows = list(self._query(
            f"SELECT ?s ?p ?o WHERE {{ GRAPH <{ASSERTED}> {{ ?s ?p ?o }} "
            f"FILTER(isBlank(?s) || isBlank(?o)) }}"
        )) if self._asserted is not self._default else list(self._query(
            "SELECT ?s ?p ?o WHERE { ?s ?p ?o FILTER(isBlank(?s) || isBlank(?o)) }"
        ))
        if not rows:
            return
        old = [ox.Quad(r["s"], r["p"], r["o"], self._asserted) for r in rows]
        new = [skolemize_triple((from_ox(r["s"], ox), from_ox(r["p"], ox), from_ox(r["o"], ox)))
               for r in rows]
        for q in old:
            self._store.remove(q)
        self._insert(new, self._asserted)
        logger.info("Skolemized %d triple(s) with blank nodes", len(rows))

    def _insert(self, triples: list[Triple], graph: Any) -> None:
        ox = self._ox
        quads = []
        for s, p, o in triples:
            if isinstance(s, Literal):
                continue  # generalized triple; not storable
            quads.append(ox.Quad(to_ox(s, ox), to_ox(p, ox), to_ox(o, ox), graph))
        if quads:
            self._store.extend(quads)

    def _has(self, t: Triple, graph: Any) -> bool:
        ox = self._ox
        s, p, o = t
        if isinstance(s, Literal):
            return False
        self.round_trips += 1
        return ox.Quad(to_ox(s, ox), to_ox(p, ox), to_ox(o, ox), graph) in self._store

    def _match(self, pattern: Pattern, graph: Any) -> Iterator[Triple]:
        ox = self._ox
        s, p, o = pattern
        if isinstance(s, Literal):
            return
        self.round_trips += 1
        for q in self._store.quads_for_pattern(
            None if s is None else to_ox(s, ox), None if p is None else to_ox(p, ox),
            None if o is None else to_ox(o, ox), graph,
        ):
            yield (from_ox(q.subject, ox), from_ox(q.predicate, ox), from_ox(q.object, ox))

    def _count(self, graph: Any) -> int:
        clause = ("{ ?s ?p ?o }" if graph is self._default
                  else f"{{ GRAPH <{ASSERTED}> {{ ?s ?p ?o }} }}")
        row = next(iter(self._query(f"SELECT (COUNT(*) AS ?n) WHERE {clause}")))
        return int(row["n"].value)

    def _query(self, q: str) -> Any:
        self.round_trips += 1
        return self._store.query(q)

    def _update(self, u: str) -> None:
        self.round_trips += 1
        self._store.update(u)

    def _record_meta(self) -> None:
        assert self._regime is not None
        self._update(
            f"DROP SILENT GRAPH <{META}>; INSERT DATA {{ GRAPH <{META}> {{ "
            f"<{_STORE}> <{_REGIME}> {Literal(self._regime.name).n3()} ; "
            f"<{_INCONSISTENT}> {Literal(self._inconsistent).n3()} }} }}"
        )

    def _recorded_regime(self) -> str | None:
        q = f"SELECT ?r WHERE {{ GRAPH <{META}> {{ <{_STORE}> <{_REGIME}> ?r }} }}"
        for row in self._query(q):
            return str(row["r"].value)
        return None

    def _recorded_inconsistent(self) -> bool:
        for row in self._query(
            f"SELECT ?i WHERE {{ GRAPH <{META}> {{ <{_STORE}> <{_INCONSISTENT}> ?i }} }}"
        ):
            return str(row["i"].value) == "true"
        return False

    def close(self) -> None:
        """Release the store (an on-disk store holds a lock while open)."""
        self._store = None

    def __enter__(self) -> OxigraphBackend:
        return self

    def __exit__(self, *exc: object) -> None:
        self.close()

    def __repr__(self) -> str:
        where = str(self._path) if self._path else "memory"
        return f"OxigraphBackend({where}, {self.size()} triples, regime={self._regime})"

materialised property

Does the store hold cl_R(G) for this regime? (Always true without one.)

store property

The underlying pyoxigraph.Store.

add(triples, *, source=None)

Assert triples and extend the closure incrementally in process.

The new triples are closed by :meth:ClosureEngine.extend with the store as the join partner, and the result is written to the closure graph; the store is never re-materialised. With source, they are also recorded in that named graph.

Source code in src/pynmms/rdf/backends/oxigraph.py 
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
def add(self, triples: Iterable[Triple], *, source: URIRef | None = None) -> int:
    """Assert triples and extend the closure incrementally in process.

    The new triples are closed by :meth:`ClosureEngine.extend` with the
    store as the join partner, and the result is written to the closure
    graph; the store is never re-materialised. With *source*, they are
    also recorded in that named graph.
    """
    triples = list(triples)
    added = [t for t in triples if not self.contains(t)]
    if source is not None and triples:
        self._insert(triples, self._ox.NamedNode(str(source)))
    if not added:
        return 0
    self._insert(added, self._asserted)
    if self._materialising:
        from pynmms.rdf.closure import ClosureEngine

        assert self._regime is not None
        new, bottom = ClosureEngine(self._regime).extend(
            added, self.closure_triples, self.closure_contains, store_join=self.join
        )
        self._insert(list(new.triples), self._default)
        if bottom and not self._inconsistent:
            self._inconsistent = True
            self._record_meta()
    else:
        self._insert(added, self._default) if self._asserted is not self._default else None
    self._generation += 1
    logger.debug("Added %d triples; generation %d", len(added), self._generation)
    return len(added)

close()

Release the store (an on-disk store holds a lock while open).

Source code in src/pynmms/rdf/backends/oxigraph.py 
601
602
603
def close(self) -> None:
    """Release the store (an on-disk store holds a lock while open)."""
    self._store = None

dump(destination, format=None)

Write the asserted graph to a file (format from the extension).

Source code in src/pynmms/rdf/backends/oxigraph.py 
299
300
301
302
303
304
305
306
307
308
def dump(self, destination: str | Path, format: str | None = None) -> None:
    """Write the asserted graph to a file (format from the extension)."""
    ox = self._ox
    path = Path(destination)
    mime = format or _FORMATS.get(path.suffix.lower())
    if mime is None:
        raise ValueError(f"Cannot guess the RDF format of {destination}")
    with open(path, "wb") as fh:
        self._store.dump(fh, format=ox.RdfFormat.from_media_type(mime),
                         from_graph=self._asserted)

graphs_of(t)

The named graphs (sources) holding t, the backend's own graphs excluded.

Source code in src/pynmms/rdf/backends/oxigraph.py 
374
375
376
377
378
379
380
381
382
383
384
385
def graphs_of(self, t: Triple) -> list[URIRef]:
    """The named graphs (sources) holding *t*, the backend's own graphs excluded."""
    ox = self._ox
    s, p, o = t
    if isinstance(s, Literal):
        return []
    out: list[URIRef] = []
    for q in self._store.quads_for_pattern(to_ox(s, ox), to_ox(p, ox), to_ox(o, ox), None):
        g = q.graph_name
        if isinstance(g, ox.NamedNode) and g.value not in (ASSERTED, META):
            out.append(URIRef(g.value))
    return sorted(out)

join(patterns, bindings)

One SELECT over the closure for the whole conjunction (an ASK when every term is bound).

Source code in src/pynmms/rdf/backends/oxigraph.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
def join(self, patterns: list[Any], bindings: dict[Any, Node]) -> Iterator[dict[Any, Node]]:
    """One ``SELECT`` over the closure for the whole conjunction (an ``ASK``
    when every term is bound)."""
    from pynmms.rdf.sparql_rules import build_select

    bgp, query, free, names = build_select(patterns, bindings)
    if not free:
        yield from ([dict(bindings)] if bool(self._query(f"ASK {{ {bgp} }}")) else [])
        return
    for row in self._query(query):
        b = dict(bindings)
        for v in free:
            val = row[names[v][1:]]
            if val is None:
                break
            b[v] = from_ox(val, self._ox)
        else:
            yield b

load(source, format=None, *, materialize=True)

Load an RDF file into the store, Skolemize, and (re)materialise.

Parsing is Oxigraph's own; format is a MIME type or guessed from the extension. Turtle @prefix lines are bound in the resolver. Returns the number of asserted triples added. With materialize=False the closure is left stale until :meth:materialize is called.

Source code in src/pynmms/rdf/backends/oxigraph.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def load(self, source: str | Path, format: str | None = None, *,
         materialize: bool = True) -> int:
    """Load an RDF file into the store, Skolemize, and (re)materialise.

    Parsing is Oxigraph's own; *format* is a MIME type or guessed from
    the extension. Turtle ``@prefix`` lines are bound in the resolver.
    Returns the number of asserted triples added. With
    ``materialize=False`` the closure is left stale until
    :meth:`materialize` is called.
    """
    ox = self._ox
    path = Path(source)
    mime = format or _FORMATS.get(path.suffix.lower())
    if mime is None:
        raise ValueError(f"Cannot guess the RDF format of {source}")
    before = self.size()
    t0 = time.perf_counter()
    with open(path, "rb") as fh:
        self._store.bulk_load(fh, format=ox.RdfFormat.from_media_type(mime),
                              to_graph=self._asserted)
    if self._skolemize:
        self._skolemize_graph()
    for pfx, ns in _file_prefixes(path).items():
        self._resolver.bind(pfx, ns)
    added = self.size() - before
    logger.info("Loaded %d triples from %s in %.1f ms", added, source,
                (time.perf_counter() - t0) * 1000)
    if self._materialising and materialize:
        self.materialize()
    else:
        self._generation += 1
    return added

load_graph(graph, *, source=None)

Load an rdflib graph (its namespace bindings included).

With source, the triples are also recorded in that named graph, so :meth:graphs_of reports them as coming from it (provenance).

Source code in src/pynmms/rdf/backends/oxigraph.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def load_graph(self, graph: Graph, *, source: URIRef | None = None) -> int:
    """Load an rdflib graph (its namespace bindings included).

    With *source*, the triples are also recorded in that named graph, so
    :meth:`graphs_of` reports them as coming from it (provenance).
    """
    for pfx, ns in graph.namespaces():
        self._resolver.bind(pfx, str(ns))
    triples = [skolemize_triple(t) if self._skolemize else t for t in graph]
    before = self.size()
    self._insert(triples, self._asserted)
    if source is not None:
        self._insert(triples, self._ox.NamedNode(str(source)))
    self.materialize()
    return self.size() - before

materialize()

Recompute cl_R(G) in the store from the asserted graph.

An in-memory store runs the rules in place. An on-disk store does so too when in_memory=False was given, or when it was left to choose and holds more than :data:IN_MEMORY_LIMIT asserted triples; otherwise it computes the closure in a temporary in-memory store and bulk-loads the result, since RocksDB writes made the rule updates five times slower than in memory, slower than the Python engine at two million closure triples.

Source code in src/pynmms/rdf/backends/oxigraph.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
def materialize(self) -> None:
    """Recompute ``cl_R(G)`` in the store from the asserted graph.

    An in-memory store runs the rules in place. An on-disk store does so
    too when ``in_memory=False`` was given, or when it was left to choose
    and holds more than :data:`IN_MEMORY_LIMIT` asserted triples;
    otherwise it computes the closure in a temporary in-memory store and
    bulk-loads the result, since RocksDB writes made the rule updates
    five times slower than in memory, slower than the Python engine at
    two million closure triples.
    """
    if not self._materialising:
        self._generation += 1
        return
    if self._path is None:
        scratch_in_memory = False
    elif self.in_memory is not None:
        scratch_in_memory = self.in_memory
    else:
        scratch_in_memory = self.size() <= IN_MEMORY_LIMIT
    if not scratch_in_memory:
        self._reset_closure()
        self._materialize()
        return
    import tempfile

    ox = self._ox
    disk = self._store
    t0 = time.perf_counter()
    with tempfile.TemporaryDirectory() as tmp:
        dump = Path(tmp) / "graph.nt"
        disk.dump(str(dump), format=ox.RdfFormat.N_TRIPLES, from_graph=self._asserted)
        scratch = ox.Store()
        scratch.bulk_load(path=str(dump), format=ox.RdfFormat.N_TRIPLES)
        self._store = scratch
        try:
            self._asserted, asserted_on_disk = self._default, self._asserted
            self._materialize(asserted=self._count(self._default), record=False)
        finally:
            self._store, self._asserted = disk, asserted_on_disk
        scratch.dump(str(dump), format=ox.RdfFormat.N_TRIPLES, from_graph=self._default)
        del scratch
        self._update("DROP SILENT DEFAULT")
        disk.bulk_load(path=str(dump), format=ox.RdfFormat.N_TRIPLES)
    self._record_meta()
    logger.info("Closure written to %s in %.1f ms total", self._path,
                (time.perf_counter() - t0) * 1000)

from_ox(t, ox)

pyoxigraph term -> rdflib term (plain strings lose their xsd:string).

Source code in src/pynmms/rdf/backends/oxigraph.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def from_ox(t: Any, ox: Any) -> Node:
    """pyoxigraph term -> rdflib term (plain strings lose their xsd:string)."""
    if isinstance(t, ox.NamedNode):
        return URIRef(t.value)
    if isinstance(t, ox.BlankNode):
        return BNode(t.value)
    if isinstance(t, ox.Literal):
        if t.language:
            return Literal(t.value, lang=t.language)
        dt = t.datatype.value if t.datatype is not None else None
        if dt is None or dt == str(XSD.string):
            return Literal(t.value)
        return Literal(t.value, datatype=URIRef(dt))
    raise TypeError(f"cannot convert {t!r} to an rdflib term")

to_ox(n, ox)

rdflib term -> pyoxigraph term.

Source code in src/pynmms/rdf/backends/oxigraph.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
def to_ox(n: Node, ox: Any) -> Any:
    """rdflib term -> pyoxigraph term."""
    if isinstance(n, URIRef):
        return ox.NamedNode(str(n))
    if isinstance(n, BNode):
        return ox.BlankNode(str(n))
    if isinstance(n, Literal):
        if n.language:
            return ox.Literal(str(n), language=n.language)
        if n.datatype is not None and n.datatype != XSD.string:
            return ox.Literal(str(n), datatype=ox.NamedNode(str(n.datatype)))
        return ox.Literal(str(n))
    raise TypeError(f"cannot convert {n!r} to an Oxigraph term")

pynmms.rdf.sparql_rules

pynmms.rdf.sparql_rules

Translate regime rules into SPARQL so a store can run the closure itself.

A pattern :class:~pynmms.rdf.rules.Rule <A, c> (def:entailmentregime) with no guard becomes one INSERT { c } WHERE { A FILTER NOT EXISTS { c } } update; running the whole set to a fixpoint materialises cl_R(G) inside the store, with the store's own join engine doing the work. A ⊥ rule becomes ASK { A }: true iff the store's graph is R-inconsistent (prop:incoherence). Rules with a Python guard and :class:~pynmms.rdf.rules.ProceduralRule\s are not translatable and are reported by :func:partition; a backend runs those in process against the store.

Terms are written in N3: rdflib nodes through Node.n3(), rule variables as ?name.

JoinQuery

Bases: NamedTuple

A backend's join() as SPARQL: the ordered BGP with bound terms inlined, the SELECT DISTINCT over it, the free variables in projection order, and each free variable's name in the query (?v0, ?v1, ...).

Source code in src/pynmms/rdf/sparql_rules.py 
85
86
87
88
89
90
91
92
93
class JoinQuery(NamedTuple):
    """A backend's ``join()`` as SPARQL: the ordered BGP with bound terms inlined,
    the ``SELECT DISTINCT`` over it, the free variables in projection order, and
    each free variable's name in the query (``?v0``, ``?v1``, ...)."""

    bgp: str
    query: str
    free: list[Any]
    names: dict[Any, str]

StoreRules dataclass

A regime split into what the store runs and what stays in process.

Source code in src/pynmms/rdf/sparql_rules.py 
171
172
173
174
175
176
177
178
179
180
181
182
@dataclass(frozen=True)
class StoreRules:
    """A regime split into what the store runs and what stays in process."""

    updates: tuple[str, ...]
    asks: tuple[str, ...]
    in_process: tuple[RuleLike, ...]
    skipped: tuple[str, ...]

    @property
    def store_side(self) -> int:
        return len(self.updates) + len(self.asks)

bgp(premises)

The basic graph pattern of a premise set.

Source code in src/pynmms/rdf/sparql_rules.py 
48
49
50
def bgp(premises: tuple[tuple[Any, Any, Any], ...]) -> str:
    """The basic graph pattern of a premise set."""
    return " . ".join(pattern_n3(p) for p in premises)

build_select(patterns, bindings)

One SELECT for a conjunction of patterns, bindings inlined.

Variables already bound are written as their values, the others as ?vN in first-seen order over the BGP as :func:order_bgp arranges it. With no free variable the projection is * (a backend may prefer an ASK { bgp } then).

Source code in src/pynmms/rdf/sparql_rules.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def build_select(patterns: list[Any], bindings: dict[Any, Any]) -> JoinQuery:
    """One ``SELECT`` for a conjunction of patterns, *bindings* inlined.

    Variables already bound are written as their values, the others as
    ``?vN`` in first-seen order over the BGP as :func:`order_bgp` arranges
    it. With no free variable the projection is ``*`` (a backend may prefer
    an ``ASK { bgp }`` then).
    """
    names: dict[Any, str] = {}
    free: list[Any] = []

    def term(x: Any) -> str:
        if isinstance(x, Var):
            if x in bindings:
                return bindings[x].n3()  # type: ignore[no-any-return]
            if x not in names:
                names[x] = f"?v{len(names)}"
                free.append(x)
            return names[x]
        return x.n3()  # type: ignore[no-any-return]

    bgp_text = " . ".join(f"{term(s)} {term(p)} {term(o)}"
                          for s, p, o in order_bgp(patterns, bindings))
    select = " ".join(names[v] for v in free) or "*"
    return JoinQuery(bgp_text, f"SELECT DISTINCT {select} WHERE {{ {bgp_text} }}", free, names)

order_bgp(patterns, bindings)

Order a basic graph pattern for a store without a statistics planner.

Oxigraph evaluates a BGP left to right, so ?x a ex:C2 . ?x ex:p ex:j7 scans every instance of the class before touching the selective triple (2.8 s against 0.4 ms on a 10⁷ graph). Most bound terms first, an rdf:type pattern with an unbound subject last, and among equals a pattern sharing a variable with an earlier one before one that does not.

Source code in src/pynmms/rdf/sparql_rules.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def order_bgp(patterns: list[Any], bindings: dict[Any, Any]) -> list[Any]:
    """Order a basic graph pattern for a store without a statistics planner.

    Oxigraph evaluates a BGP left to right, so ``?x a ex:C2 . ?x ex:p ex:j7``
    scans every instance of the class before touching the selective triple
    (2.8 s against 0.4 ms on a 10⁷ graph). Most bound terms first, an
    ``rdf:type`` pattern with an unbound subject last, and among equals a
    pattern sharing a variable with an earlier one before one that does not.
    """
    from rdflib import RDF

    def bound(t: Any) -> bool:
        return not isinstance(t, Var) or t in bindings

    remaining = list(patterns)
    ordered: list[Any] = []
    seen: set[Any] = set()
    while remaining:
        def key(p: Any) -> tuple[int, int, int]:
            s, pr, o = p
            n_bound = sum(bound(t) for t in p)
            type_scan = int(bound(pr) and pr == RDF.type and not bound(s))
            joins = int(any(isinstance(t, Var) and t in seen for t in p))
            return (-n_bound, type_scan, -joins)

        best = min(remaining, key=key)
        remaining.remove(best)
        ordered.append(best)
        seen.update(t for t in best if isinstance(t, Var))
    return ordered

partition(regime, *, skip=frozenset())

Split regime into SPARQL updates, SPARQL asks, and in-process rules.

Rule names in skip are dropped altogether (with a log line); a backend uses this for rules whose conclusions it cannot store.

Source code in src/pynmms/rdf/sparql_rules.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def partition(regime: Regime, *, skip: frozenset[str] = frozenset()) -> StoreRules:
    """Split *regime* into SPARQL updates, SPARQL asks, and in-process rules.

    Rule names in *skip* are dropped altogether (with a log line); a backend
    uses this for rules whose conclusions it cannot store.
    """
    updates: list[str] = []
    asks: list[str] = []
    in_process: list[RuleLike] = []
    skipped: list[str] = []
    for rule in regime.rules:
        if rule.name in skip:
            skipped.append(rule.name)
        elif translatable(rule):
            assert isinstance(rule, Rule)
            (asks if rule.conclusion is None else updates).append(
                rule_to_ask(rule) if rule.conclusion is None else rule_to_update(rule)
            )
        else:
            in_process.append(rule)
    if skipped:
        logger.info("Regime %s: %d rule(s) skipped for the store: %s",
                    regime.name, len(skipped), ", ".join(skipped))
    logger.debug("Regime %s: %d store-side rule(s), %d in process",
                 regime.name, len(updates) + len(asks), len(in_process))
    return StoreRules(tuple(updates), tuple(asks), tuple(in_process), tuple(skipped))

rule_to_ask(rule)

ASK { A } for a false-concluding pattern rule.

Source code in src/pynmms/rdf/sparql_rules.py 
162
163
164
165
166
167
168
def rule_to_ask(rule: Rule) -> str:
    """``ASK { A }`` for a false-concluding pattern rule."""
    if rule.conclusion is not None:
        raise ValueError(f"{rule.name} is not false-concluding")
    if not translatable(rule):
        raise ValueError(f"{rule.name} is not translatable to SPARQL")
    return f"{PROLOGUE}ASK {{ {bgp(rule.premises)}{_filter(rule)} }}"

rule_to_update(rule, *, into=None)

INSERT { c } WHERE { A FILTER NOT EXISTS { c } } for a pattern rule.

With into an IRI, the conclusions go to that named graph while the premises and the NOT EXISTS check read the default graph.

Source code in src/pynmms/rdf/sparql_rules.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def rule_to_update(rule: Rule, *, into: str | None = None) -> str:
    """``INSERT { c } WHERE { A FILTER NOT EXISTS { c } }`` for a pattern rule.

    With *into* an IRI, the conclusions go to that named graph while the
    premises and the ``NOT EXISTS`` check read the default graph.
    """
    if rule.conclusion is None:
        raise ValueError(f"{rule.name} is false-concluding; use rule_to_ask")
    if not translatable(rule):
        raise ValueError(f"{rule.name} is not translatable to SPARQL")
    head = pattern_n3(rule.conclusion)
    target = f"GRAPH <{into}> {{ {head} }}" if into else head
    return (f"{PROLOGUE}INSERT {{ {target} }} WHERE {{ {bgp(rule.premises)}{_filter(rule)} "
            f"FILTER NOT EXISTS {{ {head} }} }}")

term_n3(t)

N3 for a rule term: ?x for a variable, Node.n3() otherwise.

Source code in src/pynmms/rdf/sparql_rules.py 
35
36
37
38
39
40
41
def term_n3(t: Any) -> str:
    """N3 for a rule term: ``?x`` for a variable, ``Node.n3()`` otherwise."""
    if isinstance(t, Var):
        return f"?{t.name}"
    if isinstance(t, Node):
        return t.n3()  # type: ignore[no-any-return]
    raise TypeError(f"not a rule term: {t!r}")

translatable(rule)

Can rule run as one SPARQL update or ASK?

Pattern rules without a guard are translatable. A rule whose conclusion has a literal in subject position is not: it is a generalized triple (def:rdftriple) that a SPARQL store cannot hold.

Source code in src/pynmms/rdf/sparql_rules.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def translatable(rule: RuleLike) -> bool:
    """Can *rule* run as one SPARQL update or ASK?

    Pattern rules without a guard are translatable. A rule whose conclusion
    has a literal in subject position is not: it is a generalized triple
    (``def:rdftriple``) that a SPARQL store cannot hold.
    """
    if not isinstance(rule, Rule):
        return False
    if rule.guard is not None and (rule.guard_expr is None or rule.guard_expr.to_sparql() is None):
        return False
    if rule.conclusion is not None and isinstance(rule.conclusion[0], Literal):
        return False
    return True