Coverage for src/lumen/search/registry.py: 100%
75 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-31 18:46 +0000
1"""
2This registry manages search parameter definitions across different tenants.
3It combines static base parameters with custom parameters and handles memory caching.
4"""
6from __future__ import annotations
8import json
9from uuid import UUID
11import asyncpg
13from ..fhir.params import SearchParamDef
14from ..fhir.search_param_loader import load_registry as _load_base_registry
16# Singleton instance, initialised by ``init_registry()`` during app lifespan.
17_registry: SearchParamRegistry | None = None
20class SearchParamRegistry:
21 """Two-layer search parameter registry: immutable base + mutable custom.
23 The base layer is loaded once from the vendored R4B SearchParameter bundle
24 and shared across all tenants. The custom layer is loaded per-tenant from
25 the ``custom_search_params`` table and cached in memory. The cache for a
26 tenant is invalidated explicitly via :meth:`invalidate`.
27 """
29 def __init__(self) -> None:
30 self._base: dict[str, list[SearchParamDef]] = _load_base_registry()
31 # tenant_id → {resource_type → [SearchParamDef]}
32 self._custom_cache: dict[UUID, dict[str, list[SearchParamDef]]] = {}
34 # ── public API ─────────────────────────────────────────────────────────────
36 async def get_params(
37 self,
38 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record],
39 tenant_id: UUID,
40 resource_type: str,
41 ) -> list[SearchParamDef]:
42 """Return merged base + custom params for a single resource type."""
43 custom = await self._ensure_custom_loaded(conn, tenant_id)
44 base = self._base.get(resource_type, [])
45 extra = custom.get(resource_type, [])
46 if not extra:
47 return base
48 # Merge: base first, then custom (no shadowing — collisions prevented
49 # at creation time by the hooks layer).
50 base_codes = {sp.name for sp in base}
51 merged = list(base)
52 for sp in extra:
53 if sp.name not in base_codes:
54 merged.append(sp)
55 return merged
57 async def get_all_params(
58 self,
59 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record],
60 tenant_id: UUID,
61 ) -> dict[str, list[SearchParamDef]]:
62 """Return full merged registry for a tenant (all resource types)."""
63 custom = await self._ensure_custom_loaded(conn, tenant_id)
64 # Start with a copy of base.
65 merged: dict[str, list[SearchParamDef]] = {
66 rt: list(params) for rt, params in self._base.items()
67 }
68 for rt, custom_params in custom.items():
69 base_codes = {sp.name for sp in merged.get(rt, [])}
70 existing = merged.setdefault(rt, [])
71 for sp in custom_params:
72 if sp.name not in base_codes:
73 existing.append(sp)
74 return merged
76 def get_base_params(self, resource_type: str) -> list[SearchParamDef]:
77 """Return base R4B params only (no DB access needed)."""
78 return self._base.get(resource_type, [])
80 def is_base_param(self, resource_type: str, code: str) -> bool:
81 """Check whether a param code is a base R4B param for the given type."""
82 return any(sp.name == code for sp in self._base.get(resource_type, []))
84 def base_param_type(self, resource_type: str, code: str) -> str | None:
85 """Index type of a base param on a specific resource type, or None."""
86 for sp in self._base.get(resource_type, []):
87 if sp.name == code:
88 return sp.type
89 return None
91 def any_base_param_type(self, code: str) -> str | None:
92 """Index type of a base param by name, searched across all resource
93 types. Used by chained search when the reference target type is not
94 known up front — a standard param's index type is consistent wherever
95 it is defined, so the first match is representative.
96 """
97 for params in self._base.values():
98 for sp in params:
99 if sp.name == code:
100 return sp.type
101 return None
103 def invalidate(self, tenant_id: UUID) -> None:
104 """Clear cached custom params for a tenant, forcing reload on next access."""
105 self._custom_cache.pop(tenant_id, None)
107 def invalidate_all(self) -> None:
108 """Clear the entire custom cache (e.g. for testing)."""
109 self._custom_cache.clear()
111 # ── private ────────────────────────────────────────────────────────────────
113 async def _ensure_custom_loaded(
114 self,
115 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record],
116 tenant_id: UUID,
117 ) -> dict[str, list[SearchParamDef]]:
118 """Return cached custom params for a tenant, loading from DB if needed."""
119 if tenant_id in self._custom_cache:
120 return self._custom_cache[tenant_id]
122 rows: list[asyncpg.Record] = await conn.fetch(
123 """
124 SELECT code, resource_type, param_type, expression, fhir_resource
125 FROM custom_search_params
126 WHERE tenant_id = $1
127 """,
128 tenant_id,
129 )
131 by_type: dict[str, list[SearchParamDef]] = {}
132 for row in rows:
133 param_type = row["param_type"]
134 # Only register types we support.
135 if param_type not in ("string", "token", "date", "reference", "quantity", "number"):
136 continue
137 fhir_resource = row["fhir_resource"]
138 if isinstance(fhir_resource, str):
139 fhir_resource = json.loads(fhir_resource)
140 sp = SearchParamDef(
141 name=row["code"],
142 type=param_type, # type: ignore[arg-type] # validated above
143 paths=[p.strip() for p in row["expression"].split("|") if p.strip()],
144 definition=fhir_resource.get("url", "") if fhir_resource else "",
145 )
146 by_type.setdefault(row["resource_type"], []).append(sp)
148 self._custom_cache[tenant_id] = by_type
149 return by_type
152def init_registry() -> SearchParamRegistry:
153 """Create and store the global registry singleton. Called once during app lifespan."""
154 global _registry
155 _registry = SearchParamRegistry()
156 return _registry
159def get_registry() -> SearchParamRegistry:
160 """Return the global registry singleton. Raises if not yet initialised."""
161 if _registry is None:
162 raise RuntimeError("SearchParamRegistry not initialised — call init_registry() first")
163 return _registry