util

General utilities module.

copy_triples(src, dst=None)

Copies all triples from one graph onto another (or a new, empty Graph if none is provided).

Parameters:

Name	Type	Description	Default
src	Graph	the source Graph	required
dst	Optional[Graph]	the destination Graph (or None to create a new one)	None

Returns:

Type	Description
Graph	the destination Graph

Source code in ogc/na/util.py

def copy_triples(src: Graph, dst: Optional[Graph] = None) -> Graph:
    """
    Copies all triples from one graph onto another (or a new, empty [Graph][rdflib.Graph]
    if none is provided).

    :param src: the source Graph
    :param dst: the destination Graph (or `None` to create a new one)
    :return: the destination Graph
    """
    if dst is None:
        dst = Graph()
    for triple in src:
        dst.add(triple)
    return dst

dump_yaml(content, filename=None, ignore_alises=True, **kwargs)

Generates YAML output.

Parameters:

Name	Type	Description	Default
content	Any	content to convert to YAML.	required
filename	str | Path | None	optional filename to dump the content into. If None, string content will be returned.	None
kwargs		other args to pass to yaml.dump()	{}

Source code in ogc/na/util.py

def dump_yaml(content: Any, filename: str | Path | None = None,
              ignore_alises=True,
              **kwargs) -> str | None:
    """
    Generates YAML output.

    :param content: content to convert to YAML.
    :param filename: optional filename to dump the content into. If None, string content will be returned.
    :param kwargs: other args to pass to `yaml.dump()`
    """
    kwargs.setdefault('sort_keys', False)
    if ignore_alises:
        class Dumper(YamlDumper):
            def ignore_aliases(self, data) -> bool:
                return True
    else:
        Dumper = YamlDumper
    if filename:
        with open(filename, 'w') as f:
            return yaml.dump(content, f, Dumper=Dumper, **kwargs)
    else:
        return yaml.dump(content, Dumper=Dumper, **kwargs)

entail(g, rules, extra=None, inplace=True)

Performs SHACL entailments on a data Graph.

Parameters:

Name	Type	Description	Default
g	Graph	input data Graph	required
rules	Graph	SHACL Graph for entailments	required
extra	Optional[Graph]	Graph with additional ontological information for entailment	None
inplace	bool	if True, the source Graph will be modified, otherwise a new Graph will be created	True

Returns:

Type	Description
Graph	the resulting Graph

Source code in ogc/na/util.py

def entail(g: Graph,
           rules: Graph,
           extra: Optional[Graph] = None,
           inplace: bool = True) -> Graph:
    """
    Performs SHACL entailments on a data [Graph][rdflib.Graph].

    :param g: input data Graph
    :param rules: SHACL Graph for entailments
    :param extra: Graph with additional ontological information for entailment
    :param inplace: if `True`, the source Graph will be modified, otherwise a new
           Graph will be created
    :return: the resulting Graph
    """
    entailed_extra = None
    if extra:
        entailed_extra = copy_triples(extra)
        shacl_validate(entailed_extra, shacl_graph=rules, ont_graph=None, advanced=True, inplace=True)

    if not inplace:
        g = copy_triples(g)
    shacl_validate(g, shacl_graph=rules, ont_graph=extra, advanced=True, inplace=True)

    if entailed_extra:
        for triple in entailed_extra:
            g.remove(triple)

    return g

is_url(url, http_only=False)

Checks whether a string is a valid URL.

Parameters:

Name	Type	Description	Default
url	str	the input string	required
http_only	bool	whether to only accept HTTP and HTTPS URLs as valid	False

Returns:

Type	Description
bool	True if this is a valid URL, otherwise False

Source code in ogc/na/util.py

def is_url(url: str, http_only: bool = False) -> bool:
    """
    Checks whether a string is a valid URL.

    :param url: the input string
    :param http_only: whether to only accept HTTP and HTTPS URLs as valid
    :return: `True` if this is a valid URL, otherwise `False`
    """
    if not url:
        return False

    parsed = urlparse(url)
    if not parsed.scheme or not (parsed.netloc or parsed.path):
        return False

    if http_only and parsed.scheme not in ('http', 'https'):
        return False

    return True

load_yaml(filename=None, content=None, url=None, safe=True)

Loads a YAML file either from a file, a string or a URL.

Parameters:

Name	Type	Description	Default
filename	str | Path | None	YAML document file name	None
content	Any | None	str with YAML contents	None
url	str | None	url from which to retrieve the contents	None
safe	bool	whether to use safe YAMl loading	True

Returns:

Type	Description
dict	a dict with the loaded data

Source code in ogc/na/util.py

