diff --git a/CHANGELOG.md b/CHANGELOG.md index c98beae..92dadae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,15 @@ # Changelog +## Unreleased + +### Fixed + +- SPARQL search: an object or array under `const`, `default` or `enum` projects as one + JSON literal (`rdf:JSON`) instead of being read as JSON-LD, which failed the whole TD + when the object had a `type` key. +- SPARQL search: a TD that cannot be projected is logged and left out of the index. Before, + it failed every later write and kept the server from starting. + ## 0.1.0 - 2026-09-27 First release. diff --git a/src/thingdir/search/rdf.py b/src/thingdir/search/rdf.py index 5227163..9e7bfbe 100644 --- a/src/thingdir/search/rdf.py +++ b/src/thingdir/search/rdf.py @@ -7,10 +7,14 @@ from __future__ import annotations +import functools import json +import logging import pathlib from typing import Any +_log = logging.getLogger("thingdir") + _TD_VOCAB = "https://www.w3.org/2019/wot/td#" _REMOTE_CONTEXTS = { "https://www.w3.org/2022/wot/td/v1.1", @@ -44,6 +48,57 @@ def _localize_context(td: dict) -> dict: return out +# A TD may give these terms any JSON value (TD 1.1, 5.3.2.1), but the TD context does not +# mark them @json, so JSON-LD reads an object value as a node and its "type" key as @type. +_DATA_TERMS = frozenset({"const", "default", "enum"}) + + +@functools.cache +def _name_maps() -> frozenset[str]: + """Terms the TD context declares as index maps (``"@container": "@index"``), such as + ``properties``: a key inside one is a name, never a term.""" + found: set[str] = set() + + def walk(node: Any) -> None: + if isinstance(node, list): + for item in node: + walk(item) + elif isinstance(node, dict): + for key, value in node.items(): + if isinstance(value, dict) and value.get("@container") == "@index": + found.add(key) + walk(value) + + walk(_td_context()) + return frozenset(found) + + +def _json_literal(value: Any) -> Any: + if isinstance(value, (dict, list)): + return {"@value": value, "@type": "@json"} + return value + + +def _with_json_literals(node: Any) -> Any: + """A copy of ``node`` with each object or array value of a data term written as a + JSON-LD JSON literal, so it projects to one rdf:JSON value.""" + if isinstance(node, list): + return [_with_json_literals(item) for item in node] + if not isinstance(node, dict): + return node + out: dict[str, Any] = {} + for key, value in node.items(): + if key == "enum" and isinstance(value, list): + out[key] = [_json_literal(item) for item in value] + elif key in _DATA_TERMS: + out[key] = _json_literal(value) + elif key in _name_maps() and isinstance(value, dict): + out[key] = {name: _with_json_literals(item) for name, item in value.items()} + else: + out[key] = _with_json_literals(value) + return out + + class RdfIndex: """An embedded Oxigraph RDF index. @@ -58,14 +113,20 @@ def __init__(self, *, path: str | None = None) -> None: async def sync(self, store) -> int: """Apply store changes since this index's cursor. - A cursor of zero rebuilds the index. Repeating a sync is safe. + A cursor of zero rebuilds the index. Repeating a sync is safe. A TD that + cannot be projected is left out of the index and logged; it stays in the + store, and the cursor moves past it, so it cannot fail every later sync. """ new_cursor, changes = await store.changes_since(self._cursor) for td_id, td, _seq in changes: if td is None: self.remove(td_id) - else: + continue + try: self.insert(td_id, td) + except (SyntaxError, ValueError) as e: + self.remove(td_id) + _log.warning("thingdir: %s is stored but not searchable by SPARQL: %s", td_id, e) self._cursor = new_cursor return new_cursor @@ -87,7 +148,7 @@ def insert(self, td_id: str, td: dict) -> None: store = self._store_() graph = pyoxigraph.NamedNode(td_id) self.remove(td_id) - doc = _localize_context({**td, "id": td_id}) + doc = _localize_context(_with_json_literals({**td, "id": td_id})) store.load( json.dumps(doc).encode(), format=pyoxigraph.RdfFormat.JSON_LD, diff --git a/tests/test_rdf.py b/tests/test_rdf.py index 6f8fb0c..0b97c99 100644 --- a/tests/test_rdf.py +++ b/tests/test_rdf.py @@ -98,3 +98,71 @@ async def test_sparql_501_without_rdf(tmp_path): ) as c: r = await c.get("/search/sparql", params={"query": "SELECT * WHERE {?s ?p ?o}"}) assert r.status_code == 501 + + +DEVICE = { + "@context": "https://www.w3.org/2022/wot/td/v1.1", + "id": "urn:dev:device:1", + "title": "Device", + **_SEC, + "properties": { + "default": {"type": "string", "default": "home", "forms": [{"href": "http://device/d"}]} + }, + "actions": { + "set": { + "input": { + "type": "object", + "properties": { + "region": {"const": [{"type": 2, "size": [0.1]}]}, + "mode": {"enum": [{"type": "fast"}, "slow"]}, + }, + }, + "forms": [{"href": "http://device/set"}], + } + }, +} +JSON = "http://www.w3.org/1999/02/22-rdf-syntax-ns#JSON" +CONST = "https://www.w3.org/2019/wot/json-schema#const" +ENUM = "https://www.w3.org/2019/wot/json-schema#enum" + + +def test_object_data_values_project_as_json_literals(): + idx = RdfIndex() + idx.insert("urn:dev:device:1", DEVICE) + rows = idx.query( + f"SELECT ?p ?v ?dt WHERE {{ ?s ?p ?v FILTER(?p IN (<{CONST}>, <{ENUM}>)) " + "BIND(DATATYPE(?v) AS ?dt) }" + ) + found = sorted((r["p"], r["v"], r["dt"]) for r in rows) + assert found == [ + (CONST, '[{"size":[0.1],"type":2}]', JSON), + (ENUM, "slow", "http://www.w3.org/2001/XMLSchema#string"), + (ENUM, '{"type":"fast"}', JSON), + ] + + +def test_a_property_named_like_a_data_term_stays_a_property(): + idx = RdfIndex() + idx.insert("urn:dev:device:1", DEVICE) + names = idx.query( + "SELECT ?n WHERE { ?t ?a . " + "?a ?n }" + ) + assert names == [{"n": "default"}] + + +async def test_one_td_that_cannot_be_projected_does_not_block_the_rest(tmp_path, caplog): + store = FileStore(str(tmp_path)) + idx = RdfIndex() + broken = {**LAMP, "id": "urn:dev:broken:1", "@type": 7} + await store.put("urn:dev:broken:1", broken) + await store.put("urn:dev:pump:1", PUMP) + + await idx.sync(store) + + ids = idx.query(f"SELECT ?id WHERE {{ ?id <{TITLE}> ?t }}") + assert ids == [{"id": "urn:dev:pump:1"}] + assert "urn:dev:broken:1 is stored but not searchable" in caplog.text + await store.put("urn:dev:lamp:1", LAMP) + await idx.sync(store) + assert len(idx.query(f"SELECT ?id WHERE {{ ?id <{TITLE}> ?t }}")) == 2