Get mappings between external registries (#902)

There's an API endpoint that lets you get all if the mappings from one
external registry to another. For example, getting all mappings from OBO
Foundry to Bioportal can be done with
https://bioregistry.io/api/metaregistry/obofoundry/mapping/bioportal.

However, the implementation of this functionality is built in to the API
route itself. This PR externalizes that implementation into the
Bioregistry's manager class so it can be reused.

Related to INCATools/ontology-access-kit#588

src/bioregistry/app/api.py

@@ -283,35 +283,21 @@ def get_metaresource_external_mappings(
 ):
     """Get mappings between two external prefixes."""
     manager = request.app.manager
-    if metaprefix not in manager.metaregistry:
-        return {"bad source prefix": metaprefix}, 400
-    if target not in manager.metaregistry:
-        return {"bad target prefix": target}, 400
-    rv = {}
-    source_only = set()
-    target_only = set()
-    for resource in manager.registry.values():
-        mappings = resource.get_mappings()
-        mp1_prefix = mappings.get(metaprefix)
-        mp2_prefix = mappings.get(target)
-        if mp1_prefix and mp2_prefix:
-            rv[mp1_prefix] = mp2_prefix
-        elif mp1_prefix and not mp2_prefix:
-            source_only.add(mp1_prefix)
-        elif not mp1_prefix and mp2_prefix:
-            target_only.add(mp2_prefix)
-
+    try:
+        diff = manager.get_external_mappings(metaprefix, target)
+    except KeyError as e:
+        return {"message": str(e)}, 400
     return MappingResponse(
         meta=MappingResponseMeta(
-            len_overlap=len(rv),
+            len_overlap=len(diff.mappings),
             source=metaprefix,
             target=target,
-            len_source_only=len(source_only),
-            len_target_only=len(target_only),
-            source_only=sorted(source_only),
-            target_only=sorted(target_only),
+            len_source_only=len(diff.source_only),
+            len_target_only=len(diff.target_only),
+            source_only=sorted(diff.source_only),
+            target_only=sorted(diff.target_only),
         ),
-        mappings=rv,
+        mappings=diff.mappings,
     )
 
 

src/bioregistry/resource_manager.py

@@ -23,6 +23,7 @@
 )
 
 import curies
+from pydantic import BaseModel
 
 from .constants import (
     BIOREGISTRY_PATH,
@@ -93,6 +94,16 @@ def _safe_curie_to_str(prefix: Optional[str], identifier: Optional[str]) -> Opti
     return curie_to_str(prefix, identifier)
 
 
+class MappingsDiff(BaseModel):
+    """A difference between two mappings sets."""
+
+    source_metaprefix: str
+    source_only: Set[str]
+    target_metaprefix: str
+    target_only: Set[str]
+    mappings: Dict[str, str]
+
+
 class Manager:
     """A manager for functionality related to a metaregistry."""
 
@@ -1566,6 +1577,33 @@ def read_contributors(self, direct_only: bool = False) -> Mapping[str, Attributa
             direct_only=direct_only,
         )
 
+    def get_external_mappings(self, source_metaprefix: str, target_metaprefix: str) -> MappingsDiff:
+        """Get mappings between two external registries."""
+        if source_metaprefix not in self.metaregistry:
+            raise KeyError(f"invalid source metaprefix: {source_metaprefix}")
+        if target_metaprefix not in self.metaregistry:
+            raise KeyError(f"invalid target metaprefix: {target_metaprefix}")
+        mappings: Dict[str, str] = {}
+        source_only: Set[str] = set()
+        target_only: Set[str] = set()
+        for resource in self.registry.values():
+            metaprefix_to_prefix = resource.get_mappings()
+            mp1_prefix = metaprefix_to_prefix.get(source_metaprefix)
+            mp2_prefix = metaprefix_to_prefix.get(target_metaprefix)
+            if mp1_prefix and mp2_prefix:
+                mappings[mp1_prefix] = mp2_prefix
+            elif mp1_prefix and not mp2_prefix:
+                source_only.add(mp1_prefix)
+            elif not mp1_prefix and mp2_prefix:
+                target_only.add(mp2_prefix)
+        return MappingsDiff(
+            source_metaprefix=source_metaprefix,
+            source_only=source_only,
+            target_metaprefix=target_metaprefix,
+            target_only=target_only,
+            mappings=mappings,
+        )
+
 
 def _read_contributors(
     registry, metaregistry, collections, contexts, direct_only: bool = False

tests/test_manager.py

@@ -7,6 +7,7 @@
 import bioregistry
 from bioregistry import Manager, parse_curie
 from bioregistry.export.rdf_export import get_full_rdf
+from bioregistry.resource_manager import MappingsDiff
 
 
 class TestResourceManager(unittest.TestCase):
@@ -247,3 +248,16 @@ def test_parse_curie(self):
                 actual_prefix, actual_i = parse_curie(curie, sep=sep, use_preferred=pref)
                 self.assertEqual(p, actual_prefix)
                 self.assertEqual(i, actual_i)
+
+    def test_external_registry_mappings(self):
+        """Test external registry mappings."""
+        res = self.manager.get_external_mappings("obofoundry", "bioportal")
+        self.assertIsInstance(res, MappingsDiff)
+        self.assertEqual("obofoundry", res.source_metaprefix)
+        self.assertEqual("bioportal", res.target_metaprefix)
+        self.assertIn("gaz", res.mappings)
+        self.assertEqual("GAZ", res.mappings["gaz"])
+        # This is an obsolete OBO Foundry ontology so it won't get uploaded to BioPortal
+        self.assertIn("loggerhead", res.source_only)
+        # This is a non-ontology so it won't get in OBO Foundry
+        self.assertIn("DCTERMS", res.target_only)

tests/test_web/test_api.py

@@ -13,6 +13,7 @@
 from starlette.testclient import TestClient
 
 from bioregistry import Resource
+from bioregistry.app.api import MappingResponse
 from bioregistry.app.impl import get_app
 
 
@@ -224,3 +225,17 @@ def test_autocomplete(self):
             with self.subTest(query=q):
                 res = self.client.get(f"/api/autocomplete?q={q}")
                 self.assertEqual(200, res.status_code)
+
+    def test_external_registry_mappings(self):
+        """Test external registry mappings."""
+        url = "/api/metaregistry/obofoundry/mapping/bioportal"
+        res = self.client.get(url)
+        res_parsed = MappingResponse.parse_obj(res.json())
+        self.assertEqual("obofoundry", res_parsed.meta.source)
+        self.assertEqual("bioportal", res_parsed.meta.target)
+        self.assertIn("gaz", res_parsed.mappings)
+        self.assertEqual("GAZ", res_parsed.mappings["gaz"])
+        # This is an obsolete OBO Foundry ontology so it won't get uploaded to BioPortal
+        self.assertIn("loggerhead", res_parsed.meta.source_only)
+        # This is a non-ontology so it won't get in OBO Foundry
+        self.assertIn("DCTERMS", res_parsed.meta.target_only)