in src/graph_notebook/network/sparql/SPARQLNetwork.py [0:0]
def add_results(self, results):
"""
takes a json result from a sparql query and attempts to add all bindings
with the variables "subject" ,"predicate", "object" or "s", "p", "o"
:param results:
"""
# validate that we can process this result..
vars = []
if 'head' in results and 'vars' in results['head']:
vars = results['head']['vars']
if len(vars) < 3: # if we have less than three vars in the result, then we cannot have the three we require.
return
found_subject = False
found_predicate = False
found_object = False
subject_binding = "subject"
predicate_binding = "predicate"
object_binding = "object"
for v in vars:
if v == 'subject' and not found_subject:
subject_binding = 'subject'
found_subject = True
continue
if v == 'predicate' and not found_predicate:
predicate_binding = "predicate"
found_predicate = True
continue
if v == 'object' and not found_object:
object_binding = "object"
found_object = True
continue
if v == 's' and not found_subject:
subject_binding = 's'
found_subject = True
continue
if v == 'p' and not found_predicate:
predicate_binding = 'p'
found_predicate = True
continue
if v == 'o' and not found_object:
object_binding = 'o'
found_object = True
continue
if subject_binding == 's' and predicate_binding == 'p' and object_binding == 'o':
use_spo = True
elif subject_binding == 'subject' and predicate_binding == 'predicate' and object_binding == 'object':
use_spo = False
else:
raise InvalidBindingsCombinationError
if not (found_subject and found_predicate and found_object):
return
bindings = []
if 'results' in results and 'bindings' in results['results']:
bindings = results['results']['bindings']
# sort the bindings so we can process all triples that make up a given node (subject)
# this will reduce the number of message callbacks generated as we process larger
# result sets.
bindings.sort(key=lambda x: x[subject_binding]['value'])
if len(bindings) < 1:
return
current_subject_binding = bindings[0][subject_binding]
data = {'properties': {}}
edge_bindings = []
for b in bindings:
# just because the result vars show the needed variables doesn't mean that bindings will have them.
if subject_binding not in b or predicate_binding not in b or object_binding not in b:
if subject_binding in b:
self.parse_node(node_binding=b[subject_binding], node_id=b[subject_binding]['value'])
if object_binding in b:
self.parse_node(node_binding=b[object_binding], node_id=b[object_binding]['value'])
continue
sub = b[subject_binding]
pred = b[predicate_binding]
obj = b[object_binding]
if sub['value'] != current_subject_binding['value']:
self.parse_node(node_binding=current_subject_binding,
node_id=current_subject_binding['value'],
data=data)
data = {'properties': {}}
current_subject_binding = sub
# if obj is of type uri, and the predicate value is neither rdfs:label nor rdf:type this binding is an edge.
if (obj['type'] in NODE_TYPES or self.expand_all) and pred['value'] not in [RDFS_LABEL, RDF_TYPE]:
edge_bindings.append(b)
continue
if pred['type'] == 'uri':
prefix = self.extract_prefix(pred['value'])
value = self.extract_value(pred['value'])
obj_entry = obj['value']
if obj['type'] == 'uri':
obj_prefix = self.extract_prefix(obj['value'])
obj_value = self.extract_value(obj['value'])
obj_entry = f'{obj_prefix}:{obj_value}'
# default label/tooltip set here. May be overwritten by custom properties.
if pred['value'] == RDFS_LABEL:
title = obj_entry
label = title if len(title) <= self.label_max_length else title[:self.label_max_length - 3] + '...'
data['title'] = title
data['label'] = label
# object is a literal. Check if data has this predicate already. If it does, turn its value into an
# array and append the new value to it.
if 'properties' in data and f'{prefix}:{value}' in data['properties']:
if type(data['properties'][f'{prefix}:{value}']) is list:
data['properties'][f'{prefix}:{value}'].append(obj_entry)
else:
data['properties'][f'{prefix}:{value}'] = [data['properties'][f'{prefix}:{value}'], obj_entry]
else:
data['properties'][f'{prefix}:{value}'] = obj_entry
else:
# Check if data has this predicate already. If it does, turn its value into an
# array and append the new value to it.
if 'properties' in data and pred['value'] in data['properties']:
if type(data['properties'][pred['value']]) is list:
data['properties'][pred['value']].append(obj['value'])
else:
data['properties'][pred['value']] = [data['properties'][pred['value']], obj['value']]
else:
data['properties'][pred['value']] = obj['value']
# add the last node and all our edges
self.parse_node(node_binding=current_subject_binding,
node_id=current_subject_binding['value'],
data=data)
self.process_edge_bindings(edge_bindings, use_spo)
return