website/dokk/sparql.py

import json
import requests
from dokk import settings
from pyld import jsonld

def escape_literal(literal=''):
    """
    Escape a string literal to insert in a SPARQL query
    https://www.w3.org/TR/turtle/#sec-escapes
    """
    
    return literal.translate(str.maketrans({
        '\t':   r'\t',
        '\b':   r'\b',
        '\n':   r'\n',
        '\r':   r'\r',
        '\f':   r'\f',
        '"':    r'\"',
        "'":    r'\'',
        '\\':   r'\\',
        '\x00': r'\x00',  # NULL
        '\x1a': r'\x1a'   # Ctrl+Z
    }))

def escape_term(term=''):
    """
    Escape the local name of a term to insert in a SPARQL query.
    It's the part after the prefix, eg. "schema:name".
    https://www.w3.org/TR/turtle/#sec-escapes
    """
    
    return term.translate(str.maketrans({
        '~': r'\~',
        '.': r'\.',
        '-': r'\-',
        '!': r'\!',
        '$': r'\$',
        '&': r'\&',
        "'": r'\'',
        '(': r'\(',
        ')': r'\)',
        '*': r'\*',
        '+': r'\+',
        ',': r'\,',
        ';': r'\;',
        '=': r'\=',
        '/': r'\/',
        '?': r'\?',
        '#': r'\#',
        '@': r'\@',
        '%': r'\%',
        '_': r'\_'
    }))

def query(query_string, endpoint, timeout=60, data={}):
    """
    Query a SPARQL endpoint.
    
    :param query_string: The SPARQL query.
    :param endpoint: The URL to be queries. Must be a SPARQL endpoint.
    :param timeout: How long to wait for a reply before giving up.
    :param data: Additional parameters to be sent along with the request.
    """
    
    response = requests.post(
        endpoint,
        timeout = timeout,
        data    = { 'format': 'json',
                    'query': query_string,
                    **data })
    
    return response.json()

def update(query_string, endpoint, timeout=60):
    """
    Update a SPARQL endpoint.
    """
    
    response = requests.post(
        endpoint,
        timeout = timeout,
        data    = { 'update': query_string })
    
    return response.status_code == 200

def query_instance(query_string):
    return query(query_string, settings['sparql.instance_query'])

def query_graph(query_string):
    return query(query_string, settings['sparql.graph_query'])

def query_public(query_string):
    # ?default-graph-uri=urn:x-arq:UnionGraph is a Fuseki special graph name
    # that instructs the server to consider the default graph as the union
    # of all named graphs. We use this because the public endpoint is composed
    # of the DOKK instance graph plus other imported graphs. See the Fuseki
    # configuration file for the list of named graphs.
    return query(query_string, settings['sparql.public_query'],
                 data={'default-graph-uri': 'urn:x-arq:UnionGraph'})

def update_instance(query_string):
    return update(query_string, settings['sparql.instance_update'])

def update_graph(query_string):
    return update(query_string, settings['sparql.graph_update'])

def frame(json_data, new_frame):
    """
    Apply a new frame to a JSON-LD document.
    https://w3c.github.io/json-ld-syntax/#framed-document-form
    """
    
    return jsonld.frame(json_data, new_frame)

def expand(json_data):
    """
    Expand a JSON-LD document.
    https://w3c.github.io/json-ld-syntax/#expanded-document-form
    """
    
    return jsonld.expand(json_data)