Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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}
response = requests.get(endpoint, params=data, headers=reqHeaders, auth=auth)
glogger.debug('Response header from endpoint: ' + response.headers['Content-Type'])
# Response headers
resp = response.text
headers = {}
headers['Content-Type'] = response.headers['Content-Type']
headers['Server'] = 'grlc/' + grlc_version
return resp, 200, headers
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}
response = requests.get(endpoint, params=data, headers=reqHeaders, auth=auth)
glogger.debug('Response header from endpoint: ' + response.headers['Content-Type'])
# Response headers
resp = response.text
headers = {}
headers['Content-Type'] = response.headers['Content-Type']
headers['Server'] = 'grlc/' + grlc_version
def _getText(self, query_name):
query_uri = self.getRawRepoUri() + query_name
req = requests.get(query_uri, headers={'Authorization': 'token {}'.format(static.ACCESS_TOKEN)})
if req.status_code == 200:
return req.text
else:
return None
def __init__(self, user, repo, sha, prov):
self.user = user
self.repo = repo
self.sha = sha
self.prov = prov
gh = Github(static.ACCESS_TOKEN)
try:
self.gh_repo = gh.get_repo(user + '/' + repo, lazy=False)
except BadCredentialsException:
raise Exception('BadCredentials: have you set up github_access_token on config.ini ?')
except Exception:
raise Exception('Repo not found: ' + user + '/' + repo)
def fetchFiles(self):
api_repo_content_uri = static.GITHUB_API_BASE_URL + self.user + '/' + self.repo + '/contents'
params = {
'ref': 'master' if self.sha is None else self.sha
}
# TODO: use Github instead of requests ?
resp = requests.get(api_repo_content_uri, headers={'Authorization': 'token {}'.format(static.ACCESS_TOKEN)},
params=params)
if resp.ok:
return resp.json()
else:
raise Exception(resp.text)
glogger.debug("Detected graph names: {}".format(gnames))
glogger.debug("Detected BGP: {}".format(vtpattern))
glogger.debug("Matched triple pattern with parameter")
if gnames:
codes_subquery = re.sub("SELECT.*\{.*\}.*",
"SELECT DISTINCT ?" + v + " FROM " + gnames + " WHERE { " + vtpattern + " }", rq,
flags=re.DOTALL)
else:
codes_subquery = re.sub("SELECT.*\{.*\}.*",
"SELECT DISTINCT ?" + v + " WHERE { " + vtpattern + " }", rq,
flags=re.DOTALL)
glogger.debug("Codes subquery: {}".format(codes_subquery))
glogger.debug(endpoint)
codes_json = requests.get(endpoint, params={'query': codes_subquery},
headers={'Accept': static.mimetypes['json'],
'Authorization': 'token {}'.format(static.ACCESS_TOKEN)}, auth=auth).json()
for code in codes_json['results']['bindings']:
vcodes.append(list(code.values())[0]["value"])
else:
glogger.debug("No match between variable name and query.")
return vcodes