Skills › Scientific Agent Skills › ontology-term-resolution › id_client.py
id_client.py
316 lines · skills/ontology-term-resolution/scripts/id_client.py at 5f2bc03
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# dependencies = []
5# ///
6"""Prefix and compact-identifier helpers for Bioregistry and Identifiers.org.
8OLS checks whether a term is present and current in its loaded release. These
9services answer a different question: is this prefix real, is the local id
10well-formed, and which landing pages resolve it?
12Standard library only. Network access to https://bioregistry.io and
13https://resolver.api.identifiers.org is required for the request functions;
14helpers below the ``--- pure helpers ---`` mark are offline.
16Reviewed against official schemas and live APIs on 2026-10-01 (see
17``references/companion-apis.md``):
19* ``/api/registry/{prefix}`` accepts synonyms (``HPO`` → ``hp``) and returns
20 the canonical record. A 404 body is ``{"detail": "Prefix not found: ..."}``.
21* ``/api/reference/{CURIE}`` validates the local id against ``pattern``. A
22 404 body of ``{"detail": "invalid identifier: ..."}`` means the prefix is
23 known and the local part is wrong — not that the prefix is unknown.
24* Identifiers.org rejects synonym prefixes (``HPO:0001250`` → HTTP 400) and
25 also rejects Bioregistry's preferred prefix when that is not the MIRIAM
26 namespace (``ORPHA:558`` → 400; ``orphanet:558`` → 200). Landing pages
27 come from ``/api/reference/{CURIE}`` ``providers.miriam`` and the
28 registry ``mappings.ontobee`` field — never from templating
29 ``preferred_prefix``.
30"""
31from __future__ import annotations
33import json
34import re
35import time
36import urllib.error
37import urllib.parse
38import urllib.request
39from typing import Any
41BIOREGISTRY_BASE = "https://bioregistry.io/api"
42IDENTIFIERS_RESOLVER = "https://resolver.api.identifiers.org"
43ONTOBEE_TERM = "https://ontobee.org/ontology/{prefix}?iri={iri}"
44USER_AGENT = "scientific-agent-skills-ontology-term-resolution/1.4"
45TIMEOUT = 30
46MAX_ATTEMPTS = 3
47RETRY_STATUS = {429, 500, 502, 503, 504}
49CURIE_RE = re.compile(r"^([A-Za-z][A-Za-z0-9_.]*):(.+)$")
50PREFIX_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_.]*$")
53class IdError(RuntimeError):
54 """A request to Bioregistry or Identifiers.org failed unrecoverably."""
57class NotFoundError(IdError):
58 """The remote service answered 404 with a JSON ``detail`` body."""
60 def __init__(self, detail: str, url: str):
61 super().__init__(detail)
62 self.detail = detail
63 self.url = url
66def _request(url: str, *, return_rejection: bool = False) -> dict:
67 """GET a JSON document, retrying transient failures."""
68 request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "application/json"})
69 last: Exception | None = None
70 for attempt in range(MAX_ATTEMPTS):
71 try:
72 with urllib.request.urlopen(request, timeout=TIMEOUT) as response:
73 payload = json.load(response)
74 if not isinstance(payload, dict):
75 raise IdError(f"Expected a JSON object from {url}")
76 return payload
77 except urllib.error.HTTPError as exc:
78 body = exc.read().decode("utf-8", errors="replace")
79 if return_rejection and exc.code == 400:
80 try:
81 payload = json.loads(body)
82 except json.JSONDecodeError as error:
83 raise IdError(f"Non-JSON rejection from {url}") from error
84 if isinstance(payload, dict) and payload.get("errorMessage"):
85 return payload
86 raise IdError(f"Malformed rejection from {url}") from exc
87 if exc.code == 404:
88 detail = _detail_from_body(body) or f"HTTP 404 for {url}"
89 raise NotFoundError(detail, url) from exc
90 last = exc
91 if exc.code not in RETRY_STATUS:
92 raise IdError(f"HTTP {exc.code} for {url}: {body[:200]}") from exc
93 except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
94 last = exc
95 if attempt < MAX_ATTEMPTS - 1:
96 time.sleep(1.5 * (attempt + 1))
97 raise IdError(f"request failed after {MAX_ATTEMPTS} attempts: {url} ({last})")
100def _detail_from_body(body: str) -> str:
101 try:
102 payload = json.loads(body)
103 except json.JSONDecodeError:
104 return body.strip()[:200]
105 if isinstance(payload, dict):
106 return str(payload.get("detail") or payload.get("errorMessage") or body[:200])
107 return body.strip()[:200]
110def get_resource(prefix: str) -> dict:
111 """Return the Bioregistry record for a prefix or synonym, or raise NotFoundError."""
112 encoded = urllib.parse.quote(prefix, safe="")
113 return _request(f"{BIOREGISTRY_BASE}/registry/{encoded}")
116def get_reference(curie: str) -> dict:
117 """Resolve a CURIE to provider URLs, validating the local id.
119 Raises ``NotFoundError`` when the prefix is unknown *or* the local id
120 fails the recorded pattern — inspect ``detail`` to tell them apart.
121 """
122 encoded = urllib.parse.quote(curie, safe="")
123 return _request(f"{BIOREGISTRY_BASE}/reference/{encoded}")
126def identifiers_resolver_url(curie: str) -> str:
127 """Build the Identifiers.org resolver URL.
129 The colon in ``PREFIX:local`` must stay a colon. Encoding it as ``%3A``
130 is HTTP 400: ``NOT A NAMESPACE`` for a CURIE the service otherwise accepts.
131 """
132 encoded = urllib.parse.quote(curie, safe=":")
133 return f"{IDENTIFIERS_RESOLVER}/{encoded}"
136def resolve_identifiers(curie: str) -> dict:
137 """Ask Identifiers.org for landing pages. Returns the JSON payload.
139 A rejected compact identifier comes back as HTTP 400 with
140 ``errorMessage`` set and ``payload.resolvedResources`` null — that is
141 returned as a dict, not raised, so the caller can report it.
142 """
143 return _request(identifiers_resolver_url(curie), return_rejection=True)
146# --- pure helpers -----------------------------------------------------------
149def split_query(value: str) -> tuple[str, str | None]:
150 """Split ``PREFIX`` or ``PREFIX:local`` into ``(prefix, local_or_None)``."""
151 text = value.strip()
152 match = CURIE_RE.match(text)
153 if match:
154 return match.group(1), match.group(2)
155 return text, None
158def is_prefix(value: str) -> bool:
159 """True for a bare registry prefix, false for CURIEs and junk."""
160 return bool(PREFIX_RE.match(value.strip())) and ":" not in value.strip()
163def local_matches_pattern(local: str, pattern: str | None) -> bool | None:
164 """Check a local unique id against a Bioregistry regex.
166 Returns ``None`` when the registry has no pattern, so the caller can
167 distinguish "unchecked" from "failed".
168 """
169 if not pattern:
170 return None
171 try:
172 return re.fullmatch(pattern, local) is not None
173 except re.error:
174 return None
177def apply_uri_format(uri_format: str | None, local: str) -> str | None:
178 """Fill a Bioregistry ``uri_format`` template (``$1`` is the local id)."""
179 if not uri_format or "$1" not in uri_format:
180 return None
181 return uri_format.replace("$1", local)
184def ontobee_url(ontobee_prefix: str | None, iri: str | None) -> str | None:
185 """Build the Ontobee HTML/RDF page for a term IRI.
187 ``ontobee_prefix`` is Bioregistry ``mappings.ontobee``, not
188 ``preferred_prefix``. Ontobee has no JSON search API. The page is the
189 product: HTML for humans, RDF when the same IRI is dereferenced as
190 linked data.
191 """
192 if not ontobee_prefix or not iri:
193 return None
194 return ONTOBEE_TERM.format(
195 prefix=ontobee_prefix,
196 iri=urllib.parse.quote(iri, safe=""),
197 )
200def compact_id_from_identifiers_url(url: str) -> str | None:
201 """``https://identifiers.org/orphanet:558`` → ``orphanet:558``."""
202 if not url:
203 return None
204 path = urllib.parse.urlparse(url).path.lstrip("/")
205 return path or None
208def landing_page_urls(
209 resource: dict | None,
210 reference: dict | None,
211 iri: str | None,
212) -> tuple[str, str]:
213 """Return ``(identifiers_org, ontobee)`` from mappings, never templates.
215 Identifiers.org comes from ``/api/reference/{CURIE}`` ``providers.miriam``.
216 Ontobee comes from the registry ``mappings.ontobee`` field plus ``iri``.
217 Either side is empty when that mapping is missing — Bioregistry's
218 preferred prefix is not a substitute.
219 """
220 identifiers = ""
221 if reference:
222 providers = reference.get("providers") or {}
223 if isinstance(providers, dict):
224 identifiers = providers.get("miriam") or ""
225 ontobee = ""
226 if resource:
227 mappings = resource.get("mappings") or {}
228 ontobee = ontobee_url(mappings.get("ontobee"), iri) or ""
229 return identifiers, ontobee
232def identifiers_landing_pages(payload: dict) -> list[dict[str, Any]]:
233 """Flatten Identifiers.org ``resolvedResources`` into compact rows."""
234 resources = (payload.get("payload") or {}).get("resolvedResources") or []
235 rows = []
236 for item in resources:
237 rec = item.get("recommendation") or {}
238 rows.append(
239 {
240 "provider": item.get("providerCode") or "official",
241 "url": item.get("compactIdentifierResolvedUrl") or "",
242 "official": bool(item.get("official")),
243 "score": rec.get("recommendationIndex"),
244 }
245 )
246 rows.sort(key=lambda row: (not row["official"], -(row["score"] or 0)))
247 return rows
250def classify_prefix_query(
251 query: str,
252 resource: dict | None,
253 *,
254 local: str | None,
255 reference_detail: str | None = None,
256) -> dict:
257 """Build the status record for one prefix or CURIE lookup.
259 Pure: callers supply the Bioregistry payload (or ``None`` on 404).
260 """
261 prefix, parsed_local = split_query(query)
262 local = local if local is not None else parsed_local
263 result = {
264 "query": query,
265 "status": "ok",
266 "preferred_prefix": "",
267 "canonical_curie": "",
268 "pattern": "",
269 "example": "",
270 "name": "",
271 "default_iri": "",
272 "identifiers_org": "",
273 "ontobee": "",
274 "ols_id": "",
275 "detail": "",
276 }
277 if resource is None:
278 result["status"] = "unknown_prefix"
279 result["detail"] = reference_detail or f"no Bioregistry record for {prefix!r}"
280 return result
282 preferred = resource.get("preferred_prefix") or resource.get("prefix") or ""
283 canonical_prefix = resource.get("prefix") or ""
284 result["preferred_prefix"] = preferred
285 result["pattern"] = resource.get("pattern") or ""
286 result["example"] = resource.get("example") or ""
287 result["name"] = resource.get("name") or ""
288 mappings = resource.get("mappings") or {}
289 result["ols_id"] = mappings.get("ols") or ""
291 queried = prefix
292 accepted = {canonical_prefix.casefold(), preferred.casefold()} - {""}
293 if queried.casefold() not in accepted:
294 result["status"] = "synonym_prefix"
295 result["detail"] = f"{queried!r} is a synonym of preferred prefix {preferred}"
297 if local is None:
298 return result
300 matched = local_matches_pattern(local, result["pattern"] or None)
301 if matched is False or (reference_detail and "invalid identifier" in reference_detail):
302 result["status"] = "invalid_local"
303 result["detail"] = (
304 reference_detail
305 or f"{local!r} does not match pattern {result['pattern']}"
306 )
307 return result
309 canonical = f"{preferred}:{local}" if preferred else f"{canonical_prefix}:{local}"
310 result["canonical_curie"] = canonical
311 iri = apply_uri_format(resource.get("uri_format"), local)
312 result["default_iri"] = iri or ""
313 # Landing pages are filled by the caller from /api/reference providers
314 # and mappings.ontobee. Templating preferred_prefix here emits dead URLs
315 # (ORPHA:558, OBA:0000001).
316 return result