graph-exploring-tool/graph_exploring_tool/query.py

import json.decoder
import logging
import ssl
from dataclasses import dataclass, field
from typing import List
from urllib.error import URLError
import owly
import arklog

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 (use dict from config + merge per query)
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://purl.org/dc/elements/1.1/": "dc",
    "http://bipm.org/jcgm/vim4#": "vim4",
    "http://www.w3.org/2003/11/swrl#": "swrl",

    "http://ua.be/drivetrain/description/artifacts/artifacts#": "dart",
    "http://ua.be/drivetrain/description/ftg#": "dftg",

    "http://ua.be/sdo2l/vocabulary/base/acyclic#": "acyclic",
    "http://ua.be/sdo2l/vocabulary/base/base#": "base",
    "http://ua.be/sdo2l/vocabulary/formalisms/file#": "file",
    "http://ua.be/sdo2l/vocabulary/formalisms/ftg#": "ftg",
    "http://ua.be/sdo2l/vocabulary/formalisms/processtraces#": "traces",
    "http://ua.be/sdo2l/vocabulary/formalisms/shaclproperty#": "shaclproperty",
    "http://ua.be/sdo2l/vocabulary/formalisms/system#": "system",
    "http://ua.be/sdo2l/vocabulary/formalisms/violation#": "violation",
    "http://ua.be/sdo2l/vocabulary/formalisms/code#": "code",
    "http://ua.be/sdo2l/vocabulary/formalisms/component#": "component",
    "http://ua.be/sdo2l/vocabulary/formalisms/cs_as#": "csas",
    "http://ua.be/sdo2l/vocabulary/formalisms/dict#": "dict",
    "http://ua.be/sdo2l/vocabulary/formalisms/drawio#": "drawio",
    "http://ua.be/sdo2l/vocabulary/formalisms/object_diagram#": "object_diagram",
    "http://ua.be/sdo2l/vocabulary/formalisms/pm#": "pm",
    "http://ua.be/sdo2l/vocabulary/formalisms/requirement#": "requirement",
    "http://ua.be/sdo2l/vocabulary/formalisms/richtext#": "richtext",
    "http://ua.be/sdo2l/vocabulary/formalisms/script#": "script",
    "http://ua.be/sdo2l/vocabulary/formalisms/tabular#": "tabular",
    "http://ua.be/sdo2l/vocabulary/formalisms/text#": "text",
    "http://ua.be/sdo2l/vocabulary/formalisms/traceability_model#": "traceability_model",
    "http://ua.be/sdo2l/vocabulary/formalisms/xopp#": "xopp",
    "http://ua.be/sdo2l/vocabulary/federation#": "federation",

    "http://ua.be/sdo2l/description/artifacts/FTG_ftg#": "artifacts_ftg",
}


def perform_query(endpoint_uri: str, query_text: str, post: bool = False) -> dict:
    """"""
    endpoint = owly.Endpoint(endpoint_uri)
    try:
        result = endpoint.perform_query(query_text)
    except URLError:
        raise
    except json.decoder.JSONDecodeError:
        raise
    except ssl.SSLCertVerificationError:
        raise
    return result.as_json()