Coverage for src/lumen/fhir/invariants.py: 95%

1""" 

2This module evaluates FHIRPath constraint invariants defined in StructureDefinitions. 

3It utilizes fhirpathpy to validate structural rules and domain requirements on resources. 

4""" 

5 

6from __future__ import annotations 

7 

8import json 

9from collections.abc import Callable 

10from dataclasses import dataclass 

11from functools import lru_cache 

12from pathlib import Path 

13from typing import Any, cast 

14 

15import lumen.fhirpath as fhirpathpy 

16 

17# fhirpathpy types compile() loosely (dict/list without type args). Bind the 

18# module as Any and narrow the result to the shape we use, so the rest of the 

19# module stays strictly typed without per-call ignores. 

20_fhirpathpy: Any = fhirpathpy 

21_Evaluator = Callable[[dict[str, Any], dict[str, Any]], list[Any]] 

22 

23_DATA_PATH = Path(__file__).parent / "data" / "r4b" / "r4b-invariants.json" 

24 

25 

26@dataclass(frozen=True) 

27class Constraint: 

28 key: str 

29 severity: str # "error" | "warning" 

30 human: str 

31 expression: str 

32 path: str # element path the constraint is anchored to (e.g. "Patient" or "Patient.contact") 

33 

34 

35@dataclass(frozen=True) 

36class InvariantViolation: 

37 key: str 

38 severity: str 

39 human: str 

40 

41 

42@lru_cache(maxsize=1) 

43def _constraints() -> dict[str, list[Constraint]]: 

44 raw: dict[str, list[dict[str, Any]]] = json.loads(_DATA_PATH.read_text()) 

45 return { 

46 rt: [ 

47 Constraint( 

48 c["key"], c.get("severity", "error"), c.get("human", ""), c["expression"], c["path"] 

49 ) 

50 for c in items 

51 ] 

52 for rt, items in raw.items() 

53 } 

54 

55 

56def _silent_trace(label: str, value: Any) -> None: 

57 """No-op replacement for fhirpathpy's default trace(), which prints to stdout. 

58 

59 Several base constraints (e.g. dom-3) call FHIRPath ``trace()``; without this 

60 every $validate would emit ``TRACE:[...]`` lines to stdout. 

61 """ 

62 

63 

64@lru_cache(maxsize=2048) 

65def _compiled(expression: str) -> _Evaluator: 

66 """Compile (and cache) a FHIRPath expression into an evaluator callable.""" 

67 return cast("_Evaluator", _fhirpathpy.compile(expression, options={"traceFn": _silent_trace})) 

68 

69 

70def _is_constraint_violated( 

71 constraint: Constraint, 

72 resource: dict[str, Any], 

73 resource_type: str, 

74) -> bool: 

75 # Locate the node(s) the constraint applies to: the whole resource for a 

76 # root constraint, else the elements at its path (navigated via FHIRPath). 

77 if constraint.path == resource_type: 

78 nodes: list[Any] = [resource] 

79 else: 

80 try: 

81 nodes = _compiled(constraint.path)(resource, {"resource": resource}) 

82 except Exception: 

83 return False 

84 

85 for node in nodes: 

86 # %resource is always the root; %context is the node being validated. 

87 env = {"resource": resource, "rootResource": resource, "context": node} 

88 try: 

89 result = _compiled(constraint.expression)(node, env) 

90 except Exception: 

91 continue 

92 if result == [False]: 

93 return True 

94 return False 

95 

96 

97def validate_invariants(resource: dict[str, Any]) -> list[InvariantViolation]: 

98 """Return the invariants the resource violates (one entry per constraint key). 

99 

100 A constraint passes when its FHIRPath expression evaluates to ``true``. An 

101 expression that cannot be evaluated (unsupported function, etc.) or that does 

102 not yield a definite ``false`` is treated as a pass, so this never produces 

103 false positives that would wrongly fail a resource. 

104 """ 

105 resource_type = resource.get("resourceType") 

106 if not isinstance(resource_type, str): 

107 return [] 

108 

109 violations: list[InvariantViolation] = [] 

110 seen: set[str] = set() 

111 for constraint in _constraints().get(resource_type, []): 

112 if constraint.key in seen: 

113 continue 

114 if _is_constraint_violated(constraint, resource, resource_type): 

115 seen.add(constraint.key) 

116 violations.append( 

117 InvariantViolation(constraint.key, constraint.severity, constraint.human) 

118 ) 

119 return violations