Skills › Scientific Agent Skills › ontology-term-resolution › ols_client.py
ols_client.py
410 lines · skills/ontology-term-resolution/scripts/ols_client.py at 5f2bc03
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# dependencies = []
5# ///
6"""Minimal EBI OLS4 client plus the pure helpers the two CLIs share.
8Standard library only. Network access to https://www.ebi.ac.uk/ols4 is required
9for the request functions; every helper below the ``--- pure helpers ---`` mark
10is offline and independently testable.
12Design notes that matter for correctness (all verified against the live API):
14* ``/search`` with ``exact=true`` is exact *token* matching, not exact label
15 matching. Restrict ``queryFields`` to ``label`` (or ``label,synonym``) and
16 verify lexical equality and synonym scope locally. See ``references/ols4-api.md``.
17* ``/search`` never returns ``is_obsolete`` or ``term_replaced_by``, even when
18 they are named in ``fieldList``. Only the term-detail endpoint carries them.
19* IRI fallback templates are hypotheses only; a live term response is required.
20 EFO and Orphanet use their own namespaces.
21"""
22from __future__ import annotations
24import json
25import re
26import time
27import urllib.error
28import urllib.parse
29import urllib.request
30from typing import Any, Iterable
32OLS_BASE = "https://www.ebi.ac.uk/ols4/api"
33USER_AGENT = "scientific-agent-skills-ontology-term-resolution/1.4"
34TIMEOUT = 30
35MAX_ATTEMPTS = 3
36RETRY_STATUS = {429, 500, 502, 503, 504}
38# OLS4 now exposes the plural scoped synonym fields as well as the flat union.
39SEARCH_FIELDS = ("iri,obo_id,label,synonym,exact_synonyms,related_synonyms,"
40 "broad_synonyms,narrow_synonyms,ontology_name,is_defining_ontology,type,short_form")
42# OLS ontology ids that are not simply the lowercased CURIE prefix.
43ONTOLOGY_ID_OVERRIDES = {
44 "orphanet": "ordo",
45 "orpha": "ordo",
46}
48# IRI templates for prefixes that do not live under the OBO PURL namespace.
49# Only used by the term-detail fallback below, never to mint an IRI we then
50# trust: a wrong guess simply resolves to nothing.
51IRI_TEMPLATES = {
52 "efo": "http://www.ebi.ac.uk/efo/EFO_{local}",
53 "orphanet": "http://www.orpha.net/ORDO/Orphanet_{local}",
54 "orpha": "http://www.orpha.net/ORDO/Orphanet_{local}",
55}
56DEFAULT_IRI_TEMPLATE = "http://purl.obolibrary.org/obo/{prefix}_{local}"
59class OlsError(RuntimeError):
60 """A request to OLS failed in a way the caller cannot paper over."""
63def _request(path: str, params: dict[str, Any] | None = None) -> dict:
64 """GET a JSON document from OLS, retrying transient failures."""
65 url = f"{OLS_BASE}/{path.lstrip('/')}"
66 if params:
67 clean = {k: v for k, v in params.items() if v is not None}
68 url = f"{url}?{urllib.parse.urlencode(clean)}"
69 return _request_url(url)
72def _request_url(url: str) -> dict:
73 """Fetch an OLS HAL link, upgrading the service's HTTP links to HTTPS."""
74 parsed = urllib.parse.urlsplit(urllib.parse.urljoin(OLS_BASE + "/", url))
75 base = urllib.parse.urlsplit(OLS_BASE)
76 if parsed.netloc != base.netloc or not parsed.path.startswith(base.path + "/"):
77 raise OlsError("OLS returned a link outside its API")
78 url = urllib.parse.urlunsplit(parsed._replace(scheme="https"))
79 request = urllib.request.Request(
80 url, headers={"User-Agent": USER_AGENT, "Accept": "application/json"}
81 )
83 last: Exception | None = None
84 for attempt in range(MAX_ATTEMPTS):
85 try:
86 with urllib.request.urlopen(request, timeout=TIMEOUT) as response:
87 payload = json.load(response)
88 if not isinstance(payload, dict):
89 raise OlsError("OLS returned a non-object JSON response")
90 return payload
91 except urllib.error.HTTPError as exc:
92 if exc.code == 404:
93 raise
94 last = exc
95 if exc.code not in RETRY_STATUS:
96 break
97 except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
98 last = exc
99 if attempt < MAX_ATTEMPTS - 1:
100 time.sleep(1.5 * (attempt + 1))
101 raise OlsError(f"OLS request failed after {MAX_ATTEMPTS} attempts: {url} ({last})")
104def search(
105 text: str,
106 *,
107 ontology: str | None = None,
108 query_fields: str | None = "label,synonym",
109 exact: bool = True,
110 subtree_iri: str | None = None,
111 rows: int = 10,
112 include_obsolete: bool = False,
113) -> list[dict]:
114 """Search OLS and return the raw ``response.docs`` list.
116 ``query_fields=None`` widens the search to every indexed field. The current
117 v1 compatibility API's ``obsoletes=true`` selects obsolete terms only, so
118 ``include_obsolete=True`` merges separate current and obsolete queries.
119 """
120 docs = _request(
121 "search",
122 {
123 "q": text,
124 "ontology": ontology,
125 "queryFields": query_fields,
126 "exact": "true" if exact else None,
127 "allChildrenOf": subtree_iri,
128 "rows": rows,
129 "obsoletes": "false",
130 "type": "class",
131 "fieldList": SEARCH_FIELDS,
132 },
133 )
134 def search_docs(payload: dict) -> list[dict]:
135 response = payload.get("response")
136 if not isinstance(response, dict) or not isinstance(response.get("docs"), list):
137 raise OlsError("OLS returned an invalid search response")
138 return response["docs"]
140 found = list(search_docs(docs))
141 if include_obsolete:
142 obsolete = _request("search", {
143 "q": text, "ontology": ontology, "queryFields": query_fields,
144 "exact": "true" if exact else None, "allChildrenOf": subtree_iri,
145 "rows": rows, "obsoletes": "true", "type": "class", "fieldList": SEARCH_FIELDS,
146 })
147 found += search_docs(obsolete)
148 return found
151def candidate_iris(curie: str) -> list[str]:
152 """IRIs a CURIE might correspond to, for the term-detail fallback."""
153 match = CURIE_RE.match(curie.strip())
154 if not match:
155 return []
156 prefix, local = match.group(1), match.group(2)
157 template = IRI_TEMPLATES.get(prefix.lower())
158 if template:
159 return [template.format(prefix=prefix, local=local)]
160 return [DEFAULT_IRI_TEMPLATE.format(prefix=prefix, local=local)]
163def _terms_by_iri(iri: str) -> list[dict]:
164 """Every copy of a term across ontologies, looked up by IRI."""
165 try:
166 payload = _request("terms", {"iri": iri, "size": 100})
167 except urllib.error.HTTPError as exc:
168 if exc.code == 404:
169 return []
170 raise OlsError(f"OLS returned HTTP {exc.code} for {iri}") from exc
171 return _all_terms(payload)
174def _all_terms(payload: dict) -> list[dict]:
175 """Consume a HAL term collection without silently truncating later pages."""
176 terms: list[dict] = []
177 seen: set[str] = set()
178 while True:
179 page_terms = (payload.get("_embedded") or {}).get("terms")
180 if page_terms is None and (payload.get("page") or {}).get("totalElements") == 0:
181 page_terms = []
182 if not isinstance(page_terms, list):
183 raise OlsError("OLS returned an invalid term collection")
184 terms.extend(page_terms)
185 href = payload.get("_links", {}).get("next", {}).get("href")
186 if not href:
187 return terms
188 if href in seen:
189 raise OlsError("OLS pagination repeated a next link")
190 seen.add(href)
191 try:
192 payload = _request_url(href)
193 except urllib.error.HTTPError as exc:
194 raise OlsError(f"OLS pagination failed: HTTP {exc.code}") from exc
197def term_detail(curie: str) -> dict | None:
198 """Fetch the authoritative copy of a term by CURIE, or None if unknown.
200 This is the only endpoint that reports ``is_obsolete`` and
201 ``term_replaced_by``, so validation must come through here.
203 The ``obo_id`` index is preferred. A verified IRI fallback also handles
204 missing index entries and Orphanet/ORPHA aliases. The returned dict carries ``_resolved_via`` and
205 ``_home_ontology`` so callers can tell the two paths apart.
206 """
207 ontology = curie_to_ontology_id(curie)
208 if not ontology:
209 return None
211 try:
212 payload = _request(f"ontologies/{ontology}/terms", {"obo_id": curie})
213 terms = _all_terms(payload)
214 if terms:
215 found = dict(terms[0])
216 found["_resolved_via"] = "obo_id"
217 found["_home_ontology"] = ontology
218 return found
219 except urllib.error.HTTPError as exc:
220 if exc.code != 404:
221 raise OlsError(f"OLS returned HTTP {exc.code} for {curie}") from exc
223 copies: list[dict] = []
224 for iri in candidate_iris(curie):
225 copies.extend(_terms_by_iri(iri))
226 if not copies:
227 return None
229 home = [c for c in copies if c.get("ontology_name") == ontology]
230 defining = [c for c in copies if c.get("is_defining_ontology")]
231 best = (
232 next((c for c in home if c.get("is_defining_ontology")), None)
233 or (home[0] if home else None)
234 or (defining[0] if defining else None)
235 or copies[0]
236 )
237 found = dict(best)
238 found["_resolved_via"] = "iri"
239 found["_home_ontology"] = ontology
240 return found
243def iri_for(curie: str) -> str | None:
244 """Resolve a CURIE to its IRI via the API rather than by string templating."""
245 term = term_detail(curie)
246 return term.get("iri") if term else None
249def ancestor_curies(curie: str, *, relation: str = "hierarchical") -> set[str]:
250 """Return ancestors; hierarchical also traverses part-of/develops-from.
252 ``relation='is-a'`` uses the subclass-only ``ancestors`` relation.
253 """
254 if relation not in {"hierarchical", "is-a"}:
255 raise ValueError("relation must be hierarchical or is-a")
256 term = term_detail(curie)
257 if not term:
258 raise OlsError(f"cannot check ancestors of unknown term {curie}")
259 href = (
260 term.get("_links", {})
261 .get("hierarchicalAncestors" if relation == "hierarchical" else "ancestors", {})
262 .get("href")
263 )
264 if not href:
265 raise OlsError(f"OLS omitted the ancestor link for {curie}")
266 url = f"{href}{'&' if '?' in href else '?'}size=500"
267 try:
268 terms = _all_terms(_request_url(url))
269 except urllib.error.HTTPError as exc:
270 raise OlsError(f"ancestor lookup failed for {curie}: HTTP {exc.code}") from exc
271 return {entry.get("obo_id") or iri_to_curie(entry.get("iri", ""))
272 for entry in terms} - {None, ""}
275# --- pure helpers -----------------------------------------------------------
277CURIE_RE = re.compile(r"^([A-Za-z][A-Za-z0-9_.]*):([A-Za-z0-9_.\-]+)$")
280def is_curie(value: str) -> bool:
281 """True for ``PREFIX:local`` strings, false for IRIs, labels, and junk."""
282 return bool(CURIE_RE.match(value.strip()))
285def curie_to_ontology_id(curie: str) -> str | None:
286 """Map a CURIE to the OLS ontology id that defines it.
288 Lowercasing the prefix is right for almost every ontology; the exceptions
289 live in ``ONTOLOGY_ID_OVERRIDES`` (``Orphanet:558`` is served by ``ordo``).
290 """
291 match = CURIE_RE.match(curie.strip())
292 if not match:
293 return None
294 prefix = match.group(1).lower()
295 return ONTOLOGY_ID_OVERRIDES.get(prefix, prefix)
298def iri_to_curie(iri: str) -> str | None:
299 """Convert a recognized namespace IRI to a CURIE; unknown IRIs stay unknown.
301 Handles the three IRI shapes in use -- OBO PURLs, EFO's own namespace, and
302 Orphanet's -- plus multi-underscore prefixes such as ``APOLLO_SV_00000001``.
303 """
304 if not isinstance(iri, str) or not iri:
305 return None
306 namespaces = ("http://purl.obolibrary.org/obo/", "http://www.ebi.ac.uk/efo/EFO_",
307 "http://www.orpha.net/ORDO/Orphanet_")
308 namespace = next((value for value in namespaces if iri.startswith(value)), None)
309 if not namespace or "/" in iri[len(namespace):]:
310 return None
311 tail = iri.rsplit("/", 1)[-1]
312 if "_" not in tail:
313 return None
314 prefix, local = tail.rsplit("_", 1)
315 if not prefix or not local:
316 return None
317 candidate = f"{prefix}:{local}"
318 return candidate if is_curie(candidate) else None
321def normalize_label(text: str) -> str:
322 """Fold case and whitespace for label comparison, changing nothing else.
324 Deliberately conservative: hyphens, Greek letters, and digits carry meaning
325 in ontology labels, so only case and spacing are normalised.
326 """
327 return re.sub(r"\s+", " ", text.strip()).casefold()
330def synonyms_of(doc: dict) -> list[str]:
331 """Collect synonyms from a search doc or a term-detail doc.
333 ``/search`` returns them under ``synonym`` when requested via ``fieldList``
334 and under ``exact_synonyms`` / ``related_synonyms`` otherwise; term detail
335 uses ``synonyms``.
336 """
337 collected: list[str] = []
338 for key in ("synonym", "synonyms", "exact_synonyms", "related_synonyms", "broad_synonyms", "narrow_synonyms"):
339 value = doc.get(key)
340 if isinstance(value, str):
341 collected.append(value)
342 elif isinstance(value, Iterable):
343 collected.extend(str(item) for item in value)
344 return collected
347def synonym_scope(query: str, doc: dict) -> str | None:
348 """Retain semantic scope rather than treating every lexical synonym as exact."""
349 wanted = normalize_label(query)
350 for scope in ("exact", "related", "broad", "narrow"):
351 values = doc.get(f"{scope}_synonyms") or []
352 if isinstance(values, str):
353 values = [values]
354 values = list(values) + [entry.get("name", "") for entry in (doc.get("obo_synonym") or [])
355 if entry.get("scope") == f"has{scope.title()}Synonym"]
356 if any(normalize_label(value) == wanted for value in values):
357 return scope
358 return "unspecified" if any(normalize_label(s) == wanted for s in synonyms_of(doc)) else None
361def match_type(query: str, doc: dict) -> str:
362 """Classify a lexical hit and retain its synonym relation scope.
364 OLS ranks partial hits alongside exact ones, so the caller -- not the
365 server -- decides whether a match is exact.
366 """
367 target = normalize_label(query)
368 if normalize_label(doc.get("label") or "") == target:
369 return "exact_label"
370 scope = synonym_scope(query, doc)
371 if scope:
372 return "exact_synonym" if scope == "exact" else f"{scope}_synonym"
373 return "partial"
376def dedupe_candidates(docs: list[dict]) -> list[dict]:
377 """Collapse repeats of the same term, keeping the defining ontology's copy.
379 A search for ``liver`` returns ``UBERON:0002107`` once per ontology that
380 imports it. Only the copy with ``is_defining_ontology`` is canonical.
381 """
382 best: dict[str, dict] = {}
383 order: list[str] = []
384 for doc in docs:
385 key = doc.get("obo_id") or doc.get("iri") or doc.get("short_form")
386 if not key:
387 continue
388 if key not in best:
389 best[key] = doc
390 order.append(key)
391 elif doc.get("is_defining_ontology") and not best[key].get("is_defining_ontology"):
392 best[key] = doc
393 return [best[key] for key in order]
396def rank_candidates(query: str, docs: list[dict]) -> list[dict]:
397 """Annotate hits with ``match_type`` and sort exact matches to the front.
399 Within a match tier the server's relevance order is preserved, and terms
400 from their defining ontology outrank imported copies.
401 """
402 tier = {"exact_label": 0, "exact_synonym": 1, "related_synonym": 2,
403 "broad_synonym": 2, "narrow_synonym": 2, "unspecified_synonym": 2, "partial": 3}
404 annotated = []
405 for position, doc in enumerate(dedupe_candidates(docs)):
406 enriched = dict(doc)
407 enriched["match_type"] = match_type(query, doc)
408 annotated.append((tier[enriched["match_type"]], 0 if doc.get("is_defining_ontology") else 1, position, enriched))
409 annotated.sort(key=lambda row: row[:3])
410 return [row[3] for row in annotated]