Utilities¶

def load_graph(
    source: Union[GraphInput, list[GraphInput], tuple[GraphInput, ...]],
    *additional_graph_paths_or_str: GraphInput,
    recursive: bool = False,
) -> Graph:
    """
    Presents an RDFLib Graph from one or more existing Graphs, pickle-cached RDF
    files, RDF files or directories, remote RDF URLs, or RDF data strings.

    Multiple inputs may be supplied as positional arguments or as a list or tuple.
    Missing filesystem paths raise ``FileNotFoundError``.
    """
    # Preserve the former ``load_graph(path, recursive)`` positional call form.
    if len(additional_graph_paths_or_str) == 1 and isinstance(
        additional_graph_paths_or_str[0], bool
    ):
        recursive = additional_graph_paths_or_str[0]
        additional_graph_paths_or_str = ()

    if isinstance(source, (list, tuple)):
        graph_inputs = (*source, *additional_graph_paths_or_str)
    else:
        graph_inputs = (source, *additional_graph_paths_or_str)

    if not graph_inputs:
        return Graph()

    if len(graph_inputs) > 1:
        graph = Graph()
        for graph_input in graph_inputs:
            graph += load_graph(graph_input, recursive=recursive)
        return graph

    source = graph_inputs[0]

    # Pre-existing Graph
    if isinstance(source, Graph):
        return source

    # Serialized RDF file or dir of files, optionally using a sibling pickle cache
    if isinstance(source, Path):
        if source.is_file():
            pkl_path = source.with_suffix(".pkl")
            if pkl_path.is_file():
                with pkl_path.open("rb") as pickle_file:
                    return pickle.load(pickle_file)
            if source.suffix.lower() == ".trig":
                return _parse_dataset(source)
            return _parse_graph(source)
        elif source.is_dir():
            g = Graph()
            if recursive:
                gl = source.rglob("*.ttl")
            else:
                gl = source.glob("*.ttl")
            for f in gl:
                if f.is_file():
                    g.parse(f)
            return g
        raise FileNotFoundError(f"Graph path does not exist: {source}")

    # A remote file via HTTP
    elif isinstance(source, str) and source.startswith("http"):
        return _parse_graph(source)

    # RDF data in a string
    else:
        return _parse_graph(
            data=source,
            format=guess_format_from_data(source),
        )

def render_sparql_result(
    r: dict | str | Graph, rf: RenderFormat = RenderFormat.markdown
) -> str:
    """Renders a SPARQL result in a given render format"""
    if rf == RenderFormat.original:
        return r

    elif rf == RenderFormat.json:
        if isinstance(r, dict):
            return json.dumps(r, indent=4)
        elif isinstance(r, str):
            return json.dumps(json.loads(r), indent=4)
        elif isinstance(r, Graph):
            return r.serialize(format="json-ld", indent=4)

    elif rf == RenderFormat.markdown:
        if isinstance(r, Graph):  # CONSTRUCT: RDF GRaph
            output = "```turtle\n" + r.serialize(format="longturtle") + "```\n"
        else:  # SELECT or ASK: Python dict or JSON

            def render_sparql_value(v: dict) -> str:
                # TODO: handle v["datatype"]
                if v is None:
                    return ""
                elif isinstance(v, URIRef) or isinstance(v, str):
                    return f"[{v.split('/')[-1].split('#')[-1]}]({v})"
                elif isinstance(v, Literal):
                    return v
                elif isinstance(v, BNode):
                    return f"BN: {v:>6}"
                elif v["type"] == "uri":
                    return f"[{v['value'].split('/')[-1].split('#')[-1]}]({v['value']})"
                elif v["type"] == "literal":
                    return v["value"]
                elif v["type"] == "bnode":
                    return f"BN: {v['value']:>6}"

            if isinstance(r, str):
                r = json.loads(r)

            output = ""
            header = ["", ""]
            body = []

            if r.get("head") is not None:
                # SELECT
                if r["head"].get("vars") is not None:
                    for col in r["head"]["vars"]:
                        header[0] += f"{col} | "
                        header[1] += f"--- | "
                    output = (
                        "| " + header[0].strip() + "\n| " + header[1].strip() + "\n"
                    )

            if r.get("results"):
                if r["results"].get("bindings"):
                    for row in r["results"]["bindings"]:
                        row_cols = []
                        for k in r["head"]["vars"]:
                            v = row.get(k)
                            if v is not None:
                                # ignore the k
                                row_cols.append(render_sparql_value(v))
                            else:
                                row_cols.append("")
                        body.append(" | ".join(row_cols))

                output += "\n| ".join(body) + " |\n"

            if r.get("boolean") is not None:
                output = str(bool(r.get("boolean")))

        return output

