Coverage for src/lumen/fhirpath/engine/__init__.py: 76%

1import json 

2import numbers 

3import lumen.fhirpath.engine.util as util 

4from lumen.fhirpath.engine.nodes import TypeInfo 

5from lumen.fhirpath.engine.evaluators import evaluators 

6from lumen.fhirpath.engine.invocations import ( 

7 invocation_registry as base_invocation_registry, 

8) 

9 

10 

11def check_integer_param(val): 

12 data = util.get_data(val) 

13 if int(data) != data: 

14 raise Exception("Expected integer, got: " + json.dumps(data)) 

15 return data 

16 

17 

18def check_number_param(val): 

19 data = util.get_data(val) 

20 if not isinstance(data, numbers.Number): 

21 raise Exception("Expected number, got: " + json.dumps(data)) 

22 return data 

23 

24 

25def check_boolean_param(val): 

26 data = util.get_data(val) 

27 if data is True or data is False: 

28 return data 

29 raise Exception("Expected boolean, got: " + json.dumps(data)) 

30 

31 

32def check_string_param(val): 

33 data = util.get_data(val) 

34 if not isinstance(data, str): 

35 raise Exception("Expected string, got: " + json.dumps(data)) 

36 return data 

37 

38 

39def do_eval(ctx, parentData, node): 

40 node_type = node["type"] 

41 

42 if node_type in evaluators: 

43 evaluator = evaluators.get(node_type) 

44 return evaluator(ctx, parentData, node) 

45 

46 raise Exception("No " + node_type + " evaluator ") 

47 

48 

49def doInvoke(ctx, fn_name, data, raw_params): 

50 invocation_registry = { 

51 **base_invocation_registry, 

52 **(ctx["userInvocationTable"] or {}), 

53 } 

54 

55 if isinstance(fn_name, list) and len(fn_name) == 1: 

56 fn_name = fn_name[0] 

57 

58 if not isinstance(fn_name, str) or fn_name not in invocation_registry: 

59 raise Exception("Not implemented: " + str(fn_name)) 

60 

61 invocation = invocation_registry[fn_name] 

62 

63 if "nullable_input" in invocation and util.is_nullable(data): 

64 return [] 

65 

66 if "arity" not in invocation: 

67 if raw_params is None or util.is_empty(raw_params): 

68 res = invocation["fn"](ctx, util.arraify(data)) 

69 return util.arraify(res) 

70 

71 raise Exception(fn_name + " expects no params") 

72 

73 if invocation["fn"].__name__ == "trace_fn" and raw_params is not None: 

74 raw_params = raw_params[:1] 

75 

76 paramsNumber = 0 

77 if isinstance(raw_params, list): 

78 paramsNumber = len(raw_params) 

79 

80 if paramsNumber not in invocation["arity"]: 

81 raise Exception(fn_name + " wrong arity: got " + str(paramsNumber)) 

82 

83 params = [] 

84 argTypes = invocation["arity"][paramsNumber] 

85 

86 thisValue = ctx["$this"] if "$this" in ctx else ctx["dataRoot"] 

87 for i in range(0, paramsNumber): 

88 tp = argTypes[i] 

89 pr = raw_params[i] 

90 params.append(make_param(ctx, thisValue, tp, pr)) 

91 

92 params.insert(0, data) 

93 params.insert(0, ctx) 

94 

95 if "nullable" in invocation: 

96 if any(util.is_nullable(x) for x in params): 

97 return [] 

98 

99 # Allow per-arity function dispatch (e.g. sort() vs sort(expr)) 

100 fn = invocation.get("expr_dispatch", {}).get(paramsNumber, invocation["fn"]) 

101 res = fn(*params) 

102 

103 return util.arraify(res) 

104 

105 

106_PRIMITIVE_TYPES = frozenset( 

107 { 

108 "base64Binary", 

109 "boolean", 

110 "canonical", 

111 "code", 

112 "date", 

113 "dateTime", 

114 "decimal", 

115 "id", 

116 "instant", 

117 "integer", 

118 "markdown", 

119 "oid", 

120 "positiveInt", 

121 "string", 

122 "time", 

123 "unsignedInt", 

124 "uri", 

125 "url", 

126 "uuid", 

127 } 

128) 

129 

130 

131def is_valid_type(type_info, model): 

132 name = type_info.name 

133 ns = type_info.namespace 

134 

135 system_types = { 

136 "String", 

137 "Boolean", 

138 "Integer", 

139 "Decimal", 

140 "Date", 

141 "DateTime", 

142 "Time", 

143 "Quantity", 

144 "Concept", 

145 } 

