Coverage for src/lumen/search/search_param_hooks.py: 76%
84 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-31 18:46 +0000
1"""
2This module contains post-write hooks triggered when SearchParameter resources change.
3It updates database indices and triggers background reindexing tasks for existing resources.
4"""
6from __future__ import annotations
8from typing import Any
9from uuid import UUID
11import asyncpg
12import structlog
14from ..fhir.params import SearchParamDef
15from ..fhir.path import evaluate_path
16from .registry import SearchParamRegistry
18logger = structlog.get_logger(__name__)
20_SUPPORTED_TYPES: frozenset[str] = frozenset(
21 {"string", "token", "date", "reference", "quantity", "number"}
22)
25class SearchParamValidationError(Exception):
26 """Raised when a SearchParameter definition is invalid."""
28 def __init__(self, message: str) -> None:
29 self.message = message
30 super().__init__(message)
33def _validate_search_param(
34 resource: dict[str, Any], registry: SearchParamRegistry
35) -> tuple[str, str, str, list[str]]:
36 """Validate a SearchParameter definition.
38 Returns a tuple of (code, type, expression, base_types).
39 Raises SearchParamValidationError if invalid.
40 """
41 code = resource.get("code")
42 sp_type = resource.get("type")
43 expression = resource.get("expression")
44 base_types: list[str] = resource.get("base", [])
46 if not code or not isinstance(code, str):
47 raise SearchParamValidationError("SearchParameter.code is required")
49 if not sp_type or sp_type not in _SUPPORTED_TYPES:
50 raise SearchParamValidationError(
51 f"SearchParameter.type must be one of {sorted(_SUPPORTED_TYPES)}, got: {sp_type!r}"
52 )
54 if not expression or not isinstance(expression, str):
55 raise SearchParamValidationError("SearchParameter.expression is required")
57 if not base_types:
58 raise SearchParamValidationError(
59 "SearchParameter.base must list at least one resource type"
60 )
62 # Check for base-param shadowing.
63 for rt in base_types:
64 if registry.is_base_param(rt, code):
65 raise SearchParamValidationError(
66 f"Custom SearchParameter '{code}' shadows a base parameter"
67 )
69 # Validate FHIRPath expression by attempting evaluation against an empty
70 # resource. This catches syntax errors without needing real data.
71 for rt in base_types:
72 try:
73 evaluate_path(expression, {"resourceType": rt})
74 except Exception as exc:
75 raise SearchParamValidationError(
76 f"Invalid FHIRPath expression '{expression}': {exc}"
77 ) from exc
79 return code, sp_type, expression, base_types
82async def handle_search_param_write(
83 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record],
84 tenant_id: UUID,
85 resource: dict[str, Any],
86 registry: SearchParamRegistry,
87) -> None:
88 """Process a SearchParameter create or update.
90 Validates the definition, upserts into ``custom_search_params``, invalidates
91 the registry cache, and synchronously reindexes existing resources for the
92 new/changed parameter.
94 Raises :class:`SearchParamValidationError` on invalid definitions.
95 """
96 code, sp_type, expression, base_types = _validate_search_param(resource, registry)
97 status = resource.get("status", "active")
98 description = resource.get("description")
100 # ── Persist ────────────────────────────────────────────────────────────
102 # Only active SearchParameters are registered for search/indexing.
103 # Draft/retired ones are stored in fhir_resources but not in custom_search_params.
104 if status != "active":
105 # Remove from custom_search_params if it was previously active.
106 for rt in base_types:
107 await conn.execute(
108 """
109 DELETE FROM custom_search_params
110 WHERE tenant_id = $1 AND code = $2 AND resource_type = $3
111 """,
112 tenant_id,
113 code,
114 rt,
115 )
116 registry.invalidate(tenant_id)
117 await logger.ainfo(
118 "search_param.skipped_non_active",
119 code=code,
120 status=status,
121 tenant_id=str(tenant_id),
122 )
123 return
125 for rt in base_types:
126 await conn.execute(
127 """
128 INSERT INTO custom_search_params
129 (tenant_id, code, resource_type, param_type, expression, description, fhir_resource)
130 VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb)
131 ON CONFLICT (tenant_id, code, resource_type)
132 DO UPDATE SET
133 param_type = EXCLUDED.param_type,
134 expression = EXCLUDED.expression,
135 description = EXCLUDED.description,
136 fhir_resource = EXCLUDED.fhir_resource,
137 updated_at = now()
138 """,
139 tenant_id,
140 code,
141 rt,
142 sp_type,
143 expression,
144 description,
145 resource,
146 )
148 registry.invalidate(tenant_id)
150 await logger.ainfo(
151 "search_param.registered",
152 code=code,
153 type=sp_type,
154 base=base_types,
155 tenant_id=str(tenant_id),
156 )
158 # ── Reindex ────────────────────────────────────────────────────────────
160 sp_def = SearchParamDef(
161 name=code,
162 type=sp_type, # type: ignore[arg-type] # validated above
163 paths=[p.strip() for p in expression.split("|") if p.strip()],
164 )
165 for rt in base_types:
166 await _reindex_for_param(conn, tenant_id, rt, sp_def)
169async def handle_search_param_delete(
170 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record],
171 tenant_id: UUID,
172 resource: dict[str, Any],
173 registry: SearchParamRegistry,
174) -> None:
175 """Process a SearchParameter deletion.
177 Removes the parameter from ``custom_search_params`` and invalidates the
178 registry cache. Index rows in ``spidx_*`` tables are left as-is (they
179 become orphans and are cleaned up on next resource write).
180 """
181 code = resource.get("code")
182 base_types: list[str] = resource.get("base", [])
184 if not code or not base_types:
185 return
187 for rt in base_types:
188 await conn.execute(
189 """
190 DELETE FROM custom_search_params
191 WHERE tenant_id = $1 AND code = $2 AND resource_type = $3
192 """,
193 tenant_id,
194 code,
195 rt,
196 )
198 registry.invalidate(tenant_id)
200 await logger.ainfo(
201 "search_param.unregistered",
202 code=code,
203 base=base_types,
204 tenant_id=str(tenant_id),
205 )
208# ── reindex helper ─────────────────────────────────────────────────────────────
211async def _reindex_for_param(
212 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record],
213 tenant_id: UUID,
214 resource_type: str,
215 sp: SearchParamDef,
216) -> None:
217 """Index all existing resources of ``resource_type`` for a single new param.
219 Streams resources in batches to avoid loading everything into memory at once.
220 Only **adds** new index rows — does not touch existing indexes for other params.
221 """
222 from .indexer import _index_param # type: ignore[reportPrivateUsage]
224 batch_size = 500
225 last_id: str | None = None
226 total = 0
228 while True:
229 if last_id is None:
230 rows = await conn.fetch(
231 """
232 SELECT DISTINCT ON (resource_id) resource_id, resource
233 FROM fhir_resources
234 WHERE tenant_id = $1
235 AND resource_type = $2
236 AND deleted = FALSE
237 ORDER BY resource_id, version_id DESC
238 LIMIT $3
239 """,
240 tenant_id,
241 resource_type,
242 batch_size,
243 )
244 else:
245 rows = await conn.fetch(
246 """
247 SELECT DISTINCT ON (resource_id) resource_id, resource
248 FROM fhir_resources
249 WHERE tenant_id = $1
250 AND resource_type = $2
251 AND deleted = FALSE
252 AND resource_id > $3
253 ORDER BY resource_id, version_id DESC
254 LIMIT $4
255 """,
256 tenant_id,
257 resource_type,
258 last_id,
259 batch_size,
260 )
262 if not rows:
263 break
265 for row in rows:
266 resource_id: str = row["resource_id"]
267 raw = row["resource"]
268 resource: dict[str, Any] = (
269 __import__("json").loads(raw) if isinstance(raw, str) else dict(raw)
270 )
271 await _index_param(conn, tenant_id, resource_type, resource_id, resource, sp)
272 total += 1
274 last_id = rows[-1]["resource_id"]
275 if len(rows) < batch_size:
276 break
278 if total > 0:
279 await logger.ainfo(
280 "search_param.reindex_complete",
281 code=sp.name,
282 resource_type=resource_type,
283 count=total,
284 tenant_id=str(tenant_id),
285 )