Coverage for src/lumen/fhir/profile_validator.py: 84%
358 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-31 18:46 +0000
1"""
2Native Python Profile Validator for FHIR resources.
3Implements Level 1 (FHIRPath Invariants) and Level 2 (Cardinality)
4validation against StructureDefinitions.
5"""
7import typing
8from typing import Any
9from uuid import UUID
11import asyncpg
12import structlog
14from lumen.fhirpath import evaluate as fhirpath_evaluate # type: ignore
16from .terminology_binding import check_required_binding
18logger = structlog.get_logger(__name__)
21class ProfileDefinitionCache:
22 """Per-tenant cache of a profile's ElementDefinition list, keyed by canonical URL.
24 Mirrors `ValueSetExpansionCache` (`terminology_binding.py`): every profile
25 validation used to `fetchrow` the StructureDefinition once per profile URL
26 per write. Cached here and coarse-invalidated on any StructureDefinition
27 write/delete (see `ResourceRepository._post_write`/`_post_delete`).
29 Value is the resolved ``snapshot`` (or ``differential``) element list, or
30 ``None`` when no StructureDefinition with that url exists on the server —
31 the ``None`` case is a real, cacheable answer that callers turn into a
32 "not found on server" error, so it must be memoized too.
33 """
35 def __init__(self) -> None:
36 self._cache: dict[tuple[UUID, str], list[dict[str, Any]] | None] = {}
38 def invalidate(self, tenant_id: UUID) -> None:
39 for key in [k for k in self._cache if k[0] == tenant_id]:
40 del self._cache[key]
42 def invalidate_all(self) -> None:
43 self._cache.clear()
45 async def get_elements(
46 self,
47 conn: asyncpg.Connection | Any,
48 tenant_id: UUID,
49 profile_url: str,
50 ) -> list[dict[str, Any]] | None:
51 key = (tenant_id, profile_url)
52 if key not in self._cache:
53 self._cache[key] = await _fetch_profile_elements(conn, tenant_id, profile_url)
54 return self._cache[key]
57_profile_cache: ProfileDefinitionCache | None = None
60def get_profile_cache() -> ProfileDefinitionCache:
61 global _profile_cache
62 if _profile_cache is None:
63 _profile_cache = ProfileDefinitionCache()
64 return _profile_cache
67async def _fetch_profile_elements(
68 conn: asyncpg.Connection | Any, tenant_id: UUID, profile_url: str
69) -> list[dict[str, Any]] | None:
70 """Load a StructureDefinition by canonical url and return its snapshot (or
71 differential) element list, or ``None`` if no such profile exists."""
72 row = await conn.fetchrow(
73 """
74 SELECT resource
75 FROM fhir_resources
76 WHERE tenant_id = $1
77 AND resource_type = 'StructureDefinition'
78 AND resource->>'url' = $2
79 AND deleted = FALSE
80 ORDER BY version_id DESC
81 LIMIT 1
82 """,
83 tenant_id,
84 profile_url,
85 )
86 if not row:
87 return None
88 profile_def = row["resource"]
89 snapshot = profile_def.get("snapshot", {})
90 differential = profile_def.get("differential", {})
91 return snapshot.get("element") or differential.get("element", [])
94def _get_values_at_path(
95 data: dict[str, Any] | list[Any] | str | int | float | bool | None,
96 path_parts: list[str],
97) -> list[Any]:
98 """
99 Simple tree walker to extract values at a nested path.
100 Handles FHIR choices like `value[x]` automatically mapping to
101 `valueString`, `valueQuantity`, etc.
103 ``data`` may be any JSON node reached while descending a StructureDefinition
104 element path — including primitives (str, int, bool) once the path runs
105 past the actual data depth (e.g. the implicit `.extension`/`.id`
106 sub-elements every FHIR primitive type has). Primitives have no children,
107 so descending further simply finds nothing.
108 """
109 if not path_parts:
110 return data if isinstance(data, list) else [data]
112 current_part = path_parts[0]
113 remaining_parts = path_parts[1:]
115 # Handle paths with choices like value[x]
116 if current_part.endswith("[x]"):
117 base_name = current_part[:-3]
118 results: list[Any] = []
119 if isinstance(data, list):
120 for item in data:
121 if isinstance(item, dict):
122 item_dict = typing.cast(dict[str, Any], item)
123 for k, v in item_dict.items():
124 if k.startswith(base_name):
125 results.extend(_get_values_at_path(v, remaining_parts))
126 elif isinstance(data, dict):
127 for k, v in data.items():
128 if k.startswith(base_name):
129 results.extend(_get_values_at_path(v, remaining_parts))
130 return results
132 if isinstance(data, list):
133 results = []
134 for item in data:
135 if isinstance(item, dict):
136 item_dict = typing.cast(dict[str, Any], item)
137 val = item_dict.get(current_part)
138 if val is not None:
139 results.extend(_get_values_at_path(val, remaining_parts))
140 return results
141 elif isinstance(data, dict):
142 val = data.get(current_part)
143 if val is not None:
144 return _get_values_at_path(val, remaining_parts)
145 # Primitive value (str, int, bool, ...) with remaining path parts left to
146 # descend into — nothing further to find.
147 return []
150def _extract_pattern(el: dict[str, Any]) -> Any | None:
151 """
152 Return the pattern/fixed value discriminating a slice (e.g. ``patternCoding``,
153 ``fixedUri``), or ``None`` if the slice-root element declares none.
155 Extension-array slices are the one common case with neither: FHIR profiles
156 slice ``extension`` by which canonical URL each entry's own StructureDefinition
157 has, expressed as ``type: [{"code": "Extension", "profile": [url]}]`` on the
158 slice-root rather than a ``pattern``/``fixed`` value — but by definition every
159 instance of that extension has ``extension.url == profile[0]``, so it can be
160 expressed as an equivalent pattern for the existing `_pattern_matches` bucketing.
161 """
162 for key, val in el.items():
163 if key.startswith(("pattern", "fixed")):
164 return val
165 for t in el.get("type", []):
166 if t.get("code") == "Extension" and t.get("profile"):
167 return {"url": t["profile"][0]}
168 return None
171def _extract_type_codes(el: dict[str, Any]) -> list[str] | None:
172 """
173 Return the FHIR type codes an element's ``type`` array declares (e.g.
174 ``[{"code": "Reference"}]``), used as the discriminator for slices on a
175 choice (``[x]``) element — each type-slice of e.g. ``medication[x]``
176 re-declares ``path: "...medication[x]"`` with ``type`` narrowed to just
177 that one datatype, ``sliceName`` matching (``medicationReference``,
178 ``medicationCodeableConcept``, ...), and no pattern/fixed value.
179 """
180 types = el.get("type")
181 if not isinstance(types, list) or not types:
182 return None
183 codes: list[str] = []
184 for t in typing.cast("list[dict[str, Any]]", types):
185 code = t.get("code")
186 if isinstance(code, str):
187 codes.append(code)
188 return codes or None
191def _pattern_matches(value: Any, pattern: Any) -> bool:
192 """
193 Structural subset match: ``value`` matches ``pattern`` if every key/value pair
194 in ``pattern`` is present (recursively) in ``value``. Used to bucket array items
195 into FHIR slices discriminated by a ``pattern``/``value`` rule on ``$this``.
196 """
197 if isinstance(pattern, dict):
198 if not isinstance(value, dict):
199 return False
200 pattern_dict = typing.cast(dict[str, Any], pattern)
201 value_dict = typing.cast(dict[str, Any], value)
202 return all(
203 k in value_dict and _pattern_matches(value_dict[k], v) for k, v in pattern_dict.items()
204 )
205 return value == pattern
208def _build_slice_patterns(elements: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
209 """
210 Map ``{base_path: {sliceName: pattern}}`` from the slice-root elements
211 (``id`` like ``Composition.type.coding:KDL``, no further ``.`` after the slice
212 name). ``pattern``/``value``-on-``$this`` discriminators are supported directly;
213 for slices on a choice (``[x]``) element with no pattern/fixed value, the type
214 codes from `_extract_type_codes` are stored instead (see `_get_choice_typed_values`)
215 — this is FHIR's own type-slicing convention (e.g. ``medication[x]:medicationReference``),
216 not an ISiK-specific rule. Any other discriminator kind still falls back to `None`,
217 which downstream treats the slice as matching every item (the pre-slicing behavior).
218 """
219 slice_patterns: dict[str, dict[str, Any]] = {}
220 for el in elements:
221 slice_name = el.get("sliceName")
222 el_id = el.get("id", "")
223 if not slice_name or ":" not in el_id:
224 continue
225 tail = el_id.split(":", 1)[1]
226 if "." in tail:
227 continue # descendant of a slice, not the slice root
228 base_path = el.get("path", "")
229 pattern = _extract_pattern(el)
230 if pattern is None and base_path.split(".")[-1].endswith("[x]"):
231 pattern = _extract_type_codes(el)
232 slice_patterns.setdefault(base_path, {})[slice_name] = pattern
233 return slice_patterns
236def _extract_matching_choice_values(
237 item_dict: dict[str, Any], base_name: str, type_codes: list[str]
238) -> list[Any]:
239 results: list[Any] = []
240 for k, v in item_dict.items():
241 if k.startswith(base_name):
242 suffix = k[len(base_name) :]
243 if any(suffix.lower() == tc.lower() for tc in type_codes):
244 results.append(v)
245 return results
248def _get_choice_typed_values(
249 data: dict[str, Any] | list[Any] | str | int | float | bool | None,
250 path_parts: list[str],
251 type_codes: list[str],
252) -> list[Any]:
253 """
254 Like `_get_values_at_path`, but for a terminal ``[x]`` path part, keeps only
255 values whose originating JSON key's type suffix matches one of ``type_codes``
256 (FHIR's ``value[x]`` -> ``valueString``/``valueCoding``/... naming convention).
257 Used to discriminate type-sliced choice elements, e.g. a resource populating
258 ``medicationCodeableConcept`` must not be treated as matching the
259 ``medicationReference`` slice.
260 """
261 if not path_parts:
262 return data if isinstance(data, list) else [data]
264 current_part = path_parts[0]
265 remaining_parts = path_parts[1:]
266 is_terminal_choice = current_part.endswith("[x]") and not remaining_parts
268 candidates = data if isinstance(data, list) else [data]
269 results: list[Any] = []
270 for item in candidates:
271 if not isinstance(item, dict):
272 continue
273 item_dict = typing.cast(dict[str, Any], item)
274 if is_terminal_choice:
275 base_name = current_part[:-3]
276 results.extend(_extract_matching_choice_values(item_dict, base_name, type_codes))
277 else:
278 val = item_dict.get(current_part)
279 if val is not None:
280 results.extend(_get_choice_typed_values(val, remaining_parts, type_codes))
281 return results
284def _check_cardinality(
285 count: int, min_val: int | None, max_val: str | None, label: str
286) -> list[str]:
287 errors: list[str] = []
288 if min_val is not None and count < min_val:
289 errors.append(f"{label} min cardinality is {min_val}, but found {count}.")
290 if max_val is not None and max_val != "*":
291 try:
292 max_int = int(max_val)
293 except ValueError:
294 return errors
295 if count > max_int:
296 errors.append(f"{label} max cardinality is {max_int}, but found {count}.")
297 return errors
300def _eval_single_constraint(
301 val_ctx: Any,
302 fhirpath_expr: str | dict[str, str],
303 expr: str,
304 key: str,
305 path: str,
306 suffix: str,
307 profile_url: str,
308) -> str | None:
309 try:
310 result = typing.cast(Any, fhirpath_evaluate(val_ctx, fhirpath_expr))
311 if (isinstance(result, list) and result == [False]) or result is False:
312 return f"Profile {profile_url}: Constraint {key} ({expr}) violated at {path}{suffix}."
313 except Exception as e:
314 logger.debug("fhirpath.eval_failed", expr=expr, path=path, error=str(e))
315 return None
318def _evaluate_constraints(
319 el: dict[str, Any], found_values: list[Any], path: str, profile_url: str, suffix: str = ""
320) -> list[str]:
321 """
322 Constraints are FHIRPath invariants evaluated with the element's own type as
323 root context (e.g. ``extension.exists() != value.exists()`` on an ``Extension``
324 needs ``value`` to resolve polymorphically to ``valueCoding``/``valueString``/...
325 — that only happens if the FHIRPath engine knows the root is an ``Extension``).
326 ``found_values`` here are raw dicts sliced out of the resource by path, with no
327 such context attached, so it must be passed explicitly via the ``base`` param.
328 """
329 type_codes = _extract_type_codes(el)
330 base_type = type_codes[0] if type_codes else None
332 errors: list[str] = []
333 for constraint in el.get("constraint", []):
334 expr = constraint.get("expression")
335 key = constraint.get("key", "unknown")
336 if not expr or not found_values:
337 continue
338 fhirpath_expr: str | dict[str, str] = (
339 {"expression": expr, "base": base_type} if base_type else expr
340 )
341 for val_ctx in found_values:
342 err = _eval_single_constraint(
343 val_ctx, fhirpath_expr, expr, key, path, suffix, profile_url
344 )
345 if err:
346 errors.append(err)
347 return errors
350def _resolve_field_path_parts(path: str, resource_type: str) -> list[str]:
351 parts = path.split(".")
352 if len(parts) > 1 and parts[0] == resource_type:
353 return parts[1:]
354 return []
357def _match_slice_items(
358 el_id: str,
359 resource: dict[str, Any],
360 resource_type: str,
361 slice_patterns: dict[str, dict[str, Any]],
362) -> tuple[str, str, list[Any]]:
363 """
364 Resolve which array items on `resource` belong to a slice-scoped element's
365 slice (``id`` contains ``:sliceName``). Shared by `_validate_sliced_element`
366 (Level 4 cardinality/constraints) and `_element_scoped_values` (Level 3
367 bindings), so both see the same slice bucketing instead of Level 3 checking
368 a slice-descendant's binding against the whole unsliced array.
370 Returns ``(base_path, tail, matched)`` — ``tail`` is the id suffix after the
371 slice name (empty of further "." for a slice-root, dotted for a descendant),
372 ``matched`` is the list of resource array items bucketed into that slice.
373 """
374 # The array path is the id prefix before ":sliceName" — el["path"] already has
375 # the descendant's own field suffix appended (e.g. "...coding.code"), so it
376 # can't be used to locate the array itself. Only slice-root elements carry a
377 # "sliceName" key — descendants (e.g. "...coding:KDL.code") only encode it in
378 # "id", so the slice name must always be parsed from "id", not "sliceName".
379 base_path, tail = el_id.split(":", 1)
380 slice_name = tail.split(".", 1)[0]
381 field_path_parts = _resolve_field_path_parts(base_path, resource_type)
382 discriminator = slice_patterns.get(base_path, {}).get(slice_name)
384 if base_path.split(".")[-1].endswith("[x]") and isinstance(discriminator, list):
385 # Choice-type slice (e.g. "medication[x]:medicationReference") — discriminator
386 # is the slice's declared `type` codes, not a pattern/value to structurally match.
387 type_codes = typing.cast("list[str]", discriminator)
388 matched = _get_choice_typed_values(resource, field_path_parts, type_codes)
389 else:
390 all_items = _get_values_at_path(resource, field_path_parts)
391 matched = (
392 [item for item in all_items if _pattern_matches(item, discriminator)]
393 if discriminator is not None
394 else all_items
395 )
396 return base_path, tail, matched
399def _element_scoped_values(
400 el: dict[str, Any],
401 resource: dict[str, Any],
402 resource_type: str,
403 slice_patterns: dict[str, dict[str, Any]],
404) -> list[Any]:
405 """
406 Values an ElementDefinition's path resolves to on `resource`, respecting
407 slice membership for slice-scoped elements. Used by Level 3 binding checks
408 so a binding declared on one slice's descendant (e.g.
409 ``Patient.identifier:MRN.type``) is only checked against that slice's own
410 values, not against sibling slices sharing the same unsliced field path
411 (e.g. ``identifier:Versicherung.type``).
412 """
413 el_id = typing.cast(str, el.get("id", ""))
414 if ":" not in el_id:
415 path_parts = _resolve_field_path_parts(typing.cast(str, el.get("path", "")), resource_type)
416 return _get_values_at_path(resource, path_parts) if path_parts else []
418 _, tail, matched = _match_slice_items(el_id, resource, resource_type, slice_patterns)
419 if "." not in tail:
420 return matched
422 rel_parts = tail.split(".", 1)[1].split(".")
423 found_values: list[Any] = []
424 for item in matched:
425 found_values.extend(_get_values_at_path(item, rel_parts))
426 return found_values
429def _validate_sliced_element(
430 el: dict[str, Any],
431 resource: dict[str, Any],
432 resource_type: str,
433 slice_patterns: dict[str, dict[str, Any]],
434 profile_url: str,
435) -> list[str]:
436 """
437 Validate a slice-scoped element (``id`` contains ``:sliceName``) against only
438 the array items belonging to that slice, instead of the whole (unsliced) array
439 — see `docs/architecture/profile-validation.md` Level 4.
440 """
441 el_id = typing.cast(str, el.get("id", ""))
442 base_path, tail, matched = _match_slice_items(el_id, resource, resource_type, slice_patterns)
443 slice_name = tail.split(".", 1)[0]
444 field_path_parts = _resolve_field_path_parts(base_path, resource_type)
446 if "." not in tail:
447 # Slice root: cardinality applies to how many items matched this slice.
448 # Gate on the array's parent container existing: a required slice inside
449 # an *optional* parent (e.g. DocumentReference.category is 0..1 but its
450 # category.coding:XDS slice is 1..1) must not fire when the parent is
451 # absent — there is nothing to contain the slice. Same nesting principle
452 # as _validate_flat_element / _resolve_containers, applied to slices.
453 if len(field_path_parts) > 1:
454 parent_containers = _resolve_containers(resource, field_path_parts[:-1])
455 if not parent_containers:
456 return []
457 label = f"Profile {profile_url}: Slice {slice_name} at {base_path}"
458 return _check_cardinality(len(matched), el.get("min"), el.get("max"), label)
460 if not matched:
461 # No instance of this (optional) slice exists — descendant field rules
462 # like "system min 1" describe what a slice instance must look like if
463 # present, not that the slice itself must exist (slice-root cardinality
464 # already covers that).
465 return []
467 rel_parts = tail.split(".", 1)[1].split(".")
468 found_values: list[Any] = []
469 for item in matched:
470 found_values.extend(_get_values_at_path(item, rel_parts))
472 path = el.get("path", "")
473 label = f"Profile {profile_url}: Element {path} (slice {slice_name})"
474 errors = _check_cardinality(len(found_values), el.get("min"), el.get("max"), label)
475 errors.extend(
476 _evaluate_constraints(el, found_values, path, profile_url, suffix=f" (slice {slice_name})")
477 )
478 return errors
481def _step_into(node: Any, part: str) -> list[Any]:
482 """One step of `_resolve_containers`: read `part` off every dict in `node`
483 (or off `node` itself if it's a single dict), handling `value[x]` choices."""
484 candidates: list[Any] = typing.cast("list[Any]", node) if isinstance(node, list) else [node]
485 results: list[Any] = []
486 for candidate in candidates:
487 if not isinstance(candidate, dict):
488 continue
489 candidate_dict = typing.cast(dict[str, Any], candidate)
490 if part.endswith("[x]"):
491 base_name = part[:-3]
492 results.extend(v for k, v in candidate_dict.items() if k.startswith(base_name))
493 continue
494 val = candidate_dict.get(part)
495 if val is not None:
496 results.append(val)
497 return results
500def _resolve_containers(resource: dict[str, Any], parent_parts: list[str]) -> list[Any]:
501 """
502 Resolve every "container" instance directly enclosing a leaf field, branching
503 into every repeating (list) element along `parent_parts`. A leaf's cardinality
504 must be checked per container instance, not flattened across all of them —
505 e.g. each `attester` entry independently needs its own `mode`, regardless of
506 how many `attester` entries exist (including zero, since `attester` itself is
507 optional: zero containers means nothing to check, not a violation).
508 """
509 nodes: list[Any] = [resource]
510 for part in parent_parts:
511 next_nodes: list[Any] = []
512 for node in nodes:
513 next_nodes.extend(_step_into(node, part))
514 nodes = next_nodes
515 if not nodes:
516 return []
517 # The last step may have produced list-valued nodes (if the last parent part
518 # is itself repeating) — flatten one more level so callers get individual
519 # container instances, not the arrays holding them.
520 flattened: list[Any] = []
521 for node in nodes:
522 if isinstance(node, list):
523 flattened.extend(typing.cast("list[Any]", node))
524 else:
525 flattened.append(node)
526 return flattened
529def _validate_flat_element(
530 el: dict[str, Any], resource: dict[str, Any], resource_type: str, profile_url: str
531) -> list[str]:
532 """Validate an unsliced element's cardinality (per enclosing instance) and constraints."""
533 path = typing.cast(str, el.get("path"))
534 path_parts = _resolve_field_path_parts(path, resource_type)
535 if not path_parts:
536 return []
538 errors: list[str] = []
539 if el.get("min") is not None or el.get("max") is not None:
540 leaf = path_parts[-1]
541 label = f"Profile {profile_url}: Element {path}"
542 for container in _resolve_containers(resource, path_parts[:-1]):
543 if not isinstance(container, dict):
544 continue
545 container_dict = typing.cast(dict[str, Any], container)
546 count = len(_get_values_at_path(container_dict, [leaf]))
547 errors.extend(_check_cardinality(count, el.get("min"), el.get("max"), label))
549 if el.get("constraint"):
550 found_values = _get_values_at_path(resource, path_parts)
551 errors.extend(_evaluate_constraints(el, found_values, path, profile_url))
553 return errors
556def _validate_against_elements(
557 elements: list[dict[str, Any]], resource: dict[str, Any], resource_type: str, profile_url: str
558) -> list[str]:
559 errors: list[str] = []
560 slice_patterns = _build_slice_patterns(elements)
561 for el in elements:
562 if not el.get("path"):
563 continue
564 if ":" in el.get("id", ""):
565 # Level 4: slice-root or slice-descendant element (id contains
566 # ":sliceName") — validate against the matching bucket only, not the
567 # whole (unsliced) array.
568 errors.extend(
569 _validate_sliced_element(el, resource, resource_type, slice_patterns, profile_url)
570 )
571 else:
572 errors.extend(_validate_flat_element(el, resource, resource_type, profile_url))
573 return errors
576async def _collect_profile_errors(
577 profile_url: str,
578 resource: dict[str, Any],
579 resource_type: str,
580 tenant_id: UUID,
581 conn: asyncpg.Connection | Any,
582 profile_validation_mode: str,
583 enforced: set[str],
584) -> list[str]:
585 elements = await get_profile_cache().get_elements(conn, tenant_id, profile_url)
586 if elements is None:
587 if profile_validation_mode in ("declared", "enforced") or profile_url in enforced:
588 return [f"Profile {profile_url} not found on server."]
589 logger.warning("profile.not_found", url=profile_url)
590 return []
592 return _validate_against_elements(elements, resource, resource_type, profile_url)
595def _resolve_applicable_profiles(
596 resource: dict[str, Any], resource_type: str, enforced_profiles: dict[str, list[str]]
597) -> set[str]:
598 profiles: set[str] = set()
599 meta = resource.get("meta", {})
600 if isinstance(meta, dict):
601 meta_dict = typing.cast(dict[str, Any], meta)
602 declared = meta_dict.get("profile", [])
603 if isinstance(declared, list):
604 profiles.update(typing.cast(list[str], declared))
605 profiles.update(enforced_profiles.get(resource_type, []))
606 return profiles
609async def _load_profile_elements(
610 profile_url: str, tenant_id: UUID, conn: asyncpg.Connection | Any
611) -> list[dict[str, Any]] | None:
612 return await get_profile_cache().get_elements(conn, tenant_id, profile_url)
615async def _check_element_binding(
616 el: dict[str, Any],
617 resource: dict[str, Any],
618 resource_type: str,
619 tenant_id: UUID,
620 conn: asyncpg.Connection | Any,
621 profile_url: str,
622 slice_patterns: dict[str, dict[str, Any]],
623) -> list[str]:
624 """Level 3 check for a single ElementDefinition, if it has a required binding.
626 Values are resolved via `_element_scoped_values`, which buckets by slice
627 membership for slice-scoped elements — a binding declared on one slice's
628 descendant must not be checked against sibling slices' values.
629 """
630 path = el.get("path")
631 binding = el.get("binding")
632 if not path or not isinstance(binding, dict):
633 return []
634 binding_dict = typing.cast(dict[str, Any], binding)
635 if binding_dict.get("strength") != "required":
636 return []
637 value_set_url = binding_dict.get("valueSet")
638 if not isinstance(value_set_url, str):
639 return []
640 values = _element_scoped_values(el, resource, resource_type, slice_patterns)
641 if not values:
642 return []
643 type_codes = _extract_type_codes(el)
644 return await check_required_binding(
645 conn,
646 tenant_id,
647 value_set_url,
648 values,
649 type_codes[0] if type_codes else None,
650 path,
651 profile_url,
652 )
655async def validate_terminology_bindings(
656 resource: dict[str, Any],
657 tenant_id: UUID,
658 conn: asyncpg.Connection | Any,
659 enforced_profiles: dict[str, list[str]],
660) -> list[str]:
661 """Level 3: required-binding terminology validation (see
662 docs/architecture/profile-validation.md). Deliberately independent of
663 ``profile_validation_mode`` (which gates Level 1/2/4 structural errors) —
664 callers gate the result via ``terminology_strictness`` instead, matching
665 the existing base-binding validator's (``binding_validator.py``) behavior.
667 Only ``binding.strength == "required"`` is enforced; extensible/preferred/
668 example bindings are recognized but not validated in this iteration.
669 """
670 resource_type = resource.get("resourceType")
671 if not resource_type:
672 return []
674 profiles_to_check = _resolve_applicable_profiles(resource, resource_type, enforced_profiles)
675 if not profiles_to_check:
676 return []
678 errors: list[str] = []
679 for profile_url in profiles_to_check:
680 elements = await _load_profile_elements(profile_url, tenant_id, conn)
681 if elements is None:
682 continue # Level 1/2/4 (validate_profiles) already reports "not found".
683 slice_patterns = _build_slice_patterns(elements)
684 for el in elements:
685 errors.extend(
686 await _check_element_binding(
687 el, resource, resource_type, tenant_id, conn, profile_url, slice_patterns
688 )
689 )
690 return errors
693async def validate_profiles(
694 resource: dict[str, Any],
695 tenant_id: UUID,
696 conn: asyncpg.Connection | Any,
697 profile_validation_mode: str,
698 enforced_profiles: dict[str, list[str]],
699) -> list[str]:
700 """
701 Fetch relevant StructureDefinitions and validate the resource against them.
702 Returns a list of error messages.
703 """
704 resource_type = resource.get("resourceType")
705 if not resource_type:
706 return []
708 profiles_to_check: set[str] = set()
710 # 1. Extract from meta.profile
711 meta = resource.get("meta", {})
712 if isinstance(meta, dict):
713 meta_dict = typing.cast(dict[str, Any], meta)
714 declared = meta_dict.get("profile", [])
715 if isinstance(declared, list):
716 declared_profiles = typing.cast(list[str], declared)
717 profiles_to_check.update(declared_profiles)
719 # 2. Add enforced profiles
720 enforced = set(enforced_profiles.get(resource_type, []))
721 profiles_to_check.update(enforced)
723 if not profiles_to_check:
724 return []
726 errors: list[str] = []
727 for profile_url in profiles_to_check:
728 errors.extend(
729 await _collect_profile_errors(
730 profile_url,
731 resource,
732 resource_type,
733 tenant_id,
734 conn,
735 profile_validation_mode,
736 enforced,
737 )
738 )
740 if profile_validation_mode == "tolerant" and errors:
741 logger.warning("profile.validation_failed_lenient", errors=errors)
742 # Keep only errors from enforced profiles, but ignore missing profiles
743 return [
744 e for e in errors if any(ep in e for ep in enforced) and "not found on server" not in e
745 ]
747 return errors