Olis Graph Functions¶

def include(
    including_graph_iri: URIRef | str,
    graphs_to_include_iris: list[URIRef | str],
    system_graph_source: str | Path | Dataset | Graph = None,
    include_background_graph: bool = True,
    http_client: httpx.Client | None = None,
) -> Graph | int | None:
    """Creates a Virtual Graph with IRI of `including_graph_iri` that contains all the graphs given in `graphs_to_include_iris`.

    Args:
        including_graph_iri: The IRI of the graph that will include the other graphs.
        graphs_to_include_iris: The IRIs of each graph to be included in the including graph.
        system_graph_source: The SPARQL Endpoint, file (RDFLib Graph or Dataset), RDFLib Graph, Dataset object, or Path to a Graph or Dataset object to both read and write System Graph info to. If None, a new System Graph object will be returned.
        include_background_graph: Whether to include http://background in the subsumption
        http_client: An optional HTTPX client to contain credentials if needed to access the SPARQL endpoint. A new one is created if not given.

    Returns:
        The System Graph itself if `system_graph_source` is None or a local (non-HTTP) string, the HTTP status code if a remote write failed, or None otherwise.
    """
    # value check inputs
    if including_graph_iri is None:
        raise ValueError(
            "You must specify an including_graph_iri as a string or a URIRef"
        )

    if graphs_to_include_iris is None:
        raise ValueError(
            "You must specify at least one string or URIRef in the list variable graphs_to_include_iris"
        )

    if isinstance(including_graph_iri, str):
        including_graph_iri = URIRef(including_graph_iri)

    if isinstance(graphs_to_include_iris[0], str):
        graphs_to_include_iris = [URIRef(x) for x in graphs_to_include_iris]

    if including_graph_iri in graphs_to_include_iris:
        raise ValueError(
            "You cannot include the including_graph_iri in the list of graphs to include"
        )

    if including_graph_iri == BACKGROUND_GRAPH_IRI:
        raise ValueError("You cannot use the background graph to include other graphs")

    if (
        including_graph_iri == SYSTEM_GRAPH_IRI
        or SYSTEM_GRAPH_IRI in graphs_to_include_iris
    ):
        raise ValueError(
            "You cannot use the system graph to include other graphs or include it in other graphs"
        )

    # get/make the system graph
    """
    To extend http://graph-a to include graph-a content and http://graph-b content:

    include_graph("http://graph-a", ["http://graph-b"])

    This will:

    * learn that http://graph-a is a Virtual Graph - since it includes things
    * move any existing http://graph-a content, if it's a Real Graph, to a new Real Graph - http://graph-a-real - by convention
    * state that http://graph-a includes http://graph-a-real and all graphs in the graphs_to_include_iris list
    """
    system_graph = get_system_graph(system_graph_source, http_client=http_client)

    # alter the system graph as per the inputs
    """
    if including_graph_iri in system_graph:
        if it is an RG:
            pop out RG content and include
    else:
        create it

    include all graphs_to_include_iris
    """
    if (including_graph_iri, RDF.type, OLIS.RealGraph) in system_graph:
        # swap out all triples targeting the RG to the new RG
        new_including_graph_real_iri = URIRef(str(including_graph_iri) + "-real")

        for p, o in system_graph.subject_objects(including_graph_iri):
            system_graph.remove((including_graph_iri, p, o))
            system_graph.add((new_including_graph_real_iri, p, o))

        for s, p in system_graph.subject_predicates(including_graph_iri):
            system_graph.remove((s, including_graph_iri, p))
            system_graph.add((s, new_including_graph_real_iri, p))

        # create the new VG
        system_graph.add((including_graph_iri, RDF.type, OLIS.VirtualGraph))

        # include the new RG in the VG
        system_graph.add(
            (including_graph_iri, OLIS.includes, new_including_graph_real_iri)
        )
        system_graph.add((new_including_graph_real_iri, RDF.type, OLIS.RealGraph))
        system_graph.add(
            (
                new_including_graph_real_iri,
                SDO.description,
                Literal(f"Graph containing triples from <{str(including_graph_iri)}>"),
            )
        )
    elif (including_graph_iri, RDF.type, OLIS.VirtualGraph) in system_graph:
        pass  # we will add to the VG below
    else:  # the including_graph_iri is not present in the system_graph so create it
        system_graph.add((including_graph_iri, RDF.type, OLIS.VirtualGraph))
        system_graph.add(
            (
                including_graph_iri,
                SDO.dateCreated,
                Literal(datetime.now().isoformat()[:19], datatype=XSD.dateTime),
            )
        )

    # in all cases now...
    for graph_to_include_iri in graphs_to_include_iris:
        system_graph.add((including_graph_iri, OLIS.includes, graph_to_include_iri))

    system_graph.remove((including_graph_iri, SDO.dateModified, None))
    system_graph.add(
        (
            including_graph_iri,
            SDO.dateModified,
            Literal(datetime.now().isoformat()[:19], datatype=XSD.dateTime),
        )
    )

    if include_background_graph:
        system_graph.add((including_graph_iri, OLIS.includes, BACKGROUND_GRAPH_IRI))

    # return or write altered system graph, depending on system graph source
    return put_system_graph(system_graph, system_graph_source, http_client=http_client)

