Coverage for src/lumen/fhir/patch.py: 90%
238 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-31 18:46 +0000
1"""This module implements the FHIRPath Patch protocol (HL7 FHIR R4B standard).
3It parses a Parameters resource and applies operations (add, insert, delete, replace, move)
4to a FHIR resource.
5"""
7from __future__ import annotations
9import copy
10from typing import Any, cast
12import lumen.fhirpath as fhirpathpy # type: ignore[import-untyped]
15def parse_fhirpath_patch(parameters: dict[str, Any]) -> list[dict[str, Any]]:
16 """Parse a FHIR Parameters resource representing a FHIRPath Patch.
18 Returns a list of parsed operation dictionaries.
19 """
20 if parameters.get("resourceType") != "Parameters":
21 raise ValueError("FHIRPath Patch payload must be a Parameters resource")
23 operations: list[dict[str, Any]] = []
24 params: Any = parameters.get("parameter", [])
25 if not isinstance(params, list):
26 raise ValueError("Parameters.parameter must be a list")
28 params_list = cast(list[Any], params)
29 for p in params_list:
30 p_dict: dict[str, Any] = cast(dict[str, Any], p)
31 if p_dict.get("name") != "operation":
32 continue
34 parts: Any = p_dict.get("part")
35 if not isinstance(parts, list):
36 continue
38 parts_list = cast(list[Any], parts)
39 op: dict[str, Any] = {}
40 for part in parts_list:
41 part_dict: dict[str, Any] = cast(dict[str, Any], part)
42 part_name: Any = part_dict.get("name")
43 if not part_name:
44 continue
46 val_keys: list[str] = [k for k in part_dict if k.startswith("value")]
47 if val_keys:
48 op[part_name] = part_dict[val_keys[0]]
49 elif "resource" in part_dict:
50 op[part_name] = part_dict["resource"]
52 if "type" not in op:
53 raise ValueError("Operation part missing 'type'")
54 if "path" not in op:
55 raise ValueError("Operation part missing 'path'")
57 for int_field in ("index", "source", "destination"):
58 if int_field in op:
59 try:
60 op[int_field] = int(op[int_field])
61 except (ValueError, TypeError) as exc:
62 raise ValueError(f"Parameter '{int_field}' must be an integer") from exc
64 operations.append(op)
66 if not operations:
67 raise ValueError("No operation parameters found in Parameters resource")
69 return operations
72def split_fhirpath(path: str) -> tuple[str, str | int]:
73 """Split a FHIRPath expression into (parent_path, terminal_key_or_index)."""
74 path = path.strip()
75 if path.endswith("]"):
76 depth = 0
77 for i in range(len(path) - 1, -1, -1):
78 if path[i] == "]":
79 depth += 1
80 elif path[i] == "[":
81 depth -= 1
82 if depth == 0:
83 parent = path[:i].rstrip(".")
84 index_str = path[i + 1 : -1].strip()
85 try:
86 return parent, int(index_str)
87 except ValueError:
88 return parent, index_str
90 depth_paren = 0
91 depth_bracket = 0
92 for i in range(len(path) - 1, -1, -1):
93 ch = path[i]
94 if ch == ")":
95 depth_paren += 1
96 elif ch == "(":
97 depth_paren -= 1
98 elif ch == "]":
99 depth_bracket += 1
100 elif ch == "[":
101 depth_bracket -= 1
102 elif ch == "." and depth_paren == 0 and depth_bracket == 0:
103 parent = path[:i]
104 field = path[i + 1 :]
105 return parent, field
107 raise ValueError(f"Cannot parse parent path from: {path}")
110def build_parent_map(
111 obj: Any,
112 parent: Any = None,
113 key_or_idx: str | int | None = None,
114 parent_map: dict[int, tuple[Any, str | int]] | None = None,
115) -> dict[int, tuple[Any, str | int]]:
116 """Recursively map dictionary and list children back to their parent collections."""
117 if parent_map is None:
118 parent_map = {}
119 if isinstance(obj, dict):
120 if parent is not None and key_or_idx is not None:
121 parent_map[id(cast(Any, obj))] = (parent, key_or_idx)
122 d_obj: dict[str, Any] = cast(dict[str, Any], obj)
123 for k, v in d_obj.items():
124 build_parent_map(v, obj, k, parent_map)
125 elif isinstance(obj, list):
126 if parent is not None and key_or_idx is not None:
127 parent_map[id(cast(Any, obj))] = (parent, key_or_idx)
128 l_obj: list[Any] = cast(list[Any], obj)
129 for idx, v in enumerate(l_obj):
130 build_parent_map(v, obj, idx, parent_map)
131 return parent_map
134def _apply_delete(resource: dict[str, Any], path: str, resource_type: str) -> None:
135 fhirpathpy_any: Any = fhirpathpy
136 evaluate_fn: Any = fhirpathpy_any.evaluate
137 targets: list[Any] = cast(list[Any], evaluate_fn(resource, path))
138 if not targets:
139 return
141 parent_map: dict[int, tuple[Any, str | int]] = build_parent_map(resource)
142 for target in targets:
143 target_id: int = id(target)
144 if target_id in parent_map:
145 parent, key_or_idx = parent_map[target_id]
146 if isinstance(parent, list):
147 cast(list[Any], parent).remove(target)
148 elif isinstance(parent, dict):
149 del cast(dict[Any, Any], parent)[key_or_idx]
150 else:
151 # Target is a primitive value
152 if path == resource_type:
153 raise ValueError("Cannot delete root resource")
154 parent_path, terminal = split_fhirpath(path)
155 parents: list[Any] = cast(list[Any], evaluate_fn(resource, parent_path))
156 if not parents:
157 continue
158 parent = parents[0]
159 if (isinstance(parent, dict) and isinstance(terminal, str) and terminal in parent) or (
160 isinstance(parent, list)
161 and isinstance(terminal, int)
162 and 0 <= terminal < len(cast(list[Any], parent))
163 ):
164 if isinstance(parent, dict):
165 del cast(dict[Any, Any], parent)[terminal]
166 else:
167 del cast(list[Any], parent)[cast(int, terminal)]
170def _apply_replace(resource: dict[str, Any], path: str, value: Any, resource_type: str) -> None:
171 if path == resource_type:
172 raise ValueError("Cannot replace root resource")
173 parent_path, terminal = split_fhirpath(path)
174 fhirpathpy_any: Any = fhirpathpy
175 evaluate_fn: Any = fhirpathpy_any.evaluate
176 parents: list[Any] = cast(list[Any], evaluate_fn(resource, parent_path))
177 if not parents:
178 raise ValueError(f"Parent path not found: {parent_path}")
179 if len(parents) > 1:
180 raise ValueError(f"Path resolved to multiple parents: {parent_path}")
181 parent = parents[0]
182 if isinstance(parent, dict) and isinstance(terminal, str):
183 cast(dict[str, Any], parent)[terminal] = value
184 elif isinstance(parent, list) and isinstance(terminal, int):
185 if 0 <= terminal < len(cast(list[Any], parent)):
186 cast(list[Any], parent)[terminal] = value
187 else:
188 raise ValueError(f"Index {terminal} out of range for parent list")
189 else:
190 raise ValueError("Parent shape or terminal type mismatch")
193def _apply_add(
194 resource: dict[str, Any], path: str, value: Any, name: str | None, resource_type: str
195) -> None:
196 if path == resource_type:
197 if not name:
198 raise ValueError(
199 "For 'add' operation on the root resource, 'name' parameter is required"
200 )
201 if name in resource:
202 target: Any = resource[name]
203 if isinstance(target, list):
204 cast(list[Any], target).append(value)
205 else:
206 resource[name] = value
207 else:
208 resource[name] = value
209 return
211 parent_path, terminal = split_fhirpath(path)
212 fhirpathpy_any: Any = fhirpathpy
213 evaluate_fn: Any = fhirpathpy_any.evaluate
214 parents: list[Any] = cast(list[Any], evaluate_fn(resource, parent_path))
215 if not parents:
216 raise ValueError(f"Parent path not found: {parent_path}")
217 parent = parents[0]
219 if isinstance(parent, dict) and isinstance(terminal, str):
220 p_dict = cast(dict[str, Any], parent)
221 if terminal in p_dict:
222 target: Any = p_dict[terminal]
223 if isinstance(target, list):
224 cast(list[Any], target).append(value)
225 elif isinstance(target, dict):
226 if not name:
227 raise ValueError(
228 "For 'add' operation on an object, 'name' parameter is required"
229 )
230 cast(dict[str, Any], target)[name] = value
231 else:
232 p_dict[terminal] = value
233 else:
234 if name:
235 p_dict[terminal] = {name: value}
236 else:
237 p_dict[terminal] = value
238 else:
239 raise ValueError("Parent must be an object for adding fields")
242def _resolve_collection(resource: dict[str, Any], path: str) -> list[Any]:
243 parent_path, terminal = split_fhirpath(path)
244 fhirpathpy_any: Any = fhirpathpy
245 evaluate_fn: Any = fhirpathpy_any.evaluate
246 parents: list[Any] = cast(list[Any], evaluate_fn(resource, parent_path))
247 if not parents:
248 raise ValueError(f"Parent path not found: {parent_path}")
249 parent = parents[0]
251 if isinstance(parent, dict) and isinstance(terminal, str):
252 p_dict = cast(dict[str, Any], parent)
253 if terminal not in p_dict:
254 raise ValueError(f"Collection field {terminal} not found on parent")
255 collection: Any = p_dict[terminal]
256 elif isinstance(parent, list) and isinstance(terminal, int):
257 p_list = cast(list[Any], parent)
258 collection: Any = p_list[terminal]
259 else:
260 raise ValueError("Invalid parent shape")
262 if not isinstance(collection, list):
263 raise ValueError(f"Target at {path} is not a collection (list)")
264 return cast(list[Any], collection)
267def _apply_insert(
268 resource: dict[str, Any],
269 path: str,
270 value: Any,
271 index: int | None,
272 resource_type: str,
273) -> None:
274 if path == resource_type:
275 raise ValueError("Cannot insert into root resource path directly")
276 if index is None:
277 raise ValueError("'index' parameter is required for 'insert' operation")
278 collection = _resolve_collection(resource, path)
279 if index < 0 or index > len(collection):
280 raise ValueError(f"Index {index} out of range for collection of length {len(collection)}")
281 collection.insert(index, value)
284def _apply_move(
285 resource: dict[str, Any],
286 path: str,
287 source: int | None,
288 destination: int | None,
289 resource_type: str,
290) -> None:
291 if path == resource_type:
292 raise ValueError("Cannot move root resource")
293 if source is None or destination is None:
294 raise ValueError("'source' and 'destination' parameters are required for 'move' operation")
295 collection = _resolve_collection(resource, path)
296 if source < 0 or source >= len(collection):
297 raise ValueError(f"Source index {source} out of bounds")
298 if destination < 0 or destination >= len(collection):
299 raise ValueError(f"Destination index {destination} out of bounds")
300 item = collection.pop(source)
301 collection.insert(destination, item)
304def apply_fhirpath_patch(
305 resource: dict[str, Any],
306 patch_operations: list[dict[str, Any]],
307) -> dict[str, Any]:
308 """Apply parsed FHIRPath Patch operations to a copy of the resource.
310 Returns the updated resource.
311 """
312 resource = copy.deepcopy(resource)
313 resource_type = resource.get("resourceType", "")
315 for op in patch_operations:
316 op_type = op.get("type")
317 path = op.get("path")
318 value = op.get("value")
319 name = op.get("name")
320 index = op.get("index")
321 source = op.get("source")
322 destination = op.get("destination")
324 if not op_type or not path:
325 raise ValueError("Operation must have 'type' and 'path'")
327 if op_type == "delete":
328 _apply_delete(resource, path, resource_type)
329 elif op_type == "replace":
330 _apply_replace(resource, path, value, resource_type)
331 elif op_type == "add":
332 _apply_add(resource, path, value, name, resource_type)
333 elif op_type == "insert":
334 _apply_insert(resource, path, value, index, resource_type)
335 elif op_type == "move":
336 _apply_move(resource, path, source, destination, resource_type)
337 else:
338 raise ValueError(f"Unsupported operation type: {op_type}")
340 return resource