Coverage for src/lumen/fhir/search_param_loader.py: 95%
81 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-31 18:46 +0000
1"""
2This module loads and compiles base search parameters from the HL7 JSON bundle.
3It extracts search paths, maps parameter types to tables, and indexes them by resource type.
4"""
6from __future__ import annotations
8import json
9from functools import lru_cache
10from pathlib import Path
11from typing import Any, cast
13from .params import CompositeComponent, CompositeComponentType, SearchParamDef, SearchParamType
14from .resource_types import known_resource_types
16_BUNDLE_PATH = Path(__file__).parent / "data" / "r4b" / "r4b-search-parameters.json"
18# FHIR SearchParameter.type → our internal index type. Types not listed have
19# no index table and are skipped. "composite" resolves to a dedicated
20# spidx_composite table (see _resolve_components below); its own components
21# are individually one of the types below. "special" only ever appears as
22# Location's ``near`` in the R4B bundle — it maps to a dedicated geo table.
23_TYPE_MAP: dict[str, SearchParamType] = {
24 "string": "string",
25 "token": "token",
26 "uri": "token",
27 "date": "date",
28 "reference": "reference",
29 "quantity": "quantity",
30 "number": "number",
31 "composite": "composite",
32 "special": "geo",
33}
35# Component types a composite search parameter can be built from (MVP scope —
36# every R4B composite parameter with exactly 2 components uses these).
37_COMPOSITE_COMPONENT_TYPES = frozenset({"token", "quantity"})
39_UNIVERSAL_BASES = frozenset({"Resource", "DomainResource"})
41# FHIR "search result" parameters (spec-defined, not registered as SearchParameter
42# resources) that every resource type implicitly supports. The base R4B bundle has
43# no entries for these, so they must be added to every resource's declared param
44# list explicitly for CapabilityStatement conformance.
45#
46# ``_has`` (reverse chaining) is defined in the bundle with base=["Resource"] but
47# has no ``expression`` (it's a control parameter, not a FHIRPath-based one), so
48# the generic loader below drops it. It's fully implemented in search/parser.py
49# and search/query.py — this only fixes its CapabilityStatement declaration.
50_UNIVERSAL_RESULT_PARAMS: list[tuple[str, SearchParamType, str]] = [
51 ("_count", "number", "http://hl7.org/fhir/SearchParameter/Resource-count"),
52 ("_has", "string", "http://hl7.org/fhir/SearchParameter/Resource-has"),
53]
56def _branches_for(expression: str, target_type: str) -> list[str]:
57 """Reduce a union expression to the branches applicable to ``target_type``.
59 Branches headed by the target type are kept verbatim; branches headed by a
60 universal base (``Resource``/``DomainResource``) are rewritten onto the
61 target type. Branches for other resource types are dropped.
62 """
63 branches: list[str] = []
64 for raw in expression.split("|"):
65 branch = raw.strip()
66 if not branch:
67 continue
68 head = branch.split(".", 1)[0].lstrip("(").strip()
69 if head == target_type:
70 branches.append(branch)
71 elif head in _UNIVERSAL_BASES:
72 _, _, rest = branch.partition(".")
73 branches.append(f"{target_type}.{rest}" if rest else target_type)
74 return branches
77def _register_search_param_for_target(
78 target: str,
79 code: str,
80 internal_type: SearchParamType,
81 declared_type: str | None,
82 expression: str,
83 acc: dict[str, dict[str, SearchParamDef]],
84 definition: str = "",
85 components: tuple[CompositeComponent, ...] | None = None,
86) -> None:
87 paths = _branches_for(expression, target)
88 if not paths:
89 return
90 by_code = acc.setdefault(target, {})
91 existing = by_code.get(code)
92 if existing is None:
93 by_code[code] = SearchParamDef(
94 name=code,
95 type=internal_type,
96 paths=paths,
97 definition=definition,
98 declared_type=declared_type,
99 components=components,
100 )
101 else:
102 merged = existing.paths + [p for p in paths if p not in existing.paths]
103 by_code[code] = SearchParamDef(
104 name=code,
105 type=existing.type,
106 paths=merged,
107 definition=existing.definition or definition,
108 declared_type=existing.declared_type or declared_type,
109 components=existing.components or components,
110 )
113def _resolve_components(
114 sp: dict[str, Any], url_to_type: dict[str, str]
115) -> tuple[CompositeComponent, ...] | None:
116 """Resolve a composite SearchParameter's ``component[]`` into typed,
117 directly-usable definitions.
119 Each component references another SearchParameter by canonical URL for
120 its type; MVP only supports the (token, token) and (token, quantity)
121 shapes actually present in the R4B bundle's 2-component composites.
122 Returns None (caller skips registration) for anything wider.
123 """
124 raw_components = sp.get("component", [])
125 if len(raw_components) != 2:
126 return None
127 resolved: list[CompositeComponent] = []
128 for comp in raw_components:
129 comp_type = url_to_type.get(comp.get("definition", ""))
130 comp_expr = comp.get("expression")
131 if comp_type not in _COMPOSITE_COMPONENT_TYPES or not comp_expr:
132 return None
133 resolved.append(
134 CompositeComponent(type=cast("CompositeComponentType", comp_type), expression=comp_expr)
135 )
136 if resolved[0].type != "token":
137 return None # MVP: first component must be a token (matches all known cases)
138 return tuple(resolved)
141def _register_search_param(
142 sp: dict[str, Any],
143 concrete: frozenset[str],
144 acc: dict[str, dict[str, SearchParamDef]],
145 url_to_type: dict[str, str],
146) -> None:
147 fhir_type = sp.get("type", "")
148 internal_type = _TYPE_MAP.get(fhir_type)
149 code = sp.get("code")
150 expression = sp.get("expression")
151 definition = sp.get("url", "")
152 if internal_type is None or not code or not expression:
153 return
154 # Only carry a declared_type when it diverges from the internal index type
155 # (e.g. "uri" indexes as "token" but must still be advertised as "uri").
156 declared_type = fhir_type if fhir_type != internal_type else None
158 components: tuple[CompositeComponent, ...] | None = None
159 if internal_type == "composite":
160 components = _resolve_components(sp, url_to_type)
161 if components is None:
162 return # unsupported composite shape — skip rather than mis-declare
164 for base in sp.get("base", []):
165 targets = concrete if base in _UNIVERSAL_BASES else (base,)
166 for target in targets:
167 if target in concrete:
168 _register_search_param_for_target(
169 target,
170 code,
171 internal_type,
172 declared_type,
173 expression,
174 acc,
175 definition,
176 components,
177 )
180@lru_cache(maxsize=1)
181def load_registry() -> dict[str, list[SearchParamDef]]:
182 bundle: dict[str, Any] = json.loads(_BUNDLE_PATH.read_text())
183 concrete = known_resource_types()
185 all_search_params = [
186 entry["resource"]
187 for entry in bundle.get("entry", [])
188 if entry.get("resource", {}).get("resourceType") == "SearchParameter"
189 ]
190 url_to_type = {sp["url"]: sp["type"] for sp in all_search_params if sp.get("url")}
192 # target_type → param_code → SearchParamDef (paths merged across definitions)
193 acc: dict[str, dict[str, SearchParamDef]] = {}
195 for sp in all_search_params:
196 _register_search_param(sp, concrete, acc, url_to_type)
198 # Declare universal result parameters (_count, ...) on every resource type
199 # for CapabilityStatement conformance. Not indexed: paths=[] is a no-op in
200 # the indexer, and the query parser intercepts these before registry lookup.
201 for target in acc:
202 for code, param_type, definition in _UNIVERSAL_RESULT_PARAMS:
203 acc[target].setdefault(
204 code, SearchParamDef(name=code, type=param_type, paths=[], definition=definition)
205 )
207 return {target: list(by_code.values()) for target, by_code in acc.items()}