Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Testing Referential Opacity in RDF* triple stores
#!/usr/bin/env python3
"""
This script checks different triple stores implementing RDF*,
to see how opaque/transparent are the terms used in embedded triples.
"""
from sys import argv, stderr
import base64
try:
import requests
except:
print("""
You need to install the 'requests' python module; e.g. with
apt-get install python-requests
or
pip install requests""",
file=stderr)
exit(-1)
ENDPOINTS = [
('GraphDB', 'http://localhost:7200/repositories/superman'),
('Stardog', 'http://localhost:5820/superman'),
('Corese', 'http://localhost:8080/sparql'),
('Fuseki', 'http://localhost:3030/superman'),
];
STARDOG_CREDENTIALS = "admin:admin"
def main():
for endpoint in ENDPOINTS:
if argv[1:] and endpoint[0] not in argv[1:]:
continue
print(endpoint[0]);
sparql(endpoint, update="DELETE {?s ?p ?o} WHERE {?s ?p ?o}\n");
sparql(endpoint, update="""INSERT DATA {
<< :superman :can :fly >> :reportedBy :clark.
<< _:x rdf:type :FlyingMan >> :reportedBy :clark.
:superman owl:sameAs :clark.
:superman :birthPlace :krypton.
<< :lifeUniverseAndEverything :anwser "042"^^xsd:integer >> :accordingTo :deepThought.
}""")
print("\tloaded", file=stderr)
print("\tsupports owl:sameAs:\t", sparql(endpoint, """
ASK { :clark :birthPlace :krypton }
"""))
print("\tRDF* mode:\t", "PG" if sparql(endpoint, """
ASK { :superman :can :fly }
""") else "SA")
print("\tref. transp. IRI:\t", sparql(endpoint, """
ASK { << :clark :can :fly >> :reportedBy :clark }
"""))
print("\tref. transp. lit.:\t", sparql(endpoint, """
ASK { << :lifeUniverseAndEverything :anwser "42"^^xsd:integer >> :accordingTo :deepThought. }
"""))
print("\tref. transp. bnode:\t", sparql(endpoint, """
ASK { << _:x :can :fly >> :reportedBy :clark }
"""))
print("\tref. transp. bnode2:\t", sparql(endpoint, """
ASK { << _:y rdf:type :FlyingMan >> :reportedBy :clark }
"""))
#print("\tquery emb. triple:\t", len(sparql(endpoint, """
# SELECT ?o { <<?s rdf:type ?o>> :reportedBy :clark }
#"""))==1)
S = requests.session()
HEADERS = {
'accept': 'application/sparql-results+json',
'connection': 'keep-alive',
}
PREFIXES = """
PREFIX : <http://example.org/ns/>
PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
PREFIX owl: <http://www.w3.org/2002/07/owl#>
PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>
"""
def sparql(endpoint, query=None, update=None) -> bool:
if query is None and update is None:
raise TypeError('query() missing argument query or update')
if update is not None and update is not True:
if query is None and isinstance(update, str):
query = update
update = True
else:
raise TypeError('incompatible mix of arguments query and update')
name, url = endpoint
config = CONFIG.get(name, {})
url += config.get('update_url' if update else 'query_url', '')
headers = HEADERS.copy()
headers.update(config.get('headers', {}))
qparam = config.get('update_param', 'query') if update else 'query'
data = { qparam: PREFIXES + query }
data.update(config.get('params', {}))
res = S.post(url, data, headers=headers)
if res.status_code // 100 != 2:
print(f"{res.status_code} {res.reason}: {res.text}", file=stderr)
if update:
res.raise_for_status()
return "(ERROR)"
if not res.content or update:
return
result = res.json()
if "boolean" in result:
return result['boolean']
else:
return result['results']['bindings']
STARDOG_CREDENTIALS = base64.encodebytes(STARDOG_CREDENTIALS.encode('ascii')).decode('ascii').strip()
CONFIG = {
'Stardog': {
'query_url': '/query?reasoning=true&flushResponseHeaders=true',
'update_url': '/update?reasoning=true&flushResponseHeaders=true',
'headers': {
'authorization': 'Basic ' + STARDOG_CREDENTIALS,
}
},
'GraphDB': {
'update_url': '/statements',
'update_param': 'update',
'params': {
'infer': 'true',
'sameAs': 'true',
},
},
'Fuseki': {
'query_url': '/query',
'update_url': '/update',
'update_param': 'update',
}
}
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment