Coverage for src/lumen/db/repository.py: 91%
295 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-31 18:46 +0000
1"""
2This module provides the main repository classes for database interactions.
3It handles direct FHIR resource CRUD operations, history querying, and transaction execution.
4"""
6from __future__ import annotations
8import json
9import re
10from collections.abc import Mapping, Sequence
11from datetime import datetime
12from typing import Any, TypedDict, cast
13from uuid import UUID, uuid4
15import asyncpg
16import asyncpg.pool
17import structlog
19from ..audit.events import build_audit_event
20from ..core.exceptions import OperationOutcomeError
21from ..core.write_events import WriteEvent, record_write_event
22from ..fhir.resource_types import known_resource_types
23from ..search.indexer import delete_indexes, index_resource
24from ..search.registry import SearchParamRegistry
26logger = structlog.get_logger(__name__)
29def _decode_jsonb(value: Any) -> dict[str, Any]:
30 if isinstance(value, str):
31 return json.loads(value) # type: ignore[no-any-return]
32 return value # type: ignore[return-value]
35def extract_references(obj: Any) -> list[tuple[str, str]]:
36 """Recursively extract all local FHIR references from a resource."""
37 refs: list[tuple[str, str]] = []
38 if isinstance(obj, dict):
39 d_obj = cast(dict[str, Any], obj)
40 if "reference" in d_obj and isinstance(d_obj["reference"], str):
41 ref_str: str = d_obj["reference"]
42 if not ref_str.startswith("#"):
43 parts = ref_str.split("/")
44 if len(parts) >= 2:
45 target_type = parts[-2]
46 target_id = parts[-1]
47 if target_type in known_resource_types():
48 refs.append((target_type, target_id))
49 for v in d_obj.values():
50 refs.extend(extract_references(v))
51 elif isinstance(obj, list):
52 l_obj = cast(list[Any], obj)
53 for item in l_obj:
54 refs.extend(extract_references(item))
55 return refs
58def extract_dangling_urn_references(obj: Any) -> list[str]:
59 """Recursively find ``reference`` values still pointing at a ``urn:uuid:``.
61 Only meaningful inside a Bundle: bundle processing rewrites urn:uuid
62 references to a real ``ResourceType/id`` when the uuid matches another
63 entry's fullUrl (see ``api/bundle.py::_rewrite_refs``). One that's still
64 a bare urn:uuid after that means nothing in the bundle (or the database)
65 resolves it — extract_references() can't catch this itself, since it
66 only recognises the ``ResourceType/id`` shape.
67 """
68 refs: list[str] = []
69 if isinstance(obj, dict):
70 d_obj = cast(dict[str, Any], obj)
71 ref = d_obj.get("reference")
72 if isinstance(ref, str) and ref.startswith("urn:uuid:"):
73 refs.append(ref)
74 for v in d_obj.values():
75 refs.extend(extract_dangling_urn_references(v))
76 elif isinstance(obj, list):
77 for item in cast(list[Any], obj):
78 refs.extend(extract_dangling_urn_references(item))
79 return refs
82DATETIME_WITH_TIME_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}")
83DATETIME_WITH_TZ_RE = re.compile(
84 r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})$"
85)
88def is_datetime_missing_timezone(val: str) -> bool:
89 """Check if a string represents a FHIR dateTime/instant with time but without timezone."""
90 return bool(DATETIME_WITH_TIME_RE.match(val) and not DATETIME_WITH_TZ_RE.match(val))
93def find_timezone_violations(obj: Any) -> list[str]:
94 """Recursively search for datetime values missing timezone offsets."""
95 violations: list[str] = []
96 if isinstance(obj, dict):
97 d_obj = cast(dict[str, Any], obj)
98 for v in d_obj.values():
99 violations.extend(find_timezone_violations(v))
100 elif isinstance(obj, list):
101 l_obj = cast(list[Any], obj)
102 for item in l_obj:
103 violations.extend(find_timezone_violations(item))
104 elif isinstance(obj, str):
105 if is_datetime_missing_timezone(obj):
106 violations.append(obj)
107 return violations
110class ReadResult(TypedDict):
111 resource: dict[str, Any]
112 version_id: int
113 last_updated: datetime
116class WriteResult(TypedDict):
117 resource: dict[str, Any]
118 version_id: int
119 last_updated: datetime
122class HistoryEntry(TypedDict):
123 resource: dict[str, Any]
124 version_id: int
125 last_updated: datetime
126 deleted: bool
129class ResourceRepository:
130 """
131 FHIR resource persistence. All operations are tenant-scoped.
132 One table stores every resource version (version_id is part of the PK).
133 The newest version row per resource carries is_current = TRUE (a delete tombstone
134 included); a partial unique index enforces exactly one such row. Searches filter
135 is_current AND NOT deleted instead of aggregating MAX(version_id) at query time.
136 """
138 def __init__(
139 self,
140 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record],
141 registry: SearchParamRegistry | None = None,
142 ) -> None:
143 self._conn = conn
144 self._registry = registry
146 # ── Read ───────────────────────────────────────────────────────────────────
148 async def read(
149 self,
150 tenant_id: UUID,
151 resource_type: str,
152 resource_id: str,
153 ) -> ReadResult | None:
154 """
155 Return the current (latest, non-deleted) version or None.
156 It uses the extra index fhir_resources_current_idx to speed up the lookup.
157 """
158 row: asyncpg.Record | None = await self._conn.fetchrow(
159 """
160 SELECT resource, version_id, last_updated, deleted
161 FROM fhir_resources
162 WHERE tenant_id = $1
163 AND resource_type = $2
164 AND resource_id = $3
165 ORDER BY version_id DESC
166 LIMIT 1
167 """,
168 tenant_id,
169 resource_type,
170 resource_id,
171 )
172 if row is None or row["deleted"]:
173 return None
174 return ReadResult(
175 resource=_decode_jsonb(row["resource"]),
176 version_id=row["version_id"],
177 last_updated=row["last_updated"],
178 )
180 async def vread(
181 self,
182 tenant_id: UUID,
183 resource_type: str,
184 resource_id: str,
185 version_id: int,
186 ) -> ReadResult | None:
187 """
188 Return a specific version (including deleted markers).
189 It uses the Primary Key from the fhir_resources table.
190 """
191 row: asyncpg.Record | None = await self._conn.fetchrow(
192 """
193 SELECT resource, version_id, last_updated, deleted
194 FROM fhir_resources
195 WHERE tenant_id = $1
196 AND resource_type = $2
197 AND resource_id = $3
198 AND version_id = $4
199 """,
200 tenant_id,
201 resource_type,
202 resource_id,
203 version_id,
204 )
205 if row is None:
206 return None
207 return ReadResult(
208 resource=_decode_jsonb(row["resource"]),
209 version_id=row["version_id"],
210 last_updated=row["last_updated"],
211 )
213 # ── Write ──────────────────────────────────────────────────────────────────
215 async def create(
216 self,
217 tenant_id: UUID,
218 resource_type: str,
219 resource_id: str,
220 resource: dict[str, Any],
221 transaction_refs: set[str] | None = None,
222 ) -> WriteResult:
223 """
224 Insert as version 1, populate search indexes.
225 Raises asyncpg.UniqueViolationError if the ID exists.
226 """
227 await self._extract_binary_from_document_reference(tenant_id, resource)
228 await self._check_reference_integrity(tenant_id, resource, transaction_refs)
229 await self._check_timezone_offsets(tenant_id, resource)
230 await self._apply_derivation_rules(tenant_id, resource_type, resource)
231 row: asyncpg.Record = await self._conn.fetchrow( # type: ignore[assignment]
232 """
233 INSERT INTO fhir_resources
234 (tenant_id, resource_type, resource_id, version_id, resource)
235 VALUES ($1, $2, $3, 1, $4::jsonb)
236 RETURNING version_id, last_updated
237 """,
238 tenant_id,
239 resource_type,
240 resource_id,
241 # Pass the dict directly; the pool's jsonb codec encodes it exactly
242 # once. Calling json.dumps here too would double-encode (store a
243 # jsonb *string* scalar) and defeat the GIN index / path queries.
244 resource,
245 )
246 # handle indexing
247 params = await self._resolve_params(tenant_id, resource_type)
248 await index_resource(self._conn, tenant_id, resource_type, resource_id, resource, params)
249 await self._post_write(tenant_id, resource_type, resource)
250 logger.debug(
251 "resource.created",
252 resource_type=resource_type,
253 resource_id=resource_id,
254 version_id=row["version_id"],
255 )
256 await self._audit(tenant_id, "C", resource_type, resource_id)
257 record_write_event(
258 WriteEvent(
259 tenant_id=tenant_id,
260 operation="create",
261 resource_type=resource_type,
262 resource_id=resource_id,
263 version_id=row["version_id"],
264 last_updated=row["last_updated"],
265 resource=resource,
266 )
267 )
268 return WriteResult(
269 resource=resource, version_id=row["version_id"], last_updated=row["last_updated"]
270 )
272 async def update(
273 self,
274 tenant_id: UUID,
275 resource_type: str,
276 resource_id: str,
277 resource: dict[str, Any],
278 transaction_refs: set[str] | None = None,
279 ) -> WriteResult:
280 """Insert a new version (current_max + 1), refresh search indexes."""
281 await self._extract_binary_from_document_reference(tenant_id, resource)
282 await self._check_reference_integrity(tenant_id, resource, transaction_refs)
283 await self._check_timezone_offsets(tenant_id, resource)
284 await self._apply_derivation_rules(tenant_id, resource_type, resource)
285 try:
286 # Demote-then-insert must be two statements in a transaction, not one CTE: within a
287 # single statement the UPDATE and INSERT share one snapshot and do not see each
288 # other, so the partial unique index on is_current would see both the old and the
289 # new row as current and reject every update. Sequenced in a transaction, the
290 # INSERT sees the demoted row and the pair is atomic — a failed INSERT rolls the
291 # demote back. asyncpg nests via savepoint if a transaction is already open (bundles).
292 async with self._conn.transaction():
293 await self._conn.execute(
294 """
295 UPDATE fhir_resources SET is_current = FALSE
296 WHERE tenant_id = $1 AND resource_type = $2 AND resource_id = $3
297 AND is_current
298 """,
299 tenant_id,
300 resource_type,
301 resource_id,
302 )
303 row: asyncpg.Record = await self._conn.fetchrow( # type: ignore[assignment]
304 """
305 INSERT INTO fhir_resources
306 (tenant_id, resource_type, resource_id, version_id, resource, is_current)
307 SELECT $1, $2, $3,
308 COALESCE(MAX(version_id), 0) + 1,
309 $4::jsonb,
310 TRUE
311 FROM fhir_resources
312 WHERE tenant_id = $1
313 AND resource_type = $2
314 AND resource_id = $3
315 RETURNING version_id, last_updated
316 """,
317 tenant_id,
318 resource_type,
319 resource_id,
320 # Pass the dict directly; the pool's jsonb codec encodes it exactly
321 # once. Calling json.dumps here too would double-encode (store a
322 # jsonb *string* scalar) and defeat the GIN index / path queries.
323 resource,
324 )
325 except (asyncpg.UniqueViolationError, asyncpg.DeadlockDetectedError) as exc:
326 # Concurrent updates of the same resource conflict in one of two ways: they compute
327 # the same next version and collide on the (…, version_id) primary key or on the
328 # partial unique index that permits one is_current row per resource (migration 009);
329 # or the demote-then-insert pair of two writers interleaves into a lock cycle and
330 # Postgres aborts one as a deadlock. All are the same thing from the client's side —
331 # a write that raced — and were surfaced as a 500 before. 409 tells the client to
332 # re-read the current version and re-apply, rather than the server retrying blindly.
333 raise OperationOutcomeError(
334 409,
335 "conflict",
336 f"Concurrent update of {resource_type}/{resource_id}; "
337 "retry with the current version",
338 ) from exc
339 params = await self._resolve_params(tenant_id, resource_type)
340 await index_resource(self._conn, tenant_id, resource_type, resource_id, resource, params)
341 await self._post_write(tenant_id, resource_type, resource)
342 logger.debug(
343 "resource.updated",
344 resource_type=resource_type,
345 resource_id=resource_id,
346 version_id=row["version_id"],
347 )
348 await self._audit(tenant_id, "U", resource_type, resource_id)
349 record_write_event(
350 WriteEvent(
351 tenant_id=tenant_id,
352 operation="update",
353 resource_type=resource_type,
354 resource_id=resource_id,
355 version_id=row["version_id"],
356 last_updated=row["last_updated"],
357 resource=resource,
358 )
359 )
360 return WriteResult(
361 resource=resource, version_id=row["version_id"], last_updated=row["last_updated"]
362 )
364 async def delete(
365 self,
366 tenant_id: UUID,
367 resource_type: str,
368 resource_id: str,
369 ) -> bool:
370 """Soft-delete: inserts a tombstone version with deleted=TRUE, removes search indexes.
372 Returns False if the resource does not exist or is already deleted.
373 """
374 current = await self.read(tenant_id, resource_type, resource_id)
375 if current is None:
376 return False
378 # The tombstone becomes the current row (is_current TRUE, deleted TRUE): it is the
379 # newest version. Search filters is_current AND NOT deleted, so it is excluded there;
380 # read() checks deleted and returns None. Demote the prior current row first, in the
381 # same transaction, so exactly one is_current row remains (see update() for why this
382 # cannot be a single CTE statement).
383 async with self._conn.transaction():
384 await self._conn.execute(
385 """
386 UPDATE fhir_resources SET is_current = FALSE
387 WHERE tenant_id = $1 AND resource_type = $2 AND resource_id = $3
388 AND is_current
389 """,
390 tenant_id,
391 resource_type,
392 resource_id,
393 )
394 await self._conn.execute(
395 """
396 INSERT INTO fhir_resources
397 (tenant_id, resource_type, resource_id, version_id,
398 resource, deleted, is_current)
399 SELECT $1, $2, $3,
400 MAX(version_id) + 1,
401 $4::jsonb,
402 TRUE,
403 TRUE
404 FROM fhir_resources
405 WHERE tenant_id = $1
406 AND resource_type = $2
407 AND resource_id = $3
408 """,
409 tenant_id,
410 resource_type,
411 resource_id,
412 current["resource"], # dict; jsonb codec encodes once (see create)
413 )
414 await delete_indexes(self._conn, tenant_id, resource_type, resource_id)
415 await self._post_delete(tenant_id, resource_type, current["resource"])
416 logger.debug("resource.deleted", resource_type=resource_type, resource_id=resource_id)
417 await self._audit(tenant_id, "D", resource_type, resource_id)
418 record_write_event(
419 WriteEvent(
420 tenant_id=tenant_id,
421 operation="delete",
422 resource_type=resource_type,
423 resource_id=resource_id,
424 # The tombstone is inserted as MAX(version_id) + 1 above.
425 version_id=int(current["version_id"]) + 1,
426 last_updated=None,
427 resource=None,
428 )
429 )
430 return True
432 async def is_deleted(
433 self,
434 tenant_id: UUID,
435 resource_type: str,
436 resource_id: str,
437 ) -> bool:
438 """Return True if the resource exists but its latest version is a tombstone."""
439 row: asyncpg.Record | None = await self._conn.fetchrow(
440 """
441 SELECT deleted
442 FROM fhir_resources
443 WHERE tenant_id = $1
444 AND resource_type = $2
445 AND resource_id = $3
446 ORDER BY version_id DESC
447 LIMIT 1
448 """,
449 tenant_id,
450 resource_type,
451 resource_id,
452 )
453 if row is None:
454 return False
455 return bool(row["deleted"])
457 # ── History ────────────────────────────────────────────────────────────────
459 async def history(
460 self,
461 tenant_id: UUID,
462 resource_type: str,
463 resource_id: str,
464 ) -> list[HistoryEntry]:
465 """Return all versions, newest first."""
466 rows: list[asyncpg.Record] = await self._conn.fetch(
467 """
468 SELECT resource, version_id, last_updated, deleted
469 FROM fhir_resources
470 WHERE tenant_id = $1
471 AND resource_type = $2
472 AND resource_id = $3
473 ORDER BY version_id DESC
474 """,
475 tenant_id,
476 resource_type,
477 resource_id,
478 )
479 return [
480 HistoryEntry(
481 resource=_decode_jsonb(r["resource"]),
482 version_id=r["version_id"],
483 last_updated=r["last_updated"],
484 deleted=r["deleted"],
485 )
486 for r in rows
487 ]
489 # ── Compartment ──────────────────────────────────────────────────────────
491 async def compartment_members(
492 self,
493 tenant_id: UUID,
494 target_resource_type: str,
495 target_resource_id: str,
496 compartment_params: Mapping[str, Sequence[str]],
497 ) -> list[tuple[str, str]]:
498 """Return (resource_type, resource_id) of resources in the target's compartment.
500 Membership follows the CompartmentDefinition: ``compartment_params`` maps
501 each resource type to the search parameters whose reference to the target
502 places it in the compartment. Only references via those exact
503 (resource_type, param_name) pairs count. Used by ``$everything``.
504 """
505 types: list[str] = []
506 names: list[str] = []
507 for resource_type, params in compartment_params.items():
508 for param_name in params:
509 types.append(resource_type)
510 names.append(param_name)
511 if not types:
512 return []
513 rows: list[asyncpg.Record] = await self._conn.fetch(
514 """
515 SELECT DISTINCT s.resource_type, s.resource_id
516 FROM spidx_reference s
517 JOIN unnest($4::text[], $5::text[]) AS cp(rt, pn)
518 ON s.resource_type = cp.rt AND s.param_name = cp.pn
519 WHERE s.tenant_id = $1
520 AND s.target_resource_type = $2
521 AND s.target_resource_id = $3
522 """,
523 tenant_id,
524 target_resource_type,
525 target_resource_id,
526 types,
527 names,
528 )
529 return [(r["resource_type"], r["resource_id"]) for r in rows]
531 async def is_compartment_member(
532 self,
533 tenant_id: UUID,
534 resource_type: str,
535 resource_id: str,
536 target_resource_type: str,
537 target_resource_id: str,
538 param_names: Sequence[str],
539 ) -> bool:
540 """Check whether one specific resource belongs to another resource's compartment.
542 Narrower single-resource variant of ``compartment_members`` — used to
543 authorize a direct read/update/delete by id under a ``patient/`` scope
544 without listing the whole compartment.
545 """
546 if not param_names:
547 return False
548 row = await self._conn.fetchrow(
549 """
550 SELECT 1
551 FROM spidx_reference
552 WHERE tenant_id = $1
553 AND resource_type = $2
554 AND resource_id = $3
555 AND target_resource_type = $4
556 AND target_resource_id = $5
557 AND param_name = ANY($6::text[])
558 LIMIT 1
559 """,
560 tenant_id,
561 resource_type,
562 resource_id,
563 target_resource_type,
564 target_resource_id,
565 list(param_names),
566 )
567 return row is not None
569 async def find_current_by_url(
570 self,
571 tenant_id: UUID,
572 resource_type: str,
573 url: str,
574 ) -> ReadResult | None:
575 """Resolve a canonical resource (CodeSystem/ValueSet/ConceptMap) by its ``url``.
577 Returns the current (latest, non-deleted) version whose top-level ``url``
578 matches, or None. Used by the terminology operations.
579 """
580 row: asyncpg.Record | None = await self._conn.fetchrow(
581 """
582 SELECT DISTINCT ON (resource_id) resource, version_id, last_updated, deleted
583 FROM fhir_resources
584 WHERE tenant_id = $1
585 AND resource_type = $2
586 AND resource ->> 'url' = $3
587 ORDER BY resource_id, version_id DESC
588 """,
589 tenant_id,
590 resource_type,
591 url,
592 )
593 if row is None or row["deleted"]:
594 return None
595 return ReadResult(
596 resource=_decode_jsonb(row["resource"]),
597 version_id=row["version_id"],
598 last_updated=row["last_updated"],
599 )
601 # ── Search ─────────────────────────────────────────────────────────────────
603 async def search(
604 self,
605 tenant_id: UUID,
606 resource_type: str,
607 query: Any,
608 ) -> Any:
609 """Execute a parsed FHIR search query. ``query`` is a ``ParsedQuery``."""
610 from ..search.query import execute_search
612 params = await self._resolve_params(tenant_id, resource_type)
613 logger.debug(
614 "resource.search_executed",
615 resource_type=resource_type,
616 query=str(query.raw_query) if hasattr(query, "raw_query") else str(query),
617 )
618 return await execute_search(self._conn, tenant_id, resource_type, query, params)
620 # ── Post-write hooks ──────────────────────────────────────────────────────
622 async def _resolve_params(
623 self,
624 tenant_id: UUID,
625 resource_type: str,
626 ) -> list[Any]:
627 """
628 Return merged search params for a resource type and tenant.
629 """
630 if self._registry is None:
631 # Fallback: base params only (e.g. in tests without registry).
632 from ..fhir.search_param_loader import load_registry
634 return load_registry().get(resource_type, [])
635 return await self._registry.get_params(self._conn, tenant_id, resource_type)
637 async def _post_write(
638 self,
639 tenant_id: UUID,
640 resource_type: str,
641 resource: dict[str, Any],
642 ) -> None:
643 """Run post-write side effects for conformance resources."""
644 if resource_type == "SearchParameter" and self._registry is not None:
645 from ..search.search_param_hooks import (
646 SearchParamValidationError,
647 handle_search_param_write,
648 )
650 try:
651 await handle_search_param_write(self._conn, tenant_id, resource, self._registry)
652 except SearchParamValidationError as exc:
653 await logger.awarning(
654 "search_param.validation_failed",
655 error=exc.message,
656 resource_id=resource.get("id"),
657 )
658 raise
659 elif resource_type in ("ValueSet", "CodeSystem"):
660 from ..fhir.terminology_binding import get_valueset_cache
662 get_valueset_cache().invalidate(tenant_id)
663 if resource_type == "ValueSet":
664 from ..fhir.terminology_pre_expansion import maybe_pre_expand
666 await maybe_pre_expand(self._conn, tenant_id, resource)
667 elif resource_type == "StructureDefinition":
668 from ..fhir.profile_validator import get_profile_cache
670 get_profile_cache().invalidate(tenant_id)
672 async def _apply_derivation_rules(
673 self,
674 tenant_id: UUID,
675 resource_type: str,
676 resource: dict[str, Any],
677 ) -> None:
678 """Fill in tenant-registered missing fields via ConceptMap-based derivation.
680 No-op when no rules are registered for the tenant/resource type — the
681 common case, since no package installs any by default.
682 """
683 from ..derivation.engine import apply_derivation_rules
684 from ..derivation.registry import get_registry as get_derivation_registry
686 await apply_derivation_rules(
687 self._conn, tenant_id, resource_type, resource, get_derivation_registry()
688 )
690 async def _audit(
691 self,
692 tenant_id: UUID,
693 action: str,
694 resource_type: str,
695 resource_id: str,
696 ) -> None:
697 """Create and persist an AuditEvent for the given action."""
698 if resource_type == "AuditEvent":
699 return
701 audit_event = build_audit_event(action, resource_type, resource_id)
702 audit_id = uuid4().hex
703 audit_event["id"] = audit_id
704 await self.create(tenant_id, "AuditEvent", audit_id, audit_event)
706 async def _post_delete(
707 self,
708 tenant_id: UUID,
709 resource_type: str,
710 resource: dict[str, Any],
711 ) -> None:
712 """Run post-delete side effects for conformance resources."""
713 if resource_type == "SearchParameter" and self._registry is not None:
714 from ..search.search_param_hooks import handle_search_param_delete
716 await handle_search_param_delete(self._conn, tenant_id, resource, self._registry)
717 elif resource_type in ("ValueSet", "CodeSystem"):
718 from ..fhir.terminology_binding import get_valueset_cache
720 get_valueset_cache().invalidate(tenant_id)
721 elif resource_type == "StructureDefinition":
722 from ..fhir.profile_validator import get_profile_cache
724 get_profile_cache().invalidate(tenant_id)
726 async def _extract_binary_from_document_reference(
727 self,
728 tenant_id: UUID,
729 resource: dict[str, Any],
730 ) -> None:
731 if resource.get("resourceType") != "DocumentReference":
732 return
734 contents = resource.get("content")
735 if not isinstance(contents, list):
736 return
738 for content in cast(list[dict[str, Any]], contents):
739 attachment = content.get("attachment")
740 if not isinstance(attachment, dict):
741 continue
742 attachment_dict = cast(dict[str, Any], attachment)
744 data = attachment_dict.get("data")
745 if data and isinstance(data, str):
746 binary_id = str(uuid4())
747 binary_res: dict[str, Any] = {
748 "resourceType": "Binary",
749 "id": binary_id,
750 "contentType": str(
751 attachment_dict.get("contentType", "application/octet-stream")
752 ),
753 "data": data,
754 }
755 await self.create(tenant_id, "Binary", binary_id, binary_res)
756 attachment_dict.pop("data", None)
757 attachment_dict["url"] = f"Binary/{binary_id}"
759 async def _get_reference_integrity_setting(self, tenant_id: UUID) -> str:
760 row = await self._conn.fetchrow(
761 "SELECT settings FROM tenant_settings WHERE tenant_id = $1",
762 tenant_id,
763 )
764 if row:
765 raw_settings = row["settings"]
766 settings_dict: dict[str, Any] = {}
767 if isinstance(raw_settings, str):
768 settings_dict = cast("dict[str, Any]", json.loads(raw_settings))
769 elif isinstance(raw_settings, dict):
770 settings_dict = cast("dict[str, Any]", raw_settings)
771 if "reference_integrity" in settings_dict:
772 return str(settings_dict["reference_integrity"])
773 from ..config import get_settings
775 return get_settings().tenant.reference_integrity
777 async def _check_reference_integrity(
778 self,
779 tenant_id: UUID,
780 resource: dict[str, Any],
781 transaction_refs: set[str] | None = None,
782 ) -> None:
783 strictness = await self._get_reference_integrity_setting(tenant_id)
784 if strictness == "disabled":
785 return
787 refs = extract_references(resource)
788 # A urn:uuid reference still present at this point (after bundle
789 # processing rewrites the ones that resolve) never resolves to
790 # anything — no DB lookup needed, it's always an error.
791 errors: list[str] = list(extract_dangling_urn_references(resource))
793 # Candidate references to look up: deduped, minus any satisfied within
794 # the same transaction (bundle). Resolved in a single batched query
795 # rather than one round-trip per reference (avoids N+1 on write).
796 candidates = {
797 (t, i) for t, i in refs if not (transaction_refs and f"{t}/{i}" in transaction_refs)
798 }
799 errors.extend(await self._find_missing_references(tenant_id, candidates))
801 if errors:
802 msg = f"Referenced resource(s) not found: {', '.join(errors)}"
803 if strictness == "strict":
804 raise OperationOutcomeError(
805 status_code=422,
806 code="processing",
807 diagnostics=msg,
808 )
809 else:
810 await logger.awarning(
811 "reference_integrity.warning", message=msg, tenant_id=tenant_id
812 )
814 async def _find_missing_references(
815 self, tenant_id: UUID, candidates: set[tuple[str, str]]
816 ) -> list[str]:
817 """Return ``"Type/id"`` for each candidate reference not present (and not
818 deleted) in the tenant's resources, resolved in a single batched query.
820 The two parallel ``text[]`` arrays are zipped back into rows via
821 ``unnest`` and joined against ``fhir_resources``; whatever the join
822 doesn't return is missing. Preserves the per-reference lookup semantics
823 of the previous N+1 loop without one round-trip per reference.
824 """
825 if not candidates:
826 return []
828 types = [t for t, _ in candidates]
829 ids = [i for _, i in candidates]
830 rows = await self._conn.fetch(
831 """
832 SELECT c.rt AS rt, c.rid AS rid
833 FROM unnest($2::text[], $3::text[]) AS c(rt, rid)
834 JOIN fhir_resources f
835 ON f.tenant_id = $1
836 AND f.resource_type = c.rt
837 AND f.resource_id = c.rid
838 AND f.deleted = FALSE
839 """,
840 tenant_id,
841 types,
842 ids,
843 )
844 existing = {(row["rt"], row["rid"]) for row in rows}
845 return sorted(f"{t}/{i}" for t, i in candidates if (t, i) not in existing)
847 async def _get_timezone_strictness_setting(self, tenant_id: UUID) -> str:
848 row = await self._conn.fetchrow(
849 "SELECT settings FROM tenant_settings WHERE tenant_id = $1",
850 tenant_id,
851 )
852 if row:
853 raw_settings = row["settings"]
854 settings_dict: dict[str, Any] = {}
855 if isinstance(raw_settings, str):
856 settings_dict = cast("dict[str, Any]", json.loads(raw_settings))
857 elif isinstance(raw_settings, dict):
858 settings_dict = cast("dict[str, Any]", raw_settings)
859 if "timezone_strictness" in settings_dict:
860 return str(settings_dict["timezone_strictness"])
861 from ..config import get_settings
863 return get_settings().tenant.timezone_strictness
865 async def _check_timezone_offsets(
866 self,
867 tenant_id: UUID,
868 resource: dict[str, Any],
869 ) -> None:
870 strictness = await self._get_timezone_strictness_setting(tenant_id)
871 if strictness == "disabled":
872 return
874 violations = find_timezone_violations(resource)
875 if violations:
876 msg = f"dateTime/instant value must include timezone offset: {', '.join(violations)}"
877 if strictness == "strict":
878 raise OperationOutcomeError(
879 status_code=422,
880 code="value",
881 diagnostics=msg,
882 )
883 else:
884 await logger.awarning(
885 "timezone_validation.warning", message=msg, tenant_id=tenant_id
886 )