Coverage for src/lumen/search/parser.py: 90%

1""" 

2This module parses HTTP search query parameters into structured Python representations. 

3It supports search modifiers, prefixes, multi-value parameters, and special search fields. 

4""" 

5 

6from __future__ import annotations 

7 

8from dataclasses import dataclass, field 

9from typing import Literal 

10 

11Modifier = Literal["exact", "contains", "missing", "not", "text", "in", "not-in", ""] 

12Prefix = Literal["eq", "ne", "gt", "lt", "ge", "le", "sa", "eb", "ap", ""] 

13 

14_VALID_MODIFIERS: frozenset[str] = frozenset( 

15 {"exact", "contains", "missing", "not", "text", "in", "not-in"} 

16) 

17_VALID_PREFIXES: frozenset[str] = frozenset({"eq", "ne", "gt", "lt", "ge", "le", "sa", "eb", "ap"}) 

18 

19SortDirection = Literal["asc", "desc"] 

20SummaryMode = Literal["true", "false", "text", "data", "count"] 

21 

22 

23@dataclass(frozen=True) 

24class SearchValue: 

25 """A single search value, possibly with a comparison prefix.""" 

26 

27 raw: str # the value without prefix 

28 prefix: Prefix = "" # comparison prefix (for date/quantity/number) 

29 

30 

31@dataclass 

32class ParsedParam: 

33 """One parsed search parameter clause.""" 

34 

35 name: str # e.g. "gender", "birthdate", "_id" 

36 modifier: Modifier = "" 

37 # Each list entry is an OR group (comma-separated in raw query). 

38 # Multiple ParsedParam with the same name are ANDed together (FHIR spec). 

39 values: list[SearchValue] = field(default_factory=lambda: []) 

40 

41 

42@dataclass 

43class HasParam: 

44 """_has: reverse chaining clause. 

45 

46 Example: Patient?_has:Coverage:beneficiary:type=GKV 

47 → joining_type="Coverage", join_param="beneficiary", filter_param="type" 

48 → finds Patients referenced by a Coverage.beneficiary where Coverage.type=GKV 

49 """ 

50 

51 joining_type: str # resource type that references this type, e.g. "Coverage" 

52 join_param: str # param on joining_type pointing back here, e.g. "beneficiary" 

53 filter_param: str # param on joining_type to filter by, e.g. "type" 

54 values: list[SearchValue] = field(default_factory=lambda: []) 

55 

56 

57@dataclass 

58class ChainedParam: 

59 """Chained search modifier: base_param:chained_param=value. 

60 

61 Example: Encounter?account:identifier=sys|val 

62 → base_param="account", chained_param="identifier" 

63 → finds Encounters whose account reference target has identifier=sys|val 

64 """ 

65 

66 base_param: str # reference param on the searched type, e.g. "account" 

67 chained_param: str # param to check on the referenced resource, e.g. "identifier" 

68 values: list[SearchValue] = field(default_factory=lambda: []) 

69 # Resolved by the query layer before SQL generation (see 

70 # query._resolve_chained_params). ``target_type`` is the optional 

71 # resource-type filter from a typed chain (``subject:Patient.name`` → 

72 # "Patient"); ``chained_field`` is the bare param name on the target with 

73 # any type prefix stripped ("Patient.name" → "name"); ``chained_type`` is 

74 # that field's index type, which selects the spidx_* table to join. 

75 target_type: str | None = None 

76 chained_field: str = "" 

77 chained_type: str | None = None 

78 

79 

80@dataclass 

81class SortSpec: 

82 param: str 

83 direction: SortDirection 

84 

85 

86@dataclass 

87class ParsedQuery: 

88 params: list[ParsedParam] = field(default_factory=lambda: []) 

89 count: int = 20 # _count, default 20 

90 cursor: str | None = None # _page_token (cursor-based pagination) 

91 sort: list[SortSpec] = field(default_factory=lambda: []) 

92 summary: SummaryMode = "false" 

93 elements: list[str] = field(default_factory=lambda: []) 

94 includes: list[str] = field(default_factory=lambda: []) # _include values 

95 revincludes: list[str] = field(default_factory=lambda: []) # _revinclude values 

96 has_params: list[HasParam] = field(default_factory=lambda: []) 

97 chained_params: list[ChainedParam] = field(default_factory=lambda: []) 

98 

99 

100def _parse_sort_param(values: list[str], pq: ParsedQuery) -> None: 

101 for val in values: 

102 for part in val.split(","): 

103 part = part.strip() 

104 if part.startswith("-"): 

105 pq.sort.append(SortSpec(part[1:], "desc")) 

106 else: 

107 pq.sort.append(SortSpec(part, "asc")) 

108 

109 

110def _parse_elements_param(values: list[str], pq: ParsedQuery) -> None: 

111 for val in values: 

112 pq.elements.extend(e.strip() for e in val.split(",") if e.strip()) 

113 

114 

115def _is_special_param( 

116 key: str, values: list[str], pq: ParsedQuery, max_count: int | None = 1000 

117) -> bool: 

118 if key == "_count": 

119 import contextlib 

120 

121 with contextlib.suppress(ValueError): 

