Coverage for src/lumen/api/resources.py: 91%
282 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-31 18:46 +0000
1"""
2This file handles the main FHIR RESTful resource routes.
3It provides handlers for CRUD operations, conditional updates, patch requests, and history.
4"""
6from __future__ import annotations
8import base64
9import json
10from datetime import datetime
11from typing import Any, cast
12from uuid import UUID
14import asyncpg
15import jsonpatch # type: ignore[import-untyped]
16import jsonpointer # type: ignore[import-untyped]
17import structlog
18from fastapi import APIRouter, Request
19from fastapi.responses import Response
21from ..auth.enforcement import ScopeEnforcerDep
22from ..core.exceptions import FHIR_JSON, OperationOutcomeError
23from ..db.pool import get_pool
24from ..db.repository import ResourceRepository
25from ..fhir import apply_fhirpath_patch, parse_fhirpath_patch
26from ..fhir.binding_validator import validate_code_bindings
27from ..fhir.profile_validator import validate_profiles, validate_terminology_bindings
28from ..fhir.resource_types import known_resource_types
29from ..fhir.validator import validate_resource
30from ..observability import record_interaction
31from ..search.bundle import attach_includes, build_count_bundle, build_searchset_bundle
32from ..search.parser import ParsedQuery, parse_query
33from ..search.query import SearchResult, execute_search
34from ..search.registry import get_registry
35from ..service import ResourceService
36from ._helpers import (
37 etag as _etag,
38)
39from ._helpers import (
40 inject_meta as _inject_meta,
41)
42from ._helpers import (
43 last_modified as _last_modified,
44)
45from ._helpers import (
46 parse_if_match as _parse_if_match,
47)
48from .deps import ResolvedTenantSettings, TenantDep, TenantSettingsDep
50logger = structlog.get_logger(__name__)
52router = APIRouter(dependencies=[ScopeEnforcerDep])
55def _not_found(resource_type: str, resource_id: str) -> OperationOutcomeError:
56 return OperationOutcomeError(
57 status_code=404,
58 code="not-found",
59 diagnostics=f"{resource_type}/{resource_id} not found",
60 )
63# Media types that mean "give me the FHIR-wrapped resource", as opposed to a
64# Binary's own native content type.
65_FHIR_ACCEPT_TYPES = {
66 "application/fhir+json",
67 "application/fhir+xml",
68 "application/json",
69 "*/*",
70 "application/*",
71}
74def _binary_native_response(resource: dict[str, Any], accept: str) -> Response | None:
75 """Serve a Binary's raw ``data`` when the client's Accept header asks for
76 its native ``contentType`` instead of a FHIR representation.
78 Returns None to fall through to the normal FHIR-wrapped response (client
79 asked for a FHIR format), or raises 406 if the Accept header matches
80 neither a FHIR format nor the resource's native content type.
81 """
82 content_type = resource.get("contentType")
83 data_b64 = resource.get("data")
84 media_types = {part.split(";")[0].strip().lower() for part in accept.split(",")}
85 if media_types & _FHIR_ACCEPT_TYPES:
86 return None
87 if (
88 not isinstance(content_type, str)
89 or not isinstance(data_b64, str)
90 or content_type.split(";")[0].strip().lower() not in media_types
91 ):
92 raise OperationOutcomeError(
93 status_code=406,
94 code="not-supported",
95 diagnostics="This server supports application/fhir+json, application/fhir+xml, "
96 "or this Binary resource's own content type.",
97 )
98 try:
99 raw = base64.b64decode(data_b64)
100 except ValueError as exc:
101 raise OperationOutcomeError(
102 status_code=500, code="exception", diagnostics="Stored Binary.data is not valid base64."
103 ) from exc
104 # Set Content-Type via headers, not media_type — Starlette auto-appends
105 # "; charset=utf-8" to text/* media types, but the stored contentType
106 # must be echoed back verbatim (e.g. bare "text/plain").
107 return Response(content=raw, headers={"content-type": content_type})
110# ── Read ───────────────────────────────────────────────────────────────────────
113@router.get("/{resource_type}/{resource_id}")
114async def read_resource(
115 resource_type: str,
116 resource_id: str,
117 request: Request,
118 tenant: TenantDep,
119) -> Response:
120 # Block _history from matching this route (FastAPI evaluates in order, but
121 # explicit check guards against any ordering surprises).
122 if resource_id == "_history":
123 raise OperationOutcomeError(404, "not-found", f"{resource_type} not found")
125 record_interaction(resource_type, "read")
126 pool: asyncpg.Pool[asyncpg.Record] = get_pool()
127 async with pool.acquire() as conn:
128 repo = ResourceRepository(conn, get_registry())
129 result = await repo.read(tenant, resource_type, resource_id)
130 if result is None:
131 if await repo.is_deleted(tenant, resource_type, resource_id):
132 raise OperationOutcomeError(
133 410, "gone", f"{resource_type}/{resource_id} has been deleted"
134 )
135 raise _not_found(resource_type, resource_id)
137 etag = _etag(result["version_id"])
138 if request.headers.get("if-none-match") == etag:
139 return Response(status_code=304)
141 if resource_type == "Binary":
142 native = _binary_native_response(result["resource"], request.headers.get("accept", "*/*"))
143 if native is not None:
144 return native
146 body = _inject_meta(result["resource"], result["version_id"], result["last_updated"])
147 return Response(
148 content=json.dumps(body),
149 media_type=FHIR_JSON,
150 headers={
151 "ETag": etag,
152 "Last-Modified": _last_modified(result["last_updated"]),
153 },
154 )
157# ── Vread ──────────────────────────────────────────────────────────────────────
160@router.get("/{resource_type}/{resource_id}/_history/{version_id}")
161async def vread_resource(
162 resource_type: str,
163 resource_id: str,
164 version_id: int,
165 tenant: TenantDep,
166) -> Response:
167 pool: asyncpg.Pool[asyncpg.Record] = get_pool()
168 async with pool.acquire() as conn:
169 result = await ResourceRepository(conn).vread(
170 tenant, resource_type, resource_id, version_id
171 )
173 if result is None:
174 raise OperationOutcomeError(
175 status_code=404,
176 code="not-found",
177 diagnostics=f"{resource_type}/{resource_id}/_history/{version_id} not found",
178 )
180 body = _inject_meta(result["resource"], result["version_id"], result["last_updated"])
181 return Response(
182 content=json.dumps(body),
183 media_type=FHIR_JSON,
184 headers={
185 "ETag": _etag(result["version_id"]),
186 "Last-Modified": _last_modified(result["last_updated"]),
187 },
188 )
191# ── History ────────────────────────────────────────────────────────────────────
194@router.get("/{resource_type}/{resource_id}/_history")
195async def history_resource(
196 resource_type: str,
197 resource_id: str,
198 request: Request,
199 tenant: TenantDep,
200) -> Response:
201 pool: asyncpg.Pool[asyncpg.Record] = get_pool()
202 async with pool.acquire() as conn:
203 entries = await ResourceRepository(conn).history(tenant, resource_type, resource_id)
205 base_url = str(request.base_url).rstrip("/")
206 bundle_entries = [
207 {
208 "fullUrl": f"{base_url}/{resource_type}/{resource_id}/_history/{e['version_id']}",
209 "resource": _inject_meta(e["resource"], e["version_id"], e["last_updated"]),
210 "request": {
211 "method": "GET",
212 "url": f"{resource_type}/{resource_id}/_history/{e['version_id']}",
213 },
214 }
215 for e in entries
216 ]
218 bundle: dict[str, Any] = {
219 "resourceType": "Bundle",
220 "type": "history",
221 "total": len(bundle_entries),
222 "entry": bundle_entries,
223 }
224 return Response(content=json.dumps(bundle), media_type=FHIR_JSON)
227# ── Search ─────────────────────────────────────────────────────────────────────
230def _collect_query_params(request: Request) -> dict[str, list[str]]:
231 """Build a multi-value dict from the request query string."""
232 raw: dict[str, list[str]] = {}
233 for key, value in request.query_params.multi_items():
234 raw.setdefault(key, []).append(value)
235 return raw
238async def _filter_to_patient_compartment(
239 conn: asyncpg.pool.PoolConnectionProxy[asyncpg.Record],
240 tenant: UUID,
241 resource_type: str,
242 restriction: dict[str, Any],
243 result: SearchResult,
244) -> None:
245 """Post-filter search results to the launch patient's compartment (mutates ``result``).
247 Set by `auth.enforcement._enforce_patient_compartment` when a request is
248 covered only by a `patient/` scope. Applied after the query since
249 compartment membership can be an OR across several reference parameters,
250 which the generic search query builder doesn't express as a single filter.
251 Note: this can leave a page shorter than `_count` if entries are filtered
252 out — never leaks other patients' data, but pagination fill-rate is a
253 known limitation of this approach.
254 """
255 patient_id = restriction["patient_id"]
256 if restriction["self_only"]:
257 member_ids = {patient_id}
258 else:
259 repo = ResourceRepository(conn)
260 members = await repo.compartment_members(
261 tenant, "Patient", patient_id, {resource_type: restriction["params"]}
262 )
263 member_ids = {resource_id for _, resource_id in members}
265 keep = [i for i, res in enumerate(result.resources) if res.get("id") in member_ids]
266 result.resources = [result.resources[i] for i in keep]
267 result.version_ids = [result.version_ids[i] for i in keep]
268 result.last_updated = [result.last_updated[i] for i in keep]
269 if result.total is not None:
270 result.total = len(keep)
273async def _run_search(
274 resource_type: str,
275 pq: ParsedQuery,
276 request: Request,
277 tenant: UUID,
278 tenant_settings: ResolvedTenantSettings,
279) -> Response:
280 pool: asyncpg.Pool[asyncpg.Record] = get_pool()
281 base_url = str(request.base_url).rstrip("/")
282 request_url = str(request.url)
283 registry = get_registry()
284 patient_compartment: dict[str, Any] | None = getattr(request.state, "patient_compartment", None)
286 ignore_unknown = tenant_settings.ignore_unknown_search_parameters
287 if pq.summary == "count":
288 async with pool.acquire() as conn:
289 params = await registry.get_params(conn, tenant, resource_type)
290 result = await execute_search(conn, tenant, resource_type, pq, params, ignore_unknown)
291 if patient_compartment is not None:
292 await _filter_to_patient_compartment(
293 conn, tenant, resource_type, patient_compartment, result
294 )
295 body = build_count_bundle(result.total or 0, request_url)
296 return Response(content=json.dumps(body), media_type=FHIR_JSON)
298 async with pool.acquire() as conn:
299 params = await registry.get_params(conn, tenant, resource_type)
300 result = await execute_search(conn, tenant, resource_type, pq, params, ignore_unknown)
301 if patient_compartment is not None:
302 await _filter_to_patient_compartment(
303 conn, tenant, resource_type, patient_compartment, result
304 )
305 body = build_searchset_bundle(base_url, resource_type, result, pq, request_url)
306 await attach_includes(
307 body,
308 conn,
309 tenant,
310 base_url,
311 pq.includes,
312 pq.revincludes,
313 result.resources,
314 )
316 return Response(content=json.dumps(body), media_type=FHIR_JSON)
319@router.get("/{resource_type}")
320async def search_resource(
321 resource_type: str,
322 request: Request,
323 tenant: TenantDep,
324 tenant_settings: TenantSettingsDep,
325) -> Response:
326 if resource_type not in known_resource_types():
327 raise OperationOutcomeError(
328 status_code=404,
329 code="not-found",
330 diagnostics=f"Unknown resource type: {resource_type}",
331 )
332 pq = parse_query(_collect_query_params(request), max_count=10000)
333 return await _run_search(resource_type, pq, request, tenant, tenant_settings)
336@router.post("/{resource_type}/_search")
337async def search_resource_post(
338 resource_type: str,
339 request: Request,
340 tenant: TenantDep,
341 tenant_settings: TenantSettingsDep,
342) -> Response:
343 if resource_type not in known_resource_types():
344 raise OperationOutcomeError(
345 status_code=404,
346 code="not-found",
347 diagnostics=f"Unknown resource type: {resource_type}",
348 )
349 # application/x-www-form-urlencoded body — parse manually so we can merge
350 # body params with (any) query params per the FHIR spec.
351 raw: dict[str, list[str]] = _collect_query_params(request)
352 body_bytes = await request.body()
353 if body_bytes:
354 from urllib.parse import parse_qs
356 body_params = parse_qs(body_bytes.decode(), keep_blank_values=True)
357 for k, vs in body_params.items():
358 raw.setdefault(k, []).extend(vs)
359 pq = parse_query(raw)
360 return await _run_search(resource_type, pq, request, tenant, tenant_settings)
363# ── Write helpers ──────────────────────────────────────────────────────────────
366def _service(tenant: UUID, tenant_settings: ResolvedTenantSettings) -> ResourceService:
367 """The write engine for this request. HTTP is one of its two transports."""
368 return ResourceService(tenant, tenant_settings)
371async def _validate_and_parse_body(
372 body_bytes: bytes,
373 expected_type: str,
374 tenant_settings: ResolvedTenantSettings,
375 tenant: UUID,
376 conn: Any = None,
377) -> dict[str, Any]:
378 """Parse JSON body, check resourceType matches URL, run schema validation."""
379 return await _service(tenant, tenant_settings).validate_and_parse_body(
380 body_bytes, expected_type, conn
381 )
384def _write_response(
385 resource: dict[str, Any],
386 version_id: int,
387 last_updated: datetime,
388 status_code: int,
389 location: str | None = None,
390) -> Response:
391 body = _inject_meta(resource, version_id, last_updated)
392 headers: dict[str, str] = {
393 "ETag": _etag(version_id),
394 "Last-Modified": _last_modified(last_updated),
395 }
396 if location:
397 headers["Location"] = location
398 return Response(
399 content=json.dumps(body),
400 media_type=FHIR_JSON,
401 status_code=status_code,
402 headers=headers,
403 )
406async def _conditional_search_ids(
407 resource_type: str,
408 query_string: str,
409 tenant: UUID,
410 tenant_settings: ResolvedTenantSettings,
411) -> list[str]:
412 """Execute a conditional search query and return matching resource_ids."""
413 return await _service(tenant, tenant_settings).conditional_search_ids(
414 resource_type, query_string
415 )
418# ── POST — Create ──────────────────────────────────────────────────────────────
421@router.post("/{resource_type}")
422async def create_resource(
423 resource_type: str,
424 request: Request,
425 tenant: TenantDep,
426 tenant_settings: TenantSettingsDep,
427) -> Response:
428 if resource_type not in known_resource_types():
429 raise OperationOutcomeError(404, "not-found", f"Unknown resource type: {resource_type}")
431 record_interaction(resource_type, "create")
432 service = _service(tenant, tenant_settings)
433 body_bytes = await request.body()
434 resource = await service.validate_and_parse_body(body_bytes, resource_type)
436 outcome = await service.create(
437 resource_type, resource, if_none_exist=request.headers.get("if-none-exist")
438 )
440 if not outcome.created:
441 # If-None-Exist matched: existing resource returned unchanged (200)
442 return _write_response(outcome.resource, outcome.version_id, outcome.last_updated, 200)
444 base_url = str(request.base_url).rstrip("/")
445 return _write_response(
446 outcome.resource,
447 outcome.version_id,
448 outcome.last_updated,
449 201,
450 location=f"{base_url}/{resource_type}/{outcome.resource_id}",
451 )
454# ── PUT — Update (+ create-on-update + conditional) ───────────────────────────
457@router.put("/{resource_type}/{resource_id}")
458async def update_resource(
459 resource_type: str,
460 resource_id: str,
461 request: Request,
462 tenant: TenantDep,
463 tenant_settings: TenantSettingsDep,
464) -> Response:
465 if resource_type not in known_resource_types():
466 raise OperationOutcomeError(404, "not-found", f"Unknown resource type: {resource_type}")
468 record_interaction(resource_type, "update")
469 service = _service(tenant, tenant_settings)
470 body_bytes = await request.body()
471 resource = await service.validate_and_parse_body(body_bytes, resource_type)
473 # Body id, if present, must match URL id.
474 body_id = resource.get("id")
475 if body_id is not None and body_id != resource_id:
476 raise OperationOutcomeError(
477 400,
478 "invalid",
479 f"Resource id in body ({body_id!r}) does not match URL id ({resource_id!r})",
480 )
482 if_match = _parse_if_match(request.headers.get("if-match"))
484 outcome = await service.update(resource_type, resource_id, resource, if_match=if_match)
486 if outcome.created:
487 base_url = str(request.base_url).rstrip("/")
488 return _write_response(
489 outcome.resource,
490 outcome.version_id,
491 outcome.last_updated,
492 201,
493 location=f"{base_url}/{resource_type}/{resource_id}",
494 )
496 return _write_response(outcome.resource, outcome.version_id, outcome.last_updated, 200)
499@router.put("/{resource_type}")
500async def conditional_update_resource(
501 resource_type: str,
502 request: Request,
503 tenant: TenantDep,
504 tenant_settings: TenantSettingsDep,
505) -> Response:
506 """PUT /{type}?<search-params> — conditional update."""
507 if resource_type not in known_resource_types():
508 raise OperationOutcomeError(404, "not-found", f"Unknown resource type: {resource_type}")
510 query_string = str(request.url.query)
511 if not query_string:
512 raise OperationOutcomeError(400, "invalid", "Conditional update requires query parameters")
514 service = _service(tenant, tenant_settings)
516 # The match count is resolved before the body is read, so a query matching
517 # several resources is rejected without ever validating the payload.
518 matching_ids = await service.conditional_search_ids(resource_type, query_string)
519 if len(matching_ids) > 1:
520 raise OperationOutcomeError(
521 412,
522 "multiple-matches",
523 f"Conditional update matched {len(matching_ids)} resources; must match exactly 0 or 1",
524 )
526 body_bytes = await request.body()
527 resource = await service.validate_and_parse_body(body_bytes, resource_type)
529 outcome = await service.conditional_update(
530 resource_type, query_string, resource, matching_ids=matching_ids
531 )
533 if outcome.created:
534 base_url = str(request.base_url).rstrip("/")
535 return _write_response(
536 outcome.resource,
537 outcome.version_id,
538 outcome.last_updated,
539 201,
540 location=f"{base_url}/{resource_type}/{outcome.resource_id}",
541 )
542 return _write_response(outcome.resource, outcome.version_id, outcome.last_updated, 200)
545# ── DELETE (+ conditional) ────────────────────────────────────────────────────
548@router.delete("/{resource_type}/{resource_id}")
549async def delete_resource(
550 resource_type: str,
551 resource_id: str,
552 tenant: TenantDep,
553) -> Response:
554 record_interaction(resource_type, "delete")
555 # No tenant settings: this route never resolved them, and adding a query to
556 # it would be a performance regression on the delete path.
557 deleted = await ResourceService(tenant).delete(resource_type, resource_id)
559 if not deleted:
560 raise OperationOutcomeError(404, "not-found", f"{resource_type}/{resource_id} not found")
562 return Response(status_code=204)
565@router.delete("/{resource_type}")
566async def conditional_delete_resource(
567 resource_type: str,
568 request: Request,
569 tenant: TenantDep,
570 tenant_settings: TenantSettingsDep,
571) -> Response:
572 """DELETE /{type}?<search-params> — conditional delete."""
573 if resource_type not in known_resource_types():
574 raise OperationOutcomeError(404, "not-found", f"Unknown resource type: {resource_type}")
576 query_string = str(request.url.query)
577 if not query_string:
578 raise OperationOutcomeError(400, "invalid", "Conditional delete requires query parameters")
580 await _service(tenant, tenant_settings).conditional_delete(resource_type, query_string)
581 return Response(status_code=204)
584# ── PATCH — JSON Patch ─────────────────────────────────────────────────────────
587@router.patch("/{resource_type}/{resource_id}")
588async def patch_resource(
589 resource_type: str,
590 resource_id: str,
591 request: Request,
592 tenant: TenantDep,
593 tenant_settings: TenantSettingsDep,
594) -> Response:
595 if resource_type not in known_resource_types():
596 raise OperationOutcomeError(404, "not-found", f"Unknown resource type: {resource_type}")
598 body_bytes = await request.body()
599 try:
600 parsed_body: Any = json.loads(body_bytes)
601 except json.JSONDecodeError as exc:
602 raise OperationOutcomeError(400, "invalid", f"Invalid JSON body: {exc}") from exc
604 content_type = request.headers.get("content-type", "").lower()
605 is_fhirpath_patch = False
607 if "application/fhir+json" in content_type or (
608 isinstance(parsed_body, dict)
609 and cast(dict[str, Any], parsed_body).get("resourceType") == "Parameters"
610 ):
611 is_fhirpath_patch = True
613 if_match = _parse_if_match(request.headers.get("if-match"))
615 pool = get_pool()
616 async with pool.acquire() as conn:
617 repo = ResourceRepository(conn, get_registry())
618 existing = await repo.read(tenant, resource_type, resource_id)
620 if existing is None:
621 raise OperationOutcomeError(
622 404, "not-found", f"{resource_type}/{resource_id} not found"
623 )
625 if if_match is not None and if_match != existing["version_id"]:
626 raise OperationOutcomeError(
627 412,
628 "conflict",
629 f"Version mismatch: expected {if_match}, current is {existing['version_id']}",
630 )
632 if is_fhirpath_patch:
633 if not isinstance(parsed_body, dict):
634 raise OperationOutcomeError(
635 400, "invalid", "FHIRPath Patch payload must be a Parameters JSON object"
636 )
637 try:
638 ops = parse_fhirpath_patch(cast(dict[str, Any], parsed_body))
639 except ValueError as exc:
640 raise OperationOutcomeError(
641 400, "invalid", f"Invalid FHIRPath Patch Parameters: {exc}"
642 ) from exc
644 try:
645 patched = apply_fhirpath_patch(existing["resource"], ops)
646 except Exception as exc:
647 raise OperationOutcomeError(
648 422, "invalid", f"FHIRPath Patch failed: {exc}"
649 ) from exc
650 else:
651 if not isinstance(parsed_body, list):
652 raise OperationOutcomeError(400, "invalid", "JSON Patch body must be an array")
653 patch_ops = cast(list[dict[str, Any]], parsed_body)
654 try:
655 patch = jsonpatch.JsonPatch(patch_ops)
656 patched = cast(dict[str, Any], patch.apply(existing["resource"])) # type: ignore[reportUnknownMemberType] # jsonpatch has no stubs
657 except jsonpatch.JsonPatchTestFailed as exc:
658 raise OperationOutcomeError(
659 422, "conflict", f"JSON Patch test failed: {exc}"
660 ) from exc
661 except (jsonpatch.JsonPatchException, jsonpointer.JsonPointerException) as exc:
662 raise OperationOutcomeError(422, "invalid", f"JSON Patch failed: {exc}") from exc
664 # Run inbound validation on the patched resource before updating
665 errors = validate_resource(patched)
666 if errors:
667 if tenant_settings.validation_strictness == "strict":
668 raise OperationOutcomeError(
669 422,
670 "invalid",
671 f"Resource validation failed: {errors[0]}",
672 )
673 else:
674 logger.warning("resource.validation_failed_lenient", errors=errors)
676 binding_errors = validate_code_bindings(patched) + await validate_terminology_bindings(
677 patched, tenant, conn, tenant_settings.enforced_profiles
678 )
679 if binding_errors and tenant_settings.terminology_strictness == "strict":
680 raise OperationOutcomeError(422, "code-invalid", binding_errors[0])
682 profile_errors = await validate_profiles(
683 patched,
684 tenant,
685 conn,
686 tenant_settings.profile_validation_mode,
687 tenant_settings.enforced_profiles,
688 )
689 if profile_errors:
690 raise OperationOutcomeError(
691 422,
692 "invalid",
693 f"Profile validation failed: {profile_errors[0]}",
694 )
696 result = await repo.update(tenant, resource_type, resource_id, patched)
698 return _write_response(result["resource"], result["version_id"], result["last_updated"], 200)