Coverage for src/lumen/derivation/engine.py: 98%

1""" 

2This module applies registered derivation rules to a resource before it is 

3persisted. A rule fills in ``target_path`` from Coding values found at 

4``source_path`` by translating them through a ConceptMap — but only when the 

5target is not already present, so client-supplied data is never overwritten. 

6 

7Best-effort by design: a missing ConceptMap, an unmapped code, or any 

8evaluation problem is logged and skipped rather than failing the write. A 

9tenant with no rules registered (the default — no package installs any) pays 

10one empty-cache lookup and otherwise does nothing. 

11""" 

12 

13from __future__ import annotations 

14 

15from typing import Any, cast 

16from uuid import UUID 

17 

18import asyncpg 

19import structlog 

20 

21from ..fhir.path import evaluate_path 

22from ..fhir.terminology import translate 

23from .registry import DerivationRuleRegistry 

24 

25logger = structlog.get_logger(__name__) 

26 

27 

28async def apply_derivation_rules( 

29 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

30 tenant_id: UUID, 

31 resource_type: str, 

32 resource: dict[str, Any], 

33 registry: DerivationRuleRegistry, 

34) -> None: 

35 """Mutate ``resource`` in place, filling in any registered missing fields.""" 

36 rules = await registry.get_rules(conn, tenant_id, resource_type) 

37 if not rules: 

38 return 

39 

40 for rule in rules: 

41 if resource.get(rule.target_path): 

42 continue # never overwrite / augment client-supplied data 

43 

44 codings = _extract_codings(evaluate_path(rule.source_path, resource)) 

45 if not codings: 

46 continue 

47 

48 concept_map = await _find_concept_map(conn, tenant_id, rule.concept_map_url) 

49 if concept_map is None: 

50 await logger.awarning( 

51 "derivation.concept_map_not_found", 

52 resource_type=resource_type, 

53 target_path=rule.target_path, 

54 concept_map_url=rule.concept_map_url, 

55 tenant_id=str(tenant_id), 

56 ) 

57 continue 

58 

59 derived = _translate_first_match(concept_map, codings) 

60 if derived is not None: 

61 concept = {"coding": [derived]} 

62 # A collection target (e.g. DocumentReference.category, 0..*) must be 

63 # a JSON array of CodeableConcept; a bare object is invalid FHIR. 

64 # The target was confirmed absent above, so start a fresh list. 

65 resource[rule.target_path] = [concept] if rule.target_is_list else concept 

66 await logger.adebug( 

67 "derivation.applied", 

68 resource_type=resource_type, 

69 target_path=rule.target_path, 

70 target_is_list=rule.target_is_list, 

71 tenant_id=str(tenant_id), 

72 ) 

73 

74 

75def _extract_codings(values: list[Any]) -> list[dict[str, Any]]: 

76 """Keep only dict values that look like a FHIR Coding (have a ``code``).""" 

77 return [v for v in values if isinstance(v, dict) and cast("dict[str, Any]", v).get("code")] 

78 

79 

80def _translate_first_match( 

81 concept_map: dict[str, Any], codings: list[dict[str, Any]] 

82) -> dict[str, Any] | None: 

83 for coding in codings: 

84 matches = translate(concept_map, coding.get("system"), coding["code"]) 

85 if matches: 

86 return matches[0]["concept"] # type: ignore[no-any-return] 

87 return None 

88 

89 

90async def _find_concept_map( 

91 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

92 tenant_id: UUID, 

93 url: str, 

94) -> dict[str, Any] | None: 

95 row: asyncpg.Record | None = await conn.fetchrow( 

96 """ 

97 SELECT resource 

98 FROM fhir_resources 

99 WHERE tenant_id = $1 

100 AND resource_type = 'ConceptMap' 

101 AND resource ->> 'url' = $2 

102 AND deleted = FALSE 

103 ORDER BY resource_id, version_id DESC 

104 LIMIT 1 

105 """, 

106 tenant_id, 

107 url, 

108 ) 

109 if row is None: 

110 return None 

111 resource = row["resource"] 

112 if isinstance(resource, dict): 

113 return cast("dict[str, Any]", resource) 

114 return None