122 parsed_val = int(values[-1]) 

123 if max_count is not None: 

124 pq.count = max(1, min(parsed_val, max_count)) 

125 else: 

126 pq.count = max(1, parsed_val) 

127 return True 

128 

129 if key == "_page_token": 

130 pq.cursor = values[-1] 

131 return True 

132 

133 if key == "_sort": 

134 _parse_sort_param(values, pq) 

135 return True 

136 

137 if key == "_summary": 

138 raw_summary = values[-1].lower() 

139 if raw_summary in ("true", "false", "text", "data", "count"): 

140 pq.summary = raw_summary # type: ignore[assignment] 

141 return True 

142 

143 if key == "_elements": 

144 _parse_elements_param(values, pq) 

145 return True 

146 

147 if key == "_include": 

148 pq.includes.extend(values) 

149 return True 

150 

151 if key == "_revinclude": 

152 pq.revincludes.extend(values) 

153 return True 

154 

155 # Handled by content negotiation middleware or ignored safely. 

156 return key in ("_format", "_pretty") 

157 

158 

159def parse_query(raw_params: dict[str, list[str]], max_count: int | None = 1000) -> ParsedQuery: 

160 """Parse a multi-value dict (as returned by Starlette) into a ParsedQuery. 

161 

162 ``raw_params`` maps parameter keys to lists of string values (each call 

163 to the same key appends to the list in Starlette's QueryParams). 

164 ``max_count`` specifies the upper limit for ``_count``. Set to ``None`` for unlimited. 

165 """ 

166 pq = ParsedQuery() 

167 

168 for key, values in raw_params.items(): 

169 # _has: reverse chaining — _has:JoiningType:joinParam:filterParam=value 

170 if key.startswith("_has:"): 

171 _handle_has_param(key, values, pq) 

172 continue 

173 

174 if _is_special_param(key, values, pq, max_count=max_count): 

175 continue 

176 

177 # Split name from modifier/chained-param 

178 name, modifier, chained = _split_param_key(key) 

179 if chained: 

180 pq.chained_params.append( 

181 ChainedParam( 

182 base_param=name, 

183 chained_param=chained, 

184 values=_parse_values(values, ""), 

185 ) 

186 ) 

187 else: 

188 pq.params.append( 

189 ParsedParam( 

190 name=name, 

191 modifier=modifier, 

192 values=_parse_values(values, modifier), 

193 ) 

194 ) 

195 

196 return pq 

197 

198 

199def _handle_has_param(key: str, values: list[str], pq: ParsedQuery) -> None: 

200 """Parse _has:JoiningType:joinParam:filterParam into pq.has_params.""" 

201 parts = key.split(":") 

202 if len(parts) != 4: 

203 return # malformed _has — skip silently 

204 _, joining_type, join_param, filter_param = parts 

205 pq.has_params.append( 

206 HasParam( 

207 joining_type=joining_type, 

208 join_param=join_param, 

209 filter_param=filter_param, 

210 values=_parse_values(values, ""), 

211 ) 

212 ) 

213 

214 

215def _split_param_key(key: str) -> tuple[str, Modifier, str]: 

216 """Split a parameter key into (name, standard_modifier, chained_param). 

217 

218 Returns non-empty ``chained_param`` when the suffix is not a recognised 

219 standard modifier — indicating a chained parameter (e.g. ``account:identifier`` 

220 → ``("account", "", "identifier")``). 

221 Resource-type modifiers (e.g. ``subject:Patient``) are also treated as 

222 chained params so the query engine can apply a type filter on the reference. 

223 

224 Also supports standard FHIR dot-chaining (e.g. ``beneficiary.identifier`` 

225 → ``("beneficiary", "", "identifier")``), which resolves the reference and 

226 filters on a search parameter of the target resource — same underlying 

227 join as the colon-chained modifier, different URL syntax. 

228 """ 

229 if ":" not in key: 

230 if "." in key: 

231 name, chained = key.split(".", 1) 

232 return name, "", chained 

233 return key, "", "" 

234 name, suffix = key.split(":", 1) 

235 if suffix in _VALID_MODIFIERS: 

236 return name, suffix, "" # type: ignore[return-value] 

237 # Not a standard modifier — chained param or resource-type modifier. 

238 return name, "", suffix 

239 

240 

241def _parse_values(raw_list: list[str], modifier: Modifier) -> list[SearchValue]: 

242 """Expand comma-separated OR values and detect prefixes.""" 

243 result: list[SearchValue] = [] 

244 for raw in raw_list: 

245 for part in raw.split(","): 

246 part = part.strip() 

247 if not part: 

248 continue 

249 if modifier == "missing": 

250 # Value must be "true" or "false"; no prefix stripping. 

251 result.append(SearchValue(raw=part, prefix="")) 

252 else: 

253 prefix, bare = _strip_prefix(part) 

254 result.append(SearchValue(raw=bare, prefix=prefix)) 

255 return result 

256 

257 

258def _strip_prefix(value: str) -> tuple[Prefix, str]: 

259 if len(value) >= 2 and value[:2] in _VALID_PREFIXES: 

260 return value[:2], value[2:] # type: ignore[return-value] 

261 return "", value