handle lists of lists & expand_curie exception

- add recursive functions to iterate over lists of lists
- add method optionally mute exception for expand_curie
- address #8

src/rdf_utils/__init__.py

@@ -1,3 +1,5 @@
+# SPDX-License-Identifier:  MPL-2.0
 from importlib.metadata import version
 
-RDF_UTILS_VERSION = version("rdf-utils")
+
+__version__ = version("rdf-utils")

src/rdf_utils/collection.py

@@ -0,0 +1,58 @@
+# SPDX-License-Identifier:  MPL-2.0
+from typing import Any
+from rdflib import Graph, BNode, IdentifiedNode, Literal, URIRef
+from rdf_utils.uri import try_expand_curie
+
+
+def _load_list_re(
+    graph: Graph, col_head: BNode, node_set: set[IdentifiedNode], parse_uri: bool, quiet: bool
+) -> list[Any]:
+    """Recursive internal function to extract list of lists from RDF list containers."""
+    col_data = []
+    for col_node in graph.items(list=col_head):
+        if isinstance(col_node, Literal):
+            node_str = col_node.toPython()
+            if not parse_uri:
+                col_data.append(node_str)
+                continue
+
+            # try to expand short-form URIs,
+            # if doesn't work then just return URIRef of the string
+            uri = try_expand_curie(
+                ns_manager=graph.namespace_manager, curie_str=node_str, quiet=quiet
+            )
+            if uri is None:
+                uri = URIRef(node_str)
+
+            col_data.append(uri)
+            continue
+
+        assert isinstance(
+            col_node, BNode
+        ), f"load_collections: node '{col_node}' not a Literal or BNode, type: {type(col_node)}"
+
+        if col_node in node_set:
+            raise RuntimeError(f"Loop detected in collection at node: {col_node}")
+        node_set.add(col_node)
+
+        # recursive call
+        col_data.append(_load_list_re(graph, col_node, node_set, parse_uri, quiet))
+
+    return col_data
+
+
+def load_list_re(
+    graph: Graph, col_head: BNode, parse_uri: bool = True, quiet: bool = True
+) -> list[Any]:
+    """!Recursively iterate over RDF list containers for extracting lists of lists.
+
+    @param graph Graph object to extract the list(s) from
+    @param col_head First element in the list
+    @param parse_uri if True will try converting literals into URIRef
+    @param quiet if True will not throw exceptions other than loop detection
+    @exception RuntimeError Raised when a loop is detected
+    @exception ValueError Raised when `quiet` is `False` and short URI cannot be expanded
+    """
+    node_set = set()
+
+    return _load_list_re(graph, col_head, node_set, parse_uri, quiet)

src/rdf_utils/resolver.py

@@ -9,7 +9,7 @@
 import urllib.response
 from email.message import EmailMessage
 from rdf_utils.uri import URL_SECORO, URL_COMP_ROB2B
-from rdf_utils import RDF_UTILS_VERSION
+from rdf_utils import __version__
 
 
 __PKG_CACHE_ROOT = join(platformdirs.user_cache_dir(), "rdf-utils")
@@ -37,7 +37,7 @@ def __init__(self, url_map: dict, download: bool = True):
     def open(self, fullurl, data=None, timeout=_GLOBAL_DEFAULT_TIMEOUT):
         if isinstance(fullurl, str):
             url_req = urllib.request.Request(fullurl)
-            url_req.add_header("User-Agent", f"rdf-utils/{RDF_UTILS_VERSION}")
+            url_req.add_header("User-Agent", f"rdf-utils/{__version__}")
         elif isinstance(fullurl, urllib.request.Request):
             url_req = fullurl
         else:

src/rdf_utils/uri.py

@@ -1,4 +1,9 @@
 # SPDX-License-Identifier:  MPL-2.0
+from typing import Optional
+from rdflib import URIRef
+from rdflib.namespace import NamespaceManager
+
+
 URL_COMP_ROB2B = "https://comp-rob2b.github.io"
 URL_SECORO = "https://secorolab.github.io"
 URL_SECORO_MM = f"{URL_SECORO}/metamodels"
@@ -19,3 +24,26 @@
 URI_MM_EL = f"{URL_SECORO_MM}/behaviour/event_loop#"
 URL_MM_EL_JSON = f"{URL_SECORO_MM}/behaviour/event_loop.json"
 URL_MM_EL_SHACL = f"{URL_SECORO_MM}/behaviour/event_loop.shacl.ttl"
+
+
+def try_expand_curie(
+    ns_manager: NamespaceManager, curie_str: str, quiet: bool = False
+) -> Optional[URIRef]:
+    """!Execute rdflib `expand_curie` with exception handling
+
+    @param ns_manager NamespaceManager object, usually can use the one in the Graph object
+    @param curie_str the short URI string to be expanded
+    @param quiet if False will raise ValueError, else return None
+    @return expanded URIRef or None
+    @exception ValueError
+    """
+    try:
+        uri = ns_manager.expand_curie(curie_str)
+
+    except ValueError as e:
+        if quiet:
+            return None
+
+        raise ValueError(f"failed to expand '{curie_str}': {e}")
+
+    return uri