feat: specify backend interface and add reference implmentations

Author: simontaurus

src/oold/backend/document_store.py

@@ -0,0 +1,100 @@
+import json
+import sqlite3
+from pathlib import Path
+from typing import Dict, List, Optional, Union
+
+from oold.backend.interface import Backend, StoreResult
+
+
+class SimpleDictDocumentStore(Backend):
+    _store: Optional[Dict[str, dict]] = None
+
+    def __init__(self):
+        self._store = {}
+
+    def resolve_iris(self, iris: List[str]) -> Dict[str, Dict]:
+        jsonld_dicts = {}
+        for iri in iris:
+            jsonld_dicts[iri] = self._store.get(iri, None)
+        return jsonld_dicts
+
+    def store_jsonld_dicts(self, jsonld_dicts: Dict[str, Dict]) -> StoreResult:
+        for iri, jsonld_dict in jsonld_dicts.items():
+            self._store[iri] = jsonld_dict
+        return StoreResult(success=True)
+
+    def query():
+        pass
+
+
+class SqliteDocumentStore(Backend):
+    db_path: Union[Path, str]
+    persist_connection: bool = False
+    _conn: Optional[sqlite3.Connection] = None
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+        if self.db_path == ":memory:":
+            self.persist_connection = True
+            self._conn = sqlite3.connect(self.db_path)
+
+        # create table 'entities' if not exists
+        conn = self._conn if self.persist_connection else sqlite3.connect(self.db_path)
+        c = conn.cursor()
+        c.execute(
+            """
+            CREATE TABLE IF NOT EXISTS entities (
+                id TEXT PRIMARY KEY,
+                data JSONB
+            )
+            """
+        )
+        conn.commit()
+        if not self.persist_connection:
+            conn.close()
+
+    def close(self):
+        """Close the persistent connection if it exists."""
+        if self._conn is not None:
+            self._conn.close()
+            self._conn = None
+
+    def resolve_iris(self, iris: List[str]) -> Dict[str, Dict]:
+        jsonld_dicts = {}
+        conn = self._conn if self.persist_connection else sqlite3.connect(self.db_path)
+        c = conn.cursor()
+        c.execute(
+            """
+            SELECT id, data FROM entities WHERE id IN ({})
+            """.format(
+                ",".join("?" for _ in iris)
+            ),
+            iris,
+        )
+        rows = c.fetchall()
+        for iri, data in rows:
+            jsonld_dicts[iri] = json.loads(data)
+        if not self.persist_connection:
+            conn.close()
+        return jsonld_dicts
+
+    def store_jsonld_dicts(self, jsonld_dicts: Dict[str, Dict]) -> StoreResult:
+        conn = self._conn if self.persist_connection else sqlite3.connect(self.db_path)
+        c = conn.cursor()
+        c.executemany(
+            """
+            INSERT OR REPLACE INTO entities (id, data) VALUES (?, ?)
+            """,
+            [
+                (iri, json.dumps(jsonld_dict))
+                for iri, jsonld_dict in jsonld_dicts.items()
+            ],
+        )
+        conn.commit()
+        if not self.persist_connection:
+            conn.close()
+        return StoreResult(success=True)
+
+    def query():
+        raise NotImplementedError()

src/oold/backend/interface.py

@@ -35,7 +35,7 @@ class Resolver(BaseModel):
     model_cls: Optional[Type[GenericLinkedBaseModel]] = None
 
     @abstractmethod
-    def resolve_iri(self, iri) -> Dict:
+    def resolve_iris(self, iris: List[str]) -> Dict[str, Dict]:
         pass
 
     def resolve(self, request: ResolveParam):
@@ -47,11 +47,15 @@ def resolve(self, request: ResolveParam):
         if model_cls is None:
             raise ValueError("No model_cls provided in request or resolver")
 
+        jsonld_dicts = self.resolve_iris(request.iris)
         nodes = {}
-        for iri in request.iris:
-            # nodes[iri] = self.resolve_iri(iri)
-            jsonld_dict = self.resolve_iri(iri)
-            nodes[iri] = model_cls.from_jsonld(jsonld_dict)
+        for iri, jsonld_dict in jsonld_dicts.items():
+            if jsonld_dict is None:
+                nodes[iri] = None
+            else:
+                node = model_cls.from_jsonld(jsonld_dict)
+                nodes[iri] = node
+
         return ResolveResult(nodes=nodes)
 
 
@@ -65,7 +69,71 @@ def set_resolver(param: SetResolverParam) -> None:
 
 def get_resolver(param: GetResolverParam) -> GetResolverResult:
     # ToDo: Handle prefixes (ex:) as well as full IRIs (http://example.com/)
