Coverage for src/lumen/fhir/validator.py: 100%

1""" 

2This module performs validation on incoming resources using the official FHIR JSON Schema. 

3It uses jsonschema-rs to compile the schema once and validate writes at high performance. 

4""" 

5 

6from __future__ import annotations 

7 

8import json 

9import re 

10from functools import lru_cache 

11from pathlib import Path 

12from typing import Any 

13 

14import jsonschema_rs # type: ignore[import-untyped] # no stubs available 

15 

16_SCHEMA_PATH = Path(__file__).parent / "data" / "r4b" / "fhir.schema.json" 

17 

18 

19@lru_cache(maxsize=1) 

20def _get_base_schema() -> dict[str, Any]: 

21 """Load and cache the base FHIR schema JSON.""" 

22 return json.loads(_SCHEMA_PATH.read_text()) 

23 

24 

25@lru_cache(maxsize=1) 

26def _base_validator() -> Any: 

27 """Compiled base validator fallback.""" 

28 return jsonschema_rs.validator_for(_get_base_schema()) 

29 

30 

31@lru_cache(maxsize=200) 

32def _resource_validator(resource_type: str) -> Any: 

33 """Compile and cache a validator targeted at a specific FHIR resource type.""" 

34 schema = _get_base_schema() 

35 if not resource_type or resource_type not in schema.get("definitions", {}): 

36 return _base_validator() 

37 

38 target_schema = { 

39 "$ref": f"#/definitions/{resource_type}", 

40 "definitions": schema["definitions"], 

41 } 

42 return jsonschema_rs.validator_for(target_schema) 

43 

44 

45def clean_validation_error( 

46 err_str: str, resource_type: str | None = None, instance_path: list[str] | None = None 

47) -> str: 

48 """Strip raw JSON serialization and format jsonschema-rs errors cleanly with paths.""" 

49 # Take only the first line of the error message to discard verbose stack trace info 

50 first_line = err_str.splitlines()[0] if err_str else "" 

51 

52 path_prefix = "" 

53 if instance_path: 

54 path_prefix = f"At '{'.'.join(str(p) for p in instance_path)}': " 

55 

56 # Clean up pattern mismatch errors (hide massive regex strings) 

57 pattern_match = re.search(r'does not match\s+"[^"]+"', first_line) 

58 if pattern_match: 

59 first_line = re.sub( 

60 r'does not match\s+"[^"]+"', 

61 "does not match the required format pattern", 

62 first_line, 

63 ) 

64 

65 for marker in ["} is ", "] is ", "} was ", "] was "]: 

66 idx = first_line.find(marker) 

67 if idx != -1: 

68 suffix = first_line[idx + 1 :] 

69 res_name = f"Resource of type '{resource_type}'" if resource_type else "Resource" 

70 msg = f"{res_name}{suffix}" 

71 if "not valid under any of the schemas listed in the 'oneOf' keyword" in msg: 

72 return f"{path_prefix}{res_name} does not conform to the FHIR schema specification" 

73 return f"{path_prefix}{msg}" 

74 

75 # Fallback regex matching JSON at start 

76 match = re.match(r"^[\{\[].*?[\}\]]\s+(.*)$", first_line, re.DOTALL) 

77 if match: 

78 suffix = match.group(1) 

79 res_name = f"Resource of type '{resource_type}'" if resource_type else "Resource" 

80 msg = f"{res_name} {suffix}" 

81 if "not valid under any of the schemas listed in the 'oneOf' keyword" in msg: 

82 return f"{path_prefix}{res_name} does not conform to the FHIR schema specification" 

83 return f"{path_prefix}{msg}" 

84 

85 return f"{path_prefix}{first_line}" 

86 

87 

88def validate_resource(resource: dict[str, Any]) -> list[str]: 

89 """Return a list of validation error messages, empty if valid.""" 

90 resource_type = resource.get("resourceType") or "" 

91 validator = _resource_validator(resource_type) 

92 return [ 

93 clean_validation_error(str(err), resource_type, err.instance_path) 

94 for err in validator.iter_errors(resource) 

95 ]