146 if ns == "System": 

147 return True 

148 

149 if ns == "FHIR" or ns is None: 

150 if name in system_types: 

151 return True 

152 if name in _PRIMITIVE_TYPES or name in ( 

153 "Quantity", 

154 "Age", 

155 "Count", 

156 "Distance", 

157 "Duration", 

158 "Money", 

159 "Range", 

160 "Period", 

161 "CodeableConcept", 

162 "Reference", 

163 ): 

164 return True 

165 if model: 

166 if name in model.get("type2Parent", {}) or name in model.get("path2Type", {}).values(): 

167 return True 

168 # Lazy import to avoid coupling lumen.fhirpath → lumen.fhir at import time 

169 try: 

170 from lumen.fhir.resource_types import known_resource_types 

171 known = known_resource_types() 

172 if name in known or name.capitalize() in known: 

173 return True 

174 except Exception: 

175 pass 

176 else: 

177 return True 

178 return False 

179 

180 

181def type_specifier(ctx, parent_data, node): 

182 identifiers = node["text"].replace("`", "").split(".") 

183 namespace = None 

184 name = None 

185 

186 if len(identifiers) == 2: 

187 namespace, name = identifiers 

188 elif len(identifiers) == 1: 

189 (name,) = identifiers 

190 else: 

191 raise Exception(f"Expected TypeSpecifier node, got {node}") 

192 

193 type_info = TypeInfo(name=name, namespace=namespace) 

194 model = ctx.get("model") 

195 if not is_valid_type(type_info, model): 

196 raise ValueError(f"Unknown type specifier: {type_info.name}") 

197 return type_info 

198 

199 

200param_check_table = { 

201 "Integer": check_integer_param, 

202 "Number": check_number_param, 

203 "Boolean": check_boolean_param, 

204 "String": check_string_param, 

205} 

206 

207 

208def make_param(ctx, parentData, node_type, param): 

209 if node_type == "Expr": 

210 

211 def func(data): 

212 ctx["$this"] = util.arraify(data) 

213 return do_eval(ctx, ctx["$this"], param) 

214 

215 return func 

216 

217 if node_type == "AnyAtRoot": 

218 ctx["$this"] = ctx["$this"] if "$this" in ctx else ctx["dataRoot"] 

219 return do_eval(ctx, ctx["$this"], param) 

220 

221 if node_type == "Identifier": 

222 if param["type"] == "TermExpression": 

223 return param["text"] 

224 

225 raise Exception("Expected identifier node, got " + json.dumps(param)) 

226 

227 if node_type == "TypeSpecifier": 

228 return type_specifier(ctx, parentData, param) 

229 

230 ctx["$this"] = parentData 

231 res = do_eval(ctx, parentData, param) 

232 

233 if node_type == "Any": 

234 return res 

235 

236 if isinstance(node_type, list): 

237 if len(res) == 0: 

238 return [] 

239 else: 

240 node_type = node_type[0] 

241 

242 if len(res) > 1: 

243 raise Exception( 

244 "Unexpected collection" 

245 + json.dumps(res) 

246 + "; expected singleton of type " 

247 + node_type 

248 ) 

249 

250 if len(res) == 0: 

251 return [] 

252 

253 if node_type not in param_check_table: 

254 raise Exception("Implement me for " + node_type) 

255 

256 check = param_check_table[node_type] 

257 

258 return check(res[0]) 

259 

260 

261def infix_invoke(ctx, fn_name, data, raw_params): 

262 invocation_registry = { 

263 **base_invocation_registry, 

264 **(ctx["userInvocationTable"] or {}), 

265 } 

266 if fn_name not in invocation_registry or "fn" not in invocation_registry[fn_name]: 

267 raise Exception("Not implemented " + fn_name) 

268 

269 invocation = invocation_registry[fn_name] 

270 paramsNumber = len(raw_params) 

271 

272 if paramsNumber != 2: 

273 raise Exception("Infix invoke should have arity 2") 

274 

275 argTypes = invocation["arity"][paramsNumber] 

276 

277 if argTypes is not None: 

278 params = [ctx] 

279 

280 for i in range(0, paramsNumber): 

281 argType = argTypes[i] 

282 rawParam = raw_params[i] 

283 params.append(make_param(ctx, data, argType, rawParam)) 

284 

285 if "nullable" in invocation: 

286 if any(util.is_nullable(x) for x in params): 

287 return [] 

288 

289 res = invocation["fn"](*params) 

290 return util.arraify(res) 

291 

292 print(fn_name + " wrong arity: got " + paramsNumber) 

293 return []