Replace prefixcommons usages with curies

kgx/config.py

@@ -1,4 +1,5 @@
 import importlib
+from functools import lru_cache
 from typing import Dict, Any, Optional
 import sys
 from os import path
@@ -10,12 +11,13 @@
 import json
 import logging
 
+import curies
 from kgx.graph.base_graph import BaseGraph
 
 config: Optional[Dict[str, Any]] = None
 logger: Optional[logging.Logger] = None
 graph_store_class: Optional[BaseGraph] = None
-jsonld_context_map: Dict = {}
+jsonld_context_map: Dict[str, Dict[str, str]] = {}
 
 CONFIG_FILENAME = path.join(path.dirname(path.abspath(__file__)), "config.yml")
 
@@ -41,6 +43,13 @@ def get_config(filename: str = CONFIG_FILENAME) -> Dict:
     return config
 
 
+@lru_cache
+def get_converter(name: str = "biolink") -> curies.Converter:
+    """Get contents of a JSON-LD context."""
+    filepath = config["jsonld-context"][name]  # type: ignore
+    return curies.load_jsonld_context(filepath)
+
+
 def get_jsonld_context(name: str = "biolink"):
     """
     Get contents of a JSON-LD context.
@@ -55,19 +64,8 @@ def get_jsonld_context(name: str = "biolink"):
     if name in jsonld_context_map:
         content = jsonld_context_map[name]
     else:
-        filepath = config["jsonld-context"][name]  # type: ignore
-        if filepath.startswith("http"):
-            try:
-                content = requests.get(filepath).json()
-            except ConnectionError:
-                raise Exception(f"Unable to download JSON-LD context from {filepath}")
-        else:
-            if path.exists(filepath):
-                content = json.load(open(filepath))
-
-        if "@context" in content:
-            content = content["@context"]
-            jsonld_context_map[name] = content
+        converter = get_converter(name)
+        jsonld_context_map[name] = converter.bimap
     return content
 
 

kgx/config.yml

@@ -18,6 +18,6 @@ logging:
   format: '[%(name)s][%(filename)s][%(funcName)20s] %(levelname)s: %(message)s'
 
 jsonld-context:
-  biolink: https://raw.githubusercontent.com/biolink/biolink-model/2.2.5/context.jsonld
+  biolink: https://raw.githubusercontent.com/biolink/biolink-model/v4.0.0/project/jsonld/biolink_model.context.jsonld
   monarch_context: https://raw.githubusercontent.com/prefixcommons/biocontext/master/registry/monarch_context.jsonld
-  obo_context: https://raw.githubusercontent.com/prefixcommons/biocontext/master/registry/obo_context.jsonld
+  obo_context: https://raw.githubusercontent.com/prefixcommons/biocontext/master/registry/obo_context.jsonld

kgx/prefix_manager.py

@@ -1,7 +1,7 @@
 import re
 from typing import Dict, Optional
 
-import prefixcommons.curie_util as cu
+import curies
 from cachetools import LRUCache, cached
 
 from kgx.config import get_jsonld_context, get_logger
@@ -33,7 +33,8 @@ def __init__(self, url: str = None):
 
         """
         if url:
-            context = cu.read_remote_jsonld_context(url)
+            converter = curies.load_jsonld_context(url)
+            context = converter.bimap
         else:
             context = get_jsonld_context()
         self.set_prefix_map(context)

kgx/utils/kgx_utils.py

@@ -21,9 +21,14 @@
 from cachetools import LRUCache
 import pandas as pd
 import numpy as np
-from prefixcommons.curie_util import contract_uri
-from prefixcommons.curie_util import expand_uri
-from kgx.config import get_logger, get_jsonld_context, get_biolink_model_schema
+import curies
+
+from kgx.config import (
+    get_logger,
+    get_jsonld_context,
+    get_biolink_model_schema,
+    get_converter,
+)
 from kgx.graph.base_graph import BaseGraph
 
 curie_lookup_service = None
