Implement garbage collection (#141)

Author: jgonggrijp

backend/catalogs/triplestore.py

@@ -1,5 +1,6 @@
 """Functions that deal with adding and updating catalog records in the
 triplestore."""
+from typing import Optional
 from itertools import chain
 import datetime as dt
 from django.conf import settings
@@ -17,6 +18,11 @@
 RECORDS_GC_GRAPH_IDENTIFIER = URIRef(RECORDS_GC_GRAPH_URI)
 SCHEMA = Namespace('https://schema.org/')
 
+# When we retrieve records from a catalog, they might be duplicates of records
+# that were retrieved before. The following query gets rid of the duplicates. It
+# is meant to be executed just before the newly retrieved records are added. The
+# net effect should be that all `obsolete_records` remain in the triplestore;
+# the query is not meant for permanently deleting records.
 purge_old_update = '''
 delete {{
   graph <{records_graph}> {{
@@ -40,6 +46,40 @@
 }}
 '''.format
 
+# The following update query identifies and removes records that are not in any
+# collection and that have not been retrieved since the given `cutoff_date`.
+garbage_collect_update = '''
+delete {{
+  graph <{records_graph}> {{
+    ?r ?p1 ?o1 .
+    ?f ?p2 ?o2 .
+  }}
+  graph <{gc_graph}> {{
+    ?r schema:uploadDate ?d ;
+       schema:upvoteCount 0 .
+  }}
+}}
+where {{
+  graph <{gc_graph}> {{
+    {{
+      ?r schema:uploadDate ?d ;
+         schema:upvoteCount 0 .
+    }}
+    union
+    {{
+      ?r schema:uploadDate ?d .
+      filter not exists {{ ?r schema:upvoteCount ?c }}
+    }}
+    filter ( ?d < {cutoff_date} )
+  }}
+  graph <{records_graph}> {{
+    ?r ?p1 ?o1 ;
+       ?pt ?f .
+    optional {{?f ?p2 ?o2 .}}
+  }}
+}}
+'''.format
+
 
 def prune_recursively(graph: Graph, subject: Node):
     """Recursively prune triples """
@@ -81,3 +121,18 @@ def save_to_triplestore(content_graph: Graph, records: list[Node]) -> None:
     quads_gc = ((rec, SCHEMA.uploadDate, now, gc_graph) for rec in records)
     store.addN(chain(quads, quads_gc))
     store.commit()
+
+
+def collect_garbage(until: Optional[dt.date]=None) -> None:
+    """Forget all unused records that were added before `until`.
+
+    `until` defaults to two weeks ago."""
+    if until is None:
+        until = dt.date.today() - dt.timedelta(weeks=2)
+    store = settings.RDFLIB_STORE
+    store.update(garbage_collect_update(
+        records_graph=RECORDS_GRAPH_URI,
+        gc_graph=RECORDS_GC_GRAPH_URI,
+        cutoff_date=Literal(until).n3(),
+    ), initNs={'schema': SCHEMA})
+    store.commit()

backend/catalogs/triplestore_test.py

@@ -1,9 +1,12 @@
 import pytest
+import datetime as dt
+
 from edpop_explorer import Record, EDPOPREC, BibliographicalRecord, Field
-from rdflib import Graph, RDF, URIRef
+from rdflib import Graph, RDF, URIRef, Literal
 
 from .graphs_test import MockReader
-from .triplestore import save_to_triplestore, remove_from_triplestore, SCHEMA
+from .triplestore import collect_garbage, save_to_triplestore, \
+    remove_from_triplestore, SCHEMA, RECORDS_GC_GRAPH_IDENTIFIER
 from operator import attrgetter
 
 
@@ -29,6 +32,14 @@ def record_nodes(record_instances):
     return map(attrgetter('subject_node'), record_instances)
 
 
+@pytest.fixture
+def working_data_saved(working_data_graph, triplestore):
+    records, graph = working_data_graph
+    nodes = list(record_nodes(records))
+    save_to_triplestore(graph, nodes)
+    return nodes, records, graph
+
+
 def stored_records(triplestore):
     """ Retrieve the records currently stored in `triplestore`. """
     return list(triplestore.subjects(RDF.type, EDPOPREC.Record))
@@ -85,3 +96,50 @@ def test_remove_nonexistent_subject(working_data_graph):
     # Removing subjects that don't exist should not cause any problems
     records, _ = working_data_graph
     remove_from_triplestore(records)
+
+
+def test_gc_retain_recent(working_data_saved, triplestore):
+    cutoff = dt.date.today() - dt.timedelta(weeks=1)
+    collect_garbage(cutoff)
+    assert len(stored_records(triplestore)) == 2
+    assert stored_records_match_tracked_records(triplestore)
+
+
+def test_gc_remove_outdated(working_data_saved, triplestore):
+    cutoff = dt.date.today() + dt.timedelta(weeks=1)
+    collect_garbage(cutoff)
+    assert len(stored_records(triplestore)) == 0
+    assert stored_records_match_tracked_records(triplestore)
+
+
+def test_gc_retain_used(working_data_saved, triplestore):
+    cutoff = dt.date.today() + dt.timedelta(weeks=1)
+    nodes, _, _ = working_data_saved
+    chosen = nodes[0]
+    triplestore.addN([(
+        chosen,
+        SCHEMA.upvoteCount,
+        Literal(1),
+        Graph(identifier=RECORDS_GC_GRAPH_IDENTIFIER),
+    )])
+    collect_garbage(cutoff)
+    remaining_subjects = stored_records(triplestore)
+    assert len(remaining_subjects) == 1
+    assert remaining_subjects[0] == chosen
+    assert stored_records_match_tracked_records(triplestore)
+
+
+def test_gc_remove_obsolete(working_data_saved, triplestore):
+    cutoff = dt.date.today() + dt.timedelta(weeks=1)
+    nodes, _, _ = working_data_saved
+    chosen = nodes[0]
+    triplestore.addN([(
+        chosen,
+        SCHEMA.upvoteCount,
+        Literal(0),
+        Graph(identifier=RECORDS_GC_GRAPH_IDENTIFIER),
+    )])
+    collect_garbage(cutoff)
+    remaining_subjects = stored_records(triplestore)
+    assert len(remaining_subjects) == 0
+    assert stored_records_match_tracked_records(triplestore)