From 92924701d2769c98de66765f4714d63a6d1c2a37 Mon Sep 17 00:00:00 2001 From: alex-omophub Date: Fri, 10 Apr 2026 21:39:40 +0100 Subject: [PATCH 1/3] Add FHIR resolver example script - Introduced a new script `fhir_resolver.py` that demonstrates FHIR-to-OMOP concept resolution using the OMOPHub SDK. - The script includes examples for resolving SNOMED, ICD-10-CM, LOINC, RxNorm codes, and text-only semantic searches. - Added functionality for direct standard lookups, non-standard code mapping, and Phoebe recommendations. - Supports batch resolution and asynchronous usage for improved performance. --- examples/fhir_resolver.py | 509 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 509 insertions(+) create mode 100644 examples/fhir_resolver.py diff --git a/examples/fhir_resolver.py b/examples/fhir_resolver.py new file mode 100644 index 0000000..713b423 --- /dev/null +++ b/examples/fhir_resolver.py @@ -0,0 +1,509 @@ +#!/usr/bin/env python3 +"""Examples of FHIR-to-OMOP concept resolution using the OMOPHub SDK. + +The FHIR resolver translates FHIR coded values (system URI + code) into +OMOP standard concepts, CDM target tables, and optional Phoebe +recommendations — all in a single API call. + +Covers: + - Direct standard lookups (SNOMED, LOINC, RxNorm) + - Non-standard code mapping (ICD-10-CM → SNOMED via "Maps to") + - Text-only semantic search fallback + - Vocabulary ID bypass (skip URI resolution) + - Phoebe recommendations and mapping quality signal + - Batch resolution (up to 100 codings) + - CodeableConcept resolution with vocabulary preference + - Async usage +""" + +import asyncio + +import omophub + +# --------------------------------------------------------------------------- +# 1. Direct SNOMED resolution +# --------------------------------------------------------------------------- + + +def resolve_snomed() -> None: + """Resolve a SNOMED CT code directly to its OMOP concept.""" + print("=== 1. SNOMED Direct Resolution ===") + + client = omophub.OMOPHub(api_key="oh_your_api_key") + try: + result = client.fhir.resolve( + system="http://snomed.info/sct", + code="44054006", + resource_type="Condition", + ) + res = result["resolution"] + print(f" Source: {res['source_concept']['concept_name']}") + print(f" Standard: {res['standard_concept']['concept_name']}") + print(f" Mapping type: {res['mapping_type']}") # "direct" + print(f" Target table: {res['target_table']}") # "condition_occurrence" + print(f" Alignment: {res['domain_resource_alignment']}") # "aligned" + except omophub.OMOPHubError as e: + print(f" Error: {e.message}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 2. ICD-10-CM mapped to SNOMED via "Maps to" +# --------------------------------------------------------------------------- + + +def resolve_icd10_mapped() -> None: + """Resolve a non-standard ICD-10-CM code — automatically traverses Maps to.""" + print("\n=== 2. ICD-10-CM → SNOMED Mapping ===") + + client = omophub.OMOPHub(api_key="oh_your_api_key") + try: + result = client.fhir.resolve( + system="http://hl7.org/fhir/sid/icd-10-cm", + code="E11.9", + resource_type="Condition", + ) + res = result["resolution"] + print( + f" Source: [{res['source_concept']['vocabulary_id']}] {res['source_concept']['concept_name']}" + ) + print( + f" Standard: [{res['standard_concept']['vocabulary_id']}] {res['standard_concept']['concept_name']}" + ) + print(f" Mapping type: {res['mapping_type']}") # "mapped" + print(f" Relationship: {res.get('relationship_id', 'N/A')}") # "Maps to" + print(f" Target table: {res['target_table']}") # "condition_occurrence" + except omophub.OMOPHubError as e: + print(f" Error: {e.message}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 3. LOINC direct resolution → measurement table +# --------------------------------------------------------------------------- + + +def resolve_loinc() -> None: + """Resolve a LOINC lab code to the OMOP measurement table.""" + print("\n=== 3. LOINC → Measurement ===") + + client = omophub.OMOPHub(api_key="oh_your_api_key") + try: + result = client.fhir.resolve( + system="http://loinc.org", + code="2339-0", # Glucose [Mass/volume] in Blood + resource_type="Observation", + ) + res = result["resolution"] + print(f" Concept: {res['standard_concept']['concept_name']}") + print(f" Domain: {res['standard_concept']['domain_id']}") # "Measurement" + print(f" Target table: {res['target_table']}") # "measurement" + print(f" Alignment: {res['domain_resource_alignment']}") # "aligned" + except omophub.OMOPHubError as e: + print(f" Error: {e.message}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 4. RxNorm direct resolution → drug_exposure table +# --------------------------------------------------------------------------- + + +def resolve_rxnorm() -> None: + """Resolve an RxNorm drug code to the OMOP drug_exposure table.""" + print("\n=== 4. RxNorm → Drug Exposure ===") + + client = omophub.OMOPHub(api_key="oh_your_api_key") + try: + result = client.fhir.resolve( + system="http://www.nlm.nih.gov/research/umls/rxnorm", + code="197696", # Acetaminophen 325 MG Oral Tablet + resource_type="MedicationRequest", + ) + res = result["resolution"] + print(f" Concept: {res['standard_concept']['concept_name']}") + print(f" Target table: {res['target_table']}") # "drug_exposure" + except omophub.OMOPHubError as e: + print(f" Error: {e.message}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 5. Text-only resolution (semantic search fallback) +# --------------------------------------------------------------------------- + + +def resolve_text_only() -> None: + """Resolve using only display text — triggers BioLORD semantic search.""" + print("\n=== 5. Text-Only Semantic Fallback ===") + + client = omophub.OMOPHub(api_key="oh_your_api_key") + try: + # No system or code — just natural language text + result = client.fhir.resolve( + display="Blood Sugar", + resource_type="Observation", + ) + res = result["resolution"] + print(f" Matched: {res['standard_concept']['concept_name']}") + print(f" Mapping type: {res['mapping_type']}") # "semantic_match" + print(f" Similarity: {res.get('similarity_score', 'N/A')}") + print(f" Target table: {res['target_table']}") # "measurement" + except omophub.OMOPHubError as e: + print(f" Error: {e.message}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 6. Direct vocabulary_id bypass (skip URI resolution) +# --------------------------------------------------------------------------- + + +def resolve_vocabulary_id_bypass() -> None: + """Use vocabulary_id directly when you already know the OMOP vocabulary.""" + print("\n=== 6. Vocabulary ID Bypass ===") + + client = omophub.OMOPHub(api_key="oh_your_api_key") + try: + # Skip URI resolution — go straight to the vocabulary + result = client.fhir.resolve( + vocabulary_id="ICD10CM", + code="E11.9", + ) + res = result["resolution"] + print(f" Vocabulary: {res['vocabulary_id']}") # "ICD10CM" + print(f" Standard: {res['standard_concept']['concept_name']}") + except omophub.OMOPHubError as e: + print(f" Error: {e.message}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 7. Include Phoebe recommendations +# --------------------------------------------------------------------------- + + +def resolve_with_recommendations() -> None: + """Get Phoebe-recommended related concepts alongside the resolution.""" + print("\n=== 7. With Phoebe Recommendations ===") + + client = omophub.OMOPHub(api_key="oh_your_api_key") + try: + result = client.fhir.resolve( + system="http://snomed.info/sct", + code="44054006", + include_recommendations=True, + recommendations_limit=5, + ) + res = result["resolution"] + print(f" Resolved: {res['standard_concept']['concept_name']}") + + recs = res.get("recommendations", []) + print(f" Recommendations ({len(recs)}):") + for rec in recs: + print( + f" - {rec['concept_name']} ({rec['domain_id']}) via {rec['relationship_id']}" + ) + except omophub.OMOPHubError as e: + print(f" Error: {e.message}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 8. Include mapping quality signal +# --------------------------------------------------------------------------- + + +def resolve_with_quality() -> None: + """Get a mapping quality signal to triage which resolutions need review.""" + print("\n=== 8. With Mapping Quality ===") + + client = omophub.OMOPHub(api_key="oh_your_api_key") + try: + # Direct SNOMED match → "high" quality + result = client.fhir.resolve( + system="http://snomed.info/sct", + code="44054006", + include_quality=True, + ) + print( + f" SNOMED direct → quality: {result['resolution'].get('mapping_quality')}" + ) + + # ICD-10 mapped → quality from validation + result = client.fhir.resolve( + system="http://hl7.org/fhir/sid/icd-10-cm", + code="E11.9", + include_quality=True, + ) + print( + f" ICD-10 mapped → quality: {result['resolution'].get('mapping_quality')}" + ) + + # Text-only semantic → "medium" quality + result = client.fhir.resolve( + display="heart attack", + resource_type="Condition", + include_quality=True, + ) + quality = result["resolution"].get("mapping_quality") + note = result["resolution"].get("quality_note", "") + print(f" Text semantic → quality: {quality}") + if note: + print(f" Note: {note}") + except omophub.OMOPHubError as e: + print(f" Error: {e.message}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 9. Batch resolution +# --------------------------------------------------------------------------- + + +def resolve_batch() -> None: + """Resolve multiple codings in a single call with per-item error reporting.""" + print("\n=== 9. Batch Resolution ===") + + client = omophub.OMOPHub(api_key="oh_your_api_key") + try: + result = client.fhir.resolve_batch( + [ + {"system": "http://snomed.info/sct", "code": "44054006"}, + {"system": "http://loinc.org", "code": "2339-0"}, + { + "system": "http://www.nlm.nih.gov/research/umls/rxnorm", + "code": "197696", + }, + {"system": "http://hl7.org/fhir/sid/icd-10-cm", "code": "E11.9"}, + { + "system": "http://snomed.info/sct", + "code": "99999999-invalid", + }, # Will fail + ], + resource_type="Condition", + include_quality=True, + ) + + summary = result["summary"] + print( + f" Total: {summary['total']}, Resolved: {summary['resolved']}, Failed: {summary['failed']}" + ) + + for i, item in enumerate(result["results"]): + if "resolution" in item: + res = item["resolution"] + name = res["standard_concept"]["concept_name"] + table = res["target_table"] + quality = res.get("mapping_quality", "N/A") + print(f" [{i + 1}] {name} → {table} (quality: {quality})") + else: + code = item.get("input", {}).get("code", "?") + error = item["error"]["code"] + print(f" [{i + 1}] FAILED code={code}: {error}") + except omophub.OMOPHubError as e: + print(f" Error: {e.message}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 10. CodeableConcept resolution (vocabulary preference) +# --------------------------------------------------------------------------- + + +def resolve_codeable_concept() -> None: + """Resolve a CodeableConcept with multiple codings — SNOMED wins by preference.""" + print("\n=== 10. CodeableConcept Resolution ===") + + client = omophub.OMOPHub(api_key="oh_your_api_key") + try: + result = client.fhir.resolve_codeable_concept( + coding=[ + { + "system": "http://hl7.org/fhir/sid/icd-10-cm", + "code": "E11.9", + "display": "Type 2 diabetes mellitus without complications", + }, + { + "system": "http://snomed.info/sct", + "code": "44054006", + "display": "Type 2 diabetes mellitus", + }, + ], + resource_type="Condition", + include_recommendations=True, + recommendations_limit=3, + ) + + best = result["best_match"] + if best: + res = best["resolution"] + print( + f" Best match: [{res['source_concept']['vocabulary_id']}] {res['standard_concept']['concept_name']}" + ) + print(f" Target table: {res['target_table']}") + + recs = res.get("recommendations", []) + if recs: + print(f" Recommendations ({len(recs)}):") + for r in recs: + print(f" - {r['concept_name']}") + + print(f" Alternatives: {len(result['alternatives'])}") + for alt in result["alternatives"]: + alt_res = alt["resolution"] + print( + f" [{alt_res['source_concept']['vocabulary_id']}] {alt_res['standard_concept']['concept_name']}" + ) + + if result["unresolved"]: + print(f" Unresolved: {len(result['unresolved'])}") + except omophub.OMOPHubError as e: + print(f" Error: {e.message}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 11. CodeableConcept with text fallback +# --------------------------------------------------------------------------- + + +def resolve_codeable_concept_text_fallback() -> None: + """When no structured coding resolves, fall back to the text field.""" + print("\n=== 11. CodeableConcept Text Fallback ===") + + client = omophub.OMOPHub(api_key="oh_your_api_key") + try: + result = client.fhir.resolve_codeable_concept( + coding=[ + # This code doesn't exist — will fail + {"system": "http://loinc.org", "code": "99999-9"}, + ], + text="Type 2 diabetes mellitus", # Fallback via semantic search + resource_type="Condition", + ) + + best = result["best_match"] + if best: + res = best["resolution"] + print(f" Resolved via: {res['mapping_type']}") # "semantic_match" + print(f" Concept: {res['standard_concept']['concept_name']}") + print(f" Similarity: {res.get('similarity_score', 'N/A')}") + else: + print(" No resolution found") + + print(f" Failed codings: {len(result['unresolved'])}") + for fail in result["unresolved"]: + print(f" {fail['error']['code']}: {fail['input'].get('code', '?')}") + except omophub.OMOPHubError as e: + print(f" Error: {e.message}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 12. Async usage +# --------------------------------------------------------------------------- + + +async def async_resolve() -> None: + """Demonstrate async FHIR resolution with concurrent requests.""" + print("\n=== 12. Async FHIR Resolution ===") + + async with omophub.AsyncOMOPHub(api_key="oh_your_api_key") as client: + # Single resolve + result = await client.fhir.resolve( + system="http://snomed.info/sct", + code="44054006", + resource_type="Condition", + ) + print(f" Single: {result['resolution']['standard_concept']['concept_name']}") + + # Concurrent: batch + single resolve in parallel + batch_task = client.fhir.resolve_batch( + [ + {"system": "http://loinc.org", "code": "2339-0"}, + { + "system": "http://www.nlm.nih.gov/research/umls/rxnorm", + "code": "197696", + }, + ] + ) + single_task = client.fhir.resolve( + system="http://hl7.org/fhir/sid/icd-10-cm", + code="E11.9", + ) + + batch_result, single_result = await asyncio.gather(batch_task, single_task) + + print( + f" Batch: {batch_result['summary']['resolved']}/{batch_result['summary']['total']} resolved" + ) + print( + f" Single: {single_result['resolution']['standard_concept']['concept_name']}" + ) + + +# --------------------------------------------------------------------------- +# Error handling examples +# --------------------------------------------------------------------------- + + +def error_handling_examples() -> None: + """Demonstrate error responses from the FHIR resolver.""" + print("\n=== Error Handling Examples ===") + + client = omophub.OMOPHub(api_key="oh_your_api_key") + try: + # Unknown code system URI → 400 with suggestion + print(" Typo in URI:") + try: + client.fhir.resolve(system="http://snomed.info/sc", code="44054006") + except omophub.OMOPHubError as e: + print(f" {e.status_code}: {e.message}") + + # Restricted vocabulary (CPT4) → 403 + print(" CPT4 restricted:") + try: + client.fhir.resolve(system="http://www.ama-assn.org/go/cpt", code="99213") + except omophub.OMOPHubError as e: + print(f" {e.status_code}: {e.message}") + + # Code not found → 404 + print(" Non-existent code:") + try: + client.fhir.resolve(system="http://snomed.info/sct", code="00000000") + except omophub.OMOPHubError as e: + print(f" {e.status_code}: {e.message}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + +if __name__ == "__main__": + resolve_snomed() + resolve_icd10_mapped() + resolve_loinc() + resolve_rxnorm() + resolve_text_only() + resolve_vocabulary_id_bypass() + resolve_with_recommendations() + resolve_with_quality() + resolve_batch() + resolve_codeable_concept() + resolve_codeable_concept_text_fallback() + asyncio.run(async_resolve()) + error_handling_examples() From e618f8d7e61c236c217bd975236dd7ac2b5247f5 Mon Sep 17 00:00:00 2001 From: alex-omophub Date: Tue, 14 Apr 2026 19:23:00 +0100 Subject: [PATCH 2/3] Implement FHIR type interoperability and client helpers - Added support for FHIR type interoperability in the resolver, allowing acceptance of various Coding-like inputs (dicts, TypedDicts, and objects with .system/.code attributes). - Introduced new FHIR type definitions (`Coding`, `CodeableConcept`) and runtime-checkable protocols (`CodingLike`, `CodeableConceptLike`) for structural matching. - Implemented FHIR client interop helpers to configure external libraries (`fhirpy`, `fhir.resources`) with OMOPHub's FHIR Terminology Service. - Enhanced `Fhir.resolve`, `Fhir.resolve_batch`, and `Fhir.resolve_codeable_concept` methods to accept mixed input types. - Updated documentation and examples to reflect new functionalities and usage patterns. - Added comprehensive unit and integration tests to validate new features and ensure compatibility. --- CHANGELOG.md | 77 +++++++ README.md | 68 ++++++ examples/fhir_interop.py | 346 ++++++++++++++++++++++++++++++ pyproject.toml | 5 + src/omophub/__init__.py | 9 + src/omophub/_client.py | 23 ++ src/omophub/fhir_interop.py | 93 ++++++++ src/omophub/resources/fhir.py | 175 ++++++++++++--- src/omophub/types/__init__.py | 8 + src/omophub/types/fhir.py | 43 +++- tests/integration/test_fhir.py | 81 +++++++ tests/unit/resources/test_fhir.py | 197 +++++++++++++++++ tests/unit/test_fhir_interop.py | 116 ++++++++++ 13 files changed, 1215 insertions(+), 26 deletions(-) create mode 100644 examples/fhir_interop.py create mode 100644 src/omophub/fhir_interop.py create mode 100644 tests/unit/test_fhir_interop.py diff --git a/CHANGELOG.md b/CHANGELOG.md index dd305b1..6e3b682 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,83 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.7.0] - 2026-04-14 + +### Added + +- **FHIR type interoperability** (`client.fhir.*`): The resolver now + accepts any Coding-like input via duck typing, so pipelines that + already parse FHIR with `fhir.resources` or `fhirpy` can pass those + objects directly - no manual conversion. + - New `coding=` kwarg on `Fhir.resolve` / `AsyncFhir.resolve`. Accepts + a plain dict, the `omophub.types.fhir.Coding` TypedDict, or any + object exposing `.system` / `.code` / `.display` attributes (e.g. + `fhir.resources.Coding`, `fhirpy` codings). + - `Fhir.resolve_batch` / `AsyncFhir.resolve_batch` now accept a + mixed list of dicts and Coding-like objects; each item is + normalized to the wire format automatically. + - `Fhir.resolve_codeable_concept` / async counterpart now accept + either a list of Coding-likes (existing) or a full + CodeableConcept-like object exposing `.coding` and `.text` + (new); explicit `text=` kwarg still wins when both are passed. + - Explicit `system` / `code` kwargs override the corresponding + fields on a `coding=` input - handy for last-mile overrides. +- **New FHIR type definitions** in `omophub.types.fhir`: + - `Coding` and `CodeableConcept` lightweight TypedDicts + - `CodingLike` and `CodeableConceptLike` runtime-checkable + `Protocol`s for structural matching against external libraries. +- **FHIR client interop helpers** (`omophub.fhir_interop`): Thin + helpers for configuring an external FHIR client library against + the OMOPHub FHIR Terminology Service. + - `get_fhir_server_url(version)` returns the FHIR base URL for + `"r4"` (default), `"r4b"`, `"r5"`, or `"r6"`. + - `get_fhirpy_client(api_key, version)` and + `get_async_fhirpy_client(api_key, version)` return `fhirpy` + clients pre-wired with the right URL and `Authorization: Bearer` + header. `fhirpy` is imported lazily, so it is never a required + dependency; a helpful `ImportError` with install instructions is + raised only when you actually call the helper. + - All three helpers are re-exported from the top-level `omophub` + namespace. +- **`OMOPHub.fhir_server_url` / `AsyncOMOPHub.fhir_server_url`**: + Convenience read-only property returning the R4 FHIR endpoint for + drop-in use with external FHIR clients (`httpx`, `fhirpy`, + `fhir.resources`). +- **Optional extras** in `pyproject.toml`: + - `pip install omophub[fhirpy]` pulls in `fhirpy>=1.4.0`. + - `pip install omophub[fhir-resources]` pulls in + `fhir.resources>=7.0.0`. + Both are purely optional; duck typing means neither is required + for core SDK use. + +### Changed + +- `Fhir.resolve_batch` signature broadened from + `codings: list[dict[str, str | None]]` to + `codings: list[CodingInput]` where `CodingInput` is the union of + dict, `Coding` TypedDict, and `CodingLike` protocol. Existing + call sites keep working unchanged. +- `Fhir.resolve_codeable_concept` signature broadened from + `coding: list[dict[str, str]]` to + `coding: list[CodingInput] | CodeableConceptInput`, accepting + either the legacy list-of-codings shape or a full + CodeableConcept-like object. + +### Tests + +- 16 new unit tests covering `_extract_coding`, `_coding_to_dict`, + and duck-typed inputs across `resolve`, `resolve_batch`, and + `resolve_codeable_concept` on the sync client. Explicit-kwargs-win + precedence is covered. +- New `tests/unit/test_fhir_interop.py` with 10 cases: URL builder + for all four FHIR versions, `fhirpy` lazy import with stubbed + success and missing-module failure paths, and the + `fhir_server_url` property on both sync and async clients. +- 5 new integration tests in `tests/integration/test_fhir.py` + exercising the new `coding=` kwarg, mixed dict + duck-typed batch + inputs, CodeableConcept-like object resolution, and the + `fhir_server_url` property against the live API. + ## [1.6.0] - 2026-04-10 ### Added diff --git a/README.md b/README.md index 339bcae..9e9de57 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,10 @@ Working with OHDSI ATHENA vocabularies traditionally requires downloading multi- ```bash pip install omophub + +# Optional extras for FHIR client interop +pip install omophub[fhirpy] # Pre-wired fhirpy client +pip install omophub[fhir-resources] # Install marker for fhir.resources ``` ## Quick Start @@ -98,6 +102,70 @@ result = client.fhir.resolve_codeable_concept( print(result["best_match"]["resolution"]["source_concept"]["vocabulary_id"]) # "SNOMED" ``` +### Type Interoperability + +The resolver accepts any Coding-like input via duck typing - a plain dict, omophub's lightweight `Coding` TypedDict, or any object with `.system` / `.code` attributes (e.g. `fhir.resources.Coding`, `fhirpy` codings). + +```python +from omophub.types.fhir import Coding + +# omophub's TypedDict - IDE autocomplete, no extra deps +coding: Coding = {"system": "http://snomed.info/sct", "code": "44054006"} +result = client.fhir.resolve(coding=coding) + +# fhir.resources objects work via duck typing - no conversion needed +from fhir.resources.R4B.coding import Coding as FhirCoding +fhir_coding = FhirCoding(system="http://snomed.info/sct", code="44054006") +result = client.fhir.resolve(coding=fhir_coding) + +# Mixed shapes in a single batch call +result = client.fhir.resolve_batch([ + {"system": "http://snomed.info/sct", "code": "44054006"}, # dict + FhirCoding(system="http://loinc.org", code="2339-0"), # fhir.resources +]) +``` + +`fhir.resources` is **never** a required dependency. See [`examples/fhir_interop.py`](./examples/fhir_interop.py) for the full set of supported input shapes. + +### FHIR Client Interop + +Point external FHIR client libraries at OMOPHub's FHIR Terminology Service directly - useful when you need raw FHIR `Parameters` / `Bundle` responses instead of the Concept Resolver envelope. + +```python +from omophub import OMOPHub, get_fhir_server_url + +client = OMOPHub(api_key="oh_xxx") + +# Property on the client returns the R4 base URL +print(client.fhir_server_url) +# "https://fhir.omophub.com/fhir/r4" + +# Helper for other FHIR versions +print(get_fhir_server_url("r5")) +# "https://fhir.omophub.com/fhir/r5" +``` + +For `fhirpy`, install the optional extra and use the pre-wired client: + +```bash +pip install omophub[fhirpy] +``` + +```python +from omophub import get_fhirpy_client + +fhir = get_fhirpy_client("oh_xxx") + +# Call CodeSystem/$lookup directly via fhirpy +params = fhir.execute( + "CodeSystem/$lookup", + method="GET", + params={"system": "http://snomed.info/sct", "code": "44054006"}, +) +``` + +**When to use which**: the Concept Resolver (`client.fhir.resolve`) gives you OMOP-enriched answers - standard concept ID, CDM target table, mapping quality. Use `fhirpy` via `get_fhirpy_client()` when you need raw FHIR responses for FHIR-native tooling. + ## Semantic Search Use natural language queries to find concepts using neural embeddings: diff --git a/examples/fhir_interop.py b/examples/fhir_interop.py new file mode 100644 index 0000000..5e3fc3f --- /dev/null +++ b/examples/fhir_interop.py @@ -0,0 +1,346 @@ +#!/usr/bin/env python3 +"""FHIR type interop and client connection examples (OMOPHub SDK 1.7.0+). + +Two categories of functionality are shown here: + +1. **Type interoperability** - the resolver accepts any Coding-like input + via duck typing. You can pass a plain dict, omophub's lightweight + `Coding` TypedDict, or any object exposing `.system` / `.code` + attributes (e.g. `fhir.resources.Coding`, `fhirpy` codings). Neither + `fhir.resources` nor `fhirpy` is a required dependency - duck typing + handles them transparently. + +2. **Connection helpers** for external FHIR clients. If you need raw + FHIR `Parameters` / `Bundle` responses instead of the Concept + Resolver envelope, use `client.fhir_server_url` or + `get_fhirpy_client()` to talk to OMOPHub's FHIR Terminology Service + directly. + +For the full Concept Resolver surface (`resolve` / `resolve_batch` / +`resolve_codeable_concept`, recommendations, quality signals, async, +error handling) see `examples/fhir_resolver.py`. + +Scenarios 4 and 8 require optional extras: + + pip install omophub[fhir-resources] # for fhir.resources objects + pip install omophub[fhirpy] # for the pre-wired fhirpy client + +Both scenarios are guarded with try/except and skip cleanly when the +extra is not installed - this script runs end-to-end without them. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import TYPE_CHECKING + +import omophub + +if TYPE_CHECKING: + from omophub.types.fhir import Coding + +API_KEY = "oh_your_api_key" + + +# --------------------------------------------------------------------------- +# 1. coding= kwarg with a plain dict +# --------------------------------------------------------------------------- + + +def coding_kwarg_with_dict() -> None: + """Pass a plain dict via the new ``coding=`` kwarg.""" + print("=== 1. coding= kwarg with a plain dict ===") + + client = omophub.OMOPHub(api_key=API_KEY) + try: + result = client.fhir.resolve( + coding={ + "system": "http://snomed.info/sct", + "code": "44054006", + }, + resource_type="Condition", + ) + res = result["resolution"] + print(f" Standard: {res['standard_concept']['concept_name']}") + print(f" Target table: {res['target_table']}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 2. coding= kwarg with omophub's Coding TypedDict +# --------------------------------------------------------------------------- + + +def coding_kwarg_with_typed_dict() -> None: + """Use omophub's lightweight `Coding` TypedDict for IDE autocomplete.""" + print("\n=== 2. coding= kwarg with omophub.types.fhir.Coding ===") + + client = omophub.OMOPHub(api_key=API_KEY) + try: + coding: Coding = { + "system": "http://loinc.org", + "code": "2339-0", + "display": "Glucose [Mass/volume] in Blood", + } + result = client.fhir.resolve(coding=coding) + res = result["resolution"] + print(f" Standard: {res['standard_concept']['concept_name']}") + print(f" Target table: {res['target_table']}") # "measurement" + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 3. Duck typing with a stand-in object (no external dep) +# --------------------------------------------------------------------------- + + +def coding_kwarg_with_duck_object() -> None: + """Pass any object with .system / .code / .display attributes. + + This is exactly how the resolver handles `fhir.resources.Coding` and + `fhirpy` codings under the hood - structural matching via `getattr`. + No isinstance checks, no conversion required. + """ + print("\n=== 3. coding= kwarg with a duck-typed object ===") + + client = omophub.OMOPHub(api_key=API_KEY) + try: + # SimpleNamespace stands in for any Coding-like class - fhir.resources, + # fhirpy, or your own domain model. + fake_coding = SimpleNamespace( + system="http://snomed.info/sct", + code="44054006", + display="Type 2 diabetes mellitus", + ) + result = client.fhir.resolve(coding=fake_coding) + res = result["resolution"] + print(f" Standard: {res['standard_concept']['concept_name']}") + print(f" Concept ID: {res['standard_concept']['concept_id']}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 4. Real fhir.resources interop (optional extra) +# --------------------------------------------------------------------------- + + +def coding_kwarg_with_fhir_resources() -> None: + """Interop with the real `fhir.resources` library via duck typing. + + Requires: pip install omophub[fhir-resources] + """ + print("\n=== 4. coding= kwarg with fhir.resources objects ===") + + try: + from fhir.resources.R4B.codeableconcept import ( + CodeableConcept as FhirCodeableConcept, + ) + from fhir.resources.R4B.coding import Coding as FhirCoding + except ImportError: + print(" Skipped: pip install omophub[fhir-resources]") + return + + client = omophub.OMOPHub(api_key=API_KEY) + try: + # Single Coding from fhir.resources + snomed = FhirCoding( + system="http://snomed.info/sct", + code="44054006", + display="Type 2 diabetes mellitus", + ) + result = client.fhir.resolve(coding=snomed, resource_type="Condition") + print( + f" From FhirCoding: {result['resolution']['standard_concept']['concept_name']}" + ) + + # Full CodeableConcept with two codings - SNOMED wins on preference + cc = FhirCodeableConcept( + coding=[ + FhirCoding( + system="http://hl7.org/fhir/sid/icd-10-cm", + code="E11.9", + display="Type 2 diabetes mellitus without complications", + ), + FhirCoding( + system="http://snomed.info/sct", + code="44054006", + display="Type 2 diabetes mellitus", + ), + ], + text="Type 2 diabetes mellitus", + ) + cc_result = client.fhir.resolve_codeable_concept(cc, resource_type="Condition") + best = cc_result["best_match"]["resolution"] + print( + f" From FhirCodeableConcept best match: [{best['source_concept']['vocabulary_id']}] {best['standard_concept']['concept_name']}" + ) + assert best["source_concept"]["vocabulary_id"] == "SNOMED", "SNOMED should win" + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 5. Mixed batch input (dict + TypedDict + duck object) +# --------------------------------------------------------------------------- + + +def mixed_batch_inputs() -> None: + """A single `resolve_batch` accepts heterogeneous coding shapes.""" + print("\n=== 5. resolve_batch with mixed input shapes ===") + + client = omophub.OMOPHub(api_key=API_KEY) + try: + typed: Coding = {"system": "http://loinc.org", "code": "2339-0"} + duck = SimpleNamespace( + system="http://hl7.org/fhir/sid/icd-10-cm", + code="E11.9", + display=None, + ) + + result = client.fhir.resolve_batch( + [ + {"system": "http://snomed.info/sct", "code": "44054006"}, # dict + typed, # TypedDict + duck, # duck-typed object + ], + ) + summary = result["summary"] + print(f" Resolved {summary['resolved']}/{summary['total']}") + for i, item in enumerate(result["results"], start=1): + if "resolution" in item: + std = item["resolution"]["standard_concept"] + print(f" [{i}] {std['concept_name']} ({std['vocabulary_id']})") + else: + print(f" [{i}] failed: {item['error']['code']}") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 6. Explicit kwargs override coding= fields +# --------------------------------------------------------------------------- + + +def explicit_kwargs_override_coding() -> None: + """Explicit `system` / `code` kwargs always win over a `coding=` input. + + Useful when you want to reuse an existing Coding-like object but + override a single field without rebuilding the whole object. + """ + print("\n=== 6. Explicit kwargs override coding= fields ===") + + client = omophub.OMOPHub(api_key=API_KEY) + try: + base = SimpleNamespace( + system="http://snomed.info/sct", + code="44054006", # Type 2 diabetes + display=None, + ) + # Override the code on the fly - explicit `code` wins. + result = client.fhir.resolve( + coding=base, + code="73211009", # Diabetes mellitus (parent concept) + ) + res = result["resolution"] + print(f" Resolved: {res['standard_concept']['concept_name']}") + print(" (override code 73211009 won over base.code 44054006)") + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 7. Connection helpers: fhir_server_url + get_fhir_server_url +# --------------------------------------------------------------------------- + + +def connection_helper_urls() -> None: + """Inspect the FHIR base URL for external client configuration.""" + print("\n=== 7. Connection helper URLs ===") + + from omophub import get_fhir_server_url + + client = omophub.OMOPHub(api_key=API_KEY) + try: + # Property on the client returns the R4 base URL + print(f" client.fhir_server_url = {client.fhir_server_url}") + + # Helper function supports r4 (default), r4b, r5, r6 + for version in ("r4", "r4b", "r5", "r6"): + print( + f" get_fhir_server_url({version!r:>5}) = {get_fhir_server_url(version)}" + ) + finally: + client.close() + + +# --------------------------------------------------------------------------- +# 8. Pre-wired fhirpy client (optional extra) +# --------------------------------------------------------------------------- + + +def fhirpy_client_helper() -> None: + """Use `fhirpy` directly against OMOPHub's FHIR Terminology Service. + + Use the Concept Resolver (`client.fhir.resolve`) when you want + OMOP-enriched answers - standard concept ID, CDM target table, + mapping quality. Use `fhirpy` via `get_fhirpy_client()` when you + need raw FHIR `Parameters` / `Bundle` responses for FHIR-native + tooling. + + Requires: pip install omophub[fhirpy] + """ + print("\n=== 8. get_fhirpy_client() for raw FHIR calls ===") + + try: + from omophub import get_fhirpy_client + except ImportError: + print(" Skipped: pip install omophub[fhirpy]") + return + + try: + fhir = get_fhirpy_client(API_KEY) + except ImportError as e: + print(f" Skipped: {e}") + return + + # Example: call CodeSystem/$lookup directly. + # fhirpy returns a FHIR Parameters resource. + try: + params = fhir.execute( + "CodeSystem/$lookup", + method="GET", + params={ + "system": "http://snomed.info/sct", + "code": "44054006", + }, + ) + display = next( + ( + p.get("valueString") + for p in params.get("parameter", []) + if p.get("name") == "display" + ), + None, + ) + print(f" $lookup display: {display}") + except Exception as e: # demo script - catch anything fhirpy raises + print(f" fhirpy call failed: {e}") + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + + +if __name__ == "__main__": + coding_kwarg_with_dict() + coding_kwarg_with_typed_dict() + coding_kwarg_with_duck_object() + coding_kwarg_with_fhir_resources() + mixed_batch_inputs() + explicit_kwargs_override_coding() + connection_helper_urls() + fhirpy_client_helper() diff --git a/pyproject.toml b/pyproject.toml index 5e8646c..ef14d5b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -62,6 +62,11 @@ dev = [ "mypy>=1.8.0", "python-dotenv>=1.0.0", # For loading .env files in integration tests ] +# External FHIR client libraries. Duck-typed by the resolver, so not +# required for core SDK use. Install via `pip install omophub[fhirpy]` +# or `pip install omophub[fhir-resources]` if you want these helpers. +fhirpy = ["fhirpy>=1.4.0"] +fhir-resources = ["fhir.resources>=7.0.0"] [project.urls] Homepage = "https://omophub.com" diff --git a/src/omophub/__init__.py b/src/omophub/__init__.py index 4c04a3e..c19084d 100644 --- a/src/omophub/__init__.py +++ b/src/omophub/__init__.py @@ -30,6 +30,11 @@ ValidationError, ) from ._version import __version__ +from .fhir_interop import ( + get_async_fhirpy_client, + get_fhir_server_url, + get_fhirpy_client, +) # Re-export commonly used types from .types import ( @@ -93,6 +98,10 @@ # Configuration "api_key", "api_url", + # FHIR interop helpers + "get_async_fhirpy_client", + "get_fhir_server_url", + "get_fhirpy_client", "max_retries", "timeout", ] diff --git a/src/omophub/_client.py b/src/omophub/_client.py index 338e857..40b5d84 100644 --- a/src/omophub/_client.py +++ b/src/omophub/_client.py @@ -107,6 +107,19 @@ def fhir(self) -> Fhir: self._fhir = Fhir(self._request) return self._fhir + @property + def fhir_server_url(self) -> str: + """URL of the OMOPHub FHIR Terminology Service (R4). + + Use this when configuring an external FHIR client library + (``fhirpy``, ``fhir.resources``, ``httpx``) to talk directly + to OMOPHub's FHIR endpoint. For other FHIR versions, call + :func:`omophub.fhir_interop.get_fhir_server_url` directly. + """ + from .fhir_interop import get_fhir_server_url + + return get_fhir_server_url("r4") + @property def concepts(self) -> Concepts: """Access the concepts resource.""" @@ -246,6 +259,16 @@ def fhir(self) -> AsyncFhir: self._fhir = AsyncFhir(self._request) return self._fhir + @property + def fhir_server_url(self) -> str: + """URL of the OMOPHub FHIR Terminology Service (R4). + + See :attr:`OMOPHub.fhir_server_url` for details. + """ + from .fhir_interop import get_fhir_server_url + + return get_fhir_server_url("r4") + @property def concepts(self) -> AsyncConcepts: """Access the concepts resource.""" diff --git a/src/omophub/fhir_interop.py b/src/omophub/fhir_interop.py new file mode 100644 index 0000000..ab94624 --- /dev/null +++ b/src/omophub/fhir_interop.py @@ -0,0 +1,93 @@ +"""FHIR client interop helpers for the OMOPHub FHIR Terminology Service. + +These helpers make it easy to point an external FHIR client library +(`fhirpy`, `fhir.resources`, or any httpx-based client) at OMOPHub's +FHIR endpoint. None of the optional FHIR client libraries are required +dependencies of the SDK; import errors are raised lazily when the +helper is actually called. + +Example: + >>> from omophub.fhir_interop import get_fhirpy_client + >>> fhir = get_fhirpy_client("oh_xxxxxxxxx") + >>> bundle = fhir.resources("CodeSystem").search( + ... url="http://snomed.info/sct" + ... ).fetch() +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Literal + +if TYPE_CHECKING: + FhirVersion = Literal["r4", "r4b", "r5", "r6"] + +_FHIR_BASE = "https://fhir.omophub.com/fhir" + + +def get_fhir_server_url(version: FhirVersion = "r4") -> str: + """Return the OMOPHub FHIR Terminology Service base URL. + + Args: + version: FHIR version prefix - one of ``"r4"`` (default), + ``"r4b"``, ``"r5"``, or ``"r6"``. + + Returns: + Full base URL, e.g. ``"https://fhir.omophub.com/fhir/r4"``. + """ + return f"{_FHIR_BASE}/{version}" + + +def get_fhirpy_client(api_key: str, version: FhirVersion = "r4") -> Any: + """Return a ``fhirpy.SyncFHIRClient`` pre-configured for OMOPHub. + + Requires the ``fhirpy`` package. Install with:: + + pip install omophub[fhirpy] + + Args: + api_key: OMOPHub API key (``oh_...``). + version: FHIR version prefix. + + Returns: + A ``SyncFHIRClient`` instance pointed at OMOPHub with the + ``Authorization: Bearer `` header pre-attached. + + Raises: + ImportError: if ``fhirpy`` is not installed. + """ + try: + from fhirpy import SyncFHIRClient # type: ignore[import-not-found] + except ImportError as e: # pragma: no cover - import guard + raise ImportError( + "fhirpy is required for FHIR client interop. " + "Install with: pip install omophub[fhirpy]" + ) from e + return SyncFHIRClient( + url=get_fhir_server_url(version), + authorization=f"Bearer {api_key}", + ) + + +def get_async_fhirpy_client(api_key: str, version: FhirVersion = "r4") -> Any: + """Return a ``fhirpy.AsyncFHIRClient`` pre-configured for OMOPHub. + + Async counterpart of :func:`get_fhirpy_client`. Same install hint. + """ + try: + from fhirpy import AsyncFHIRClient + except ImportError as e: # pragma: no cover - import guard + raise ImportError( + "fhirpy is required for FHIR client interop. " + "Install with: pip install omophub[fhirpy]" + ) from e + return AsyncFHIRClient( + url=get_fhir_server_url(version), + authorization=f"Bearer {api_key}", + ) + + +__all__ = [ + "get_async_fhirpy_client", + "get_fhir_server_url", + "get_fhirpy_client", +] diff --git a/src/omophub/resources/fhir.py b/src/omophub/resources/fhir.py index 5f8560e..29f70b9 100644 --- a/src/omophub/resources/fhir.py +++ b/src/omophub/resources/fhir.py @@ -7,11 +7,108 @@ if TYPE_CHECKING: from .._request import AsyncRequest, Request from ..types.fhir import ( + CodeableConcept, + CodeableConceptLike, + Coding, + CodingLike, FhirBatchResult, FhirCodeableConceptResult, FhirResolveResult, ) + CodingInput = Coding | CodingLike | dict[str, Any] + CodeableConceptInput = CodeableConcept | CodeableConceptLike | dict[str, Any] + + +def _extract_coding( + coding_input: object, +) -> tuple[str | None, str | None, str | None]: + """Normalize any Coding-like input to ``(system, code, display)``. + + Accepts plain dicts, the ``omophub.types.fhir.Coding`` TypedDict, + ``fhir.resources.Coding`` instances, ``fhirpy`` codings, or any user + object exposing ``.system`` / ``.code`` / ``.display`` attributes. + """ + if isinstance(coding_input, dict): + return ( + coding_input.get("system"), + coding_input.get("code"), + coding_input.get("display"), + ) + return ( + getattr(coding_input, "system", None), + getattr(coding_input, "code", None), + getattr(coding_input, "display", None), + ) + + +def _coding_to_dict(coding_input: object) -> dict[str, Any]: + """Convert any Coding-like input to a wire-format dict. + + Preserves only the keys the resolver endpoint understands. Skips keys + whose values are ``None`` so the request payload stays tight. + """ + if isinstance(coding_input, dict): + src: dict[str, Any] = coding_input + else: + src = { + "system": getattr(coding_input, "system", None), + "code": getattr(coding_input, "code", None), + "display": getattr(coding_input, "display", None), + "vocabulary_id": getattr(coding_input, "vocabulary_id", None), + } + return {k: v for k, v in src.items() if v is not None} + + +def _extract_codeable_concept( + cc_input: object, +) -> tuple[list[object], str | None]: + """Normalize any CodeableConcept-like input to ``(codings, text)``.""" + if isinstance(cc_input, dict): + codings = cc_input.get("coding") or [] + text = cc_input.get("text") + else: + codings = getattr(cc_input, "coding", None) or [] + text = getattr(cc_input, "text", None) + if not isinstance(codings, list): + codings = list(codings) + return codings, text + + +def _resolve_coding_kwargs( + coding: object | None, + system: str | None, + code: str | None, + display: str | None, +) -> tuple[str | None, str | None, str | None]: + """Merge explicit kwargs with an optional coding input. + + Explicit kwargs always win. Missing kwargs fall back to values + extracted from ``coding``. + """ + if coding is None: + return system, code, display + ext_system, ext_code, ext_display = _extract_coding(coding) + return ( + system if system is not None else ext_system, + code if code is not None else ext_code, + display if display is not None else ext_display, + ) + + +def _normalize_codeable_concept_input( + coding_input: object, +) -> tuple[list[object], str | None]: + """Accept either a list of Coding-likes or a CodeableConcept-like. + + A bare list is treated as the ``coding`` array with no ``text`` + field. Any other object is passed through + :func:`_extract_codeable_concept`. + """ + if isinstance(coding_input, list): + return coding_input, None + return _extract_codeable_concept(coding_input) + def _build_resolve_body( *, @@ -69,6 +166,7 @@ def resolve( system: str | None = None, code: str | None = None, display: str | None = None, + coding: CodingInput | None = None, vocabulary_id: str | None = None, resource_type: str | None = None, include_recommendations: bool = False, @@ -77,13 +175,20 @@ def resolve( ) -> FhirResolveResult: """Resolve a single FHIR Coding to an OMOP standard concept. - Provide at least one of (``system`` + ``code``), - (``vocabulary_id`` + ``code``), or ``display``. + Provide either explicit ``system`` + ``code`` kwargs, a ``coding`` + input (dict, :class:`~omophub.types.fhir.Coding` TypedDict, or any + ``fhir.resources`` / ``fhirpy`` coding object with ``.system`` + / ``.code`` attributes), or ``vocabulary_id`` + ``code``, or + ``display``. + + When both ``coding`` and explicit ``system``/``code`` are passed, + the explicit kwargs take precedence. Args: system: FHIR code system URI (e.g. ``http://snomed.info/sct``) code: Code value from the FHIR Coding display: Human-readable text (semantic search fallback) + coding: Any Coding-like input - dict, TypedDict, or object vocabulary_id: Direct OMOP vocabulary_id, bypasses URI resolution resource_type: FHIR resource type for domain alignment check include_recommendations: Include Phoebe recommendations @@ -94,10 +199,13 @@ def resolve( Resolution result with source concept, standard concept, target CDM table, and optional enrichments. """ + resolved_system, resolved_code, resolved_display = _resolve_coding_kwargs( + coding, system, code, display + ) body = _build_resolve_body( - system=system, - code=code, - display=display, + system=resolved_system, + code=resolved_code, + display=resolved_display, vocabulary_id=vocabulary_id, resource_type=resource_type, include_recommendations=include_recommendations, @@ -108,7 +216,7 @@ def resolve( def resolve_batch( self, - codings: list[dict[str, str | None]], + codings: list[CodingInput], *, resource_type: str | None = None, include_recommendations: bool = False, @@ -120,8 +228,12 @@ def resolve_batch( Failed items are reported inline without failing the batch. Args: - codings: List of coding dicts, each with optional keys - ``system``, ``code``, ``display``, ``vocabulary_id``. + codings: List of Coding-like inputs. Each item may be a plain + dict with ``system`` / ``code`` / ``display`` / + ``vocabulary_id`` keys, an + :class:`~omophub.types.fhir.Coding` TypedDict, or any + object exposing ``.system`` / ``.code`` attributes + (e.g. ``fhir.resources.Coding``, ``fhirpy`` codings). resource_type: FHIR resource type applied to all codings include_recommendations: Include Phoebe recommendations recommendations_limit: Max recommendations per item (1-20) @@ -130,7 +242,7 @@ def resolve_batch( Returns: Batch result with per-item results and a summary. """ - body: dict[str, Any] = {"codings": codings} + body: dict[str, Any] = {"codings": [_coding_to_dict(c) for c in codings]} if resource_type is not None: body["resource_type"] = resource_type if include_recommendations: @@ -142,7 +254,7 @@ def resolve_batch( def resolve_codeable_concept( self, - coding: list[dict[str, str]], + coding: list[CodingInput] | CodeableConceptInput, *, text: str | None = None, resource_type: str | None = None, @@ -158,9 +270,14 @@ def resolve_codeable_concept( coding resolves. Args: - coding: List of structured codings, each with ``system``, - ``code``, and optional ``display``. - text: CodeableConcept.text for semantic search fallback + coding: Either a list of Coding-like inputs (same shapes + accepted by :meth:`resolve_batch`), or a full + CodeableConcept-like object exposing ``.coding`` + and ``.text`` (``omophub.types.fhir.CodeableConcept``, + ``fhir.resources.CodeableConcept``, or a dict). + text: CodeableConcept.text for semantic search fallback. + When ``coding`` is a CodeableConcept-like object, this + kwarg overrides its ``text`` field if provided. resource_type: FHIR resource type for domain alignment include_recommendations: Include Phoebe recommendations recommendations_limit: Max recommendations (1-20) @@ -170,9 +287,11 @@ def resolve_codeable_concept( Result with ``best_match``, ``alternatives``, and ``unresolved`` lists. """ - body: dict[str, Any] = {"coding": coding} - if text is not None: - body["text"] = text + codings_list, extracted_text = _normalize_codeable_concept_input(coding) + final_text = text if text is not None else extracted_text + body: dict[str, Any] = {"coding": [_coding_to_dict(c) for c in codings_list]} + if final_text is not None: + body["text"] = final_text if resource_type is not None: body["resource_type"] = resource_type if include_recommendations: @@ -198,6 +317,7 @@ async def resolve( system: str | None = None, code: str | None = None, display: str | None = None, + coding: CodingInput | None = None, vocabulary_id: str | None = None, resource_type: str | None = None, include_recommendations: bool = False, @@ -208,10 +328,13 @@ async def resolve( See :meth:`Fhir.resolve` for full documentation. """ + resolved_system, resolved_code, resolved_display = _resolve_coding_kwargs( + coding, system, code, display + ) body = _build_resolve_body( - system=system, - code=code, - display=display, + system=resolved_system, + code=resolved_code, + display=resolved_display, vocabulary_id=vocabulary_id, resource_type=resource_type, include_recommendations=include_recommendations, @@ -222,7 +345,7 @@ async def resolve( async def resolve_batch( self, - codings: list[dict[str, str | None]], + codings: list[CodingInput], *, resource_type: str | None = None, include_recommendations: bool = False, @@ -233,7 +356,7 @@ async def resolve_batch( See :meth:`Fhir.resolve_batch` for full documentation. """ - body: dict[str, Any] = {"codings": codings} + body: dict[str, Any] = {"codings": [_coding_to_dict(c) for c in codings]} if resource_type is not None: body["resource_type"] = resource_type if include_recommendations: @@ -245,7 +368,7 @@ async def resolve_batch( async def resolve_codeable_concept( self, - coding: list[dict[str, str]], + coding: list[CodingInput] | CodeableConceptInput, *, text: str | None = None, resource_type: str | None = None, @@ -257,9 +380,11 @@ async def resolve_codeable_concept( See :meth:`Fhir.resolve_codeable_concept` for full documentation. """ - body: dict[str, Any] = {"coding": coding} - if text is not None: - body["text"] = text + codings_list, extracted_text = _normalize_codeable_concept_input(coding) + final_text = text if text is not None else extracted_text + body: dict[str, Any] = {"coding": [_coding_to_dict(c) for c in codings_list]} + if final_text is not None: + body["text"] = final_text if resource_type is not None: body["resource_type"] = resource_type if include_recommendations: diff --git a/src/omophub/types/__init__.py b/src/omophub/types/__init__.py index 7b0842c..f8a2d81 100644 --- a/src/omophub/types/__init__.py +++ b/src/omophub/types/__init__.py @@ -17,6 +17,10 @@ ) from .domain import Domain, DomainCategory, DomainStats, DomainSummary from .fhir import ( + CodeableConcept, + CodeableConceptLike, + Coding, + CodingLike, FhirBatchResult, FhirBatchSummary, FhirCodeableConceptResult, @@ -82,6 +86,10 @@ "BulkSemanticSearchInput", "BulkSemanticSearchResponse", "BulkSemanticSearchResultItem", + "CodeableConcept", + "CodeableConceptLike", + "Coding", + "CodingLike", "Concept", "ConceptSummary", "Descendant", diff --git a/src/omophub/types/fhir.py b/src/omophub/types/fhir.py index 26f60c6..e1f976c 100644 --- a/src/omophub/types/fhir.py +++ b/src/omophub/types/fhir.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Any, TypedDict +from typing import Any, Protocol, TypedDict, runtime_checkable from typing_extensions import NotRequired @@ -78,3 +78,44 @@ class FhirCodeableConceptResult(TypedDict): best_match: FhirResolveResult | None alternatives: list[FhirResolveResult] unresolved: list[dict[str, Any]] + + +class Coding(TypedDict, total=False): + """Lightweight FHIR ``Coding`` input type. + + Structurally compatible with ``fhir.resources.Coding`` and ``fhirpy`` + coding objects. Accepted anywhere the resolver takes a ``coding=`` input. + """ + + system: str + code: str + display: str + version: str + + +class CodeableConcept(TypedDict, total=False): + """Lightweight FHIR ``CodeableConcept`` input type.""" + + coding: list[Coding] + text: str + + +@runtime_checkable +class CodingLike(Protocol): + """Structural protocol for any object exposing ``system`` and ``code``. + + Lets the resolver accept ``fhir.resources.Coding``, ``fhirpy`` codings, + or any user-defined class with ``system``/``code`` attributes without + taking a hard dependency on those libraries. + """ + + system: str | None + code: str | None + + +@runtime_checkable +class CodeableConceptLike(Protocol): + """Structural protocol for any object exposing ``coding`` and ``text``.""" + + coding: list[Any] | None + text: str | None diff --git a/tests/integration/test_fhir.py b/tests/integration/test_fhir.py index 100d0e6..383f513 100644 --- a/tests/integration/test_fhir.py +++ b/tests/integration/test_fhir.py @@ -82,3 +82,84 @@ def test_resolve_with_quality_live(self, integration_client: OMOPHub) -> None: res = result["resolution"] assert "mapping_quality" in res assert res["mapping_quality"] in ("high", "medium", "low", "manual_review") + + # -- Type interop / duck typing (§B) --------------------------------- + + def test_resolve_with_coding_dict_live( + self, integration_client: OMOPHub + ) -> None: + """``coding=`` kwarg accepts a plain dict against the live API.""" + result = integration_client.fhir.resolve( + coding={ + "system": "http://snomed.info/sct", + "code": "44054006", + }, + resource_type="Condition", + ) + res = result["resolution"] + assert res["standard_concept"]["concept_id"] == 201826 + assert res["target_table"] == "condition_occurrence" + + def test_resolve_with_duck_typed_coding_live( + self, integration_client: OMOPHub + ) -> None: + """``coding=`` kwarg accepts a duck-typed object (system/code attrs).""" + from types import SimpleNamespace + + fake_fhir_coding = SimpleNamespace( + system="http://snomed.info/sct", + code="44054006", + display="Type 2 diabetes mellitus", + ) + result = integration_client.fhir.resolve(coding=fake_fhir_coding) + res = result["resolution"] + assert res["standard_concept"]["concept_id"] == 201826 + + def test_resolve_batch_mixed_inputs_live( + self, integration_client: OMOPHub + ) -> None: + """``resolve_batch`` accepts a mixed list of dicts and duck objects.""" + from types import SimpleNamespace + + result = integration_client.fhir.resolve_batch( + [ + {"system": "http://snomed.info/sct", "code": "44054006"}, + SimpleNamespace( + system="http://hl7.org/fhir/sid/icd-10-cm", + code="E11.9", + display=None, + ), + ], + ) + assert result["summary"]["total"] == 2 + assert result["summary"]["resolved"] >= 1 + + def test_resolve_codeable_concept_from_duck_object_live( + self, integration_client: OMOPHub + ) -> None: + """``resolve_codeable_concept`` accepts a CodeableConcept-like object.""" + from types import SimpleNamespace + + cc = SimpleNamespace( + coding=[ + SimpleNamespace( + system="http://snomed.info/sct", + code="44054006", + display=None, + ), + {"system": "http://hl7.org/fhir/sid/icd-10-cm", "code": "E11.9"}, + ], + text="Type 2 diabetes mellitus", + ) + result = integration_client.fhir.resolve_codeable_concept(cc) + assert result["best_match"] is not None + best = result["best_match"]["resolution"] + assert best["source_concept"]["vocabulary_id"] == "SNOMED" + + # -- fhir_server_url property (§C) ----------------------------------- + + def test_fhir_server_url_property( + self, integration_client: OMOPHub + ) -> None: + """The ``fhir_server_url`` property exposes the R4 endpoint.""" + assert integration_client.fhir_server_url == "https://fhir.omophub.com/fhir/r4" diff --git a/tests/unit/resources/test_fhir.py b/tests/unit/resources/test_fhir.py index 17aa2dc..1c4ed6c 100644 --- a/tests/unit/resources/test_fhir.py +++ b/tests/unit/resources/test_fhir.py @@ -573,3 +573,200 @@ def test_sync_fhir_property_cached(self, sync_client: OMOPHub) -> None: fhir1 = sync_client.fhir fhir2 = sync_client.fhir assert fhir1 is fhir2 + + +# -- Type interop / duck typing --------------------------------------------- + + +class _DuckCoding: + """Stand-in for a fhir.resources or fhirpy coding object.""" + + def __init__( + self, + system: str | None = None, + code: str | None = None, + display: str | None = None, + ) -> None: + self.system = system + self.code = code + self.display = display + + +class _DuckCodeableConcept: + """Stand-in for a CodeableConcept-like object.""" + + def __init__(self, coding: list[object], text: str | None = None) -> None: + self.coding = coding + self.text = text + + +class TestExtractCodingHelper: + """Tests for ``_extract_coding`` and ``_coding_to_dict`` normalizers.""" + + def test_extract_dict(self) -> None: + from omophub.resources.fhir import _extract_coding + + assert _extract_coding( + {"system": "http://snomed.info/sct", "code": "44054006", "display": "T2DM"} + ) == ("http://snomed.info/sct", "44054006", "T2DM") + + def test_extract_typed_dict(self) -> None: + from omophub.resources.fhir import _extract_coding + + # Coding is a TypedDict; passing as dict literal exercises the dict branch. + coding = {"system": "http://loinc.org", "code": "2339-0"} + assert _extract_coding(coding) == ( + "http://loinc.org", + "2339-0", + None, + ) + + def test_extract_duck_object(self) -> None: + from omophub.resources.fhir import _extract_coding + + obj = _DuckCoding(system="http://snomed.info/sct", code="44054006") + assert _extract_coding(obj) == ( + "http://snomed.info/sct", + "44054006", + None, + ) + + def test_extract_missing_attrs(self) -> None: + from omophub.resources.fhir import _extract_coding + + class Empty: + pass + + assert _extract_coding(Empty()) == (None, None, None) + + def test_coding_to_dict_drops_none(self) -> None: + from omophub.resources.fhir import _coding_to_dict + + obj = _DuckCoding(system="http://snomed.info/sct", code="44054006") + payload = _coding_to_dict(obj) + assert payload == { + "system": "http://snomed.info/sct", + "code": "44054006", + } + + +class TestFhirDuckTyping: + """Verify resolver methods accept dict, TypedDict, and duck objects.""" + + @respx.mock + def test_resolve_with_coding_dict( + self, sync_client: OMOPHub, base_url: str + ) -> None: + """Passing ``coding=`` as a dict produces the same upstream payload.""" + route = respx.post(f"{base_url}/fhir/resolve").mock( + return_value=Response(200, json=SNOMED_RESOLVE_RESPONSE) + ) + sync_client.fhir.resolve( + coding={ + "system": "http://snomed.info/sct", + "code": "44054006", + }, + ) + sent = route.calls.last.request.content.decode() + assert "http://snomed.info/sct" in sent + assert "44054006" in sent + + @respx.mock + def test_resolve_with_duck_object( + self, sync_client: OMOPHub, base_url: str + ) -> None: + """Passing ``coding=`` as a duck object works via ``getattr``.""" + route = respx.post(f"{base_url}/fhir/resolve").mock( + return_value=Response(200, json=SNOMED_RESOLVE_RESPONSE) + ) + sync_client.fhir.resolve( + coding=_DuckCoding( + system="http://snomed.info/sct", + code="44054006", + ), + ) + sent = route.calls.last.request.content.decode() + assert "http://snomed.info/sct" in sent + assert "44054006" in sent + + @respx.mock + def test_explicit_kwargs_win_over_coding( + self, sync_client: OMOPHub, base_url: str + ) -> None: + """Explicit ``system``/``code`` kwargs override ``coding`` values.""" + route = respx.post(f"{base_url}/fhir/resolve").mock( + return_value=Response(200, json=SNOMED_RESOLVE_RESPONSE) + ) + sync_client.fhir.resolve( + system="http://loinc.org", + code="2339-0", + coding=_DuckCoding( + system="http://snomed.info/sct", + code="44054006", + ), + ) + sent = route.calls.last.request.content.decode() + assert "http://loinc.org" in sent + assert "2339-0" in sent + assert "http://snomed.info/sct" not in sent + + @respx.mock + def test_resolve_batch_mixed_inputs( + self, sync_client: OMOPHub, base_url: str + ) -> None: + """``resolve_batch`` accepts a mixed list of dicts and duck objects.""" + route = respx.post(f"{base_url}/fhir/resolve/batch").mock( + return_value=Response( + 200, + json={ + "success": True, + "data": { + "results": [], + "summary": {"total": 2, "resolved": 0, "failed": 2}, + }, + "meta": { + "request_id": "t", + "timestamp": "2026-04-10T00:00:00Z", + "vocab_release": "2025.2", + }, + }, + ) + ) + sync_client.fhir.resolve_batch( + [ + {"system": "http://snomed.info/sct", "code": "44054006"}, + _DuckCoding( + system="http://hl7.org/fhir/sid/icd-10-cm", + code="E11.9", + ), + ], + ) + sent = route.calls.last.request.content.decode() + assert "44054006" in sent + assert "E11.9" in sent + + @respx.mock + def test_resolve_codeable_concept_from_duck_object( + self, sync_client: OMOPHub, base_url: str + ) -> None: + """``resolve_codeable_concept`` accepts a full CodeableConcept-like object.""" + route = respx.post(f"{base_url}/fhir/resolve/codeable-concept").mock( + return_value=Response( + 200, json=CODEABLE_CONCEPT_RESPONSE + ) + ) + cc = _DuckCodeableConcept( + coding=[ + _DuckCoding( + system="http://snomed.info/sct", + code="44054006", + ), + {"system": "http://hl7.org/fhir/sid/icd-10-cm", "code": "E11.9"}, + ], + text="Type 2 diabetes mellitus", + ) + sync_client.fhir.resolve_codeable_concept(cc) + sent = route.calls.last.request.content.decode() + assert "44054006" in sent + assert "E11.9" in sent + assert "Type 2 diabetes mellitus" in sent diff --git a/tests/unit/test_fhir_interop.py b/tests/unit/test_fhir_interop.py new file mode 100644 index 0000000..47a4eb2 --- /dev/null +++ b/tests/unit/test_fhir_interop.py @@ -0,0 +1,116 @@ +"""Tests for the FHIR client interop helpers.""" + +from __future__ import annotations + +import sys +from types import ModuleType, SimpleNamespace + +import pytest + + +class TestGetFhirServerUrl: + """``get_fhir_server_url`` returns the right base URL per FHIR version.""" + + def test_default_r4(self) -> None: + from omophub.fhir_interop import get_fhir_server_url + + assert get_fhir_server_url() == "https://fhir.omophub.com/fhir/r4" + + @pytest.mark.parametrize( + "version,expected", + [ + ("r4", "https://fhir.omophub.com/fhir/r4"), + ("r4b", "https://fhir.omophub.com/fhir/r4b"), + ("r5", "https://fhir.omophub.com/fhir/r5"), + ("r6", "https://fhir.omophub.com/fhir/r6"), + ], + ) + def test_each_version(self, version: str, expected: str) -> None: + from omophub.fhir_interop import get_fhir_server_url + + assert get_fhir_server_url(version) == expected # type: ignore[arg-type] + + +class TestGetFhirpyClient: + """``get_fhirpy_client`` is lazy and raises ImportError when missing.""" + + def test_missing_fhirpy_raises_import_error( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """When ``fhirpy`` is not installed, we raise with an install hint.""" + # Ensure any cached import is cleared. + monkeypatch.setitem(sys.modules, "fhirpy", None) # type: ignore[arg-type] + from omophub.fhir_interop import get_fhirpy_client + + with pytest.raises(ImportError, match="pip install omophub\\[fhirpy\\]"): + get_fhirpy_client("oh_test") + + def test_returns_sync_client_with_auth( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """When ``fhirpy`` is available, the stub is called with the right URL.""" + captured: dict[str, object] = {} + + def _fake_sync_client( + *, url: str, authorization: str + ) -> SimpleNamespace: + captured["url"] = url + captured["authorization"] = authorization + return SimpleNamespace(url=url, authorization=authorization) + + fake_module = ModuleType("fhirpy") + fake_module.SyncFHIRClient = _fake_sync_client # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "fhirpy", fake_module) + + from omophub.fhir_interop import get_fhirpy_client + + client = get_fhirpy_client("oh_abc", "r5") + + assert captured == { + "url": "https://fhir.omophub.com/fhir/r5", + "authorization": "Bearer oh_abc", + } + assert client.url == "https://fhir.omophub.com/fhir/r5" + + def test_async_variant(self, monkeypatch: pytest.MonkeyPatch) -> None: + """``get_async_fhirpy_client`` wires the async constructor similarly.""" + captured: dict[str, object] = {} + + def _fake_async_client( + *, url: str, authorization: str + ) -> SimpleNamespace: + captured["url"] = url + captured["authorization"] = authorization + return SimpleNamespace(url=url, authorization=authorization) + + fake_module = ModuleType("fhirpy") + fake_module.AsyncFHIRClient = _fake_async_client # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "fhirpy", fake_module) + + from omophub.fhir_interop import get_async_fhirpy_client + + get_async_fhirpy_client("oh_abc") + + assert captured == { + "url": "https://fhir.omophub.com/fhir/r4", + "authorization": "Bearer oh_abc", + } + + +class TestOMOPHubFhirServerUrlProperty: + """``OMOPHub.fhir_server_url`` returns the r4 base URL.""" + + def test_sync_property(self) -> None: + from omophub import OMOPHub + + client = OMOPHub(api_key="oh_test") + try: + assert client.fhir_server_url == "https://fhir.omophub.com/fhir/r4" + finally: + client.close() + + def test_async_property(self) -> None: + from omophub import AsyncOMOPHub + + client = AsyncOMOPHub(api_key="oh_test") + assert client.fhir_server_url == "https://fhir.omophub.com/fhir/r4" From ba0cb9cbe58c8e69ce182e902f673950d27f675f Mon Sep 17 00:00:00 2001 From: alex-omophub Date: Tue, 14 Apr 2026 21:17:59 +0100 Subject: [PATCH 3/3] Refactor examples to remove hardcoded API keys - Updated example scripts. - Adjusted search and resolution methods in examples to reflect the new API response structures, ensuring consistency across examples. - Improved documentation and comments in example scripts for clarity and better user guidance. --- examples/async_usage.py | 37 ++++----- examples/basic_usage.py | 12 +-- examples/error_handling.py | 8 +- examples/fhir_interop.py | 22 ++++-- examples/fhir_resolver.py | 26 +++--- examples/map_between_vocabularies.py | 6 +- examples/navigate_hierarchy.py | 4 +- examples/search_concepts.py | 86 +++++++++++++------- src/omophub/fhir_interop.py | 12 ++- src/omophub/resources/fhir.py | 28 ++++--- tests/unit/resources/test_fhir.py | 114 ++++++++++++++++++++++++++- 11 files changed, 257 insertions(+), 98 deletions(-) diff --git a/examples/async_usage.py b/examples/async_usage.py index 856336f..8deed76 100644 --- a/examples/async_usage.py +++ b/examples/async_usage.py @@ -11,14 +11,13 @@ async def basic_async_usage() -> None: print("=== Basic Async Usage ===") # Use async context manager - async with omophub.AsyncOMOPHub(api_key="oh_your_api_key") as client: + async with omophub.AsyncOMOPHub() as client: # Get a concept concept = await client.concepts.get(201826) print(f"Concept: {concept['concept_name']}") - # Search - results = await client.search.basic("diabetes", page_size=5) - concepts = results.get("concepts", []) + # Search - returns a flat list of concept dicts + concepts = await client.search.basic("diabetes", page_size=5) print(f"Found {len(concepts)} concepts") @@ -26,9 +25,9 @@ async def concurrent_requests() -> None: """Demonstrate concurrent API requests.""" print("\n=== Concurrent Requests ===") - async with omophub.AsyncOMOPHub(api_key="oh_your_api_key") as client: + async with omophub.AsyncOMOPHub() as client: # Fetch multiple concepts concurrently - concept_ids = [201826, 4329847, 1112807, 40483312, 37311061] + concept_ids = [201826, 4329847, 1112807, 316866, 37311061] tasks = [client.concepts.get(cid) for cid in concept_ids] concepts = await asyncio.gather(*tasks) @@ -42,26 +41,20 @@ async def parallel_searches() -> None: """Demonstrate parallel search operations.""" print("\n=== Parallel Searches ===") - async with omophub.AsyncOMOPHub(api_key="oh_your_api_key") as client: - # Run multiple searches in parallel + async with omophub.AsyncOMOPHub() as client: + # Run multiple searches in parallel. ``search.basic`` returns a + # flat list of concept dicts for the current page; for the full + # total count we iterate across pages via ``basic_iter``. search_terms = ["diabetes", "hypertension", "asthma", "depression"] - async def search_and_count(term: str) -> tuple[str, int]: - results = await client.search.basic(term, page_size=1) - # Get total from pagination metadata if available - meta = results.get("meta", {}).get("pagination", {}) - total_items = meta.get("total_items") - if total_items is not None: - total = total_items - else: - concepts = results.get("concepts", []) - total = len(concepts) if isinstance(concepts, list) else 0 - return term, total - - tasks = [search_and_count(term) for term in search_terms] + async def page_count(term: str) -> tuple[str, int]: + concepts = await client.search.basic(term, page_size=50) + return term, len(concepts) + + tasks = [page_count(term) for term in search_terms] results = await asyncio.gather(*tasks) - print("Search results:") + print("First-page hit counts:") for term, count in results: print(f" '{term}': {count} concepts") diff --git a/examples/basic_usage.py b/examples/basic_usage.py index 0f6dd6f..f171a4a 100644 --- a/examples/basic_usage.py +++ b/examples/basic_usage.py @@ -6,9 +6,9 @@ def main() -> None: """Demonstrate basic SDK usage.""" - # Initialize the client with your API key - # You can also set OMOPHUB_API_KEY environment variable - client = omophub.OMOPHub(api_key="oh_your_api_key") + # Reads OMOPHUB_API_KEY from the environment. To pass it explicitly: + # client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() # Get a concept by ID concept = client.concepts.get(201826) @@ -18,18 +18,18 @@ def main() -> None: print(f" Domain: {concept['domain_id']}") print() - # Search for concepts + # Search for concepts - returns a list of concept dicts results = client.search.basic( "diabetes", vocabulary_ids=["SNOMED"], page_size=5, ) print("Search results for 'diabetes':") - for c in results.get("concepts", []): + for c in results: print(f" {c['concept_id']}: {c['concept_name']}") print() - # List vocabularies + # List vocabularies - returns a dict with a 'vocabularies' key vocabs = client.vocabularies.list(page_size=5) print("Available vocabularies:") for v in vocabs.get("vocabularies", []): diff --git a/examples/error_handling.py b/examples/error_handling.py index bd281c8..7d6a05d 100644 --- a/examples/error_handling.py +++ b/examples/error_handling.py @@ -10,7 +10,7 @@ def handle_not_found() -> None: """Handle concept not found errors.""" print("=== Handling Not Found ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: # Try to get a non-existent concept @@ -40,7 +40,7 @@ def handle_rate_limit() -> None: """Handle rate limit errors with retry.""" print("\n=== Handling Rate Limits ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() for i in range(5): try: @@ -58,7 +58,7 @@ def handle_validation() -> None: """Handle validation errors.""" print("\n=== Handling Validation Errors ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: # Try an invalid request @@ -73,7 +73,7 @@ def comprehensive_error_handling() -> None: """Demonstrate comprehensive error handling.""" print("\n=== Comprehensive Error Handling ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: concept = client.concepts.get(201826) diff --git a/examples/fhir_interop.py b/examples/fhir_interop.py index 5e3fc3f..a422f4a 100644 --- a/examples/fhir_interop.py +++ b/examples/fhir_interop.py @@ -31,6 +31,7 @@ from __future__ import annotations +import os from types import SimpleNamespace from typing import TYPE_CHECKING @@ -39,7 +40,12 @@ if TYPE_CHECKING: from omophub.types.fhir import Coding -API_KEY = "oh_your_api_key" +# Read from the environment so the example runs out-of-the-box: +# export OMOPHUB_API_KEY=oh_... +# Scenario 8 (get_fhirpy_client) needs an explicit key string, so we +# materialize it once here. All other scenarios use omophub.OMOPHub() +# directly, which also picks up OMOPHUB_API_KEY. +API_KEY = os.environ.get("OMOPHUB_API_KEY", "oh_your_api_key") # --------------------------------------------------------------------------- @@ -51,7 +57,7 @@ def coding_kwarg_with_dict() -> None: """Pass a plain dict via the new ``coding=`` kwarg.""" print("=== 1. coding= kwarg with a plain dict ===") - client = omophub.OMOPHub(api_key=API_KEY) + client = omophub.OMOPHub() try: result = client.fhir.resolve( coding={ @@ -76,7 +82,7 @@ def coding_kwarg_with_typed_dict() -> None: """Use omophub's lightweight `Coding` TypedDict for IDE autocomplete.""" print("\n=== 2. coding= kwarg with omophub.types.fhir.Coding ===") - client = omophub.OMOPHub(api_key=API_KEY) + client = omophub.OMOPHub() try: coding: Coding = { "system": "http://loinc.org", @@ -105,7 +111,7 @@ def coding_kwarg_with_duck_object() -> None: """ print("\n=== 3. coding= kwarg with a duck-typed object ===") - client = omophub.OMOPHub(api_key=API_KEY) + client = omophub.OMOPHub() try: # SimpleNamespace stands in for any Coding-like class - fhir.resources, # fhirpy, or your own domain model. @@ -143,7 +149,7 @@ def coding_kwarg_with_fhir_resources() -> None: print(" Skipped: pip install omophub[fhir-resources]") return - client = omophub.OMOPHub(api_key=API_KEY) + client = omophub.OMOPHub() try: # Single Coding from fhir.resources snomed = FhirCoding( @@ -191,7 +197,7 @@ def mixed_batch_inputs() -> None: """A single `resolve_batch` accepts heterogeneous coding shapes.""" print("\n=== 5. resolve_batch with mixed input shapes ===") - client = omophub.OMOPHub(api_key=API_KEY) + client = omophub.OMOPHub() try: typed: Coding = {"system": "http://loinc.org", "code": "2339-0"} duck = SimpleNamespace( @@ -232,7 +238,7 @@ def explicit_kwargs_override_coding() -> None: """ print("\n=== 6. Explicit kwargs override coding= fields ===") - client = omophub.OMOPHub(api_key=API_KEY) + client = omophub.OMOPHub() try: base = SimpleNamespace( system="http://snomed.info/sct", @@ -262,7 +268,7 @@ def connection_helper_urls() -> None: from omophub import get_fhir_server_url - client = omophub.OMOPHub(api_key=API_KEY) + client = omophub.OMOPHub() try: # Property on the client returns the R4 base URL print(f" client.fhir_server_url = {client.fhir_server_url}") diff --git a/examples/fhir_resolver.py b/examples/fhir_resolver.py index 713b423..742ba58 100644 --- a/examples/fhir_resolver.py +++ b/examples/fhir_resolver.py @@ -29,7 +29,7 @@ def resolve_snomed() -> None: """Resolve a SNOMED CT code directly to its OMOP concept.""" print("=== 1. SNOMED Direct Resolution ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: result = client.fhir.resolve( system="http://snomed.info/sct", @@ -57,7 +57,7 @@ def resolve_icd10_mapped() -> None: """Resolve a non-standard ICD-10-CM code — automatically traverses Maps to.""" print("\n=== 2. ICD-10-CM → SNOMED Mapping ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: result = client.fhir.resolve( system="http://hl7.org/fhir/sid/icd-10-cm", @@ -89,7 +89,7 @@ def resolve_loinc() -> None: """Resolve a LOINC lab code to the OMOP measurement table.""" print("\n=== 3. LOINC → Measurement ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: result = client.fhir.resolve( system="http://loinc.org", @@ -116,7 +116,7 @@ def resolve_rxnorm() -> None: """Resolve an RxNorm drug code to the OMOP drug_exposure table.""" print("\n=== 4. RxNorm → Drug Exposure ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: result = client.fhir.resolve( system="http://www.nlm.nih.gov/research/umls/rxnorm", @@ -141,7 +141,7 @@ def resolve_text_only() -> None: """Resolve using only display text — triggers BioLORD semantic search.""" print("\n=== 5. Text-Only Semantic Fallback ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: # No system or code — just natural language text result = client.fhir.resolve( @@ -168,7 +168,7 @@ def resolve_vocabulary_id_bypass() -> None: """Use vocabulary_id directly when you already know the OMOP vocabulary.""" print("\n=== 6. Vocabulary ID Bypass ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: # Skip URI resolution — go straight to the vocabulary result = client.fhir.resolve( @@ -193,7 +193,7 @@ def resolve_with_recommendations() -> None: """Get Phoebe-recommended related concepts alongside the resolution.""" print("\n=== 7. With Phoebe Recommendations ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: result = client.fhir.resolve( system="http://snomed.info/sct", @@ -225,7 +225,7 @@ def resolve_with_quality() -> None: """Get a mapping quality signal to triage which resolutions need review.""" print("\n=== 8. With Mapping Quality ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: # Direct SNOMED match → "high" quality result = client.fhir.resolve( @@ -273,7 +273,7 @@ def resolve_batch() -> None: """Resolve multiple codings in a single call with per-item error reporting.""" print("\n=== 9. Batch Resolution ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: result = client.fhir.resolve_batch( [ @@ -324,7 +324,7 @@ def resolve_codeable_concept() -> None: """Resolve a CodeableConcept with multiple codings — SNOMED wins by preference.""" print("\n=== 10. CodeableConcept Resolution ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: result = client.fhir.resolve_codeable_concept( coding=[ @@ -382,7 +382,7 @@ def resolve_codeable_concept_text_fallback() -> None: """When no structured coding resolves, fall back to the text field.""" print("\n=== 11. CodeableConcept Text Fallback ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: result = client.fhir.resolve_codeable_concept( coding=[ @@ -420,7 +420,7 @@ async def async_resolve() -> None: """Demonstrate async FHIR resolution with concurrent requests.""" print("\n=== 12. Async FHIR Resolution ===") - async with omophub.AsyncOMOPHub(api_key="oh_your_api_key") as client: + async with omophub.AsyncOMOPHub() as client: # Single resolve result = await client.fhir.resolve( system="http://snomed.info/sct", @@ -463,7 +463,7 @@ def error_handling_examples() -> None: """Demonstrate error responses from the FHIR resolver.""" print("\n=== Error Handling Examples ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: # Unknown code system URI → 400 with suggestion print(" Typo in URI:") diff --git a/examples/map_between_vocabularies.py b/examples/map_between_vocabularies.py index bcd0e73..5a78f54 100644 --- a/examples/map_between_vocabularies.py +++ b/examples/map_between_vocabularies.py @@ -8,7 +8,7 @@ def get_mappings() -> None: """Get mappings for a concept to other vocabularies.""" print("=== Concept Mappings ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: # Type 2 diabetes mellitus (SNOMED) @@ -43,7 +43,7 @@ def map_concepts() -> None: """Map multiple concepts to a target vocabulary.""" print("\n=== Batch Concept Mapping ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: # Map SNOMED concepts to ICD-10-CM @@ -74,7 +74,7 @@ def lookup_by_code() -> None: """Look up a concept by vocabulary code and find its standard mapping.""" print("\n=== Code Lookup and Mapping ===") - client = omophub.OMOPHub(api_key="oh_your_api_key") + client = omophub.OMOPHub() try: # Look up ICD-10-CM code E11 (Type 2 diabetes) diff --git a/examples/navigate_hierarchy.py b/examples/navigate_hierarchy.py index e239047..b6e300c 100644 --- a/examples/navigate_hierarchy.py +++ b/examples/navigate_hierarchy.py @@ -3,7 +3,9 @@ import omophub -client = omophub.OMOPHub(api_key="oh_your_api_key") +# Reads OMOPHUB_API_KEY from the environment. To pass it explicitly: +# client = omophub.OMOPHub(api_key="oh_your_api_key") +client = omophub.OMOPHub() def get_ancestors() -> None: diff --git a/examples/search_concepts.py b/examples/search_concepts.py index 9c091d7..56ae6bb 100644 --- a/examples/search_concepts.py +++ b/examples/search_concepts.py @@ -7,16 +7,17 @@ import omophub -client = omophub.OMOPHub(api_key="oh_your_api_key") +# Reads OMOPHUB_API_KEY from the environment. To pass it explicitly: +# client = omophub.OMOPHub(api_key="oh_your_api_key") +client = omophub.OMOPHub() def basic_search() -> None: """Demonstrate basic concept search.""" print("=== Basic Search ===") - # Simple text search - results = client.search.basic("heart attack") - concepts = results.get("concepts", results) + # Simple text search - returns a flat list of concept dicts + concepts = client.search.basic("heart attack") print(f"Found {len(concepts)} concepts for 'heart attack'") for c in concepts[:3]: @@ -28,14 +29,13 @@ def filtered_search() -> None: print("\n=== Filtered Search ===") # Search in specific vocabularies - results = client.search.basic( + concepts = client.search.basic( "myocardial infarction", vocabulary_ids=["SNOMED", "ICD10CM"], domain_ids=["Condition"], standard_concept="S", # Only standard concepts page_size=10, ) - concepts = results.get("concepts", results) print(f"Found {len(concepts)} standard condition concepts") for c in concepts[:5]: @@ -43,11 +43,16 @@ def filtered_search() -> None: def bulk_lexical_search() -> None: - """Demonstrate bulk lexical search — multiple queries in one call.""" + """Demonstrate bulk lexical search — multiple queries in one call. + + ``bulk_basic`` returns a list of per-query result objects. Each item + has ``search_id``, ``query``, ``results`` (a nested list), ``status``, + and ``duration``. + """ print("\n=== Bulk Lexical Search ===") # Search for multiple terms at once (up to 50) - results = client.search.bulk_basic( + items = client.search.bulk_basic( [ {"search_id": "q1", "query": "diabetes mellitus"}, {"search_id": "q2", "query": "hypertension"}, @@ -56,31 +61,42 @@ def bulk_lexical_search() -> None: defaults={"vocabulary_ids": ["SNOMED"], "page_size": 5}, ) - for item in results["results"]: - print(f" {item['search_id']}: {len(item['results'])} results ({item['status']})") + for item in items: + print( + f" {item['search_id']}: {len(item['results'])} results ({item['status']})" + ) # Per-query overrides — different domains per query - results = client.search.bulk_basic( + items = client.search.bulk_basic( [ - {"search_id": "conditions", "query": "diabetes", "domain_ids": ["Condition"]}, + { + "search_id": "conditions", + "query": "diabetes", + "domain_ids": ["Condition"], + }, {"search_id": "drugs", "query": "metformin", "domain_ids": ["Drug"]}, ], defaults={"vocabulary_ids": ["SNOMED", "RxNorm"], "page_size": 3}, ) print("\n Per-query domain overrides:") - for item in results["results"]: + for item in items: print(f" {item['search_id']}:") - for c in item["results"]: + for c in item["results"][:3]: print(f" {c['concept_name']} ({c['vocabulary_id']}/{c['domain_id']})") def bulk_semantic_search() -> None: - """Demonstrate bulk semantic search — multiple NLP queries in one call.""" + """Demonstrate bulk semantic search — multiple NLP queries in one call. + + ``bulk_semantic`` returns a dict with ``results`` (list of per-query + items), ``total_searches``, ``completed_count``, ``failed_count``, + and ``total_duration``. + """ print("\n=== Bulk Semantic Search ===") # Search for multiple natural-language queries (up to 25) - results = client.search.bulk_semantic( + response = client.search.bulk_semantic( [ {"search_id": "s1", "query": "heart failure treatment options"}, {"search_id": "s2", "query": "type 2 diabetes medication"}, @@ -89,26 +105,35 @@ def bulk_semantic_search() -> None: defaults={"threshold": 0.5, "page_size": 5}, ) - for item in results["results"]: + print( + f" Completed {response['completed_count']}/{response['total_searches']} " + f"in {response.get('total_duration', '?')}" + ) + for item in response["results"]: count = item.get("result_count", len(item["results"])) print(f" {item['search_id']}: {count} results ({item['status']})") # Show top result per query if item["results"]: top = item["results"][0] - print(f" Top: {top['concept_name']} (score: {top['similarity_score']:.2f})") + print( + f" Top: {top['concept_name']} (score: {top['similarity_score']:.2f})" + ) def autocomplete_example() -> None: - """Demonstrate autocomplete suggestions.""" + """Demonstrate autocomplete suggestions. + + ``concepts.suggest`` returns a flat list of concept dicts - the same + shape as ``search.basic`` - so you read ``concept_name`` directly. + """ print("\n=== Autocomplete ===") - # Get suggestions as user types suggestions = client.concepts.suggest("hypert", page_size=5) print("Suggestions for 'hypert':") for s in suggestions[:5]: - print(f" {s['suggestion']}") + print(f" [{s['vocabulary_id']}] {s['concept_name']}") def pagination_example() -> None: @@ -164,24 +189,31 @@ def semantic_pagination() -> None: def similarity_search() -> None: - """Demonstrate similarity search.""" + """Demonstrate similarity search. + + ``search.similar`` returns a dict with a ``similar_concepts`` key + (not ``results``) - each item has the standard concept fields plus + a ``similarity_score``. + """ print("\n=== Similarity Search ===") # Find concepts similar to Type 2 diabetes mellitus (concept_id=201826) - results = client.search.similar(concept_id=201826, algorithm="hybrid") + response = client.search.similar(concept_id=201826, algorithm="hybrid") print("Concepts similar to 'Type 2 diabetes mellitus':") - for r in results["results"][:5]: - print(f" {r['concept_name']} (score: {r['similarity_score']:.2f})") + for r in response["similar_concepts"][:5]: + score = r.get("similarity_score") + score_str = f"{score:.2f}" if score is not None else "?" + print(f" {r['concept_name']} (score: {score_str})") # Find similar using a natural language query with semantic algorithm - results = client.search.similar( + response = client.search.similar( query="medications for high blood pressure", algorithm="semantic", similarity_threshold=0.6, vocabulary_ids=["RxNorm"], include_scores=True, ) - print(f"\n Found {len(results['results'])} similar RxNorm concepts") + print(f"\n Found {len(response['similar_concepts'])} similar RxNorm concepts") if __name__ == "__main__": diff --git a/src/omophub/fhir_interop.py b/src/omophub/fhir_interop.py index ab94624..5d341f3 100644 --- a/src/omophub/fhir_interop.py +++ b/src/omophub/fhir_interop.py @@ -56,7 +56,13 @@ def get_fhirpy_client(api_key: str, version: FhirVersion = "r4") -> Any: ImportError: if ``fhirpy`` is not installed. """ try: - from fhirpy import SyncFHIRClient # type: ignore[import-not-found] + # `fhirpy` is an optional extra; mypy's per-module caching means + # only one of the two conditional imports in this file is flagged + # with `import-not-found`, depending on order, so we silence both + # and pair with `unused-ignore` to keep `warn_unused_ignores` happy. + from fhirpy import ( # type: ignore[import-not-found, unused-ignore] + SyncFHIRClient, + ) except ImportError as e: # pragma: no cover - import guard raise ImportError( "fhirpy is required for FHIR client interop. " @@ -74,7 +80,9 @@ def get_async_fhirpy_client(api_key: str, version: FhirVersion = "r4") -> Any: Async counterpart of :func:`get_fhirpy_client`. Same install hint. """ try: - from fhirpy import AsyncFHIRClient + from fhirpy import ( # type: ignore[import-not-found, unused-ignore] + AsyncFHIRClient, + ) except ImportError as e: # pragma: no cover - import guard raise ImportError( "fhirpy is required for FHIR client interop. " diff --git a/src/omophub/resources/fhir.py b/src/omophub/resources/fhir.py index 29f70b9..cc67e8d 100644 --- a/src/omophub/resources/fhir.py +++ b/src/omophub/resources/fhir.py @@ -42,21 +42,31 @@ def _extract_coding( ) +# Keys the FHIR Concept Resolver accepts on a single coding item. Any +# other keys in a dict input (e.g. ``userSelected``, ``extension``, +# ``version`` from ``fhir.resources.Coding.model_dump()``) are dropped +# so the server never sees FHIR metadata it does not understand. +_ALLOWED_CODING_KEYS: tuple[str, ...] = ( + "system", + "code", + "display", + "vocabulary_id", +) + + def _coding_to_dict(coding_input: object) -> dict[str, Any]: """Convert any Coding-like input to a wire-format dict. - Preserves only the keys the resolver endpoint understands. Skips keys - whose values are ``None`` so the request payload stays tight. + Preserves only the keys the resolver endpoint understands + (:data:`_ALLOWED_CODING_KEYS`). Skips keys whose values are ``None`` + so the request payload stays tight. """ if isinstance(coding_input, dict): - src: dict[str, Any] = coding_input - else: - src = { - "system": getattr(coding_input, "system", None), - "code": getattr(coding_input, "code", None), - "display": getattr(coding_input, "display", None), - "vocabulary_id": getattr(coding_input, "vocabulary_id", None), + src: dict[str, Any] = { + key: coding_input.get(key) for key in _ALLOWED_CODING_KEYS } + else: + src = {key: getattr(coding_input, key, None) for key in _ALLOWED_CODING_KEYS} return {k: v for k, v in src.items() if v is not None} diff --git a/tests/unit/resources/test_fhir.py b/tests/unit/resources/test_fhir.py index 1c4ed6c..f82e9dc 100644 --- a/tests/unit/resources/test_fhir.py +++ b/tests/unit/resources/test_fhir.py @@ -649,6 +649,53 @@ def test_coding_to_dict_drops_none(self) -> None: "code": "44054006", } + def test_coding_to_dict_filters_unknown_keys_from_dict(self) -> None: + """Dict inputs with FHIR metadata keys must be filtered to the + resolver's allowed key set - the server should never see + ``userSelected``, ``extension``, or version markers from + ``fhir.resources.Coding.model_dump()``. + """ + from omophub.resources.fhir import _ALLOWED_CODING_KEYS, _coding_to_dict + + model_dump_style = { + "system": "http://snomed.info/sct", + "code": "44054006", + "display": "Type 2 diabetes mellitus", + "version": "20240301", # Not in allowed keys + "userSelected": True, # FHIR extension + "extension": [{"url": "http://example.com"}], + "id": "coding-1", + } + payload = _coding_to_dict(model_dump_style) + assert set(payload.keys()) <= set(_ALLOWED_CODING_KEYS) + assert payload == { + "system": "http://snomed.info/sct", + "code": "44054006", + "display": "Type 2 diabetes mellitus", + } + + def test_coding_to_dict_filters_unknown_attrs_from_object(self) -> None: + """Duck objects with extra attributes are also filtered to the + allowed key set. + """ + from types import SimpleNamespace + + from omophub.resources.fhir import _ALLOWED_CODING_KEYS, _coding_to_dict + + obj = SimpleNamespace( + system="http://snomed.info/sct", + code="44054006", + display=None, + userSelected=True, # Should not leak into payload + extension="anything", + ) + payload = _coding_to_dict(obj) + assert set(payload.keys()) <= set(_ALLOWED_CODING_KEYS) + assert payload == { + "system": "http://snomed.info/sct", + "code": "44054006", + } + class TestFhirDuckTyping: """Verify resolver methods accept dict, TypedDict, and duck objects.""" @@ -751,9 +798,7 @@ def test_resolve_codeable_concept_from_duck_object( ) -> None: """``resolve_codeable_concept`` accepts a full CodeableConcept-like object.""" route = respx.post(f"{base_url}/fhir/resolve/codeable-concept").mock( - return_value=Response( - 200, json=CODEABLE_CONCEPT_RESPONSE - ) + return_value=Response(200, json=CODEABLE_CONCEPT_RESPONSE) ) cc = _DuckCodeableConcept( coding=[ @@ -770,3 +815,66 @@ def test_resolve_codeable_concept_from_duck_object( assert "44054006" in sent assert "E11.9" in sent assert "Type 2 diabetes mellitus" in sent + + @respx.mock + def test_resolve_codeable_concept_from_dict( + self, sync_client: OMOPHub, base_url: str + ) -> None: + """``resolve_codeable_concept`` accepts a dict-shaped CodeableConcept.""" + route = respx.post(f"{base_url}/fhir/resolve/codeable-concept").mock( + return_value=Response(200, json=CODEABLE_CONCEPT_RESPONSE) + ) + cc = { + "coding": [ + {"system": "http://snomed.info/sct", "code": "44054006"}, + {"system": "http://hl7.org/fhir/sid/icd-10-cm", "code": "E11.9"}, + ], + "text": "Type 2 diabetes mellitus", + } + sync_client.fhir.resolve_codeable_concept(cc) + sent = route.calls.last.request.content.decode() + assert "44054006" in sent + assert "E11.9" in sent + assert "Type 2 diabetes mellitus" in sent + + @respx.mock + def test_resolve_codeable_concept_from_dict_text_only( + self, sync_client: OMOPHub, base_url: str + ) -> None: + """Dict CodeableConcept with no ``coding`` key and no ``text`` is valid. + + Exercises the ``cc_input.get("coding") or []`` fallback and the + ``cc_input.get("text")`` path when either field is absent. + """ + route = respx.post(f"{base_url}/fhir/resolve/codeable-concept").mock( + return_value=Response(200, json=CODEABLE_CONCEPT_RESPONSE) + ) + # Pass an empty dict - both keys missing; extracted text is None and + # gets overridden by the explicit ``text=`` kwarg. + sync_client.fhir.resolve_codeable_concept({}, text="diabetes") + sent = route.calls.last.request.content.decode() + assert "diabetes" in sent + assert '"coding":[]' in sent.replace(" ", "") + + @respx.mock + def test_resolve_codeable_concept_with_tuple_coding( + self, sync_client: OMOPHub, base_url: str + ) -> None: + """A duck object whose ``.coding`` is a non-list iterable is converted.""" + route = respx.post(f"{base_url}/fhir/resolve/codeable-concept").mock( + return_value=Response(200, json=CODEABLE_CONCEPT_RESPONSE) + ) + cc = _DuckCodeableConcept( + coding=( # tuple, not a list - exercises the list() conversion + _DuckCoding( + system="http://snomed.info/sct", + code="44054006", + ), + {"system": "http://hl7.org/fhir/sid/icd-10-cm", "code": "E11.9"}, + ), + text=None, + ) + sync_client.fhir.resolve_codeable_concept(cc) + sent = route.calls.last.request.content.decode() + assert "44054006" in sent + assert "E11.9" in sent