Source code for brickschema.persistent

from datetime import datetime
import logging
from collections import OrderedDict
import uuid
import time
from contextlib import contextmanager
from rdflib import ConjunctiveGraph
from rdflib.graph import BatchAddGraph
from rdflib import plugin
from import Store
from rdflib_sqlalchemy import registerplugins
import pickle
from .graph import Graph, BrickBase

logger = logging.getLogger(__name__)

changeset_table_defn = """CREATE TABLE IF NOT EXISTS changesets (
    id TEXT,
    timestamp TIMESTAMP NOT NULL,
    graph TEXT NOT NULL,
    is_insertion BOOLEAN NOT NULL,
    triple BLOB NOT NULL
changeset_table_idx = """CREATE INDEX IF NOT EXISTS changesets_idx
                        ON changesets (graph, timestamp);"""
redo_table_defn = """CREATE TABLE IF NOT EXISTS redos (
    id TEXT,
    timestamp TIMESTAMP NOT NULL,
    graph TEXT NOT NULL,
    is_insertion BOOLEAN NOT NULL,
    triple BLOB NOT NULL

_remove_params = ["_delay_init", "brick_version", "load_brick", "load_brick_nightly"]

[docs]class PersistentGraph(Graph): def __init__(self, uri: str, *args, **kwargs): store = plugin.get("SQLAlchemy", Store)( identifier="brickschema_persistent_graph" ) kwargs.update({"_delay_init": True}) super().__init__(store, *args, **kwargs) kwargs.update({"create": True}) for k in _remove_params: kwargs.pop(k, None) super().open(uri, **kwargs) self.uri = uri super()._graph_init()
[docs]class Changeset(Graph): def __init__(self, graph_name): super().__init__() = graph_name self.uid = uuid.uuid4() self.additions = [] self.deletions = []
[docs] def add(self, triple): """ Add a triple to the changeset """ self.additions.append(triple) super().add(triple)
[docs] def load_file(self, filename): g = Graph() g.parse(filename, format="turtle") self.additions.extend(g.triples((None, None, None))) self += g # propagate namespaces for pfx, ns in g.namespace_manager.namespaces(): self.bind(pfx, ns)
[docs] def remove(self, triple): self.deletions.append(triple) super().remove(triple)
[docs]class VersionedGraphCollection(ConjunctiveGraph, BrickBase): def __init__(self, uri: str, *args, **kwargs): """ To create an in-memory store, use uri="sqlite://" """ store = plugin.get("SQLAlchemy", Store)(identifier="my_store") super().__init__(store, *args, **kwargs), create=True) self._precommit_hooks = OrderedDict() self._postcommit_hooks = OrderedDict() with self.conn() as conn: conn.execute("PRAGMA journal_mode=WAL;") # conn.execute("PRAGMA synchronous=OFF;") conn.execute(changeset_table_defn) conn.execute(changeset_table_idx) conn.execute(redo_table_defn) @property def latest_version(self): with self.conn() as conn: rows = conn.execute( "SELECT id, timestamp from changesets " "ORDER BY timestamp DESC LIMIT 1" ) res = rows.fetchone() return res def __len__(self): # need to override __len__ because the rdflib-sqlalchemy # backend doesn't support .count() for recent versions of # SQLAlchemy return len(list(self.triples((None, None, None))))
[docs] def undo(self): """ Undoes the given changeset. If no changeset is given, undoes the most recent changeset. """ if self.latest_version is None: raise Exception("No changesets to undo") with self.conn() as conn: changeset_id = self.latest_version["id"]"Undoing changeset {changeset_id}") self._graph_at(self, conn, self.latest_version["timestamp"]) conn.execute( "INSERT INTO redos(id, timestamp, graph, is_insertion, triple) SELECT id, timestamp, graph, is_insertion, triple FROM changesets WHERE id = ?", (changeset_id,), ) conn.execute("DELETE FROM changesets WHERE id = ?", (changeset_id,))
[docs] def redo(self): """ Redoes the most recent changeset. """ with self.conn() as conn: redo_record = conn.execute( "SELECT * from redos " "ORDER BY timestamp ASC LIMIT 1" ).fetchone() if redo_record is None: raise Exception("No changesets to redo") changeset_id = redo_record["id"]"Redoing changeset {changeset_id}") conn.execute( "INSERT INTO changesets SELECT * FROM redos WHERE id = ?", (changeset_id,), ) conn.execute("DELETE FROM redos WHERE id = ?", (changeset_id,)) self._graph_at(self, conn, redo_record["timestamp"]) for row in conn.execute( "SELECT * from changesets WHERE id = ?", (changeset_id,) ): triple = pickle.loads(row["triple"]) if row["is_insertion"]: self.remove((triple[0], triple[1], triple[2])) else: self.add((triple[0], triple[1], triple[2]))
[docs] def versions(self, graph=None): """ Return a list of all versions of the provided graph; defaults to the union of all graphs """ with self.conn() as conn: if graph is None: rows = conn.execute( "SELECT DISTINCT id, graph, timestamp from changesets " "ORDER BY timestamp DESC" ) else: rows = conn.execute( "SELECT DISTINCT id, graph, timestamp from changesets " "WHERE graph = ? ORDER BY timestamp DESC", (graph,), ) return list(rows)
[docs] def add_precommit_hook(self, hook): self._precommit_hooks[hook.__name__] = hook
[docs] def add_postcommit_hook(self, hook): self._postcommit_hooks[hook.__name__] = hook
[docs] @contextmanager def conn(self): yield
[docs] @contextmanager def new_changeset(self, graph_name, ts=None): namespaces = [] with self.conn() as conn: transaction_start = time.time() cs = Changeset(graph_name) yield cs if ts is None: ts = # delta by the user. We need to invert the changes so that they are expressed as a "backward" # delta. This means that we save the deletions in the changeset as "inserts", and the additions # as "deletions". if cs.deletions: conn.exec_driver_sql( "INSERT INTO changesets VALUES (?, ?, ?, ?, ?)", [ (str(cs.uid), ts, graph_name, True, pickle.dumps(triple)) for triple in cs.deletions ], ) graph = self.get_context(graph_name) for triple in cs.deletions: graph.remove(triple) if cs.additions: conn.exec_driver_sql( "INSERT INTO changesets VALUES (?, ?, ?, ?, ?)", [ (str(cs.uid), ts, graph_name, False, pickle.dumps(triple)) for triple in cs.additions ], ) with BatchAddGraph( self.get_context(graph_name), batch_size=10000 ) as graph: for triple in cs.additions: graph.add(triple) # take care of precommit hooks transaction_end = time.time() for hook in self._precommit_hooks.values(): hook(self) # keep track of namespaces so we can add them to the graph # after the commit namespaces.extend(cs.namespace_manager.namespaces()) # # finally, remove all of the 'redos' # conn.execute("DELETE FROM redos") # # and remove all of the 'changesets' that come after us f"Committing after {transaction_end - transaction_start} seconds" ) # update namespaces for pfx, ns in namespaces: self.bind(pfx, ns) for hook in self._postcommit_hooks.values(): hook(self) self._latest_version = ts
[docs] def latest(self, graph): return self.get_context(graph)
[docs] def graph_at(self, timestamp=None, graph=None): # setup graph and bind namespaces g = Graph() for pfx, ns in self.namespace_manager.namespaces(): g.bind(pfx, ns) # if graph is specified, only copy triples from that graph. # otherwise, copy triples from all graphs. if graph is not None: for t in self.get_context(graph).triples((None, None, None)): g.add(t) else: for t in self.triples((None, None, None)): g.add(t) with self.conn() as conn: return self._graph_at(g, conn, timestamp, graph)
def _graph_at(self, alter_graph, conn, timestamp=None, graph=None): """ Return *copy* of the graph at the given timestamp. Chooses the most recent timestamp that is less than or equal to the given timestamp. """ if timestamp is None: timestamp = if graph is not None: rows = conn.execute( "SELECT * FROM changesets WHERE graph = ? AND timestamp >= ? ORDER BY timestamp DESC", (graph, timestamp), ) else: rows = conn.execute( "SELECT * FROM changesets WHERE timestamp >= ? ORDER BY timestamp DESC", (timestamp,), ) for row in rows: triple = pickle.loads(row["triple"]) if row["is_insertion"]: alter_graph.add((triple[0], triple[1], triple[2])) else: alter_graph.remove((triple[0], triple[1], triple[2])) return alter_graph