Coverage for src/lumen/fhir/terminology_pre_expansion.py: 97%

1""" 

2Stage 3b (docs/architecture/profile-validation.md, Level 3): external 

3Pre-Expansion for ValueSets the local expansion engine can't resolve — nested 

4`compose.include.valueSet` refs, `filter` rules, or a referenced CodeSystem too 

5large to load locally (e.g. LOINC/SNOMED). Calls a public FHIR terminology 

6server's `$expand` operation once, at ValueSet-write time, and caches the flat 

7result in `valueset_expansion_cache`. Resource *validation* itself never makes 

8a network call — it only ever reads that cache table (see 

9`terminology_binding.py::_resolve_value_set`). 

10 

11Opt-in per tenant (`terminology_pre_expansion_enabled`, default off): this 

12makes an outbound call to a third-party server Lumen doesn't control, which 

13deployments that need to stay air-gapped must explicitly allow. A failure here 

14(unreachable server, timeout, malformed response) never fails the ValueSet 

15write itself — best-effort only, logged and otherwise ignored. 

16""" 

17 

18from __future__ import annotations 

19 

20import json 

21import typing 

22from typing import Any 

23from uuid import UUID 

24 

25import asyncpg 

26import httpx 

27import structlog 

28 

29from ..config import get_settings 

30from .terminology_binding import JsonDict, expand_locally 

31 

32logger = structlog.get_logger(__name__) 

33 

34 

35async def _is_pre_expansion_enabled( 

36 conn: asyncpg.Connection[asyncpg.Record] | Any, tenant_id: UUID 

37) -> bool: 

38 row = await conn.fetchrow( 

39 "SELECT settings FROM tenant_settings WHERE tenant_id = $1", tenant_id 

40 ) 

41 if row: 

42 raw = row["settings"] 

43 settings_dict = json.loads(raw) if isinstance(raw, str) else dict(raw) 

44 if "terminology_pre_expansion_enabled" in settings_dict: 

45 return bool(settings_dict["terminology_pre_expansion_enabled"]) 

46 return get_settings().tenant.terminology_pre_expansion_enabled 

47 

48 

49async def fetch_external_expansion( 

50 server_url: str, value_set_url: str, timeout_seconds: float 

51) -> list[tuple[str, str]] | None: 

52 """Call a public terminology server's `$expand` for `value_set_url`. 

53 

54 Best-effort: returns None (never raises) on any failure — unreachable 

55 server, timeout, non-2xx response, or a malformed body. 

56 """ 

57 try: 

58 async with httpx.AsyncClient(timeout=timeout_seconds) as client: 

59 resp = await client.get( 

60 f"{server_url.rstrip('/')}/ValueSet/$expand", 

61 params={"url": value_set_url}, 

62 ) 

63 resp.raise_for_status() 

64 body: JsonDict = resp.json() 

65 except Exception as exc: 

66 await logger.awarning( 

67 "terminology.pre_expansion_failed", value_set_url=value_set_url, error=str(exc) 

68 ) 

69 return None 

70 

71 expansion = body.get("expansion") 

72 contains = ( 

73 typing.cast(JsonDict, expansion).get("contains") if isinstance(expansion, dict) else None 

74 ) 

75 if not isinstance(contains, list): 

76 return None 

77 out: list[tuple[str, str]] = [] 

78 for raw_entry in typing.cast("list[Any]", contains): 

79 if not isinstance(raw_entry, dict): 

80 continue 

81 entry = typing.cast(JsonDict, raw_entry) 

82 system = entry.get("system") 

83 code = entry.get("code") 

84 if isinstance(system, str) and isinstance(code, str): 

85 out.append((system, code)) 

86 return out 

87 

88 

89async def store_expansion_cache( 

90 conn: asyncpg.Connection[asyncpg.Record] | Any, 

91 tenant_id: UUID, 

92 value_set_url: str, 

93 contains: list[tuple[str, str]], 

94) -> None: 

95 await conn.execute( 

96 """ 

97 INSERT INTO valueset_expansion_cache (tenant_id, value_set_url, contains, fetched_at) 

98 VALUES ($1, $2, $3::jsonb, now()) 

99 ON CONFLICT (tenant_id, value_set_url) 

100 DO UPDATE SET contains = EXCLUDED.contains, fetched_at = EXCLUDED.fetched_at 

101 """, 

102 tenant_id, 

103 value_set_url, 

104 [{"system": s, "code": c} for s, c in contains], 

105 ) 

106 

107 

108async def load_expansion_cache( 

109 conn: asyncpg.Connection[asyncpg.Record] | Any, 

110 tenant_id: UUID, 

111 value_set_url: str, 

112) -> list[tuple[str, str]] | None: 

113 row = await conn.fetchrow( 

114 "SELECT contains FROM valueset_expansion_cache WHERE tenant_id = $1 AND value_set_url = $2", 

115 tenant_id, 

116 value_set_url, 

117 ) 

118 if row is None: 

119 return None 

120 contains = row["contains"] 

121 return [(c["system"], c["code"]) for c in contains if "system" in c and "code" in c] 

122 

123 

124async def maybe_pre_expand( 

125 conn: asyncpg.Connection[asyncpg.Record] | Any, 

126 tenant_id: UUID, 

127 value_set: JsonDict, 

128) -> None: 

129 """On a ValueSet write: if pre-expansion is enabled for this tenant and the 

130 local expansion engine can't resolve it, fetch + cache an external 

131 expansion. Best-effort — never raises, so a ValueSet write never fails 

132 because of this. 

133 """ 

134 url = value_set.get("url") 

135 if not isinstance(url, str): 

136 return 

137 if not await _is_pre_expansion_enabled(conn, tenant_id): 

138 return 

139 if await expand_locally(conn, tenant_id, value_set) is not None: 

140 return # locally resolvable already — no need for an external call 

141 

142 settings = get_settings().terminology 

143 contains = await fetch_external_expansion(settings.server_url, url, settings.timeout_seconds) 

144 if contains is not None: 

145 await store_expansion_cache(conn, tenant_id, url, contains)