+    # ToDo: Handle list of IRIs with mixed domains
     iri = param.iri.split(":")[0]
     if iri not in _resolvers:
         raise ValueError(f"No resolvers found for {iri}")
     return GetResolverResult(resolver=_resolvers[iri])
+
+
+class SetBackendParam(BaseModel):
+    iri: str
+    backend: "Backend"
+
+
+class GetBackendParam(BaseModel):
+    iri: str
+
+
+class GetBackendResult(BaseModel):
+    backend: "Backend"
+
+
+class StoreParam(BaseModel):
+    model_config = {
+        "arbitrary_types_allowed": True,
+    }
+    nodes: Dict[str, Union[None, GenericLinkedBaseModel]]
+
+
+class StoreResult(BaseModel):
+    success: bool
+
+
+class Query(BaseModel):
+    pass
+
+
+class Backend(Resolver):
+    def store(self, param: StoreParam) -> StoreResult:
+        jsonld_dicts = {}
+        for iri, node in param.nodes.items():
+            if node is None:
+                jsonld_dicts[iri] = None
+            else:
+                jsonld_dicts[iri] = node.to_jsonld()
+        return self.store_jsonld_dicts(jsonld_dicts)
+
+    @abstractmethod
+    def store_jsonld_dicts(self, jsonld_dicts: Dict[str, Dict]) -> StoreResult:
+        pass
+
+    @abstractmethod
+    def query(self, query: Query) -> ResolveResult:
+        """Query the backend and return a ResolveResult."""
+        pass
+
+
+global _backends
+_backends = {}
+
+
+def set_backend(param: SetBackendParam) -> None:
+    _backends[param.iri] = param.backend
+
+
+def get_backend(param: GetBackendParam) -> GetBackendResult:
+    iri = param.iri.split(":")[0]
+    if iri not in _backends:
+        raise ValueError(f"No backends found for {iri}")
+    return GetBackendResult(backend=_backends[iri])

src/oold/backend/sparql.py

@@ -1,11 +1,11 @@
 import json
-from typing import Dict, Optional
+from typing import Dict, List, Optional
 
 from pydantic import ConfigDict
 from rdflib import Graph
 from SPARQLWrapper import JSONLD, SPARQLWrapper
 
-from oold.backend.interface import Resolver
+from oold.backend.interface import Backend, Resolver, StoreResult
 
 
 class LocalSparqlResolver(Resolver):
@@ -18,30 +18,60 @@ def __init__(self, **kwargs):
         if self.graph is None:
             self.graph = Graph()
 
-    def resolve_iri(self, iri) -> Dict:
+    def resolve_iris(self, iris: List[str]) -> Dict[str, Dict]:
         # sparql query to get a node by IRI with all its properties
         # using CONSTRUCT to get the full node
         # format the result as json-ld
