add functions to parse iterables of N3 strings

- also include a function to parse N3 strings
- fix minor arg in python module to match loader protocol

src/rdf_utils/models/python.py

@@ -30,7 +30,7 @@ def import_attr_from_node(graph: Graph, uri: URIRef | str) -> Any:
     return getattr(import_module(module_name), attr_name, None)
 
 
-def load_py_module_attr(graph: Graph, model: ModelBase, quiet: bool = True) -> None:
+def load_py_module_attr(graph: Graph, model: ModelBase, quiet: bool = True, **kwargs: Any) -> None:
     """Load relevant attributes of a `ModuleAttribute` node into a model object.
 
     Parameters:

src/rdf_utils/uri.py

@@ -1,6 +1,8 @@
 # SPDX-License-Identifier:  MPL-2.0
-from typing import Optional
+from typing import Iterable, Optional, Union
 from rdflib import URIRef
+from rdflib.util import from_n3
+from rdflib.term import Node as RDFNode
 from rdflib.namespace import NamespaceManager
 
 
@@ -33,10 +35,11 @@
 def try_expand_curie(
     ns_manager: NamespaceManager, curie_str: str, quiet: bool = False
 ) -> Optional[URIRef]:
-    """Execute rdflib `expand_curie` with exception handling
+    """Execute rdflib `expand_curie` with exception handling.
 
     Parameters:
-        ns_manager: NamespaceManager object, usually can use the one in the Graph object
+        ns_manager: `NamespaceManager` maps prefixes to namespaces,
+                    usually can use the one in the Graph object.
         curie_str: The short URI string to be expanded
         quiet: If False will raise ValueError, else return None
 
@@ -56,3 +59,55 @@ def try_expand_curie(
         raise ValueError(f"failed to expand '{curie_str}': {e}")
 
     return uri
+
+
+def try_parse_n3_string(
+    n3_str: str, ns_manager: NamespaceManager, quiet: bool = False
+) -> Optional[Union[RDFNode, str]]:
+    """Parse N3 string using rdflib.util.from_n3 with exception handling.
+
+    Parameters:
+        n3_str: N3 string to be parsed.
+        ns_manager: `NamespaceManager` maps prefixes to namespaces.
+        quiet: If False will raise ValueError, else return None.
+
+    Raises:
+        ValueError: When not `quiet` and `from_n3` throws an exception
+    """
+    res = None
+    try:
+        res = from_n3(s=n3_str, nsm=ns_manager)
+    except Exception as e:
+        if not quiet:
+            raise ValueError(f"Unable to parse N3 string '{n3_str}', got exception: {e}")
+    return res
+
+
+def try_parse_n3_iterable(
+    n3_str_iterable: Iterable[str], ns_manager: NamespaceManager, quiet: bool = False
+) -> Optional[list[Union[RDFNode, str]]]:
+    """Parse an iterable of N3 strings using rdflib.util.from_n3 with exception handling.
+
+    Parameters:
+        n3_str_iterable: Iterable of N3 strings to be parsed.
+        ns_manager: `NamespaceManager` maps prefixes to namespaces.
+        quiet: If False will raise ValueError, else return None.
+
+    Raises:
+        ValueError: When not `quiet` and `try_parse_n3_string` throws an exception
+    """
+    if isinstance(n3_str_iterable, str):
+        res = try_parse_n3_string(n3_str=n3_str_iterable, ns_manager=ns_manager, quiet=quiet)
+        if res is None:
+            return None
+        return [res]
+
+    res_list = []
+    for n3_str in n3_str_iterable:
+        res = try_parse_n3_string(n3_str=n3_str, ns_manager=ns_manager, quiet=quiet)
+        if res is None:
+            return None
+
+        res_list.append(res)
+
+    return res_list