Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
#!/usr/bin/env python
# cache.py: grlc spec caching utilities
import json
import urllib.request, urllib.error, urllib.parse
import grlc.glogging as glogging
glogger = glogging.getGrlcLogger(__name__)
# Name of the cache json file
CACHE_NAME = "db-cache.json"
def init_cache():
'''
Initializes the grlc cache (json file)
'''
cache_obj = json.loads("{}")
try:
with open(CACHE_NAME, 'r') as cache_file:
try:
cache_obj = json.load(cache_file)
except ValueError:
print("The cache file seems to be empty, starting with flushed cache")
except IOError:
from pythonql.parser.Preprocessor import makeProgramFromString
from six import PY3
import grlc.glogging as glogging
glogger = glogging.getGrlcLogger(__name__)
def project(dataIn, projectionScript):
'''Programs may make use of data in the `dataIn` variable and should
produce data on the `dataOut` variable.'''
# We don't really need to initialize it, but we do it to avoid linter errors
dataOut = {}
try:
projectionScript = str(projectionScript)
program = makeProgramFromString(projectionScript)
if PY3:
loc = {
'dataIn': dataIn,
'dataOut': dataOut
}
exec(program, {}, loc)
dataOut = loc['dataOut']
#!/usr/bin/env python
# prov.py: class generating grlc related W3C prov triples
from rdflib import Graph, URIRef, Namespace, RDF, Literal
from datetime import datetime
from subprocess import check_output
from six import PY3
# grlc modules
import grlc.static as static
import grlc.glogging as glogging
glogger = glogging.getGrlcLogger(__name__)
class grlcPROV():
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))
import yaml
import json
from rdflib.plugins.sparql.parser import Query, UpdateUnit
from rdflib.plugins.sparql.processor import translateQuery
from flask import request, has_request_context
from pyparsing import ParseException
from pprint import pformat
import traceback
import re
import requests
# grlc modules
import grlc.static as static
import grlc.glogging as glogging
glogger = glogging.getGrlcLogger(__name__)
XSD_PREFIX = 'PREFIX xsd: '
import SPARQLTransformer
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
from SPARQLWrapper import SPARQLWrapper, CSV, JSON
from flask import jsonify
from collections import defaultdict
import static as static
import grlc.glogging as glogging
glogger = glogging.getGrlcLogger(__name__)
# Default value is JSON
SUPPORTED_MIME_FORMATS = defaultdict(
lambda: JSON, {
'text/csv': CSV,
'application/json': JSON
}
)
MIME_FORMAT = {
format: mime for mime, format in SUPPORTED_MIME_FORMATS.items()
}
def getResponseText(endpoint, query, requestedMimeType):
'''
endpoint - URL of sparql endpoint