@@ -220,12 +225,28 @@ def format_biolink_slots(s: str) -> str:
         return f"biolink:{formatted}"
 
 
+MONARCH_CONVERTER = get_converter("monarch_context")
+OBO_CONVERTER = get_converter("obo_context")
+DEFAULT_CONVERTER = curies.chain([MONARCH_CONVERTER, OBO_CONVERTER])
+
+
+def _resolve_converter(prefix_maps: Optional[List[Dict[str, str]]] = None) -> curies.Converter:
+    if not prefix_maps:
+        return DEFAULT_CONVERTER
+    return curies.chain([
+        *(
+            curies.load_prefix_map(prefix_map)
+            for prefix_map in prefix_maps
+        ),
+        DEFAULT_CONVERTER,
+    ])
+
+
 def contract(
-    uri: str, prefix_maps: Optional[List[Dict]] = None, fallback: bool = True
+    uri: str, prefix_maps: Optional[List[Dict[str, str]]] = None, fallback: bool = True
 ) -> str:
     """
     Contract a given URI to a CURIE, based on mappings from `prefix_maps`.
-    If no prefix map is provided then will use defaults from prefixcommons-py.
 
     This method will return the URI as the CURIE if there is no mapping found.
 
@@ -236,39 +257,19 @@ def contract(
     prefix_maps: Optional[List[Dict]]
         A list of prefix maps to use for mapping
     fallback: bool
-        Determines whether to fallback to default prefix mappings, as determined
-        by `prefixcommons.curie_util`, when URI prefix is not found in `prefix_maps`.
+        Defunct option. New implementation always chains with default prefix maps.
 
     Returns
     -------
     str
         A CURIE corresponding to the URI
 
     """
-    curie = uri
-    default_curie_maps = [
-        get_jsonld_context("monarch_context"),
-        get_jsonld_context("obo_context"),
-    ]
-    if prefix_maps:
-        curie_list = contract_uri(uri, prefix_maps)
-        if len(curie_list) == 0:
-            if fallback:
-                curie_list = contract_uri(uri, default_curie_maps)
-                if curie_list:
-                    curie = curie_list[0]
-        else:
-            curie = curie_list[0]
-    else:
-        curie_list = contract_uri(uri, default_curie_maps)
-        if len(curie_list) > 0:
-            curie = curie_list[0]
-
-    return curie
+    return _resolve_converter(prefix_maps).compress(uri, passthrough=True)
 
 
 def expand(
-    curie: str, prefix_maps: Optional[List[dict]] = None, fallback: bool = True
+    curie: str, prefix_maps: Optional[List[Dict[str, str]]] = None, fallback: bool = True
 ) -> str:
     """
     Expand a given CURIE to an URI, based on mappings from `prefix_map`.
@@ -282,27 +283,15 @@ def expand(
     prefix_maps: Optional[List[dict]]
         A list of prefix maps to use for mapping
     fallback: bool
-        Determines whether to fallback to default prefix mappings, as determined
-        by `prefixcommons.curie_util`, when CURIE prefix is not found in `prefix_maps`.
+        Defunct option. New implementation always chains with default prefix maps.
 
     Returns
     -------
     str
         A URI corresponding to the CURIE
 
     """
-    default_curie_maps = [
-        get_jsonld_context("monarch_context"),
-        get_jsonld_context("obo_context"),
-    ]
-    if prefix_maps:
-        uri = expand_uri(curie, prefix_maps)
-        if uri == curie and fallback:
-            uri = expand_uri(curie, default_curie_maps)
-    else:
-        uri = expand_uri(curie, default_curie_maps)
-
-    return uri
+    return _resolve_converter(prefix_maps).expand(curie, passthrough=True)
 
 
 _default_toolkit = None

tests/unit/__init__.py

@@ -1,4 +1,3 @@
-import docker
 import pytest
 from neo4j import GraphDatabase
 
@@ -17,6 +16,8 @@ def check_container():
     is currently running.
     """
     try:
+        import docker
+
         client = docker.from_env()
         status = False
         try: