Coverage for src/lumen/fhir/terminology_binding.py: 96%

1""" 

2Level 3 profile validation: required-binding terminology checks (see 

3docs/architecture/profile-validation.md). Resolves a ValueSet's expansion using 

4only locally-available data (explicit `compose` codes + locally-loaded 

5CodeSystems, via `terminology.expand_valueset`) and caches it per tenant — 

6external pre-expansion against a public terminology server (e.g. tx.fhir.org) 

7for implicit/huge ValueSets is a deliberately separate, not-yet-built step. 

8""" 

9 

10from __future__ import annotations 

11 

12import typing 

13from typing import Any 

14from uuid import UUID 

15 

16import asyncpg 

17 

18from .terminology import TerminologyError, expand_valueset, referenced_systems 

19 

20JsonDict = dict[str, Any] 

21 

22# (system, code) pairs, or None if the ValueSet couldn't be resolved with only 

23# local data (not found, or a `compose` rule this server doesn't support yet — 

24# nested `valueSet` references, `filter` rules). None means "can't verify 

25# locally" and is never treated as a violation. 

26_ExpansionResult = frozenset[tuple[str, str]] | None 

27 

28 

29class ValueSetExpansionCache: 

30 """Per-tenant cache of resolved ValueSet expansions, keyed by canonical URL. 

31 

32 Mirrors the coarse invalidation granularity of `SearchParamRegistry` 

33 (`search/registry.py`): a write to any ValueSet/CodeSystem drops the whole 

34 tenant's cache rather than tracking fine-grained dependencies. 

35 """ 

36 

37 def __init__(self) -> None: 

38 self._cache: dict[tuple[UUID, str], _ExpansionResult] = {} 

39 

40 def invalidate(self, tenant_id: UUID) -> None: 

41 for key in [k for k in self._cache if k[0] == tenant_id]: 

42 del self._cache[key] 

43 

44 def invalidate_all(self) -> None: 

45 self._cache.clear() 

46 

47 async def resolve( 

48 self, 

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

50 tenant_id: UUID, 

51 value_set_url: str, 

52 ) -> _ExpansionResult: 

53 key = (tenant_id, value_set_url) 

54 if key not in self._cache: 

55 self._cache[key] = await resolve_value_set(conn, tenant_id, value_set_url) 

56 return self._cache[key] 

57 

58 

59_cache: ValueSetExpansionCache | None = None 

60 

61 

62def get_valueset_cache() -> ValueSetExpansionCache: 

63 global _cache 

64 if _cache is None: 

65 _cache = ValueSetExpansionCache() 

66 return _cache 

67 

68 

69async def _find_current_by_url( 

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

71 tenant_id: UUID, 

72 resource_type: str, 

73 url: str, 

74) -> JsonDict | None: 

75 row = await conn.fetchrow( 

76 """ 

77 SELECT resource 

78 FROM fhir_resources 

79 WHERE tenant_id = $1 

80 AND resource_type = $2 

81 AND resource->>'url' = $3 

82 AND deleted = FALSE 

83 ORDER BY version_id DESC 

84 LIMIT 1 

85 """, 

86 tenant_id, 

87 resource_type, 

88 url, 

89 ) 

90 return row["resource"] if row else None 

91 

92 

93async def expand_locally( 

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

95 tenant_id: UUID, 

96 value_set: JsonDict, 

97) -> frozenset[tuple[str, str]] | None: 

98 """Expand a ValueSet using only local data (explicit `compose` codes + 

99 locally-loaded CodeSystems). None if `compose` has a rule the local engine 

100 doesn't support (nested `valueSet` refs, `filter` rules) or references a 

101 CodeSystem that isn't loaded locally — the case Stage 3b's external 

102 Pre-Expansion cache (`terminology_pre_expansion.py`) exists to cover. 

103 """ 

104 systems: dict[str, JsonDict] = {} 

105 for system_url in referenced_systems(value_set): 

106 cs = await _find_current_by_url(conn, tenant_id, "CodeSystem", system_url) 

107 if cs is not None: 

108 systems[system_url] = cs 

109 

110 try: 

111 contains = expand_valueset(value_set, systems) 

112 except TerminologyError: 

113 return None 

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

115 

116 

117async def resolve_value_set( 

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

119 tenant_id: UUID, 

120 value_set_url: str, 

121) -> _ExpansionResult: 

122 value_set = await _find_current_by_url(conn, tenant_id, "ValueSet", value_set_url) 

123 if value_set is None: 

124 return None 

125 

126 local = await expand_locally(conn, tenant_id, value_set) 

127 if local is not None: 

128 return local 

129 

130 # Stage 3b: fall back to a previously pre-expanded external cache. This is 

131 # a pure DB read — no network call happens during validation; see 

132 # `terminology_pre_expansion.py` for the (opt-in, write-time) fetch itself. 

133 from .terminology_pre_expansion import load_expansion_cache 

134 

135 cached = await load_expansion_cache(conn, tenant_id, value_set_url) 

136 return frozenset(cached) if cached is not None else None 

137 

138 

139def value_matches(value: Any, type_code: str | None, codes: frozenset[tuple[str, str]]) -> bool: 

140 """Check one extracted value against a resolved ValueSet expansion. 

141 

142 Only `code` (bare string), `Coding`, and `CodeableConcept` are checked — 

143 other bound types (e.g. `Quantity.code`/`.system` for UCUM bindings) are 

144 left unvalidated for now (documented limitation, not a false positive risk 

145 since unmatched shapes are treated as passing). 

146 """ 

147 if type_code == "CodeableConcept" and isinstance(value, dict): 

148 value_dict = typing.cast(JsonDict, value) 

149 codings = value_dict.get("coding") 

150 if not isinstance(codings, list): 

151 return False 

152 for coding in typing.cast("list[Any]", codings): 

153 if isinstance(coding, dict) and coding_matches(typing.cast(JsonDict, coding), codes): 

154 return True 

155 return False 

156 if type_code == "Coding" and isinstance(value, dict): 

157 return coding_matches(typing.cast(JsonDict, value), codes) 

158 if isinstance(value, str): 

159 return any(code == value for _, code in codes) 

160 return True 

161 

162 

163def coding_matches(coding: JsonDict, codes: frozenset[tuple[str, str]]) -> bool: 

164 code = coding.get("code") 

165 if not isinstance(code, str): 

166 return False 

167 system = coding.get("system") 

168 if system is None: 

169 return any(c == code for _, c in codes) 

170 return (system, code) in codes 

171 

172 

173async def check_required_binding( 

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

175 tenant_id: UUID, 

176 value_set_url: str, 

177 values: list[Any], 

178 type_code: str | None, 

179 path: str, 

180 profile_url: str, 

181) -> list[str]: 

182 """Validate already-extracted `values` at `path` against a `required` binding. 

183 

184 Returns [] if the ValueSet can't be resolved with local data — that's "not 

185 verifiable", never a violation (see module docstring). 

186 """ 

187 codes = await get_valueset_cache().resolve(conn, tenant_id, value_set_url) 

188 if codes is None: 

189 return [] 

190 errors: list[str] = [] 

191 for value in values: 

192 if not value_matches(value, type_code, codes): 

193 errors.append( 

194 f"Profile {profile_url}: {path} = {value!r} is not in the required " 

195 f"binding {value_set_url}." 

196 ) 

197 return errors