graph-exploring-tool/graph_exploring_tool/query.py

from dataclasses import dataclass, field
from typing import List
from urllib.error import URLError

import arklog
from SPARQLWrapper import SPARQLWrapper, JSON, POSTDIRECTLY
from SPARQLWrapper.SPARQLExceptions import QueryBadFormed

arklog.set_config_logging()

@dataclass(init=True, repr=True, order=False, frozen=True)
class Prefix:
    name: str
    namespace: str

    def __str__(self) -> str:
        """Return the prefix in the SPARQL format."""
        return f"PREFIX {self.name}: <{self.namespace}>"


@dataclass(init=True, repr=True, order=False, frozen=True)
class Replacement:
    placeholder: str
    description: str
    suggestion: str

@dataclass(init=True, repr=True, order=False, frozen=True)
class QueryTemplate:
    group: str
    name: str
    prefix: str
    query: str
    description: str
    visual_support: bool = False
    replacements: List[Replacement] = field(default_factory=list)
    modifies: bool = False

# TODO Rewrite this
reverse_prefix = {
    "https://ontology.rys.app/dt/function/": "dtf",
    "http://www.w3.org/2002/07/owl#": "owl",
    "http://www.w3.org/1999/02/22-rdf-syntax-ns#": "rdf",
    "http://www.w3.org/2001/XMLSchema#": "xsd",
    "http://www.w3.org/2000/01/rdf-schema#": "rdfs",
    "http://ua.be/drivetrain/description/artifacts/artifacts#": "art",
    "http://ua.be/sdo2l/vocabulary/base/tabular#": "tabular",
    "http://purl.org/dc/elements/1.1/": "dc",
    "http://ua.be/sdo2l/vocabulary/ftg#": "ftg",
    "http://ua.be/sdo2l/vocabulary/base#": "base",
    "http://bipm.org/jcgm/vim4#": "vim4",
    "http://ua.be/sdo2l/vocabulary/component#": "comp",
    "http://ua.be/sdo2l/vocabulary/base/code#":"code",
    "http://ua.be/sdo2l/vocabulary/base/script#": "script",
    "http://ua.be/sdo2l/vocabulary/base/file#": "file",
    "http://www.w3.org/2003/11/swrl#": "swrl",
    "http://ua.be/sdo2l/vocabulary/processtraces#":"traces",
    "http://ua.be/sdo2l/vocabulary/workflow#": "wf",
    "http://ua.be/sdo2l/vocabulary/base/text#": "text",
    "http://ua.be/sdo2l/vocabulary/federation#": "federation",
    "http://ua.be/drivetrain/description/ftg#": "dftg",
}

def perform_query(endpoint: str, query_text: str, post: bool = False) -> dict:
    """"""
    sparql = SPARQLWrapper(endpoint)
    sparql.setReturnFormat(JSON)
    if post:
        sparql.setRequestMethod(POSTDIRECTLY)
    sparql.setQuery(query_text)
    try:
        ret = sparql.query().convert()
    except QueryBadFormed:
        raise
    except URLError:
        raise
    return ret