Coverage for src/lumen/derivation/registry.py: 91%

1""" 

2This registry caches tenant-scoped derivation rules loaded from the 

3``derivation_rules`` table. There is no base/built-in layer — unlike search 

4parameters, FHIR defines no standard derivation rules; every rule is 

5tenant-specific data registered via the admin API. 

6""" 

7 

8from __future__ import annotations 

9 

10from uuid import UUID 

11 

12import asyncpg 

13 

14from .models import DerivationRuleDef 

15 

16# Singleton instance, initialised by ``init_registry()`` during app lifespan. 

17_registry: DerivationRuleRegistry | None = None 

18 

19 

20class DerivationRuleRegistry: 

21 """Per-tenant cache of derivation rules, invalidated explicitly on write.""" 

22 

23 def __init__(self) -> None: 

24 # tenant_id → resource_type → [DerivationRuleDef] 

25 self._cache: dict[UUID, dict[str, list[DerivationRuleDef]]] = {} 

26 

27 async def get_rules( 

28 self, 

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

30 tenant_id: UUID, 

31 resource_type: str, 

32 ) -> list[DerivationRuleDef]: 

33 """Return derivation rules registered for a resource type, loading from DB if needed.""" 

34 by_type = await self._ensure_loaded(conn, tenant_id) 

35 return by_type.get(resource_type, []) 

36 

37 def invalidate(self, tenant_id: UUID) -> None: 

38 """Clear cached rules for a tenant, forcing reload on next access.""" 

39 self._cache.pop(tenant_id, None) 

40 

41 def invalidate_all(self) -> None: 

42 """Clear the entire cache (e.g. for testing).""" 

43 self._cache.clear() 

44 

45 async def _ensure_loaded( 

46 self, 

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

48 tenant_id: UUID, 

49 ) -> dict[str, list[DerivationRuleDef]]: 

50 if tenant_id in self._cache: 

51 return self._cache[tenant_id] 

52 

53 rows: list[asyncpg.Record] = await conn.fetch( 

54 """ 

55 SELECT resource_type, target_path, source_path, concept_map_url, 

56 description, target_is_list 

57 FROM derivation_rules 

58 WHERE tenant_id = $1 

59 """, 

60 tenant_id, 

61 ) 

62 

63 by_type: dict[str, list[DerivationRuleDef]] = {} 

64 for row in rows: 

65 rule = DerivationRuleDef( 

66 resource_type=row["resource_type"], 

67 target_path=row["target_path"], 

68 source_path=row["source_path"], 

69 concept_map_url=row["concept_map_url"], 

70 description=row["description"] or "", 

71 target_is_list=row["target_is_list"], 

72 ) 

73 by_type.setdefault(rule.resource_type, []).append(rule) 

74 

75 self._cache[tenant_id] = by_type 

76 return by_type 

77 

78 

79def init_registry() -> DerivationRuleRegistry: 

80 """Create and store the global registry singleton. Called once during app lifespan.""" 

81 global _registry 

82 _registry = DerivationRuleRegistry() 

83 return _registry 

84 

85 

86def get_registry() -> DerivationRuleRegistry: 

87 """Return the global registry singleton. Raises if not yet initialised.""" 

88 if _registry is None: 

89 raise RuntimeError("DerivationRuleRegistry not initialised — call init_registry() first") 

90 return _registry