def exclude(
    including_graph_iri: URIRef | str,
    graphs_to_exclude_iris: list[URIRef | str],
    system_graph_source: str | Path | Dataset | Graph = None,
    http_client: httpx.Client | None = None,
) -> Graph | int | None | ValueError:
    """Remove the `includes` statement from a Virtual Graph to one or more Named Graphs.

    See `include` for parameter details.

    Returns:
        The System Graph itself if `system_graph_source` is None or a local (non-HTTP) string, the HTTP status code if a remote write failed, or None with no errors otherwise. A `ValueError` instance if `including_graph_iri` is not a known VirtualGraph.
    """
    # value check inputs
    if including_graph_iri is None:
        raise ValueError(
            "You must specify an including_graph_iri as a string or a URIRef"
        )

    if graphs_to_exclude_iris is None:
        raise ValueError(
            "You must specify at least one string or URIRef in the list variable graphs_to_include_iris"
        )

    if isinstance(including_graph_iri, str):
        including_graph_iri = URIRef(including_graph_iri)

    if isinstance(graphs_to_exclude_iris[0], str):
        graphs_to_exclude_iris = [URIRef(x) for x in graphs_to_exclude_iris]

    if including_graph_iri in graphs_to_exclude_iris:
        raise ValueError(
            "You cannot include the including_graph_iri in the list of graphs to exclude"
        )

    if including_graph_iri == BACKGROUND_GRAPH_IRI:
        raise ValueError("You cannot exclude graphs from the background graph")

    if (
        including_graph_iri == SYSTEM_GRAPH_IRI
        or SYSTEM_GRAPH_IRI in graphs_to_exclude_iris
    ):
        raise ValueError(
            "You cannot exclude graph from the system graph or include it in other graphs"
        )

    # get/make the system graph
    system_graph = get_system_graph(system_graph_source, http_client=http_client)

    # alter the system graph as per the inputs
    if not (including_graph_iri, RDF.type, OLIS.VirtualGraph) in system_graph:
        return ValueError(
            "The including_graph_iri is not known as a VirtualGraph in the system_graph_source"
        )

    system_graph.remove((including_graph_iri, SDO.dateModified, None))
    system_graph.add(
        (
            including_graph_iri,
            SDO.dateModified,
            Literal(datetime.now().isoformat()[:19], datatype=XSD.dateTime),
        )
    )

    for graph_to_exclude_iri in graphs_to_exclude_iris:
        system_graph.remove((including_graph_iri, OLIS.includes, graph_to_exclude_iri))

    # return or write altered system graph, depending on system graph source
    return put_system_graph(system_graph, system_graph_source, http_client=http_client)

def validate_system_graph(
    system_graph_source: str | Path | Dataset | Graph, http_client: httpx.Client = None
) -> tuple[bool, Graph, str, Graph]:
    """Validate a System Graph using the Olis System Graph SHACL validator.

    Args:
        system_graph_source: The System Graph to validate - a SPARQL Endpoint, RDF file, Path, Graph, or Dataset.
        http_client: An optional HTTPX client to contain credentials if needed. A new one is created if not given.

    Returns:
        Tuple of the validation status, results graph, message, and summary graph.
    """
    system_graph = get_system_graph(system_graph_source, http_client=http_client)
    v = validate(system_graph, "https://olis.dev/sg-validator")
    return v