Coverage for src/lumen/service/resources.py: 90%
228 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-31 18:46 +0000
1"""
2The write path, one level below HTTP.
4Everything here was extracted verbatim from ``lumen.api.resources`` — the
5validation stack, conditional resolution, id assignment and the
6create/update/delete flows including create-on-update. The HTTP routes are the
7reference implementation and became thin adapters over this module; they still
8own everything that is genuinely HTTP (status codes, ETag/Location headers,
9interaction metrics, request parsing).
11Anything an in-process caller needs that HTTP does not have goes *on top* of
12this module, never into it.
13"""
15from __future__ import annotations
17import json
18from dataclasses import dataclass
19from datetime import datetime
20from typing import Any, cast
21from uuid import UUID
23import asyncpg
24import asyncpg.pool
25import structlog
27from ..api._helpers import inject_meta, new_id
28from ..api.deps import ResolvedTenantSettings
29from ..core.exceptions import OperationOutcomeError
30from ..core.write_events import WriteEvent, collect_write_events
31from ..db.pool import get_pool
32from ..db.repository import ReadResult, ResourceRepository
33from ..fhir.binding_validator import validate_code_bindings
34from ..fhir.profile_validator import validate_profiles, validate_terminology_bindings
35from ..fhir.validator import validate_resource
36from ..observability import record_validation_error
37from ..search.parser import parse_query
38from ..search.query import execute_search
39from ..search.registry import get_registry
41logger = structlog.get_logger(__name__)
43# PEP 695: lazily evaluated, so the subscripted asyncpg types stay annotation-only.
44type _Conn = (
45 asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record] | None
46)
49@dataclass(frozen=True)
50class WriteOutcome:
51 """The result of a write, independent of how it will be reported."""
53 resource: dict[str, Any]
54 version_id: int
55 last_updated: datetime
56 resource_id: str
57 #: True when a new resource was created (HTTP: 201 + ``Location``), False
58 #: when an existing one was updated or returned unchanged (HTTP: 200).
59 created: bool
61 @property
62 def stored_resource(self) -> dict[str, Any]:
63 """The resource as a client would read it back, with ``meta`` filled in."""
64 return inject_meta(self.resource, self.version_id, self.last_updated)
67class ResourceService:
68 """Write and lookup operations for one tenant, without a transport."""
70 def __init__(
71 self,
72 tenant_id: UUID,
73 tenant_settings: ResolvedTenantSettings | None = None,
74 pool: asyncpg.Pool[asyncpg.Record] | None = None,
75 ) -> None:
76 self._tenant = tenant_id
77 self._resolved_settings = tenant_settings
78 self._pool = pool
80 @property
81 def tenant_id(self) -> UUID:
82 return self._tenant
84 @property
85 def _settings(self) -> ResolvedTenantSettings:
86 """Tenant settings, required by everything except plain id-based delete.
88 ``delete_resource`` is the one HTTP route that never needed them, and
89 resolving them there would add a query to a path that had none.
90 """
91 if self._resolved_settings is None:
92 raise RuntimeError("This ResourceService was built without tenant settings")
93 return self._resolved_settings
95 @property
96 def tenant_settings(self) -> ResolvedTenantSettings:
97 return self._settings
99 @property
100 def pool(self) -> asyncpg.Pool[asyncpg.Record]:
101 return self._pool if self._pool is not None else get_pool()
103 def new_id(self) -> str:
104 return new_id()
106 # ── Validation ────────────────────────────────────────────────────────────
108 async def validate_and_parse_body(
109 self,
110 body_bytes: bytes,
111 expected_type: str,
112 conn: _Conn = None,
113 ) -> dict[str, Any]:
114 """Parse JSON body, check resourceType matches URL, run schema validation."""
115 try:
116 parsed = json.loads(body_bytes)
117 except json.JSONDecodeError as exc:
118 raise OperationOutcomeError(400, "invalid", f"Invalid JSON: {exc}") from exc
120 if not isinstance(parsed, dict):
121 raise OperationOutcomeError(400, "invalid", "Body must be a JSON object")
122 return await self.validate_resource_dict(cast(dict[str, Any], parsed), expected_type, conn)
124 async def validate_resource_dict(
125 self,
126 resource: dict[str, Any],
127 expected_type: str,
128 conn: _Conn = None,
129 ) -> dict[str, Any]:
130 """Run the full inbound validation stack over an already-parsed resource."""
131 rt = resource.get("resourceType")
132 if rt != expected_type:
133 raise OperationOutcomeError(
134 400,
135 "invalid",
136 f"resourceType in body ({rt!r}) does not match URL ({expected_type!r})",
137 )
139 errors = validate_resource(resource)
140 if errors:
141 record_validation_error(expected_type)
142 if self._settings.validation_strictness == "strict":
143 raise OperationOutcomeError(
144 422,
145 "invalid",
146 f"Resource validation failed: {errors[0]}",
147 )
148 else:
149 logger.warning("resource.validation_failed_lenient", errors=errors)
151 binding_errors = validate_code_bindings(resource)
153 if conn:
154 binding_errors, profile_errors = await self._db_validations(
155 resource, conn, binding_errors
156 )
157 else:
158 async with self.pool.acquire() as temp_conn:
159 binding_errors, profile_errors = await self._db_validations(
160 resource, temp_conn, binding_errors
161 )
163 if binding_errors:
164 if self._settings.terminology_strictness == "strict":
165 raise OperationOutcomeError(422, "code-invalid", binding_errors[0])
166 else:
167 logger.warning("resource.terminology_failed_lenient", errors=binding_errors)
169 if profile_errors:
170 raise OperationOutcomeError(
171 422,
172 "invalid",
173 f"Profile validation failed: {profile_errors[0]}",
174 )
176 return resource
178 async def _db_validations(
179 self,
180 resource: dict[str, Any],
181 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record],
182 binding_errors: list[str],
183 ) -> tuple[list[str], list[str]]:
184 """Terminology-binding and profile validation, both of which need the DB."""
185 binding_errors = binding_errors + await validate_terminology_bindings(
186 resource, self._tenant, conn, self._settings.enforced_profiles
187 )
188 profile_errors = await validate_profiles(
189 resource,
190 self._tenant,
191 conn,
192 self._settings.profile_validation_mode,
193 self._settings.enforced_profiles,
194 )
195 return binding_errors, profile_errors
197 # ── Conditional resolution ────────────────────────────────────────────────
199 async def conditional_search_ids(self, resource_type: str, query_string: str) -> list[str]:
200 """Execute a conditional search query and return matching resource_ids."""
201 from urllib.parse import parse_qs
203 raw = parse_qs(query_string, keep_blank_values=True)
204 multi: dict[str, list[str]] = {k: v for k, v in raw.items()}
205 pq = parse_query(multi)
206 registry = get_registry()
207 ignore_unknown = self._settings.ignore_unknown_search_parameters
208 async with self.pool.acquire() as conn:
209 params = await registry.get_params(conn, self._tenant, resource_type)
210 result = await execute_search(
211 conn, self._tenant, resource_type, pq, params, ignore_unknown
212 )
213 return [r["id"] for r in result.resources if "id" in r]
215 # ── Reads ─────────────────────────────────────────────────────────────────
217 async def read(self, resource_type: str, resource_id: str) -> ReadResult | None:
218 """Return the current version, or None if absent or deleted."""
219 async with self.pool.acquire() as conn:
220 return await ResourceRepository(conn, get_registry()).read(
221 self._tenant, resource_type, resource_id
222 )
224 async def search(
225 self,
226 resource_type: str,
227 params: dict[str, list[str]],
228 max_count: int | None = 1000,
229 ) -> list[dict[str, Any]]:
230 """Run a parsed search and return the matching resources."""
231 pq = parse_query(params, max_count=max_count)
232 registry = get_registry()
233 ignore_unknown = self._settings.ignore_unknown_search_parameters
234 async with self.pool.acquire() as conn:
235 search_params = await registry.get_params(conn, self._tenant, resource_type)
236 result = await execute_search(
237 conn, self._tenant, resource_type, pq, search_params, ignore_unknown
238 )
239 return result.resources
241 async def search_pandas(
242 self,
243 resource_type: str,
244 params: dict[str, list[str]] | None = None,
245 max_records: int | None = 50000,
246 **kwargs: Any,
247 ) -> Any:
248 """Run a search and return matching resources as a flat Pandas DataFrame.
250 Accepts either a FHIR search param dictionary (e.g. ``{"code": ["8310-5"]}``)
251 or keyword arguments (e.g. ``code="8310-5"``, ``_count=5000``).
252 Safely loads up to ``max_records`` (default 50,000) in a single in-process DB query.
253 """
254 import pandas as pd
256 merged_params: dict[str, list[str]] = {}
257 if params:
258 for k, v in params.items():
259 merged_params[k] = list(v) if isinstance(v, (list, tuple)) else [str(v)]
260 for k, v in kwargs.items():
261 norm_k = k.replace("_", "-") if k not in ("_count", "_sort", "_summary") else k
262 merged_params[norm_k] = list(v) if isinstance(v, (list, tuple)) else [str(v)]
264 if "_count" not in merged_params:
265 merged_params["_count"] = [str(max_records)]
267 resources = await self.search(resource_type, merged_params, max_count=max_records)
268 if not resources:
269 return pd.DataFrame()
271 return pd.json_normalize(resources)
273 async def search_arrow(
274 self,
275 resource_type: str,
276 params: dict[str, list[str]] | None = None,
277 max_records: int | None = 100000,
278 **kwargs: Any,
279 ) -> Any:
280 """Run a search and return matching resources as an Apache Arrow Table (pyarrow.Table).
282 Provides zero-copy binary memory format for high-performance ML, DuckDB, Polars & PyTorch.
283 Safely loads up to ``max_records`` (default 100,000) in a single in-process DB query.
284 """
285 import pandas as pd
286 import pyarrow as pa
288 merged_params: dict[str, list[str]] = {}
289 if params:
290 for k, v in params.items():
291 merged_params[k] = list(v) if isinstance(v, (list, tuple)) else [str(v)]
292 for k, v in kwargs.items():
293 norm_k = k.replace("_", "-") if k not in ("_count", "_sort", "_summary") else k
294 merged_params[norm_k] = list(v) if isinstance(v, (list, tuple)) else [str(v)]
296 if "_count" not in merged_params:
297 merged_params["_count"] = [str(max_records)]
299 resources = await self.search(resource_type, merged_params, max_count=max_records)
300 df = pd.json_normalize(resources)
301 return pa.Table.from_pandas(df)
303 # ── Writes ────────────────────────────────────────────────────────────────
305 async def create(
306 self,
307 resource_type: str,
308 resource: dict[str, Any],
309 *,
310 if_none_exist: str | None = None,
311 ) -> WriteOutcome:
312 """Create a resource, honouring ``If-None-Exist`` conditional-create semantics."""
313 with collect_write_events() as events:
314 outcome = await self._create(resource_type, resource, if_none_exist)
315 await self._dispatch(events)
316 return outcome
318 async def _create(
319 self,
320 resource_type: str,
321 resource: dict[str, Any],
322 if_none_exist: str | None,
323 ) -> WriteOutcome:
324 if if_none_exist:
325 matching_ids = await self.conditional_search_ids(resource_type, if_none_exist)
326 if len(matching_ids) == 1:
327 # Return existing resource unchanged (200)
328 async with self.pool.acquire() as conn:
329 existing = await ResourceRepository(conn, get_registry()).read(
330 self._tenant, resource_type, matching_ids[0]
331 )
332 if existing:
333 return WriteOutcome(
334 resource=existing["resource"],
335 version_id=existing["version_id"],
336 last_updated=existing["last_updated"],
337 resource_id=matching_ids[0],
338 created=False,
339 )
340 elif len(matching_ids) > 1:
341 raise OperationOutcomeError(
342 412,
343 "duplicate",
344 f"If-None-Exist matched {len(matching_ids)} resources; expected 0 or 1",
345 )
347 resource_id = self.new_id()
348 resource = {**resource, "id": resource_id}
350 async with self.pool.acquire() as conn:
351 result = await ResourceRepository(conn, get_registry()).create(
352 self._tenant, resource_type, resource_id, resource
353 )
354 return WriteOutcome(
355 resource=result["resource"],
356 version_id=result["version_id"],
357 last_updated=result["last_updated"],
358 resource_id=resource_id,
359 created=True,
360 )
362 async def update(
363 self,
364 resource_type: str,
365 resource_id: str,
366 resource: dict[str, Any],
367 *,
368 if_match: int | None = None,
369 ) -> WriteOutcome:
370 """Update a resource; creates it when create-on-update is enabled and it is absent."""
371 with collect_write_events() as events:
372 outcome = await self._update(resource_type, resource_id, resource, if_match)
373 await self._dispatch(events)
374 return outcome
376 async def _update(
377 self,
378 resource_type: str,
379 resource_id: str,
380 resource: dict[str, Any],
381 if_match: int | None,
382 ) -> WriteOutcome:
383 resource = {**resource, "id": resource_id}
385 async with self.pool.acquire() as conn:
386 repo = ResourceRepository(conn, get_registry())
387 existing = await repo.read(self._tenant, resource_type, resource_id)
389 if existing is None:
390 if not self._settings.create_on_update:
391 raise OperationOutcomeError(
392 404, "not-found", f"{resource_type}/{resource_id} not found"
393 )
394 # A previously-deleted id already has version history (a tombstone
395 # row), so create()'s hardcoded version_id=1 INSERT would collide
396 # with it (UniqueViolationError). update()'s MAX(version_id)+1
397 # continues the history correctly either way.
398 if await repo.is_deleted(self._tenant, resource_type, resource_id):
399 result = await repo.update(self._tenant, resource_type, resource_id, resource)
400 else:
401 result = await repo.create(self._tenant, resource_type, resource_id, resource)
402 return WriteOutcome(
403 resource=result["resource"],
404 version_id=result["version_id"],
405 last_updated=result["last_updated"],
406 resource_id=resource_id,
407 created=True,
408 )
410 if if_match is not None and if_match != existing["version_id"]:
411 raise OperationOutcomeError(
412 412,
413 "conflict",
414 f"Version mismatch: expected {if_match}, current is {existing['version_id']}",
415 )
417 result = await repo.update(self._tenant, resource_type, resource_id, resource)
419 return WriteOutcome(
420 resource=result["resource"],
421 version_id=result["version_id"],
422 last_updated=result["last_updated"],
423 resource_id=resource_id,
424 created=False,
425 )
427 async def conditional_update(
428 self,
429 resource_type: str,
430 query_string: str,
431 resource: dict[str, Any],
432 *,
433 matching_ids: list[str] | None = None,
434 ) -> WriteOutcome:
435 """``PUT /{type}?<search-params>`` — update the single match, or create.
437 ``matching_ids`` lets a caller that already resolved the query (the HTTP
438 route rejects a multiple match before it reads the body) pass the result
439 in rather than searching twice.
440 """
441 with collect_write_events() as events:
442 outcome = await self._conditional_update(
443 resource_type, query_string, resource, matching_ids
444 )
445 await self._dispatch(events)
446 return outcome
448 async def _conditional_update(
449 self,
450 resource_type: str,
451 query_string: str,
452 resource: dict[str, Any],
453 matching_ids: list[str] | None,
454 ) -> WriteOutcome:
455 if matching_ids is None:
456 matching_ids = await self.conditional_search_ids(resource_type, query_string)
458 if len(matching_ids) > 1:
459 raise OperationOutcomeError(
460 412,
461 "multiple-matches",
462 f"Conditional update matched {len(matching_ids)} resources; "
463 "must match exactly 0 or 1",
464 )
466 if len(matching_ids) == 0:
467 # No match → create
468 resource_id = cast(str, resource.get("id") or self.new_id())
469 resource = {**resource, "id": resource_id}
470 async with self.pool.acquire() as conn:
471 result = await ResourceRepository(conn, get_registry()).create(
472 self._tenant, resource_type, resource_id, resource
473 )
474 return WriteOutcome(
475 resource=result["resource"],
476 version_id=result["version_id"],
477 last_updated=result["last_updated"],
478 resource_id=resource_id,
479 created=True,
480 )
482 # One match → update
483 resource_id = matching_ids[0]
484 resource = {**resource, "id": resource_id}
485 async with self.pool.acquire() as conn:
486 result = await ResourceRepository(conn, get_registry()).update(
487 self._tenant, resource_type, resource_id, resource
488 )
489 return WriteOutcome(
490 resource=result["resource"],
491 version_id=result["version_id"],
492 last_updated=result["last_updated"],
493 resource_id=resource_id,
494 created=False,
495 )
497 async def delete(self, resource_type: str, resource_id: str) -> bool:
498 """Delete a resource. False means it never existed (HTTP: 404).
500 Deleting an already-deleted resource is idempotent and returns True.
501 """
502 with collect_write_events() as events:
503 async with self.pool.acquire() as conn:
504 repo = ResourceRepository(conn, get_registry())
505 if await repo.is_deleted(self._tenant, resource_type, resource_id):
506 # Already deleted — idempotent
507 return True
508 deleted = await repo.delete(self._tenant, resource_type, resource_id)
509 await self._dispatch(events)
510 return deleted
512 async def conditional_delete(self, resource_type: str, query_string: str) -> int:
513 """``DELETE /{type}?<search-params>`` — delete every match. Returns the count."""
514 matching_ids = await self.conditional_search_ids(resource_type, query_string)
515 if not matching_ids:
516 return 0
518 with collect_write_events() as events:
519 async with self.pool.acquire() as conn:
520 repo = ResourceRepository(conn, get_registry())
521 for rid in matching_ids:
522 await repo.delete(self._tenant, resource_type, rid)
523 await self._dispatch(events)
524 return len(matching_ids)
526 # ── Write-time extensions ─────────────────────────────────────────────────
528 async def _dispatch(self, events: list[WriteEvent]) -> None:
529 """Hand committed writes to the plugin host (no-op when no plugin listens).
531 The settings are passed through as-is — possibly ``None`` — so that the
532 dispatcher resolves them only if a hook actually matches. With no
533 plugins registered this costs one dict lookup.
534 """
535 if not events:
536 return
537 from ..cortex.dispatch import dispatch_write_events
539 await dispatch_write_events(events, self._resolved_settings)