1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
|
"""RDF utilities
This module is a collection of functions that handle SPARQL queries.
"""
import json
from SPARQLWrapper import SPARQLWrapper
from pyld import jsonld # type: ignore
PREFIXES = {
"dcat": "http://www.w3.org/ns/dcat#",
"dct": "http://purl.org/dc/terms/",
"ex": "http://example.org/stuff/1.0/",
"fabio": "http://purl.org/spar/fabio/",
"foaf": "http://xmlns.com/foaf/0.1/",
"generif": "http://www.ncbi.nlm.nih.gov/gene?cmd=Retrieve&dopt=Graphics&list_uids=",
"genotype": "http://genenetwork.org/genotype/",
"gn": "http://genenetwork.org/id/",
"gnc": "http://genenetwork.org/category/",
"gnt": "http://genenetwork.org/term/",
"owl": "http://www.w3.org/2002/07/owl#",
"phenotype": "http://genenetwork.org/phenotype/",
"prism": "http://prismstandard.org/namespaces/basic/2.0/",
"publication": "http://genenetwork.org/publication/",
"pubmed": "http://rdf.ncbi.nlm.nih.gov/pubmed/",
"rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
"rdfs": "http://www.w3.org/2000/01/rdf-schema#",
"skos": "http://www.w3.org/2004/02/skos/core#",
"taxon": "https://www.ncbi.nlm.nih.gov/Taxonomy/Browser/wwwtax.cgi?mode=Info&id=",
"up": "http://purl.uniprot.org/core/",
"xkos": "http://rdf-vocabulary.ddialliance.org/xkos#",
"xsd": "http://www.w3.org/2001/XMLSchema#",
}
RDF_PREFIXES = "\n".join([f"PREFIX {key}: <{value}>"
for key, value in PREFIXES.items()])
def sparql_construct_query(query: str, endpoint: str) -> dict:
"""Query virtuoso using a CONSTRUCT query and return a json-ld
dictionary"""
sparql = SPARQLWrapper(endpoint)
sparql.setQuery(query)
results = sparql.queryAndConvert()
return json.loads(results.serialize(format="json-ld")) # type: ignore
def query_frame_and_compact(query: str, context: dict, endpoint: str) -> dict:
"""Frame and then compact the results given a context"""
results = sparql_construct_query(query, endpoint)
if not results:
return {}
return jsonld.compact(jsonld.frame(results, context), context)
def query_and_compact(query: str, context: dict, endpoint: str) -> dict:
"""Compact the results given a context"""
results = sparql_construct_query(query, endpoint)
if not results:
return {}
return jsonld.compact(results, context)
def query_and_frame(query: str, context: dict, endpoint: str) -> dict:
"""Frame the results given a context"""
results = sparql_construct_query(query, endpoint)
if not results:
return {}
return jsonld.frame(results, context)
|
