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_guess_endpoint(self):
with self.app.test_request_context('/?endpoint=http://url-endpoint/from-url/sparql'):
endpoint, _ = gquery.guess_endpoint_uri('', self.loader)
self.assertIn('from-url', endpoint,
'Should match endpoint given in url')
with self.app.test_request_context('/'):
endpoint, _ = gquery.guess_endpoint_uri('', self.loader)
self.assertIn('from-file', endpoint,
'Should match endpoint in endpoint.txt')
rq, _ = self.loader.getTextForName('test-rq')
endpoint, _ = gquery.guess_endpoint_uri(rq, self.loader)
self.assertIn('from-decorator', endpoint,
'Should match endpoint in test-rq.rq')
def dispatchTPFQuery(raw_tpf_query, loader, acceptHeader, content):
endpoint, auth = gquery.guess_endpoint_uri(raw_tpf_query, loader)
glogger.debug("=====================================================")
glogger.debug("Sending query to TPF endpoint: {}".format(endpoint))
glogger.debug("=====================================================")
# TODO: pagination for TPF
# Preapre HTTP request
reqHeaders = {'Accept': acceptHeader, 'Authorization': 'token {}'.format(static.ACCESS_TOKEN)}
if content:
reqHeaders = {'Accept': static.mimetypes[content], 'Authorization': 'token {}'.format(static.ACCESS_TOKEN)}
tpf_list = re.split('\n|=', raw_tpf_query)
subject = tpf_list[tpf_list.index('subject') + 1]
predicate = tpf_list[tpf_list.index('predicate') + 1]
object = tpf_list[tpf_list.index('object') + 1]
data = {'subject': subject, 'predicate': predicate, 'object': object}
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':
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 ""