def load_yaml(filename: str | Path | None = None,
              content: Any | None = None,
              url: str | None = None,
              safe: bool = True) -> dict:
    """
    Loads a YAML file either from a file, a string or a URL.

    :param filename: YAML document file name
    :param content: str with YAML contents
    :param url: url from which to retrieve the contents
    :param safe: whether to use safe YAMl loading
    :return: a dict with the loaded data
    """

    if bool(filename) + bool(content) + bool(url) > 1:
        raise ValueError("One (and only one) of filename, contents and url must be provided")

    if filename:
        with open(filename, 'r') as f:
            return yaml.load(f, Loader=SafeYamlLoader if safe else YamlLoader)
    else:
        if url:
            content = requests.get(url).text
        return yaml.load(content, Loader=SafeYamlLoader if safe else YamlLoader)

merge_contexts(a, b, fix_nest=True)

Merges two JSON-lD contexts, updating the first one passed to this function (and returning it).

Source code in ogc/na/util.py

def merge_contexts(a: dict, b: dict, fix_nest=True) -> dict[str, Any]:
    '''
    Merges two JSON-lD contexts, updating the first one passed to this function (and returning it).
    '''
    if not b:
        return a
    if not a:
        if isinstance(a, dict):
            a.update(b)
            return a
        return b
    for term in list(a.keys()):
        va = a[term]
        vb = b.get(term)
        if term not in JSON_LD_KEYWORDS:
            if isinstance(va, str):
                va = {'@id': va}
                a[term] = va
            if isinstance(vb, str):
                vb = {'@id': vb}
            if vb:
                for vb_term, vb_term_val in vb.items():
                    if vb_term != '@context':
                        va[vb_term] = vb_term_val
                if '@context' in vb:
                    if '@context' not in va:
                        va['@context'] = vb['@context']
                    elif isinstance(va['@context'], list):
                        if isinstance(vb['@context'], list):
                            va['@context'].extend(vb['@context'])
                        else:
                            va['@context'].append(vb['@context'])
                    elif isinstance(vb['@context'], list):
                        va['@context'] = [va['@context'], *vb['@context']]
                    else:
                        va['@context'] = merge_contexts(va['@context'], vb['@context'])
        elif vb:
            a[term] = vb
    for t, tb in b.items():
        if t not in a:
            a[t] = tb

    if fix_nest:
        # fix nested @context inside @nest terms
        # spec is unclear, but our tooling of interest (json-ld playground, rdflib) do not support it
        # see: https://github.com/json-ld/json-ld.org/issues/737
        pending_merges = []
        for term in list(a.keys()):
            if isinstance(a[term], dict) and a[term].get('@id') == '@nest':
                nested_ctx = a[term].pop('@context', None)
                if nested_ctx:
                    pending_merges.append(nested_ctx)
                a[term] = '@nest'
        for pm in pending_merges:
            a = merge_contexts(a, pm)

    return a

parse_resources(src)

Join one or more RDF documents or Graph's together into a new Graph.

Parameters:

Name	Type	Description	Default
src	Union[str, Graph, list[Union[str, Graph]]]	a path or Graph, or list thereof	required

Returns:

Type	Description
Graph	a union Graph

Source code in ogc/na/util.py

def parse_resources(src: Union[str, Graph, list[Union[str, Graph]]]) -> Graph:
    """
    Join one or more RDF documents or [Graph][rdflib.Graph]'s together into
    a new Graph.
    :param src: a path or [Graph][rdflib.Graph], or list thereof
    :return: a union Graph
    """
    if not isinstance(src, list):
        src = [src]

    result = Graph()
    for s in src:
        if not isinstance(s, Graph):
            s = Graph().parse(s)
        copy_triples(s, result)

    return result

validate(g, shacl_graph, extra=None, **kwargs)

Perform SHACL validation on a data Graph.

Parameters:

Name	Type	Description	Default
g	Graph	input data Graph	required
shacl_graph	Graph	SHACL graph for validation	required
extra	Optional[Graph]	Graph with additional ontological information for validation	None

Returns:

Type	Description
ValidationReport	the resulting

Source code in ogc/na/util.py

def validate(g: Graph, shacl_graph: Graph, extra: Optional[Graph] = None,
             **kwargs) -> ValidationReport:
    """
    Perform SHACL validation on a data [Graph][rdflib.Graph].

    :param g: input data Graph
    :param shacl_graph: SHACL graph for validation
    :param extra: Graph with additional ontological information for validation
    :return: the resulting [][ogc.na.validation.ValidationReport]
    """
    return ValidationReport(shacl_validate(data_graph=g,
                                           shacl_graph=shacl_graph,
                                           ont_graph=extra,
                                           inference='rdfs',
                                           advanced=True,
                                           **kwargs))