Coverage for src/lumen/fhirpath/__init__.py: 95%

1from __future__ import annotations 

2 

3from collections.abc import Callable, Mapping, Sequence 

4from typing import Any, Protocol, TypeVar, cast 

5 

6from lumen.fhirpath.engine import do_eval 

7from lumen.fhirpath.engine.invocations.constants import constants 

8from lumen.fhirpath.engine.nodes import FP_Type, ResourceNode 

9from lumen.fhirpath.engine.util import arraify, get_data, process_user_invocation_table, set_paths 

10from lumen.fhirpath.parser import parse 

11 

12__title__ = "lumen.fhirpath" 

13__version__ = "2.2.0" 

14__author__ = "beda.software" 

15__license__ = "MIT" 

16__copyright__ = "Copyright 2026 beda.software" 

17 

18# Version synonym 

19VERSION = __version__ 

20 

21ResourceType = Mapping[str, Any] 

22ContextType = Mapping[str, Any] | None 

23 

24 

25def apply_parsed_path(resource, parsedPath, context=None, model=None, options=None): 

26 constants.reset() 

27 dataRoot = arraify(resource) 

28 

29 """ 

30 do_eval takes a "ctx" object, and we store things in that as we parse, so we 

31 need to put user-provided variable data in a sub-object, ctx['vars']. 

32 Set up default standard variables, and allow override from the variables. 

33 However, we'll keep our own copy of dataRoot for internal processing. 

34 """ 

35 vars = {"context": resource, "ucum": "http://unitsofmeasure.org"} 

36 vars.update(context or {}) 

37 

38 ctx = { 

39 "dataRoot": dataRoot, 

40 "vars": vars, 

41 "model": model, 

42 "userInvocationTable": process_user_invocation_table( 

43 (options or {}).get("userInvocationTable", {}) 

44 ), 

45 } 

46 

47 # Add trace callback if provided in options 

48 if options and "traceFn" in options: 

49 ctx["traceFn"] = options["traceFn"] 

50 

51 node = do_eval(ctx, dataRoot, parsedPath["children"][0]) 

52 

53 # Resolve any internal "ResourceNode" instances. Continue to let FP_Type 

54 # subclasses through. 

55 

56 if options and options.get("returnRawData", False): 

57 if isinstance(node, list): 

58 res = [] 

59 # Filter out intenal representation of primitive extensions 

60 # even in this raw data mode (as they are not a part of the output) 

61 for item in node: 

62 if isinstance(item, ResourceNode): 

63 if isinstance(item.data, dict): 

64 keys = list(item.data.keys()) 

65 if keys == ["extension"]: 

66 continue 

67 res.append(item) 

68 return res 

69 return node 

70 

71 def visit(node): 

72 data = get_data(node) 

73 

74 if isinstance(node, list): 

75 res = [] 

76 for item in data: 

77 # Filter out intenal representation of primitive extensions 

78 i = visit(item) 

79 if isinstance(i, dict): 

80 keys = list(i.keys()) 

81 if keys == ["extension"]: 

82 continue 

83 res.append(i) 

84 return res 

85 

86 if isinstance(data, dict) and not isinstance(data, FP_Type): 

87 for key, value in data.items(): 

88 data[key] = visit(value) 

89 

90 return data 

91 

92 return visit(node) 

93 

94 

95def evaluate( 

96 resource: ResourceType | list, 

97 path: str | dict, 

98 context: dict | None = None, 

99 model: dict | None = None, 

100 options: dict | None = None, 

101) -> list: 

102 if model is None: 

103 from lumen.fhirpath.models import models 

104 model = models["r4"] 

105 default_context = { 

106 "sct": "http://snomed.info/sct", 

107 "loinc": "http://loinc.org", 

108 "ucum": "http://unitsofmeasure.org", 

109 "vs-administrative-gender": "http://hl7.org/fhir/ValueSet/administrative-gender", 

110 "ext-patient-birthTime": "http://hl7.org/fhir/StructureDefinition/patient-birthTime", 

111 } 