-        iri_filter = f"FILTER (?s = {iri})"
-        # check if the iri is a full IRI or a prefix
-        if iri.startswith("http"):
-            iri_filter = f"FILTER (?s = <{iri}>)"
-        qres = self.graph.query(
-            """
-            PREFIX ex: <https://example.com/>
-            CONSTRUCT {
-                ?s ?p ?o .
-            }
-            WHERE {
-                ?s ?p ?o .
-                {{{iri_filter}}}
-            }
-            """.replace(
+        jsonld_dicts = {}
+        for iri in iris:
+            iri_filter = f"FILTER (?s = {iri})"
+            # check if the iri is a full IRI or a prefix
+            if iri.startswith("http"):
+                iri_filter = f"FILTER (?s = <{iri}>)"
+            qres = self.graph.query(
+                """
+                PREFIX ex: <https://example.com/>
+                CONSTRUCT {
+                    ?s ?p ?o .
+                }
+                WHERE {
+                    ?s ?p ?o .
+                    {{{iri_filter}}}
+                }
+                """.replace(
+                    "{{{iri_filter}}}", iri_filter
+                )
+            )
+            jsonld_dict = json.loads(qres.serialize(format="json-ld"))[0]
+            jsonld_dicts[iri] = jsonld_dict
+        return jsonld_dicts
+
+
+class LocalSparqlBackend(LocalSparqlResolver, Backend):
+    def store_jsonld_dicts(self, jsonld_dicts: Dict[str, Dict]) -> StoreResult:
+        # delete all triples with the given iris as subject
+        for iri in jsonld_dicts.keys():
+            iri_filter = f"{iri}"
+            # check if the iri is a full IRI or a prefix
+            if iri.startswith("http"):
+                iri_filter = f"<{iri}>"
+            query = """
+                PREFIX ex: <https://example.com/>
+                DELETE WHERE {
+                    {{{iri_filter}}} ?p ?o .
+                }
+                """.replace(
                 "{{{iri_filter}}}", iri_filter
             )
-        )
-        jsonld_dict = json.loads(qres.serialize(format="json-ld"))[0]
-        return jsonld_dict
+            self.graph.update(query)
+            # convert jsonld_dict to rdflib triples and add to graph
+            g = Graph()
+            g.parse(data=json.dumps(jsonld_dicts[iri]), format="json-ld")
+            self.graph += g
+        return StoreResult(success=True)
+
+    def query():
+        raise NotImplementedError()
 
 
 class WikiDataSparqlResolver(Resolver):
@@ -54,36 +84,39 @@ def __init__(self, **kwargs):
 
         self._sparql = SPARQLWrapper(self.endpoint)
 
-    def resolve_iri(self, iri):
+    def resolve_iri(self, iris: List[str]) -> Dict[str, Dict]:
         # sparql query to get a node by IRI with all its properties
         # using CONSTRUCT to get the full node
         # format the result as json-ld
-        iri_filter = f"FILTER (?s = {iri})"
-        # check if the iri is a full IRI or a prefix
-        if iri.startswith("http"):
-            iri_filter = f"FILTER (?s = <{iri}>)"
-        self._sparql.setQuery(
-            """
-            PREFIX ex: <https://example.com/>
-            PREFIX Item: <http://www.wikidata.org/entity/>
-            CONSTRUCT {
-                ?s ?p ?o .
-            }
-            WHERE {
-                ?s ?p ?o .
-                {{{iri_filter}}}
-            }
-            """.replace(
-                "{{{iri_filter}}}", iri_filter
+        jsonld_dicts = {}
+        for iri in iris:
+            iri_filter = f"FILTER (?s = {iri})"
+            # check if the iri is a full IRI or a prefix
+            if iri.startswith("http"):
+                iri_filter = f"FILTER (?s = <{iri}>)"
+            self._sparql.setQuery(
+                """
+                PREFIX ex: <https://example.com/>
+                PREFIX Item: <http://www.wikidata.org/entity/>
+                CONSTRUCT {
+                    ?s ?p ?o .
+                }
+                WHERE {
+                    ?s ?p ?o .
+                    {{{iri_filter}}}
+                }
+                """.replace(
+                    "{{{iri_filter}}}", iri_filter
+                )
             )
-        )
-        self._sparql.setReturnFormat(JSONLD)
-        result: Graph = self._sparql.query().convert()
-        jsonld_dict = json.loads(result.serialize(format="json-ld"))[0]
-        # replace http://www.wikidata.org/prop/direct/P31 with @type
-        if "http://www.wikidata.org/prop/direct/P31" in jsonld_dict:
-            jsonld_dict["@type"] = jsonld_dict.pop(
-                "http://www.wikidata.org/prop/direct/P31"
-            )[0]["@id"]
-
-        return jsonld_dict
+            self._sparql.setReturnFormat(JSONLD)
+            result: Graph = self._sparql.query().convert()
+            jsonld_dict = json.loads(result.serialize(format="json-ld"))[0]
+            # replace http://www.wikidata.org/prop/direct/P31 with @type
+            if "http://www.wikidata.org/prop/direct/P31" in jsonld_dict:
+                jsonld_dict["@type"] = jsonld_dict.pop(
+                    "http://www.wikidata.org/prop/direct/P31"
+                )[0]["@id"]
+            jsonld_dicts[iri] = jsonld_dict
+
+        return jsonld_dicts

src/oold/model/__init__.py

@@ -5,7 +5,14 @@
 from pydantic import BaseModel
 from typing_extensions import Self
 
-from oold.backend.interface import GetResolverParam, ResolveParam, get_resolver
+from oold.backend.interface import (
+    GetBackendParam,
+    GetResolverParam,
+    ResolveParam,
+    StoreParam,
+    get_backend,
+    get_resolver,
+)
 from oold.static import GenericLinkedBaseModel, export_jsonld, import_jsonld
 
 # pydantic v2
@@ -325,6 +332,14 @@ def _resolve(iris):
         ).nodes
         return node_dict
 
+    def _store(self):
+        backend = get_backend(GetBackendParam(iri=self.get_iri())).backend
+        backend.store(StoreParam(nodes={self.get_iri(): self}))
+
+    def store_jsonld(self):
+        """Store the model instance in a backend matching its IRI."""
+        self._store()
+
     # pydantic v2
     def model_dump_json(
         self,