Add DRS compact identifier support

dannon · dannon · commit 298521ad47ca · 2025-09-08T13:25:25.000-04:00
Support compact identifiers (e.g., drs://drs.anv0:object-id) by
resolving prefixes via identifiers.org. Includes caching and tests.
diff --git a/lib/galaxy/files/sources/util.py b/lib/galaxy/files/sources/util.py
@@ -1,5 +1,13 @@
+import json
+import logging
 import time
-from typing import Optional
+from typing import (
+    Dict,
+    List,
+    Optional,
+    Tuple,
+)
+from urllib.parse import quote
 
 from galaxy import exceptions
 from galaxy.files import (
@@ -18,6 +26,8 @@
 from galaxy.util.config_parsers import IpAllowedListEntryT
 from galaxy.util.path import StrPath
 
+log = logging.getLogger(__name__)
+
 
 def _not_implemented(drs_uri: str, desc: str) -> NotImplementedError:
     missing_client_func = f"Galaxy client cannot currently fetch URIs {desc}."
@@ -72,6 +82,134 @@ def _get_access_info(obj_url: str, access_method: dict, headers=None) -> tuple[s
     return url, headers_as_dict
 
 
+class CompactIdentifierResolver:
+    _instance: Optional["CompactIdentifierResolver"] = None
+
+    def __new__(cls, *args, **kwargs):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+        return cls._instance
+
+    def __init__(self, cache_ttl: int = 86400):
+        if not hasattr(self, "_cache"):
+            self._cache: Dict[str, Dict] = {}
+            self._cache_ttl = cache_ttl
+
+    @classmethod
+    def _reset_singleton(cls):
+        """Reset the singleton instance - for testing only."""
+        cls._instance = None
+
+    def _is_cached(self, prefix: str) -> bool:
+        if prefix not in self._cache:
+            return False
+        cached_time = self._cache[prefix].get("timestamp", 0)
+        return (time.time() - cached_time) < self._cache_ttl
+
+    def _cache_result(self, prefix: str, url_pattern: str):
+        self._cache[prefix] = {"url_pattern": url_pattern, "timestamp": time.time()}
+
+    def _query_identifiers_org(self, prefix: str) -> Optional[str]:
+        try:
+            namespace_url = (
+                f"https://registry.api.identifiers.org/restApi/namespaces/search/findByPrefix?prefix={prefix}"
+            )
+            response = requests.get(namespace_url, timeout=DEFAULT_SOCKET_TIMEOUT)
+            response.raise_for_status()
+
+            namespace_data = response.json()
+            if not namespace_data or "_links" not in namespace_data:
+                return None
+
+            if "resources" in namespace_data["_links"]:
+                resources_url = namespace_data["_links"]["resources"]["href"]
+            else:
+                return None
+            response = requests.get(resources_url, timeout=DEFAULT_SOCKET_TIMEOUT)
+            response.raise_for_status()
+
+            resources = response.json()
+            if "_embedded" in resources and "resources" in resources["_embedded"]:
+                official_resource = None
+                fallback_resource = None
+
+                for resource in resources["_embedded"]["resources"]:
+                    if "urlPattern" in resource:
+                        if resource.get("official", False):
+                            official_resource = resource
+                            break
+                        elif fallback_resource is None:
+                            fallback_resource = resource
+
+                best_resource = official_resource or fallback_resource
+                if best_resource:
+                    return best_resource["urlPattern"]
+
+        except requests.exceptions.RequestException as e:
+            log.warning(f"Failed to query identifiers.org for prefix {prefix}: {e}")
+        except (KeyError, json.JSONDecodeError) as e:
+            log.warning(f"Invalid response from identifiers.org for prefix {prefix}: {e}")
+
+        return None
+
+    def resolve_prefix(self, prefix: str) -> Optional[str]:
+        if self._is_cached(prefix):
+            return self._cache[prefix]["url_pattern"]
+
+        url_pattern = self._query_identifiers_org(prefix)
+
+        if url_pattern:
+            self._cache_result(prefix, url_pattern)
+            log.info(f"Resolved DRS prefix '{prefix}' to URL pattern: {url_pattern}")
+        else:
+            log.warning(f"Could not resolve DRS prefix '{prefix}' via identifiers.org")
+
+        return url_pattern
+
+
+def parse_compact_identifier(drs_uri: str) -> Tuple[str, str]:
+    if not drs_uri.startswith("drs://"):
+        raise ValueError(f"Not a valid DRS URI: {drs_uri}")
+
+    rest_of_uri = drs_uri[len("drs://") :]
+
+    colon_idx = rest_of_uri.find(":")
+    if colon_idx == -1:
+        raise ValueError(f"Invalid compact identifier format (missing colon): {drs_uri}")
+
+    prefix = rest_of_uri[:colon_idx]
+    accession = rest_of_uri[colon_idx + 1 :]
+
+    if not all(c.islower() or c.isdigit() or c in "._" for c in prefix):
+        raise ValueError(
+            f"Invalid prefix format '{prefix}': must contain only lowercase letters, numbers, dots, and underscores"
+        )
+
+    if not prefix or not accession:
+        raise ValueError(f"Empty prefix or accession in compact identifier: {drs_uri}")
+
+    return prefix, accession
+
+
+def resolve_compact_identifier_to_url(drs_uri: str, resolver: Optional[CompactIdentifierResolver] = None) -> str:
+    prefix, accession = parse_compact_identifier(drs_uri)
+
+    if resolver is None:
+        resolver = CompactIdentifierResolver()
+
+    url_pattern = resolver.resolve_prefix(prefix)
+    if not url_pattern:
+        raise ValueError(f"Could not resolve prefix '{prefix}' via identifiers.org")
+
+    encoded_accession = quote(accession, safe="")
+    resolved_url = url_pattern.replace("{$id}", encoded_accession)
+
+    if not resolved_url.startswith(("http://", "https://")):
+        raise ValueError(f"Resolved URL is not HTTP(S): {resolved_url}")
+
+    return resolved_url
+
+
 def fetch_drs_to_file(
     drs_uri: str,
     target_path: StrPath,
@@ -84,16 +222,23 @@ def fetch_drs_to_file(
     """Fetch contents of drs:// URI to a target path."""
     if not drs_uri.startswith("drs://"):
         raise ValueError(f"Unknown scheme for drs_uri {drs_uri}")
+
     rest_of_drs_uri = drs_uri[len("drs://") :]
-    if "/" not in rest_of_drs_uri:
-        # DRS URI uses compact identifiers, not yet implemented.
-        # https://ga4gh.github.io/data-repository-service-schemas/preview/release/drs-1.2.0/docs/more-background-on-compact-identifiers.html
-        raise _not_implemented(drs_uri, "that use compact identifiers")
-    netspec, object_id = rest_of_drs_uri.split("/", 1)
-    scheme = "https"
-    if force_http:
-        scheme = "http"
-    get_url = f"{scheme}://{netspec}/ga4gh/drs/v1/objects/{object_id}"
+
+    if "/" not in rest_of_drs_uri and ":" in rest_of_drs_uri:
+        try:
+            get_url = resolve_compact_identifier_to_url(drs_uri)
+            log.info(f"Resolved compact identifier DRS URI {drs_uri} to {get_url}")
+        except ValueError as e:
+            raise ValueError(f"Failed to resolve compact identifier DRS URI {drs_uri}: {str(e)}")
+    elif "/" in rest_of_drs_uri:
+        netspec, object_id = rest_of_drs_uri.split("/", 1)
+        scheme = "https"
+        if force_http:
+            scheme = "http"
+        get_url = f"{scheme}://{netspec}/ga4gh/drs/v1/objects/{object_id}"
+    else:
+        raise ValueError(f"Invalid DRS URI format: {drs_uri}")
     response = retry_and_get(get_url, retry_options or RetryOptions(), headers=headers)
     response.raise_for_status()
     response_object = response.json()
diff --git a/test/integration/test_drs_compact_identifiers.py b/test/integration/test_drs_compact_identifiers.py
@@ -0,0 +1,98 @@
+import os
+import tempfile
+from unittest.mock import (
+    Mock,
+    patch,
+)
+
+import pytest
+
+from galaxy.files.sources.util import fetch_drs_to_file
+
+
+class TestDRSCompactIdentifiersIntegration:
+    @patch("galaxy.files.sources.util.requests.get")
+    @patch("galaxy.files.sources.util.stream_url_to_file")
+    def test_fetch_compact_identifier_drs_file(self, mock_stream, mock_get):
+        namespace_response = Mock()
+        namespace_response.json.return_value = {
+            "_links": {
+                "self": {"href": "https://registry.api.identifiers.org/restApi/namespaces/123"},
+                "resources": {"href": "https://registry.api.identifiers.org/restApi/namespaces/123/resources"},
+            }
+        }
+        namespace_response.raise_for_status.return_value = None
+
+        resources_response = Mock()
+        resources_response.json.return_value = {
+            "_embedded": {
+                "resources": [{"urlPattern": "https://example-drs.com/ga4gh/drs/v1/objects/{$id}", "official": True}]
+            }
+        }
+        resources_response.raise_for_status.return_value = None
+
+        drs_response = Mock()
+        drs_response.json.return_value = {
+            "id": "test-object-id",
+            "access_methods": [
+                {
+                    "type": "https",
+                    "access_url": {
+                        "url": "https://download.example.com/test-file.txt",
+                        "headers": ["Authorization: Bearer test-token"],
+                    },
+                }
+            ],
+        }
+        drs_response.raise_for_status.return_value = None
+        drs_response.status_code = 200
+
+        mock_get.side_effect = [namespace_response, resources_response, drs_response]
+
+        with tempfile.NamedTemporaryFile(delete=False) as tmp:
+            try:
+                fetch_drs_to_file("drs://drs.test:test-object-id", tmp.name, None)
+
+                assert mock_get.call_count == 3
+                mock_get.assert_any_call(
+                    "https://registry.api.identifiers.org/restApi/namespaces/search/findByPrefix?prefix=drs.test",
+                    timeout=600,
+                )
+                mock_get.assert_any_call(
+                    "https://registry.api.identifiers.org/restApi/namespaces/123/resources",
+                    timeout=600,
+                )
+
+                mock_get.assert_any_call(
+                    "https://example-drs.com/ga4gh/drs/v1/objects/test-object-id", timeout=600, headers=None
+                )
+
+                mock_stream.assert_called_once()
+                call_args = mock_stream.call_args
+                assert call_args[0][0] == "https://download.example.com/test-file.txt"
+                assert call_args[1]["target_path"] == tmp.name
+
+            finally:
+                if os.path.exists(tmp.name):
+                    os.unlink(tmp.name)
+
+    def test_compact_identifier_error_handling(self):
+        invalid_uris = [
+            "drs://",  # No identifier
+            "drs://no-colon",  # Missing colon
+            "drs://:only-accession",  # Missing prefix
+            "drs://prefix:",  # Missing accession
+            "drs://UPPERCASE:accession",  # Invalid prefix format
+            "https://not-drs:accession",  # Wrong scheme
+        ]
+
+        for uri in invalid_uris:
+            with pytest.raises(ValueError):
+                fetch_drs_to_file(uri, "/tmp/output", None)
+
+    @patch("galaxy.files.sources.util.CompactIdentifierResolver._query_identifiers_org")
+    def test_meta_resolver_failure_handling(self, mock_identifiers):
+        mock_identifiers.return_value = None
+
+        with pytest.raises(ValueError, match="Failed to resolve compact identifier DRS URI"):
+            fetch_drs_to_file("drs://unknown.prefix:accession", "/tmp/output", None)
diff --git a/test/unit/files/test_drs_compact_identifiers.py b/test/unit/files/test_drs_compact_identifiers.py
