From cfbd5b90791d44bcb5d2bd587b484bed6b962900 Mon Sep 17 00:00:00 2001 From: Sunny Yang Date: Mon, 6 Jul 2026 03:45:23 -0600 Subject: [PATCH] feat: export/import workspace knowledge in .tgx bundles (#877 Phase 2) (#1024) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: streaming N-Quads serializer for wire-format triples Groundwork for Phase 2 of #877 (knowledge export). Hand-rolled N-Triples term encoding: rdflib's term.n3() emits Turtle-style forms (numeric shorthand, unescaped newlines) that are invalid in line-oriented N-Quads, so literals are escaped per the ECHAR grammar and IRIs validated for representability. Round-trip tests parse the output back with rdflib's nquads parser and compare term-for-term. * feat: export/import workspace knowledge in .tgx bundles (#877) Phase 2 of the workspace bundle commands: tg-export-workspace now includes the workspace's knowledge by default — per-collection knowledge-graph triples as N-Quads (the collection names the graph, streamed through a tempfile so memory stays flat regardless of knowledge-base size) and the document library (metadata plus content, fetched one document at a time). --config-only skips knowledge on both sides; --triples-limit bounds very large graphs; -f/--flow-id selects the flow the triples services run through. tg-import-workspace streams triples back through the bulk import per collection and recreates library documents (children after parents). Knowledge import is additive, unlike config's skip-existing semantics. Embedding vectors are not carried in bundles: --process re-runs imported documents through the flow, which regenerates extraction output and embeddings; --process-collection targets it. Round-trip covered by unit tests over real archives: export with a mocked Api, re-import, and assert the bulk triples stream and library add calls reproduce the original values (including datatyped literals via the N-Quads path). --- tests/unit/test_cli/conftest.py | 16 +- tests/unit/test_cli/test_nquads.py | 97 +++++ .../test_workspace_bundle_commands.py | 405 ++++++++++++++---- .../trustgraph/cli/export_workspace.py | 260 ++++++++--- .../trustgraph/cli/import_workspace.py | 306 +++++++++++-- trustgraph-cli/trustgraph/cli/nquads.py | 137 ++++++ 6 files changed, 1040 insertions(+), 181 deletions(-) create mode 100644 tests/unit/test_cli/test_nquads.py create mode 100644 trustgraph-cli/trustgraph/cli/nquads.py diff --git a/tests/unit/test_cli/conftest.py b/tests/unit/test_cli/conftest.py index b085345f..ff293844 100644 --- a/tests/unit/test_cli/conftest.py +++ b/tests/unit/test_cli/conftest.py @@ -45,4 +45,18 @@ def sample_metadata(): "metadata": [], "user": "test-user", "collection": "test-collection" - } \ No newline at end of file + } + +def iri(v): + """Wire-format IRI term dict, as triples_query_stream yields them.""" + return {"t": "i", "i": v} + + +def lit(v, d=None, lang=None): + """Wire-format literal term dict (optional datatype / language).""" + t = {"t": "l", "v": v} + if d: + t["d"] = d + if lang: + t["l"] = lang + return t diff --git a/tests/unit/test_cli/test_nquads.py b/tests/unit/test_cli/test_nquads.py new file mode 100644 index 00000000..d395b324 --- /dev/null +++ b/tests/unit/test_cli/test_nquads.py @@ -0,0 +1,97 @@ +""" +Round-trip tests for the streaming N-Quads serializer: wire-format triples +are serialized line-by-line, then parsed back with rdflib's nquads parser +and compared term-for-term — proving the output is valid N-Quads and the +encoding (escaping, datatypes, language tags, unicode) is lossless. +""" + +import io + +import rdflib + +from trustgraph.cli.nquads import serialize_nquads, triple_to_nquad + +from tests.unit.test_cli.conftest import iri, lit + +GRAPH = "urn:trustgraph:collection:default" + + +def roundtrip(batches): + """Serialize then parse back; return (parsed_dataset, written, skipped).""" + out = io.StringIO() + written, skipped = serialize_nquads(batches, GRAPH, out) + ds = rdflib.Dataset() + ds.parse(data=out.getvalue(), format="nquads") + return ds, written, skipped + + +class TestNquadsRoundTrip: + + def test_iri_and_literal_flavours_survive_roundtrip(self): + batches = [[ + {"s": iri("http://example.com/s"), "p": iri("http://example.com/p"), + "o": iri("http://example.com/o")}, + {"s": iri("http://example.com/s"), "p": iri("http://example.com/label"), + "o": lit("plain value")}, + {"s": iri("http://example.com/s"), "p": iri("http://example.com/label"), + "o": lit("bonjour", lang="fr")}, + {"s": iri("http://example.com/s"), "p": iri("http://example.com/count"), + "o": lit("42", d="http://www.w3.org/2001/XMLSchema#integer")}, + ]] + ds, written, skipped = roundtrip(batches) + + assert (written, skipped) == (4, 0) + quads = list(ds.quads((None, None, None, None))) + assert len(quads) == 4 + g = rdflib.URIRef(GRAPH) + assert all(q[3] == g for q in quads) + + objects = {q[2] for q in quads} + assert rdflib.URIRef("http://example.com/o") in objects + assert rdflib.Literal("plain value") in objects + assert rdflib.Literal("bonjour", lang="fr") in objects + assert rdflib.Literal( + "42", datatype=rdflib.URIRef("http://www.w3.org/2001/XMLSchema#integer") + ) in objects + + def test_hostile_literal_content_is_escaped_losslessly(self): + nasty = 'line1\nline2\t"quoted" back\\slash 中文 emoji\U0001f680' + batches = [[{ + "s": iri("http://example.com/s"), + "p": iri("http://example.com/note"), + "o": lit(nasty), + }]] + ds, written, skipped = roundtrip(batches) + + assert (written, skipped) == (1, 0) + obj = next(iter(ds.quads((None, None, None, None))))[2] + assert str(obj) == nasty + + def test_malformed_and_unrepresentable_terms_are_skipped_not_emitted(self): + batches = [[ + # IRI with a space (matches graph_to_turtle's malformed skip) + {"s": iri("http://example.com/bad iri"), "p": iri("http://example.com/p"), + "o": lit("x")}, + # RDF-star quoted triple: no N-Quads encoding + {"s": iri("http://example.com/s"), "p": iri("http://example.com/p"), + "o": {"t": "r", "r": {}}}, + # literal in predicate position: invalid RDF + {"s": iri("http://example.com/s"), "p": lit("not-a-predicate"), + "o": lit("x")}, + # one good triple to prove the stream continues past skips + {"s": iri("http://example.com/s"), "p": iri("http://example.com/p"), + "o": lit("good")}, + ]] + ds, written, skipped = roundtrip(batches) + + assert (written, skipped) == (1, 3) + assert len(list(ds.quads((None, None, None, None)))) == 1 + + def test_streaming_shape_one_line_per_triple(self): + line = triple_to_nquad( + {"s": iri("http://example.com/s"), "p": iri("http://example.com/p"), + "o": lit("v")}, + f"<{GRAPH}>", + ) + assert line.endswith(" .\n") + assert line.count("\n") == 1 diff --git a/tests/unit/test_cli/test_workspace_bundle_commands.py b/tests/unit/test_cli/test_workspace_bundle_commands.py index e45c0ef4..915e0dee 100644 --- a/tests/unit/test_cli/test_workspace_bundle_commands.py +++ b/tests/unit/test_cli/test_workspace_bundle_commands.py @@ -3,19 +3,25 @@ Tests for tg-export-workspace / tg-import-workspace (.tgx bundle commands). The Api class is mocked in each command module's namespace (same pattern as test_config_commands.py); bundles are written to and read from tmp_path so -the archive format itself is exercised end-to-end. +the archive format itself is exercised end-to-end, including the Phase-2 +knowledge tree (per-collection N-Quads + library documents). """ +import datetime +import io import json import tarfile +from types import SimpleNamespace from unittest.mock import Mock, patch import pytest -from trustgraph.api.types import ConfigValue +from trustgraph.api.types import ConfigValue, Triple from trustgraph.cli.export_workspace import export_workspace from trustgraph.cli.import_workspace import import_workspace +from tests.unit.test_cli.conftest import iri, lit + SAMPLE_CONFIG = { "prompt": { "extract-concepts": json.dumps({"template": "Extract {{q}}"}), @@ -26,25 +32,110 @@ SAMPLE_CONFIG = { }, } +# Wire-format triples for one collection, incl. a datatyped literal. +WIRE_BATCHES = [[ + {"s": iri("http://ex.com/s"), "p": iri("http://ex.com/p"), + "o": iri("http://ex.com/o")}, + {"s": iri("http://ex.com/s"), "p": iri("http://ex.com/count"), + "o": lit("42", d="http://www.w3.org/2001/XMLSchema#integer")}, +]] -def make_mock_api(): - mock_config = Mock() +DOC = SimpleNamespace( + id="doc-1", + time=datetime.datetime(2026, 7, 1, 12, 0, 0), + kind="text/plain", + title="Policy", + comments="returns policy", + metadata=[Triple(s="http://ex.com/doc-1", p="http://ex.com/about", + o="returns")], + tags=["policy"], + parent_id="", + document_type="source", +) + + +def make_mock_api(collections=(), batches=(), docs=(), contents=None): + """Full-surface Api mock; returns (mock_api, mock_config).""" mock_api = Mock() + + mock_config = Mock() mock_api.config.return_value = mock_config + mock_config.all.return_value = (SAMPLE_CONFIG, "v42") + + mock_api.collection.return_value.list_collections.return_value = \ + list(collections) + + flow = mock_api.socket.return_value.flow.return_value + flow.triples_query_stream.return_value = iter(batches) + + library = mock_api.library.return_value + library.get_documents.return_value = list(docs) + library.get_document_content.side_effect = \ + lambda id: (contents or {}).get(id, b"") + return mock_api, mock_config +def export_bundle(path, collections=(), batches=(), docs=(), contents=None, + **kwargs): + """Export SAMPLE_CONFIG (+ optional knowledge mocks) to path.""" + mock_api, mock_config = make_mock_api( + collections=collections, batches=batches, docs=docs, + contents=contents, + ) + with patch("trustgraph.cli.export_workspace.Api") as api_cls: + api_cls.return_value = mock_api + export_workspace( + url="http://api/", workspace="source-ws", output=str(path), + **kwargs, + ) + return mock_api, mock_config + + +DEFAULT_MANIFEST = { + "format": "tgx", "format_version": 1, "workspace": "w", + "contents": {"config": True, "knowledge": True}, +} + + +def write_bundle(path, members, manifest=DEFAULT_MANIFEST): + """Write a raw .tgx from name -> bytes (manifest added first).""" + entries = {"manifest.json": json.dumps(manifest).encode(), **members} + with tarfile.open(path, "w:gz") as tar: + for name, data in entries.items(): + info = tarfile.TarInfo(name) + info.size = len(data) + tar.addfile(info, io.BytesIO(data)) + return path + + +def run_import(mock_api, input, **kwargs): + """Run import_workspace against a mocked Api.""" + with patch("trustgraph.cli.import_workspace.Api") as api_cls: + api_cls.return_value = mock_api + import_workspace(url="http://api/", input=str(input), **kwargs) + return api_cls + + @pytest.fixture def bundle(tmp_path): - """Export SAMPLE_CONFIG to a real .tgx and yield its path.""" + """Config-only-shaped bundle (no collections/docs mocked).""" path = tmp_path / "ws.tgx" - with patch("trustgraph.cli.export_workspace.Api") as api_cls: - mock_api, mock_config = make_mock_api() - api_cls.return_value = mock_api - mock_config.all.return_value = (SAMPLE_CONFIG, "v42") - export_workspace( - url="http://api/", workspace="source-ws", output=str(path), - ) + export_bundle(path) + return path + + +@pytest.fixture +def knowledge_bundle(tmp_path): + """Bundle with one collection (2 triples) and one document.""" + path = tmp_path / "kws.tgx" + export_bundle( + path, + collections=[SimpleNamespace(collection="research")], + batches=WIRE_BATCHES, + docs=[DOC], + contents={"doc-1": b"Customers may return items."}, + ) return path @@ -58,7 +149,10 @@ class TestExportWorkspace: assert manifest["format"] == "tgx" assert manifest["workspace"] == "source-ws" assert manifest["config_version"] == "v42" - assert manifest["contents"] == {"config": True, "knowledge": False} + assert manifest["contents"] == {"config": True, "knowledge": True} + assert manifest["knowledge"] == { + "collections": [], "documents": 0, "triples": {}, + } assert "config/prompt/extract-concepts.json" in names assert "config/prompt/answer.json" in names @@ -79,12 +173,12 @@ class TestExportWorkspace: def test_path_unsafe_keys_are_quoted_in_filenames(self, tmp_path): path = tmp_path / "ws.tgx" + mock_api, mock_config = make_mock_api() + mock_config.all.return_value = ( + {"prompt": {"a/b": json.dumps({"x": 1})}}, "v1", + ) with patch("trustgraph.cli.export_workspace.Api") as api_cls: - mock_api, mock_config = make_mock_api() api_cls.return_value = mock_api - mock_config.all.return_value = ( - {"prompt": {"a/b": json.dumps({"x": 1})}}, "v1", - ) export_workspace( url="http://api/", workspace="ws", output=str(path), ) @@ -94,16 +188,63 @@ class TestExportWorkspace: assert "config/prompt/a%2Fb.json" in names assert entry["key"] == "a/b" + def test_knowledge_tree_written_per_collection_and_document( + self, knowledge_bundle): + with tarfile.open(knowledge_bundle, "r:gz") as tar: + names = tar.getnames() + manifest = json.load(tar.extractfile("manifest.json")) + nq = tar.extractfile( + "knowledge/research/triples.nq").read().decode() + meta = json.load( + tar.extractfile("knowledge/library/doc-1.meta.json")) + content = tar.extractfile( + "knowledge/library/doc-1.content").read() + + assert manifest["contents"]["knowledge"] is True + assert manifest["knowledge"] == { + "collections": ["research"], "documents": 1, + "triples": {"research": 2}, + } + assert "knowledge/research/triples.nq" in names + # N-Quads: one line per triple, graph = the collection IRI, and + # the datatyped literal keeps its full quoted form. + lines = [ln for ln in nq.splitlines() if ln] + assert len(lines) == 2 + assert all("" in ln + for ln in lines) + assert '"42"^^' in nq + + assert meta["id"] == "doc-1" + assert meta["title"] == "Policy" + assert meta["metadata"] == [{ + "s": "http://ex.com/doc-1", "p": "http://ex.com/about", + "o": "returns", + }] + assert content == b"Customers may return items." + + def test_config_only_skips_knowledge(self, tmp_path): + path = tmp_path / "co.tgx" + mock_api, _ = export_bundle( + path, + collections=[SimpleNamespace(collection="research")], + config_only=True, + ) + with tarfile.open(path, "r:gz") as tar: + names = tar.getnames() + manifest = json.load(tar.extractfile("manifest.json")) + assert manifest["contents"] == {"config": True, "knowledge": False} + assert "knowledge" not in manifest + assert not any(n.startswith("knowledge/") for n in names) + mock_api.collection.return_value.list_collections.assert_not_called() + mock_api.socket.assert_not_called() + mock_api.library.assert_not_called() + class TestImportWorkspace: def test_roundtrip_puts_all_values_with_overwrite(self, bundle): - with patch("trustgraph.cli.import_workspace.Api") as api_cls: - mock_api, mock_config = make_mock_api() - api_cls.return_value = mock_api - import_workspace( - url="http://api/", input=str(bundle), overwrite=True, - ) + mock_api, mock_config = make_mock_api() + api_cls = run_import(mock_api, bundle, overwrite=True) # Target workspace defaults to the manifest's workspace. api_cls.assert_called_once_with( @@ -123,27 +264,22 @@ class TestImportWorkspace: assert all(isinstance(v, ConfigValue) for v in values) def test_workspace_flag_renames_target(self, bundle): - with patch("trustgraph.cli.import_workspace.Api") as api_cls: - mock_api, mock_config = make_mock_api() - api_cls.return_value = mock_api - import_workspace( - url="http://api/", input=str(bundle), workspace="staging", - overwrite=True, - ) + mock_api, mock_config = make_mock_api() + api_cls = run_import( + mock_api, bundle, workspace="staging", overwrite=True, + ) api_cls.assert_called_once_with( "http://api/", token=None, workspace="staging", ) def test_default_skips_existing_keys(self, bundle): """WorkspaceInit re-run semantics: only missing keys are written.""" - with patch("trustgraph.cli.import_workspace.Api") as api_cls: - mock_api, mock_config = make_mock_api() - api_cls.return_value = mock_api - mock_config.list.side_effect = lambda t: { - "prompt": ["extract-concepts"], - "tool": [], - }[t] - import_workspace(url="http://api/", input=str(bundle)) + mock_api, mock_config = make_mock_api() + mock_config.list.side_effect = lambda t: { + "prompt": ["extract-concepts"], + "tool": [], + }[t] + run_import(mock_api, bundle) values = mock_config.put.call_args.args[0] assert sorted((v.type, v.key) for v in values) == [ @@ -152,17 +288,11 @@ class TestImportWorkspace: ] def test_dry_run_writes_nothing(self, bundle, capsys): - with patch("trustgraph.cli.import_workspace.Api") as api_cls: - mock_api, mock_config = make_mock_api() - api_cls.return_value = mock_api - import_workspace( - url="http://api/", input=str(bundle), overwrite=True, - dry_run=True, - ) + mock_api, mock_config = make_mock_api() + run_import(mock_api, bundle, overwrite=True, dry_run=True) mock_config.put.assert_not_called() out = capsys.readouterr().out assert "would import prompt/extract-concepts" in out - assert "3 item(s) would be imported" in out def test_rejects_bundle_without_manifest(self, tmp_path): path = tmp_path / "bad.tgx" @@ -173,52 +303,147 @@ class TestImportWorkspace: import_workspace(url="http://api/", input=str(path)) def test_rejects_newer_format_version(self, tmp_path): - import io - path = tmp_path / "future.tgx" - manifest = json.dumps({ - "format": "tgx", "format_version": 99, "workspace": "w", - "contents": {"config": True, "knowledge": False}, - }).encode() - with tarfile.open(path, "w:gz") as tar: - info = tarfile.TarInfo("manifest.json") - info.size = len(manifest) - tar.addfile(info, io.BytesIO(manifest)) + path = write_bundle( + tmp_path / "future.tgx", {}, + manifest={**DEFAULT_MANIFEST, "format_version": 99}, + ) with patch("trustgraph.cli.import_workspace.Api"): with pytest.raises(RuntimeError, match="newer than this tool"): import_workspace(url="http://api/", input=str(path)) - def test_refuses_knowledge_bundle_without_config_only(self, tmp_path): - import io - path = tmp_path / "knowledge.tgx" - manifest = json.dumps({ - "format": "tgx", "format_version": 1, "workspace": "w", - "contents": {"config": True, "knowledge": True}, - }).encode() - with tarfile.open(path, "w:gz") as tar: - info = tarfile.TarInfo("manifest.json") - info.size = len(manifest) - tar.addfile(info, io.BytesIO(manifest)) - with patch("trustgraph.cli.import_workspace.Api"): - with pytest.raises(RuntimeError, match="--config-only"): - import_workspace(url="http://api/", input=str(path)) - def test_config_only_flag_allows_knowledge_bundle(self, tmp_path): - import io - path = tmp_path / "knowledge.tgx" - manifest = json.dumps({ - "format": "tgx", "format_version": 1, "workspace": "w", - "contents": {"config": True, "knowledge": True}, - }).encode() - with tarfile.open(path, "w:gz") as tar: - info = tarfile.TarInfo("manifest.json") - info.size = len(manifest) - tar.addfile(info, io.BytesIO(manifest)) - with patch("trustgraph.cli.import_workspace.Api") as api_cls: - mock_api, mock_config = make_mock_api() - api_cls.return_value = mock_api - import_workspace( - url="http://api/", input=str(path), config_only=True, - overwrite=True, - ) - # No config entries in this bundle; nothing written, no error. - mock_config.put.assert_not_called() +class TestImportKnowledge: + + def test_roundtrip_imports_triples_and_documents(self, knowledge_bundle): + mock_api, mock_config = make_mock_api() + run_import(mock_api, knowledge_bundle, overwrite=True) + + # Triples land in the bulk import stream for the right collection. + bulk = mock_api.bulk.return_value + call = bulk.import_triples.call_args + assert call.args[0] == "default" # flow id + triples = sorted(list(call.args[1]), key=lambda t: t.p) + assert triples == [ + Triple(s="http://ex.com/s", p="http://ex.com/count", o="42"), + Triple(s="http://ex.com/s", p="http://ex.com/p", + o="http://ex.com/o"), + ] + assert call.kwargs["metadata"]["collection"] == "research" + + # The document is recreated with its metadata and content. + add = mock_api.library.return_value.add_document.call_args + assert add.kwargs["id"] == "doc-1" + assert add.kwargs["document"] == b"Customers may return items." + assert add.kwargs["title"] == "Policy" + assert add.kwargs["kind"] == "text/plain" + assert add.kwargs["tags"] == ["policy"] + assert add.kwargs["metadata"] == [ + Triple(s="http://ex.com/doc-1", p="http://ex.com/about", + o="returns"), + ] + # No processing unless asked: embeddings re-derivation is opt-in. + mock_api.library.return_value.start_processing.assert_not_called() + + def test_config_only_skips_knowledge_on_import(self, knowledge_bundle): + mock_api, mock_config = make_mock_api() + run_import(mock_api, knowledge_bundle, overwrite=True, + config_only=True) + mock_api.bulk.assert_not_called() + mock_api.library.return_value.add_document.assert_not_called() + mock_config.put.assert_called_once() + + def test_dry_run_covers_knowledge(self, knowledge_bundle, capsys): + mock_api, mock_config = make_mock_api() + run_import(mock_api, knowledge_bundle, overwrite=True, dry_run=True) + mock_api.bulk.assert_not_called() + mock_api.library.return_value.add_document.assert_not_called() + out = capsys.readouterr().out + assert "would import 2 triple(s) into collection 'research'" in out + assert "would import document doc-1" in out + + def test_process_flag_reprocesses_documents(self, knowledge_bundle): + mock_api, mock_config = make_mock_api() + run_import(mock_api, knowledge_bundle, overwrite=True, process=True, + process_collection="research") + proc = mock_api.library.return_value.start_processing.call_args + assert proc.kwargs["document_id"] == "doc-1" + assert proc.kwargs["flow"] == "default" + assert proc.kwargs["collection"] == "research" + + +class TestImportDocumentSkipOverwrite: + """Live-verified semantics: existing documents skip by default, replace + with --overwrite (remove + add; the library API has no content update).""" + + def _bundle_with_doc(self, tmp_path): + return write_bundle(tmp_path / "doc.tgx", { + "knowledge/library/doc-1.meta.json": json.dumps( + {"id": "doc-1", "title": "T", "metadata": []}).encode(), + "knowledge/library/doc-1.content": b"hello", + }) + + def test_existing_document_is_skipped_not_fatal(self, tmp_path): + path = self._bundle_with_doc(tmp_path) + mock_api, mock_config = make_mock_api( + docs=[SimpleNamespace(id="doc-1")], + ) + run_import(mock_api, path, overwrite=False) + lib = mock_api.library.return_value + lib.add_document.assert_not_called() + lib.remove_document.assert_not_called() + + def test_overwrite_replaces_existing_document(self, tmp_path): + path = self._bundle_with_doc(tmp_path) + mock_api, mock_config = make_mock_api( + docs=[SimpleNamespace(id="doc-1")], + ) + run_import(mock_api, path, overwrite=True) + lib = mock_api.library.return_value + lib.remove_document.assert_called_once_with("doc-1") + lib.add_document.assert_called_once() + + +class TestCollectionDiscovery: + """Live-verified: implicitly-created collections (raw triple loads) are + queryable but unlisted, so export merges --collection extras and import + registers what it restores.""" + + def test_export_includes_extra_unregistered_collections(self, tmp_path): + path = tmp_path / "ws.tgx" + mock_api, _ = export_bundle( + path, + collections=[SimpleNamespace(collection="default")], + batches=[[]], + extra_collections=["research"], + ) + with tarfile.open(path, "r:gz") as tar: + manifest = json.load(tar.extractfile("manifest.json")) + assert manifest["knowledge"]["collections"] == ["default", "research"] + + def test_import_registers_each_restored_collection(self, tmp_path): + path = write_bundle(tmp_path / "kb.tgx", { + "knowledge/research/triples.nq": ( + b' "v" ' + b' .\n' + ), + }) + mock_api, mock_config = make_mock_api() + run_import(mock_api, path) + mock_api.collection.return_value.update_collection.assert_called_once_with( + "research", name="research", + ) + + def test_registered_collections_are_not_reregistered(self, tmp_path): + """update_collection is an upsert that clears omitted fields, so a + collection already in the registry must be left untouched.""" + path = write_bundle(tmp_path / "kb.tgx", { + "knowledge/research/triples.nq": ( + b' "v" ' + b' .\n' + ), + }) + mock_api, mock_config = make_mock_api( + collections=[SimpleNamespace(collection="research")], + ) + run_import(mock_api, path) + mock_api.collection.return_value.update_collection.assert_not_called() diff --git a/trustgraph-cli/trustgraph/cli/export_workspace.py b/trustgraph-cli/trustgraph/cli/export_workspace.py index 41d3403d..2474ce36 100644 --- a/trustgraph-cli/trustgraph/cli/export_workspace.py +++ b/trustgraph-cli/trustgraph/cli/export_workspace.py @@ -1,14 +1,14 @@ """ -Exports a workspace's full configuration state as a portable .tgx bundle -(a gzipped tar archive) for backup, migration between deployments, or -sharing a pre-configured workspace. +Exports a workspace's full state as a portable .tgx bundle (a gzipped tar +archive) for backup, migration between deployments, or sharing a +pre-configured workspace. -The bundle is human-readable: a manifest.json plus one pretty-printed JSON -file per config key under config//, so it can be inspected and -hand-edited before import. Each entry file embeds its own type and key, so -filenames are cosmetic. Knowledge export (triples, documents, embeddings) -is not yet included; the manifest records that so future importers can -distinguish config-only bundles. +The bundle carries the workspace configuration (one pretty-printed JSON +file per config key) and, by default, its knowledge: per-collection +knowledge-graph triples as N-Quads (the collection names the graph) and +the document library (metadata plus content). Pass --config-only to +export just the configuration. Embedding vectors are not exported — +re-processing imported documents through a flow regenerates them. """ import argparse @@ -17,11 +17,14 @@ import json import os import sys import tarfile +import tempfile import time from urllib.parse import quote from trustgraph.api import Api +from . nquads import serialize_nquads + default_url = os.getenv("TRUSTGRAPH_URL", 'http://localhost:8088/') default_token = os.getenv("TRUSTGRAPH_TOKEN", None) default_workspace = os.getenv("TRUSTGRAPH_WORKSPACE", "default") @@ -29,6 +32,10 @@ default_workspace = os.getenv("TRUSTGRAPH_WORKSPACE", "default") MANIFEST_FORMAT = "tgx" MANIFEST_FORMAT_VERSION = 1 +# triples_query_stream is bounded by a limit; exports want "everything", so +# default high and let --triples-limit override for truly huge graphs. +DEFAULT_TRIPLES_LIMIT = 1_000_000 + def _add_bytes(tar, name, data): info = tarfile.TarInfo(name=name) @@ -37,56 +44,176 @@ def _add_bytes(tar, name, data): tar.addfile(info, io.BytesIO(data)) -def export_workspace(url, workspace, output, token=None): - - api = Api(url, token=token, workspace=workspace).config() - - config, version = api.all() - - manifest = { - "format": MANIFEST_FORMAT, - "format_version": MANIFEST_FORMAT_VERSION, - "workspace": workspace, - "config_version": version, - "exported_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), - "contents": {"config": True, "knowledge": False}, - } - +def _export_config(tar, config): + """Write one self-describing JSON file per config key; return count.""" count = 0 + for type_, entries in sorted(config.items()): + for key, raw in sorted(entries.items()): + + # Config values are stored as JSON strings; parse so the + # bundle is pretty-printed and hand-editable. A value that + # isn't valid JSON is preserved verbatim. + try: + value = json.loads(raw) + except (TypeError, json.JSONDecodeError): + value = raw + + entry = {"type": type_, "key": key, "value": value} + + # Keys may contain path-unsafe characters; the entry embeds + # the real key, so the quoted filename is cosmetic only. + name = f"config/{quote(type_, safe='')}/{quote(key, safe='')}.json" + + _add_bytes( + tar, name, + json.dumps(entry, indent=2).encode("utf-8"), + ) + + count += 1 + return count + + +def _export_triples(tar, api, flow_id, collections, triples_limit): + """Stream each collection's triples into knowledge//triples.nq. + + N-Quads are written to a tempfile first (tar members need their size + upfront), so memory stays flat regardless of knowledge-base size. + Returns {collection: written} and a total skipped count. + """ + counts = {} + skipped_total = 0 + socket = api.socket() + try: + flow = socket.flow(flow_id) + for c in collections: + graph_iri = f"urn:trustgraph:collection:{quote(c, safe='')}" + tmp = tempfile.NamedTemporaryFile( + "w", encoding="utf-8", suffix=".nq", delete=False, + ) + try: + with tmp: + written, skipped = serialize_nquads( + flow.triples_query_stream( + s=None, p=None, o=None, + collection=c, + limit=triples_limit, + batch_size=100, + ), + graph_iri, + tmp, + ) + if written: + tar.add( + tmp.name, + arcname=( + f"knowledge/{quote(c, safe='')}/triples.nq" + ), + ) + counts[c] = written + skipped_total += skipped + finally: + os.unlink(tmp.name) + finally: + socket.close() + return counts, skipped_total + + +def _export_library(tar, api): + """Write each library document's metadata + content; return count.""" + library = api.library() + count = 0 + for doc in library.get_documents(include_children=True): + # Content is fetched one document at a time so memory is bounded + # by the largest single document, not the whole library. + content = library.get_document_content(doc.id) + meta = { + "id": doc.id, + "time": doc.time.isoformat() if doc.time else None, + "kind": doc.kind, + "title": doc.title, + "comments": doc.comments, + "metadata": [ + {"s": t.s, "p": t.p, "o": t.o} for t in (doc.metadata or []) + ], + "tags": list(doc.tags or []), + "parent_id": doc.parent_id or "", + "document_type": getattr(doc, "document_type", "") or "", + } + base = f"knowledge/library/{quote(doc.id, safe='')}" + _add_bytes( + tar, f"{base}.meta.json", + json.dumps(meta, indent=2).encode("utf-8"), + ) + _add_bytes(tar, f"{base}.content", content or b"") + count += 1 + return count + + +def export_workspace( + url, workspace, output, token=None, config_only=False, + flow_id="default", triples_limit=DEFAULT_TRIPLES_LIMIT, + extra_collections=(), +): + + api = Api(url, token=token, workspace=workspace) + + config, version = api.config().all() + + # Collection discovery is registry-based: collections created implicitly + # by raw triple loads (e.g. tg-load-knowledge) are queryable but not + # listed, so they would silently drop out of the bundle. --collection + # names them explicitly; the enumeration is printed so what's included + # is never a guess. + collections = [] + if not config_only: + registered = [c.collection for c in api.collection().list_collections()] + collections = sorted(set(registered) | set(extra_collections)) + print(f"Exporting collections: {', '.join(collections)}", flush=True) with tarfile.open(output, "w:gz") as tar: + config_count = _export_config(tar, config) + + triple_counts = {} + skipped = 0 + doc_count = 0 + if not config_only: + triple_counts, skipped = _export_triples( + tar, api, flow_id, collections, triples_limit, + ) + doc_count = _export_library(tar, api) + + manifest = { + "format": MANIFEST_FORMAT, + "format_version": MANIFEST_FORMAT_VERSION, + "workspace": workspace, + "config_version": version, + "exported_at": time.strftime( + "%Y-%m-%dT%H:%M:%SZ", time.gmtime(), + ), + "contents": {"config": True, "knowledge": not config_only}, + } + if not config_only: + manifest["knowledge"] = { + "collections": collections, + "documents": doc_count, + "triples": triple_counts, + } + _add_bytes( tar, "manifest.json", json.dumps(manifest, indent=2).encode("utf-8"), ) - for type_, entries in sorted(config.items()): - for key, raw in sorted(entries.items()): - - # Config values are stored as JSON strings; parse so the - # bundle is pretty-printed and hand-editable. A value that - # isn't valid JSON is preserved verbatim. - try: - value = json.loads(raw) - except (TypeError, json.JSONDecodeError): - value = raw - - entry = {"type": type_, "key": key, "value": value} - - # Keys may contain path-unsafe characters; the entry embeds - # the real key, so the quoted filename is cosmetic only. - name = f"config/{quote(type_, safe='')}/{quote(key, safe='')}.json" - - _add_bytes( - tar, name, - json.dumps(entry, indent=2).encode("utf-8"), - ) - - count += 1 - - print(f"Exported {count} config item(s) from workspace " - f"'{workspace}' to {output}", flush=True) + summary = f"Exported {config_count} config item(s)" + if not config_only: + summary += ( + f", {sum(triple_counts.values())} triple(s) across " + f"{len(triple_counts)} collection(s), {doc_count} document(s)" + ) + if skipped: + summary += f" ({skipped} triple(s) not representable, skipped)" + print(f"{summary} from workspace '{workspace}' to {output}", flush=True) def main(): @@ -114,12 +241,41 @@ def main(): help=f'Workspace to export (default: {default_workspace})', ) + parser.add_argument( + '-c', '--collection', + action='append', + default=[], + help='Additionally export this collection even if it is not ' + 'registered in collection management (repeatable)', + ) + parser.add_argument( '-o', '--output', required=True, help='Output bundle path, e.g. workspace-default.tgx', ) + parser.add_argument( + '-f', '--flow-id', + default="default", + help='Flow to query triples through (default: default)', + ) + + parser.add_argument( + '--config-only', + action='store_true', + help='Export only the configuration, skipping knowledge ' + '(triples and library documents)', + ) + + parser.add_argument( + '--triples-limit', + type=int, + default=DEFAULT_TRIPLES_LIMIT, + help='Maximum triples to export per collection ' + f'(default: {DEFAULT_TRIPLES_LIMIT})', + ) + args = parser.parse_args() try: @@ -129,6 +285,10 @@ def main(): workspace=args.workspace, output=args.output, token=args.token, + config_only=args.config_only, + flow_id=args.flow_id, + triples_limit=args.triples_limit, + extra_collections=args.collection, ) except Exception as e: diff --git a/trustgraph-cli/trustgraph/cli/import_workspace.py b/trustgraph-cli/trustgraph/cli/import_workspace.py index 71885bfb..13fabdd1 100644 --- a/trustgraph-cli/trustgraph/cli/import_workspace.py +++ b/trustgraph-cli/trustgraph/cli/import_workspace.py @@ -3,10 +3,16 @@ Imports a workspace bundle (.tgx, produced by tg-export-workspace) into a TrustGraph deployment. The target workspace defaults to the name recorded in the bundle's manifest and can be renamed with --workspace. -By default existing (type, key) entries in the target workspace are left -untouched and only missing keys are added, matching WorkspaceInit's -re-run behaviour; pass --overwrite to replace every imported key. Use ---dry-run to show what would be written without changing anything. +Configuration import follows WorkspaceInit's re-run behaviour: existing +(type, key) entries are left untouched and only missing keys are added; +pass --overwrite to replace every imported key. Knowledge import (triples +and library documents) is additive — triples are streamed into the target +collection and documents are added to the library; re-importing the same +bundle twice will duplicate knowledge, not merge it. Use --dry-run to +show what would be written without changing anything, --config-only to +skip a bundle's knowledge, and --process to re-run imported documents +through a flow (which regenerates embeddings, so bundles don't carry +vectors). """ import argparse @@ -14,9 +20,12 @@ import json import os import sys import tarfile +import uuid +from urllib.parse import unquote from trustgraph.api import Api -from trustgraph.api.types import ConfigValue +from trustgraph.api.types import ConfigValue, Triple +from trustgraph.cli.nquads import parse_nquads default_url = os.getenv("TRUSTGRAPH_URL", 'http://localhost:8088/') default_token = os.getenv("TRUSTGRAPH_TOKEN", None) @@ -26,10 +35,21 @@ SUPPORTED_FORMAT_VERSION = 1 def _read_bundle(path): - """Read manifest and config entries from a .tgx bundle.""" + """Read manifest, config entries, triples and documents from a .tgx. + + Returns (manifest, config_entries, triples_by_collection, documents) + where triples_by_collection maps collection -> list[Triple] and + documents is a list of {meta: dict, content: bytes}. + """ manifest = None - entries = [] + config_entries = [] + triples = {} + doc_meta = {} + doc_content = {} + + def member_id(name, prefix, suffix): + return unquote(name[len(prefix):-len(suffix)]) with tarfile.open(path, "r:gz") as tar: for member in tar.getmembers(): @@ -39,11 +59,28 @@ def _read_bundle(path): if f is None: continue data = f.read() - if member.name == "manifest.json": + name = member.name + + if name == "manifest.json": manifest = json.loads(data) - elif member.name.startswith("config/") and \ - member.name.endswith(".json"): - entries.append(json.loads(data)) + + elif name.startswith("config/") and name.endswith(".json"): + config_entries.append(json.loads(data)) + + elif name.startswith("knowledge/library/") and \ + name.endswith(".meta.json"): + doc_id = member_id(name, "knowledge/library/", ".meta.json") + doc_meta[doc_id] = json.loads(data) + + elif name.startswith("knowledge/library/") and \ + name.endswith(".content"): + doc_id = member_id(name, "knowledge/library/", ".content") + doc_content[doc_id] = data + + elif name.startswith("knowledge/") and \ + name.endswith("/triples.nq"): + collection = member_id(name, "knowledge/", "/triples.nq") + triples[collection] = parse_nquads(data) if manifest is None: raise RuntimeError("not a workspace bundle: manifest.json missing") @@ -60,27 +97,18 @@ def _read_bundle(path): "upgrade trustgraph-cli" ) - return manifest, entries + documents = [ + {"meta": meta, "content": doc_content.get(doc_id, b"")} + for doc_id, meta in doc_meta.items() + ] + + return manifest, config_entries, triples, documents -def import_workspace( - url, input, workspace=None, overwrite=False, config_only=False, - dry_run=False, token=None, -): +def _import_config(api, entries, overwrite, dry_run): + """Import config entries; returns (imported, skipped) counts.""" - manifest, entries = _read_bundle(input) - - # Knowledge import (triples, documents, embeddings) is not implemented - # yet; refuse to silently drop it from a bundle that carries it. - if manifest.get("contents", {}).get("knowledge") and not config_only: - raise RuntimeError( - "bundle contains knowledge data, which this tool cannot import " - "yet; re-run with --config-only to import just the configuration" - ) - - target = workspace or manifest.get("workspace") or "default" - - api = Api(url, token=token, workspace=target).config() + config = api.config() # Mirror WorkspaceInit's re-run behaviour: without --overwrite, keys # already present in the target workspace are skipped (per key, not per @@ -88,7 +116,7 @@ def import_workspace( existing = {} if not overwrite: for type_ in sorted({e["type"] for e in entries}): - existing[type_] = set(api.list(type_)) + existing[type_] = set(config.list(type_)) values = [] skipped = 0 @@ -106,16 +134,191 @@ def import_workspace( if dry_run: for v in values: print(f"would import {v.type}/{v.key}", flush=True) - print(f"Dry run: {len(values)} item(s) would be imported into " - f"workspace '{target}', {skipped} skipped as existing", - flush=True) - return + elif values: + config.put(values) - if values: - api.put(values) + return len(values), skipped - print(f"Imported {len(values)} config item(s) into workspace " - f"'{target}', {skipped} skipped as existing", flush=True) + +def _import_triples(api, flow_id, triples_by_collection, dry_run): + """Stream each collection's triples into the flow; returns count.""" + + # Collections restored by the bulk triples path (unlike document + # processing) are not auto-registered, and an unregistered collection + # would silently drop out of a future export. Register only the missing + # ones: update_collection is an upsert whose omitted fields clear the + # description/tags that _import_config may have just restored. + # Best-effort — a registry hiccup shouldn't fail a completed triple + # import (tg-set-collection is the manual remedy). + registered = None + if triples_by_collection and not dry_run: + try: + registered = { + c.collection for c in api.collection().list_collections() + } + except Exception as e: + print( + f"warning: could not list collections for registration: " + f"{e} — register manually with tg-set-collection", + flush=True, + ) + + total = 0 + for collection, triples in sorted(triples_by_collection.items()): + if dry_run: + print( + f"would import {len(triples)} triple(s) into " + f"collection '{collection}'", + flush=True, + ) + else: + api.bulk().import_triples( + flow_id, + triples, + metadata={ + "id": f"workspace-import-{uuid.uuid4()}", + "metadata": [], + "collection": collection, + }, + ) + if registered is not None and collection not in registered: + try: + api.collection().update_collection( + collection, name=collection, + ) + except Exception as e: + print( + f"warning: could not register collection " + f"'{collection}': {e} — register manually with " + f"tg-set-collection", + flush=True, + ) + total += len(triples) + return total + + +def _import_documents( + api, documents, flow_id, process, process_collection, overwrite, + dry_run, +): + """Add library documents back (children after parents). + + Mirrors the config semantics: documents already present in the target + workspace are skipped unless --overwrite, which replaces them (the + library API has no in-place content update, so replace = remove + add). + Returns (imported, skipped). + """ + + library = api.library() + + existing = set() + if documents: + existing = {d.id for d in library.get_documents(include_children=True)} + + # Parents must exist before their children. + ordered = sorted( + documents, key=lambda d: bool(d["meta"].get("parent_id")), + ) + + count = 0 + doc_skipped = 0 + for doc in ordered: + meta = doc["meta"] + + if meta["id"] in existing and not overwrite: + doc_skipped += 1 + if dry_run: + print(f"would skip existing document {meta['id']}", flush=True) + continue + + if dry_run: + print(f"would import document {meta['id']}", flush=True) + count += 1 + continue + + if meta["id"] in existing: + library.remove_document(meta["id"]) + + metadata = [ + Triple(s=t["s"], p=t["p"], o=t["o"]) + for t in meta.get("metadata", []) + ] + if meta.get("parent_id"): + library.add_child_document( + document=doc["content"], + id=meta["id"], + parent_id=meta["parent_id"], + title=meta.get("title", ""), + comments=meta.get("comments", ""), + kind=meta.get("kind", "text/plain"), + tags=meta.get("tags", []), + metadata=metadata, + ) + else: + library.add_document( + document=doc["content"], + id=meta["id"], + metadata=metadata, + title=meta.get("title", ""), + comments=meta.get("comments", ""), + kind=meta.get("kind", "text/plain"), + tags=meta.get("tags", []), + ) + + # Re-processing regenerates extraction output and embeddings for + # the imported content (bundles carry no vectors). + if process: + library.start_processing( + id=f"proc-{uuid.uuid4()}", + document_id=meta["id"], + flow=flow_id, + collection=process_collection, + tags=meta.get("tags", []), + ) + + count += 1 + return count, doc_skipped + + +def import_workspace( + url, input, workspace=None, overwrite=False, config_only=False, + flow_id="default", process=False, process_collection="default", + dry_run=False, token=None, +): + + manifest, config_entries, triples, documents = _read_bundle(input) + + target = workspace or manifest.get("workspace") or "default" + + api = Api(url, token=token, workspace=target) + + imported, skipped = _import_config( + api, config_entries, overwrite, dry_run, + ) + + triple_count = 0 + doc_count = 0 + has_knowledge = bool(triples or documents) + if has_knowledge and not config_only: + triple_count = _import_triples(api, flow_id, triples, dry_run) + doc_count, doc_skipped = _import_documents( + api, documents, flow_id, process, process_collection, overwrite, + dry_run, + ) + + verb = "Dry run:" if dry_run else "Imported" + summary = ( + f"{verb} {imported} config item(s) into workspace '{target}', " + f"{skipped} skipped as existing" + ) + if has_knowledge and not config_only: + summary += ( + f"; {triple_count} triple(s), {doc_count} document(s)" + f" ({doc_skipped} skipped as existing)" + ) + elif has_knowledge: + summary += "; knowledge skipped (--config-only)" + print(summary, flush=True) def main(): @@ -150,11 +353,18 @@ def main(): 'bundle manifest)', ) + parser.add_argument( + '-f', '--flow-id', + default="default", + help='Flow to import triples through and process documents with ' + '(default: default)', + ) + parser.add_argument( '--overwrite', action='store_true', - help='Replace existing keys in the target workspace (default: ' - 'keep existing keys and only add missing ones)', + help='Replace existing config keys in the target workspace ' + '(default: keep existing keys and only add missing ones)', ) parser.add_argument( @@ -164,6 +374,19 @@ def main(): 'in the bundle', ) + parser.add_argument( + '--process', + action='store_true', + help='Re-process imported documents through the flow after import ' + '(regenerates extraction output and embeddings)', + ) + + parser.add_argument( + '--process-collection', + default="default", + help='Collection that --process targets (default: default)', + ) + parser.add_argument( '--dry-run', action='store_true', @@ -180,6 +403,9 @@ def main(): workspace=args.workspace, overwrite=args.overwrite, config_only=args.config_only, + flow_id=args.flow_id, + process=args.process, + process_collection=args.process_collection, dry_run=args.dry_run, token=args.token, ) diff --git a/trustgraph-cli/trustgraph/cli/nquads.py b/trustgraph-cli/trustgraph/cli/nquads.py new file mode 100644 index 00000000..f38a7747 --- /dev/null +++ b/trustgraph-cli/trustgraph/cli/nquads.py @@ -0,0 +1,137 @@ +""" +N-Quads serialization and parsing for workspace knowledge bundles: the +wire-format triples yielded by triples_query_stream go out one line per +triple (so an export never holds a whole graph in memory), and bundle +members come back as api Triple values. Term encoding is hand-rolled to +the N-Triples grammar: rdflib's term.n3() emits Turtle-style forms +(numeric shorthand, unescaped newlines) that are not valid in +line-oriented N-Quads. +""" + +import re + +import rdflib + +from trustgraph.schema import IRI, LITERAL +from trustgraph.api.types import Triple + +# RDF-star quoted triples have no standard N-Quads encoding; they are +# skipped with a count so callers can surface the omission. + +# N-Triples string-literal escapes (ECHAR): backslash first, then the rest. +_ESCAPES = [ + ("\\", "\\\\"), + ('"', '\\"'), + ("\n", "\\n"), + ("\r", "\\r"), + ("\t", "\\t"), +] + +# Characters the IRIREF production cannot carry: controls/space plus the +# explicitly forbidden set. One compiled scan keeps this off the profile +# for large exports (it runs per term). +_BAD_IRI = re.compile(r'[\x00-\x20<>"{}|^\x60]') + + +def _escape_literal(value): + for raw, esc in _ESCAPES: + value = value.replace(raw, esc) + return value + + +def _encode_iri(iri): + """, or None for values the grammar cannot carry.""" + if not iri or _BAD_IRI.search(iri): + return None + return f"<{iri}>" + + +def encode_term(term, is_object=False): + """Encode one wire-format term dict for an N-Quads line. + + :param is_object: literals are only valid in object position; + subjects and predicates must be IRIs (bnodes never appear on + the wire). + :returns: encoded string, or None when the term can't be represented. + """ + if term is None: + return None + + t = term.get("t", "") + + if t == IRI: + return _encode_iri(term.get("i", "")) + + if t == LITERAL and is_object: + value = _escape_literal(term.get("v", "")) + language = term.get("l") + datatype = term.get("d") + if language: + return f'"{value}"@{language}' + if datatype: + dt = _encode_iri(datatype) + if dt is None: + return None + return f'"{value}"^^{dt}' + return f'"{value}"' + + # literals outside object position, RDF-star, unknown types + return None + + +def triple_to_nquad(triple, graph_encoded): + """One wire-format triple dict -> an N-Quads line, or None to skip. + + :param triple: {"s": term, "p": term, "o": term} wire dict + :param graph_encoded: pre-encoded string + """ + s = encode_term(triple.get("s")) + p = encode_term(triple.get("p")) + o = encode_term(triple.get("o"), is_object=True) + if s is None or p is None or o is None: + return None + return f"{s} {p} {o} {graph_encoded} .\n" + + +def serialize_nquads(batches, graph_iri, out): + """Write wire-format triple batches to a text file-like as N-Quads. + + :param batches: iterable of lists of wire triple dicts + (e.g. triples_query_stream output) + :param graph_iri: graph name for every quad (str) + :param out: text-mode file-like with .write() + :returns: (written, skipped) counts + """ + g = _encode_iri(graph_iri) + if g is None: + raise ValueError(f"graph IRI not representable in N-Quads: {graph_iri!r}") + written = 0 + skipped = 0 + for batch in batches: + for t in batch: + line = triple_to_nquad(t, g) + if line is None: + skipped += 1 + else: + out.write(line) + written += 1 + return written, skipped + + +def parse_nquads(data): + """Parse N-Quads bytes back into api Triple values. + + Terms are stringified with str(), the same convention tg-load-knowledge + uses, so values survive the store round trip unchanged. The whole + member is materialized in memory (bundles are bounded by + --triples-limit at export); line-streaming is a possible follow-up. + + :param data: N-Quads bytes (one bundle member) + :returns: list of Triple + """ + ds = rdflib.Dataset() + ds.parse(data=data.decode("utf-8"), format="nquads") + return [ + Triple(s=str(s), p=str(p), o=str(o)) + for s, p, o, _g in ds.quads((None, None, None, None)) + ]