Skip to content

Instantly share code, notes, and snippets.

@roelj
Created January 29, 2024 13:51
Show Gist options
  • Save roelj/a7b2f1751d8fe23fab96404ec69a10c0 to your computer and use it in GitHub Desktop.
Save roelj/a7b2f1751d8fe23fab96404ec69a10c0 to your computer and use it in GitHub Desktop.
RDFLib BerkeleyDB deletion problem.
from datetime import datetime
from rdflib import Dataset
from rdflib.plugins.stores import sparqlstore
import shutil
db_directory = "/tmp/bdb-test"
dataset = Dataset("BerkeleyDB")
dataset.open (db_directory, create=True)
if not isinstance (dataset, Dataset):
print ("Failed to setup BerkeleyDB store.")
print (f"# Triples in the store: {len(dataset)}")
query_0 = """
PREFIX test: <http://test.nonexisting/>
PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>
INSERT DATA {
GRAPH <test://default> {
<dataset:1> rdf:type test:Dataset .
<dataset:1> test:title "Hello"^^xsd:string .
<dataset:1> test:modified_date "2024-01-29T12:59:54Z"^^xsd:dateTime .
}
}
"""
dataset.update (query_0)
print (f"Triples in the store: {len(dataset)}")
for row in dataset:
print (f"Row: {row}")
def generate_query ():
current_time = datetime.strftime (datetime.now(), "%Y-%m-%dT%H:%M:%SZ")
query = f"""
PREFIX test: <http://test.nonexisting/>
PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>
DELETE {{
GRAPH <test://default> {{
?dataset test:title ?title .
?dataset test:modified_date ?modified_date .
}}
}}
INSERT {{
GRAPH <test://default> {{
?dataset test:title "Test"^^xsd:string .
?dataset test:modified_date "{current_time}"^^xsd:dateTime .
}}
}}
WHERE {{
GRAPH <test://default> {{
?dataset rdf:type test:Dataset .
?dataset test:title ?title .
?dataset test:modified_date ?modified_date .
}}
}}
"""
return query
query_1 = generate_query()
dataset.update (query_1)
print (f"Triples in the store: {len(dataset)}")
query_2 = generate_query()
dataset.update (query_2)
print (f"# Triples in the store: {len(dataset)}")
for row in dataset:
print (f"Row: {row}")
dataset.close()
shutil.rmtree (db_directory)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment