CurieNamespace catalog that is not a singleton and that uses curies

linkml_runtime/utils/curienamespace.py

@@ -1,17 +1,109 @@
-from typing import Optional, Union
+from logging import warning
+from typing import Optional, Union, Dict, List
 
 from rdflib import Namespace, URIRef
+from curies import Converter, Record
+
+class CurieNamespaceCatalog(object):
+    """
+    A CurieNamespaceCatalog is a catalog of CurieNamespace objects
+    its main purpose is to convert between uri's and curies for the namespaces in the catalog
+    """
+    def __init__(self) -> None:
+        self.namespaces = []
+        self._converter: Optional[Converter] = None
+
+    @property
+    def converter(self):
+        """
+        return a curies.Converter that knows all namespaces. 
+        When multiple namespaces have the same prefix, they are added as uri synonyms to the converter.
+        When there are two prefixes leading to the same uri, they are added as prefix synonyms to the converter.
+        """
+        if not self._converter:
+            self._converter = self._buildConverter()
+        return self._converter
+
+    def _buildConverter(self):
+        records = []
+        namespaces_to_treat = self.namespaces[:]
+        while len(namespaces_to_treat) > 0:
+            ns = namespaces_to_treat.pop(0)
+            prefix = ns.prefix
+            uri = str(ns)
+            all_prefixes = [prefix]
+            all_uris = [uri]
+            iteration_needed = True
+            while iteration_needed:
+                iteration_needed = False
+                for possible_synonym in namespaces_to_treat[:]:
+                    if possible_synonym.prefix in all_prefixes:
+                        all_uris.append(str(possible_synonym))
+                        namespaces_to_treat.remove(possible_synonym)
+                        iteration_needed = True
+                    if str(possible_synonym) in all_uris:
+                        all_prefixes.append(possible_synonym.prefix)
+                        namespaces_to_treat.remove(possible_synonym)
+                        iteration_needed = True
+            records.append(Record(prefix, uri , [x for x in all_prefixes if not x == prefix], [x for x in all_uris if not x == uri]))
+        return Converter(records=records)
+
+
+
+
+    def to_curie(self, uri: Union[str, URIRef]) -> Optional[str]:
+        """
+        Compress a URI to a CURIE, if possible.
+
+        :param uri:
+            A string representing a valid uniform resource identifier (URI)
+        :returns:
+            A compact URI if this converter could find an appropriate URI prefix, otherwise None.
+
+        """
+        if isinstance(uri, URIRef):
+            uri = str(uri)
+        return self.converter.compress(uri)
+
+    def to_uri(self, curie: str) -> Optional[URIRef]:
+        """
+        Expand a CURIE to a URI, if possible.
+
+        :param curie:
+            A string representing a compact URI
+        :returns:
+            A URIRef if this converter contains a URI prefix for the prefix in this CURIE, otherwise None
+        """
+        expanded = self.converter.expand(curie)
+        return None if expanded is None else URIRef(expanded) 
+
+    def add_namespace(self,ns: "CurieNamespace"):
+        """
+        Adds a new namespace to the catalog.
+        """
+        self.namespaces.append(ns)
+        self._converter = None
+
+    @classmethod
+    def create(cls, *namespaces: List["CurieNamespace"]):
+        """
+        creates a new catalog from the given namespaces
+        """
+        cat = CurieNamespaceCatalog()
+        [cat.add_namespace(x) for x in namespaces]
+        return cat
+
 
 
 class CurieNamespace(Namespace):
-    def __new__(cls, prefix: str, value: Union[str, URIRef]):
-        value = str(value)
-        try:
-            rt = str.__new__(cls, value)
-        except UnicodeDecodeError:
-            rt = str.__new__(cls, value, 'utf-8')
+    def __new__(cls, prefix: str, ns: Union[str, URIRef]):
+        rt = Namespace.__new__(cls, str(ns) if not isinstance(ns, bytes) else ns)
         rt.prefix = prefix
         return rt
 
     def curie(self, reference: Optional[str] = '') -> str:
         return self.prefix + ':' + reference
+
+    def addTo(self, catalog: CurieNamespaceCatalog) -> "CurieNamespace":
+        catalog.add_namespace(self)
+        return self

pyproject.toml

@@ -53,7 +53,7 @@ pyyaml = "*"
 rdflib = ">=6.0.0"
 requests = "*"
 prefixmaps = ">=0.1.4"
-curies = "^0.4.0"
+curies = "^0.4.3"
 
 [tool.poetry.dev-dependencies]
 coverage = "^6.2"