112 if context: 

113 default_context.update(context) 

114 

115 if isinstance(path, dict): 

116 node = parse(path["expression"]) 

117 if "base" in path: 

118 resource = ResourceNode.create_node(resource, path["base"]) 

119 else: 

120 node = parse(path) 

121 

122 return apply_parsed_path(resource, node, default_context, model, options) 

123 

124 

125def compile( 

126 path: str, model: dict | None = None, options: dict | None = None 

127) -> Callable[[ResourceType, ContextType], list]: 

128 if model is None: 

129 from lumen.fhirpath.models import models 

130 model = models["r4"] 

131 return set_paths(apply_parsed_path, parsedPath=parse(path), model=model, options=options) 

132 

133 

134InputType = TypeVar("InputType") 

135OutputType = TypeVar("OutputType") 

136# Contravariant: a callable accepting a wider input type is a valid subtype. 

137_I_contra = TypeVar("_I_contra", contravariant=True) 

138# Covariant: a callable returning a narrower output type is a valid subtype. 

139# Sequence (read-only) is required here because list is invariant and rejects covariant TypeVars. 

140_O_co = TypeVar("_O_co", covariant=True) 

141 

142 

143class CompiledFirst(Protocol[_I_contra, _O_co]): 

144 def __call__(self, resource: _I_contra, context: ContextType = ...) -> _O_co | None: ... 

145 

146 

147class CompiledArray(Protocol[_I_contra, _O_co]): 

148 def __call__(self, resource: _I_contra, context: ContextType = ...) -> Sequence[_O_co]: ... 

149 

150 

151def compile_as_array( 

152 expression: str, input_type: type[InputType], output_type: type[OutputType] 

153) -> CompiledArray[InputType, OutputType]: 

154 path_fn = compile(expression) 

155 

156 def fn(resource: Any, context: ContextType = None) -> Any: 

157 return _format_result( 

158 path_fn(_prepare_data(resource, input_type), context), 

159 output_type, 

160 is_array=True, 

161 ) 

162 

163 return cast(CompiledArray[InputType, OutputType], fn) 

164 

165 

166def compile_as_first( 

167 expression: str, input_type: type[InputType], output_type: type[OutputType] 

168) -> CompiledFirst[InputType, OutputType]: 

169 path_fn = compile(expression) 

170 

171 def fn(resource: Any, context: ContextType = None) -> Any: 

172 return _format_result( 

173 path_fn(_prepare_data(resource, input_type), context), 

174 output_type, 

175 is_array=False, 

176 ) 

177 

178 return cast(CompiledFirst[InputType, OutputType], fn) 

179 

180 

181def _prepare_data(resource: Any, input_type: type[InputType]) -> ResourceType: 

182 if not isinstance(resource, input_type): 

183 raise Exception( 

184 f"Resource type is {type(resource).__name__}, expected {input_type.__name__}" 

185 ) 

186 

187 if isinstance(resource, dict): 

188 return resource 

189 

190 if hasattr(resource, "model_dump"): 

191 return resource.model_dump() 

192 

193 raise Exception(f"Don't know how to work with type {type(resource).__name__}") 

194 

195 

196def _format_result(result: list, output_type: type[OutputType], is_array=False) -> Any: 

197 result = [_format_item(item, output_type) for item in result] 

198 

199 if is_array: 

200 return result 

201 

202 if len(result) == 0: 

203 return None 

204 

205 return result[0] 

206 

207 

208def _format_item(item: Any, output_type: type[OutputType]) -> OutputType: 

209 if hasattr(output_type, "model_validate"): 

210 return cast(Any, output_type).model_validate(item) 

211 

212 if not isinstance(item, output_type): 

213 raise Exception( 

214 f"Expected result to be {output_type.__name__}, but got {type(item).__name__}" 

215 ) 

216 

217 return item