in src/graph_notebook/magics/graph_magic.py [0:0]
def sparql(self, line='', cell='', local_ns: dict = None):
parser = argparse.ArgumentParser()
parser.add_argument('query_mode', nargs='?', default='query',
help='query mode (default=query) [query|explain]')
parser.add_argument('--path', '-p', default='',
help='prefix path to sparql endpoint. For example, if "foo/bar" were specified, '
'the endpoint called would be host:port/foo/bar')
parser.add_argument('--expand-all', action='store_true')
parser.add_argument('--explain-type', default='dynamic',
help='explain mode to use when using the explain query mode',
choices=['dynamic', 'static', 'details'])
parser.add_argument('--explain-format', default='text/html', help='response format for explain query mode',
choices=['text/csv', 'text/html'])
parser.add_argument('-g', '--group-by', type=str, default='',
help='Property used to group nodes.')
parser.add_argument('-d', '--display-property', type=str, default='',
help='Property to display the value of on each node.')
parser.add_argument('-de', '--edge-display-property', type=str, default='',
help='Property to display the value of on each edge.')
parser.add_argument('-t', '--tooltip-property', type=str, default='',
help='Property to display the value of on each node tooltip.')
parser.add_argument('-te', '--edge-tooltip-property', type=str, default='',
help='Property to display the value of on each edge tooltip.')
parser.add_argument('-l', '--label-max-length', type=int, default=10,
help='Specifies max length of vertex labels, in characters. Default is 10')
parser.add_argument('-le', '--edge-label-max-length', type=int, default=10,
help='Specifies max length of edge labels, in characters. Default is 10')
parser.add_argument('--store-to', type=str, default='', help='store query result to this variable')
parser.add_argument('--ignore-groups', action='store_true', default=False, help="Ignore all grouping options")
parser.add_argument('-sp', '--stop-physics', action='store_true', default=False,
help="Disable visualization physics after the initial simulation stabilizes.")
parser.add_argument('-sd', '--simulation-duration', type=int, default=1500,
help='Specifies maximum duration of visualization physics simulation. Default is 1500ms')
parser.add_argument('--silent', action='store_true', default=False, help="Display no query output.")
parser.add_argument('--no-scroll', action='store_true', default=False,
help="Display the entire output without a scroll bar.")
args = parser.parse_args(line.split())
mode = str_to_query_mode(args.query_mode)
if args.no_scroll:
sparql_layout = UNRESTRICTED_LAYOUT
else:
sparql_layout = DEFAULT_LAYOUT
if not args.silent:
tab = widgets.Tab()
titles = []
children = []
first_tab_output = widgets.Output(layout=sparql_layout)
children.append(first_tab_output)
path = args.path if args.path != '' else self.graph_notebook_config.sparql.path
logger.debug(f'using mode={mode}')
if mode == QueryMode.EXPLAIN:
res = self.client.sparql_explain(cell, args.explain_type, args.explain_format, path=path)
res.raise_for_status()
explain = res.content.decode('utf-8')
store_to_ns(args.store_to, explain, local_ns)
if not args.silent:
sparql_metadata = build_sparql_metadata_from_query(query_type='explain', res=res)
titles.append('Explain')
first_tab_html = sparql_explain_template.render(table=explain)
else:
query_type = get_query_type(cell)
headers = {} if query_type not in ['SELECT', 'CONSTRUCT', 'DESCRIBE'] else {
'Accept': 'application/sparql-results+json'}
query_res = self.client.sparql(cell, path=path, headers=headers)
query_res.raise_for_status()
results = query_res.json()
store_to_ns(args.store_to, results, local_ns)
try:
res = query_res.json()
except JSONDecodeError:
res = query_res.content.decode('utf-8')
store_to_ns(args.store_to, res, local_ns)
if not args.silent:
# Assign an empty value so we can always display to table output.
# We will only add it as a tab if the type of query allows it.
# Because of this, the table_output will only be displayed on the DOM if the query was of type SELECT.
first_tab_html = ""
query_type = get_query_type(cell)
if query_type in ['SELECT', 'CONSTRUCT', 'DESCRIBE']:
logger.debug('creating sparql network...')
titles.append('Table')
sparql_metadata = build_sparql_metadata_from_query(query_type='query', res=query_res,
results=results, scd_query=True)
sn = SPARQLNetwork(group_by_property=args.group_by,
display_property=args.display_property,
edge_display_property=args.edge_display_property,
tooltip_property=args.tooltip_property,
edge_tooltip_property=args.edge_tooltip_property,
label_max_length=args.label_max_length,
edge_label_max_length=args.edge_label_max_length,
ignore_groups=args.ignore_groups,
expand_all=args.expand_all)
sn.extract_prefix_declarations_from_query(cell)
try:
sn.add_results(results)
except ValueError as value_error:
logger.debug(value_error)
logger.debug(f'number of nodes is {len(sn.graph.nodes)}')
if len(sn.graph.nodes) > 0:
self.graph_notebook_vis_options['physics']['disablePhysicsAfterInitialSimulation'] \
= args.stop_physics
self.graph_notebook_vis_options['physics']['simulationDuration'] = args.simulation_duration
f = Force(network=sn, options=self.graph_notebook_vis_options)
titles.append('Graph')
children.append(f)
logger.debug('added sparql network to tabs')
rows_and_columns = sparql_get_rows_and_columns(results)
if rows_and_columns is not None:
table_id = f"table-{str(uuid.uuid4())[:8]}"
first_tab_html = sparql_table_template.render(columns=rows_and_columns['columns'],
rows=rows_and_columns['rows'], guid=table_id)
# Handling CONSTRUCT and DESCRIBE on their own because we want to maintain the previous result
# pattern of showing a tsv with each line being a result binding in addition to new ones.
if query_type == 'CONSTRUCT' or query_type == 'DESCRIBE':
lines = []
for b in results['results']['bindings']:
lines.append(f'{b["subject"]["value"]}\t{b["predicate"]["value"]}\t{b["object"]["value"]}')
raw_output = widgets.Output(layout=sparql_layout)
with raw_output:
html = sparql_construct_template.render(lines=lines)
display(HTML(html))
children.append(raw_output)
titles.append('Raw')
else:
sparql_metadata = build_sparql_metadata_from_query(query_type='query', res=query_res,
results=results)
json_output = widgets.Output(layout=sparql_layout)
with json_output:
print(json.dumps(results, indent=2))
children.append(json_output)
titles.append('JSON')
if not args.silent:
metadata_output = widgets.Output(layout=sparql_layout)
children.append(metadata_output)
titles.append('Query Metadata')
if first_tab_html == "":
tab.children = children[1:] # the first tab is empty, remove it and proceed
else:
tab.children = children
for i in range(len(titles)):
tab.set_title(i, titles[i])
display(tab)
with metadata_output:
display(HTML(sparql_metadata.to_html()))
if first_tab_html != "":
with first_tab_output:
display(HTML(first_tab_html))