def get_system_graph(
    system_graph_source: str | Path | Dataset | Graph = None,
    http_client: httpx.Client | None = None,
):
    """Returns a System Graph, graph and can accept many source options"""
    system_graph = Graph(identifier=SYSTEM_GRAPH_IRI)
    system_graph.bind("olis", OLIS)
    if system_graph_source is None:
        # no incoming System Graph
        pass
    elif isinstance(system_graph_source, Path):
        # we have a Graph or Dataset file, so read it
        if not system_graph_source.is_file():
            raise ValueError(
                f"system_graph_source must be an existing RDF file. Value supplied was {system_graph_source}"
            )

        if system_graph_source.suffix == ".trig":
            system_graph += _parse_dataset(
                system_graph_source, format="trig"
            ).graph(SYSTEM_GRAPH_IRI)
        else:
            system_graph += load_graph(system_graph_source)
    elif isinstance(system_graph_source, Graph):
        # we have a Graph, so assume it's a System Graph and load it
        system_graph += system_graph_source
    elif isinstance(system_graph_source, Dataset):
        # we have a Dataset object, so load its system Graph
        system_graph += system_graph_source.graph(SYSTEM_GRAPH_IRI)
    elif system_graph_source and system_graph_source.startswith("http"):
        # we have a remote SPARQL Endpoint, so read the System Graph
        # this is simplified GSP get()
        close_http_client = False
        if http_client is None:
            http_client = httpx.Client()
            close_http_client = True

        r = http_client.get(
            str(system_graph_source),
            params={"graph": SYSTEM_GRAPH_IRI},
            headers={"Accept": "text/turtle"},
        )

        if close_http_client:
            http_client.close()

        if r.is_success:
            system_graph += Graph().parse(data=r.text, format="turtle")
        else:
            return r.status_code
    elif system_graph_source and not system_graph_source.startswith("http"):
        system_graph += load_graph(system_graph_source)
    else:
        raise ValueError(
            "The parameter system_graph_source must be either None, a Path to an RDF Graph or Dataset serialised "
            "in Turtle or Trig, an RDFLib Graph object assumed to be a System Graph, an RDFLib Dataset object containing"
            "a System Graph or a string URL for a SPARQL Endpoint."
        )

    return system_graph

def make_system_specific_sparql_endpoint(
    sparql_endpoint: str,
    q: str = None,
    statement: SparqlStatementType = None,
    gsp_query_type: GspType = None,
) -> str:
    """Alters a given SPARQL Endpoint to meet specific system requirements.

    e.g. GraphDB using /statements at the end of the base SPARQL Endpoint for updates"""

    # GraphDB SPARQL
    if q is not None and statement is not None:
        # GraphDB: Update
        if (
            "/repositories/" in sparql_endpoint
            and is_update_query(q, statement)
            and not sparql_endpoint.endswith("/statements")
        ):
            return sparql_endpoint + "/statements"

    # GraphDB GSP
    if gsp_query_type is not None:
        if "/repositories/" in sparql_endpoint:
            return sparql_endpoint + "/rdf-graphs/service"

    return sparql_endpoint