Add derived dataset provenance to the ro-crate

pyproject.toml

@@ -0,0 +1,4 @@
+[tool.isort]
+profile = "black"
+multi_line_output = 3
+

src/crate_builder_script/api_calls.py

@@ -0,0 +1,90 @@
+"""Handle requests to HuBMAP APIs."""
+import logging
+import os
+from functools import lru_cache
+from pprint import pformat
+from typing import Any
+
+import requests
+
+LOGGER = logging.getLogger(__name__)
+
+ENTITY_API = "https://entity.api.hubmapconsortium.org"
+ASSETS_API = "https://assets.hubmapconsortium.org"
+UUID_API = "https://uuid.api.hubmapconsortium.org"
+
+AUTH_TOK = os.environ["AUTH_TOK"]
+
+HUBMAP = "https://hubmapconsortium.org/"
+
+HUBMAP_ORG_ENTITY = HUBMAP  # for lack of a better choice
+
+ENTITY_BASE = "https://portal.hubmapconsortium.org/browse/"
+
+HUBMAP_ORG_ENTITY = HUBMAP  # for lack of a better choice
+
+
+def entity_iri(entity: Any) -> str:
+    """Return a resolvable IRI for a HuBMAP entity (dataset / sample / donor).
+
+    Raw datasets and multi-assay splits carry `doi_url`; processed datasets, samples
+    and donors do not, and fall back to the portal browse URL, which resolves by
+    HuBMAP ID for every entity type.
+    """
+    return entity.get("doi_url") or ENTITY_BASE + (entity.get("hubmap_id") or "")
+
+
+@lru_cache(maxsize=8)
+def fetch_entity_info(target_id: str) -> dict[str, Any]:
+    """Fetch a dataset's entity information."""
+    LOGGER.debug("FETCH %s", target_id)
+    try:
+        resp = requests.get(
+            ENTITY_API + f"/entities/{target_id}",
+            headers={"Authorization": f"Bearer {AUTH_TOK}"},
+        )
+        resp.raise_for_status()
+    except Exception as excp:
+        raise RuntimeError(f"Fetch of entity {target_id} failed: {excp}") from excp
+    ds_info = resp.json()
+    LOGGER.debug("TOP LEVEL for %s:\n%s", target_id, pformat(ds_info, depth=1))
+    LOGGER.debug(
+        "INGEST METADATA:\n%s", pformat(ds_info.get("ingest_metadata", {}), depth=2)
+    )
+    LOGGER.debug("METADATA:\n%s", pformat(ds_info.get("metadata", {}), depth=2))
+    LOGGER.debug(
+        "DIRECT ANCESTORS: %s\n%s",
+        [elt["hubmap_id"] for elt in ds_info.get("direct_ancestors", [])],
+        pformat(ds_info.get("direct_ancestors"), depth=2),
+    )
+    LOGGER.debug(
+        "DIRECT ANCESTOR:\n%s", pformat(ds_info.get("direct_ancestor"), depth=2)
+    )
+    if "direct_ancestors" in ds_info:
+        first_ancestor = ds_info["direct_ancestors"][0]
+    elif "direct_ancestor" in ds_info:
+        first_ancestor = ds_info["direct_ancestor"]
+    else:
+        first_ancestor = {}
+    LOGGER.debug(
+        "ANCESTOR INGEST MD\n%s", pformat(first_ancestor.get("ingest_metadata", {}))
+    )
+    LOGGER.debug("ANCESTOR MD\n%s", pformat(first_ancestor.get("metadata", {})))
+    return ds_info
+
+
+@lru_cache(maxsize=8)
+def fetch_uuid_files_info(target_id: str) -> dict[str, Any]:
+    """Fetch files informaion from the UUID API."""
+    resp = requests.get(
+        UUID_API + f"/{target_id}/files",
+        headers={"Authorization": f"Bearer {AUTH_TOK}"},
+    )
+    resp.raise_for_status()
+    # LOGGER.debug("UUID FILES first 10:\n%s", pformat(resp.json()[:10]))
+    return resp.json()
+
+
+def asset_url(uuid: str, rel_path: str) -> str:
+    """Build the URL by which a file should be available from the ASSETS API."""
+    return f"{ASSETS_API}/{uuid}/{rel_path}"