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

1""" 

2This file implements a lightweight FHIRPath evaluator tailored for search parameter extraction. 

3It parses dot paths, handles choices, and resolves basic collections without full engine overhead. 

4""" 

5 

6from __future__ import annotations 

7 

8import re 

9from typing import Any, cast 

10 

11# Infix "expr as Type" → "expr.as(Type)" so it parses as a trailing step. 

12_INFIX_AS_RE = re.compile(r"^(.*\S)\s+as\s+(\w+)$") 

13 

14 

15def evaluate_path(expression: str, resource: dict[str, Any]) -> list[Any]: 

16 """Evaluate a FHIRPath expression and return a flat list of primitive/dict values. 

17 

18 The leading resource-type segment (e.g. ``Patient.``) is stripped 

19 automatically when it matches ``resource["resourceType"]``. 

20 """ 

21 parts = [p.strip() for p in expression.split("|")] 

22 results: list[Any] = [] 

23 seen: set[int] = set() 

24 

25 for part in parts: 

26 for val in _eval_single(_normalize(part.strip()), resource): 

27 if isinstance(val, dict): 

28 d: dict[str, Any] = cast("dict[str, Any]", val) 

29 obj_id = id(d) 

30 if obj_id not in seen: 

31 seen.add(obj_id) 

32 results.append(d) 

33 elif val not in results: 

34 results.append(val) 

35 

36 return results 

37 

38 

39# ── internals ───────────────────────────────────────────────────────────────── 

40 

41 

42def _normalize(expr: str) -> str: 

43 """Strip wrapping parens and rewrite infix ``as`` into a trailing ``.as()`` step.""" 

44 expr = expr.strip() 

45 while expr.startswith("(") and expr.endswith(")"): 

46 expr = expr[1:-1].strip() 

47 m = _INFIX_AS_RE.match(expr) 

48 if m: 

49 expr = f"{m.group(1).strip()}.as({m.group(2)})" 

50 return expr 

51 

52 

53def _eval_single(expr: str, resource: dict[str, Any]) -> list[Any]: 

54 """Evaluate a single (non-union) FHIRPath expression.""" 

55 steps = _parse_steps(expr) 

56 

57 # Strip leading resource-type step when it matches (case-sensitive per FHIR spec). 

58 resource_type = resource.get("resourceType", "") 

59 if steps and steps[0] == ("field", resource_type): 

60 steps = steps[1:] 

61 

62 current: list[Any] = [resource] 

63 for kind, arg in steps: 

64 if kind == "field": 

65 current = _step_field(current, arg) 

66 elif kind in ("ofType", "as"): 

67 current = _step_of_type(current, arg) 

68 elif kind in ("where", "resolve"): 

69 pass 

70 if not current: 

71 break 

72 

73 return current 

74 

75 

76def _parse_steps(expr: str) -> list[tuple[str, str]]: 

77 """Parse a dot-notation expression into a list of (kind, arg) steps.""" 

78 steps: list[tuple[str, str]] = [] 

79 raw = _split_dot_respecting_parens(expr) 

80 for token in raw: 

81 token = token.strip() 

82 if not token: 

83 continue 

84 if token.startswith("ofType(") and token.endswith(")"): 

85 steps.append(("ofType", token[7:-1].strip())) 

86 elif token.startswith("as(") and token.endswith(")"): 

87 steps.append(("as", token[3:-1].strip())) 

88 elif token.startswith("where(") and token.endswith(")"): 

89 steps.append(("where", token[6:-1].strip())) 

90 elif token.startswith("extension(") and token.endswith(")"): 

91 # FHIRPath shorthand: extension('url') is essentially extension.where(url='url') 

92 # Our evaluator ignores 'where' clauses but this ensures the 'extension' field 

93 # is correctly matched. 

94 steps.append(("field", "extension")) 

95 steps.append(("where", token[10:-1].strip())) 

96 elif token == "resolve()": 

97 steps.append(("resolve", "")) 

98 else: 

99 steps.append(("field", token)) 

100 return steps 

101 

102 

103def _split_dot_respecting_parens(expr: str) -> list[str]: 

104 """Split on '.' while ignoring dots inside parentheses.""" 

105 parts: list[str] = [] 

106 depth = 0 

107 start = 0 

108 for i, ch in enumerate(expr): 

109 if ch == "(": 

110 depth += 1 

111 elif ch == ")": 

112 depth -= 1 

113 elif ch == "." and depth == 0: 

114 parts.append(expr[start:i]) 

115 start = i + 1 

116 parts.append(expr[start:]) 

117 return parts 

118 

119 

120def _get_field_values(d: dict[str, Any], field_name: str) -> list[Any]: 

121 values: list[Any] = [] 

122 if field_name in d: 

123 values.append(d[field_name]) 

124 else: 

125 for key in d: 

126 if ( 

127 key.startswith(field_name) 

128 and len(key) > len(field_name) 

129 and key[len(field_name)].isupper() 

130 ): 

131 values.append(d[key]) 

132 return values 

133 

134 

135def _step_field(nodes: list[Any], field_name: str) -> list[Any]: 

136 """Traverse ``field_name`` on each node, flattening arrays. 

137 

138 When the literal field is absent, expand FHIR choice elements: ``onset`` 

139 matches ``onsetDateTime``, ``onsetPeriod``, … (a suffix starting uppercase). 

140 """ 

141 result: list[Any] = [] 

142 for node in nodes: 

143 if not isinstance(node, dict): 

144 continue 

145 d: dict[str, Any] = cast("dict[str, Any]", node) 

146 values = _get_field_values(d, field_name) 

147 for value in values: 

148 if value is None: 

149 continue 

150 if isinstance(value, list): 

151 result.extend(cast("list[Any]", value)) 

152 else: 

153 result.append(value) 

154 return result 

155 

156 

157# FHIR primitive types — when an ofType()/as() filter names one of these, the 

158# matching choice value is a JSON primitive (str/number/bool), not an object. 

159_PRIMITIVE_TYPES: frozenset[str] = frozenset( 

160 { 

161 "base64Binary", "boolean", "canonical", "code", "date", "dateTime", 

162 "decimal", "id", "instant", "integer", "markdown", "oid", "positiveInt", 

163 "string", "time", "unsignedInt", "uri", "url", "uuid", 

164 } 

165) # fmt: skip 

166 

167 

168def _step_of_type(nodes: list[Any], type_name: str) -> list[Any]: 

169 """Filter nodes to those matching the given FHIR type, by JSON shape.""" 

170 if type_name in _PRIMITIVE_TYPES: 

171 return [n for n in nodes if not isinstance(n, dict)] 

172 if type_name in ("Quantity", "Age", "Count", "Distance", "Duration", "Money"): 

173 return [n for n in nodes if isinstance(n, dict) and "value" in cast("dict[str, Any]", n)] 

174 if type_name in ("Period", "Range"): 

175 d_keys = ("start", "end", "low", "high") 

176 return [ 

177 n 

178 for n in nodes 

179 if isinstance(n, dict) and any(k in cast("dict[str, Any]", n) for k in d_keys) 

180 ] 

181 if type_name == "CodeableConcept": 

182 return [ 

183 n 

184 for n in nodes 

185 if isinstance(n, dict) 

186 and ("coding" in cast("dict[str, Any]", n) or "text" in cast("dict[str, Any]", n)) 

187 ] 

188 if type_name == "Reference": 

189 return [ 

190 n for n in nodes if isinstance(n, dict) and "reference" in cast("dict[str, Any]", n) 

191 ] 

192 return [n for n in nodes if isinstance(n, dict)]