Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def guess_endpoint_uri(rq, gh_repo):
"""
Guesses the endpoint URI from (in this order):
- An endpoint parameter in URL
- An #+endpoint decorator
- A endpoint.txt file in the repo
Otherwise assigns a default one
"""
auth = (static.DEFAULT_ENDPOINT_USER, static.DEFAULT_ENDPOINT_PASSWORD)
if auth == ('none', 'none'):
auth = None
if has_request_context() and "endpoint" in request.args:
endpoint = request.args['endpoint']
glogger.info("Endpoint provided in request: " + endpoint)
return endpoint, auth
# Decorator
try:
decorators = get_yaml_decorators(rq)
endpoint = decorators['endpoint']
auth = None
glogger.info("Decorator guessed endpoint: " + endpoint)
except (TypeError, KeyError):
# File
def __init__(self, user, repo):
"""
Default constructor
"""
self.user = user
self.repo = repo
self.prov_g = Graph()
prov_uri = URIRef("http://www.w3.org/ns/prov#")
self.prov = Namespace(prov_uri)
self.prov_g.bind('prov', self.prov)
self.agent = URIRef("http://{}".format(static.SERVER_NAME))
self.entity_d = URIRef("http://{}/api/{}/{}/spec".format(static.SERVER_NAME, self.user, self.repo))
self.activity = URIRef(self.entity_d + "-activity")
self.init_prov_graph()
def getRawRepoUri(self):
raw_repo_uri = static.GITHUB_RAW_BASE_URL + self.user + '/' + self.repo
if self.sha is None:
raw_repo_uri += '/master/'
else:
raw_repo_uri += '/{}/'.format(self.sha)
return raw_repo_uri
if 'text/turtle' in request.headers['Accept']:
resp_spec = make_response(utils.turtleize(swag))
resp_spec.headers['Content-Type'] = 'text/turtle'
else:
resp_spec = make_response(jsonify(swag))
resp_spec.headers['Content-Type'] = 'application/json'
resp_spec.headers['Cache-Control'] = static.CACHE_CONTROL_POLICY # Caching JSON specs for 15 minutes
glogger.info("-----> API spec generation for /{}/{} on commit {} complete".format(user, repo, sha))
return resp_spec
if __name__ == '__main__':
app.run(host=static.DEFAULT_HOST, port=static.DEFAULT_PORT, debug=static.LOG_DEBUG_MODE)
vlang = None
vdatatype = None
vformat = None
mtype = match.group('type')
muserdefined = match.group('userdefined')
if mtype in ['number', 'literal', 'string']:
vtype = mtype
elif mtype in ['iri']: # TODO: proper form validation of input parameter uris
vtype = 'string'
vformat = 'iri'
elif mtype:
vtype = 'string'
if mtype in static.XSD_DATATYPES:
vdatatype = 'xsd:{}'.format(mtype)
elif len(mtype) == 2:
vlang = mtype
elif muserdefined:
vdatatype = '{}:{}'.format(mtype, muserdefined)
parameters[vname] = {
'original': '?{}'.format(v),
'required': vrequired,
'name': vname,
'type': vtype
}
# Possibly None parameter attributes
if vcodes is not None:
parameters[vname]['enum'] = sorted(vcodes)
# 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")
del g
# Check for INSERT/POST
elif query_metadata['type'] == 'InsertData':
glogger.debug("Processing INSERT query")
# Rewrite INSERT
rewritten_query = rewritten_query.replace("?_g_iri", "{}".format(formData.get('g')))
rewritten_query = rewritten_query.replace("<s> <p> ", formData.get('data'))</p></s>