-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathfhir_interop.py
More file actions
352 lines (284 loc) · 12.3 KB
/
fhir_interop.py
File metadata and controls
352 lines (284 loc) · 12.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
#!/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
import os
from types import SimpleNamespace
from typing import TYPE_CHECKING
import omophub
if TYPE_CHECKING:
from omophub.types.fhir import Coding
# 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")
# ---------------------------------------------------------------------------
# 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()
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()
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()
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()
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()
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()
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()
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()