Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_get_metadata(self):
rq, _ = self.loader.getTextForName('test-sparql')
metadata = gquery.get_metadata(rq, '')
self.assertIn('type', metadata, 'Should have a type field')
self.assertIn('variables', metadata, 'Should have a variables field')
self.assertEqual(metadata['type'], 'SelectQuery',
'Should be type SelectQuery')
self.assertIsInstance(
metadata['variables'], list, 'Should be a list of variables')
for var in metadata['variables']:
self.assertIsInstance(var, rdflib.term.Variable,
'Should be of type Variable')
def process_sparql_query_text(query_text, loader, call_name, extraMetadata):
# We get the endpoint name first, since some query metadata fields (eg enums) require it
endpoint, auth = gquery.guess_endpoint_uri(query_text, loader)
glogger.debug("Read query endpoint: {}".format(endpoint))
try:
query_metadata = gquery.get_metadata(query_text, endpoint)
except Exception:
raw_repo_uri = loader.getRawRepoUri()
raw_query_uri = raw_repo_uri + ' / ' + call_name
glogger.error("Could not parse query at {}".format(raw_query_uri))
glogger.error(traceback.print_exc())
return None
tags = query_metadata['tags'] if 'tags' in query_metadata else []
summary = query_metadata['summary'] if 'summary' in query_metadata else ""
description = query_metadata['description'] if 'description' in query_metadata else ""
method = query_metadata['method'].lower() if 'method' in query_metadata else ""
if method not in ['get', 'post', 'head', 'put', 'delete', 'options', 'connect']:
method = ""
rewritten_query = gquery.rewrite_query(query_metadata['original_query'], query_metadata['parameters'], requestArgs)
# Rewrite query using pagination
if query_metadata['type'] == 'SelectQuery' and 'pagination' in query_metadata:
rewritten_query = gquery.paginate_query(rewritten_query, query_metadata['pagination'], requestArgs)
resp = None
headers = {}
# If we have a mime field, we load the remote dump and query it locally
if 'mime' in query_metadata and query_metadata['mime']:
glogger.debug(
"Detected {} MIME type, proceeding with locally loading remote dump".format(query_metadata['mime']))
g = Graph()
try:
query_metadata = gquery.get_metadata(raw_sparql_query, endpoint)
g.parse(endpoint, format=query_metadata['mime'])
glogger.debug("Local RDF graph loaded successfully with {} triples".format(len(g)))
except Exception as e:
glogger.error(e)
results = g.query(rewritten_query, result='sparql')
# Prepare return format as requested
resp_string = ""
if 'application/json' in acceptHeader or (content and 'application/json' in static.mimetypes[content]):
resp_string = results.serialize(format='json')
glogger.debug("Results of SPARQL query against locally loaded dump: {}".format(resp_string))
elif 'text/csv' in acceptHeader or (content and 'text/csv' in static.mimetypes[content]):
resp_string = results.serialize(format='csv')
glogger.debug("Results of SPARQL query against locally loaded dump: {}".format(resp_string))
else:
return 'Unacceptable requested format', 415, {}
glogger.debug("Finished processing query against RDF dump, end of use case")
def dispatchSPARQLQuery(raw_sparql_query, loader, requestArgs, acceptHeader, content, formData, requestUrl):
endpoint, auth = gquery.guess_endpoint_uri(raw_sparql_query, loader)
if endpoint == '':
return 'No SPARQL endpoint indicated', 407, {}
glogger.debug("=====================================================")
glogger.debug("Sending query to SPARQL endpoint: {}".format(endpoint))
glogger.debug("=====================================================")
query_metadata = gquery.get_metadata(raw_sparql_query, endpoint)
acceptHeader = 'application/json' if isinstance(raw_sparql_query, dict) else acceptHeader
pagination = query_metadata['pagination'] if 'pagination' in query_metadata else ""
rewritten_query = query_metadata['query']
# Rewrite query using parameter values
if query_metadata['type'] == 'SelectQuery' or query_metadata['type'] == 'ConstructQuery':
rewritten_query = gquery.rewrite_query(query_metadata['original_query'], query_metadata['parameters'], requestArgs)
# Rewrite query using pagination
if query_metadata['type'] == 'SelectQuery' and 'pagination' in query_metadata:
rewritten_query = gquery.paginate_query(rewritten_query, query_metadata['pagination'], requestArgs)
resp = None
headers = {}