Coverage for src/lumen/api/operations.py: 67%

1""" 

2This route handler manages the execution of custom FHIR operations. 

3It matches operations at the system, type, and instance levels and routes them appropriately. 

4""" 

5 

6from __future__ import annotations 

7 

8import json 

9from collections.abc import Awaitable, Callable 

10from dataclasses import dataclass 

11from enum import StrEnum 

12from typing import Any, cast 

13from uuid import UUID 

14 

15import asyncpg 

16from fastapi import APIRouter, Request 

17from fastapi.responses import Response 

18 

19from ..auth.enforcement import ScopeEnforcerDep 

20from ..core.exceptions import FHIR_JSON, OperationOutcomeError 

21from ..db.pool import get_pool 

22from ..db.repository import ResourceRepository 

23from ..fhir.binding_validator import validate_code_bindings 

24from ..fhir.compartment import compartment_params 

25from ..fhir.invariants import validate_invariants 

26from ..fhir.resource_types import known_resource_types 

27from ..fhir.validator import validate_resource 

28from ..search.registry import get_registry 

29from ._helpers import inject_meta as _inject_meta 

30from ._helpers import new_id as _new_id 

31from .deps import TenantDep 

32 

33router = APIRouter(dependencies=[ScopeEnforcerDep]) 

34 

35 

36class OpScope(StrEnum): 

37 SYSTEM = "system" 

38 TYPE = "type" 

39 INSTANCE = "instance" 

40 

41 

42@dataclass(frozen=True) 

43class OperationContext: 

44 """Everything an operation handler needs about the invocation.""" 

45 

46 request: Request 

47 scope: OpScope 

48 resource_type: str | None 

49 resource_id: str | None 

50 body: bytes 

51 tenant_id: UUID 

52 

53 

54OperationHandler = Callable[[OperationContext], Awaitable[Response]] 

55 

56_REGISTRY: dict[tuple[str, OpScope], OperationHandler] = {} 

57 

58# Operations advertised per resource type in the CapabilityStatement. 

59# The "*" key applies to every resource type; named keys add type-specific ones. 

60_FHIR_OP_BASE = "http://hl7.org/fhir/OperationDefinition" 

61_ISIK_OP_BASE = "https://gematik.de/fhir/isik/OperationDefinition" 

62 

63CAPABILITY_OPERATIONS: dict[str, list[dict[str, str]]] = { 

64 "*": [ 

65 {"name": "validate", "definition": f"{_FHIR_OP_BASE}/Resource-validate"}, 

66 {"name": "meta", "definition": f"{_FHIR_OP_BASE}/Resource-meta"}, 

67 {"name": "meta-add", "definition": f"{_FHIR_OP_BASE}/Resource-meta-add"}, 

68 {"name": "meta-delete", "definition": f"{_FHIR_OP_BASE}/Resource-meta-delete"}, 

69 ], 

70 "Patient": [{"name": "everything", "definition": f"{_FHIR_OP_BASE}/Patient-everything"}], 

71 "Encounter": [{"name": "everything", "definition": f"{_FHIR_OP_BASE}/Encounter-everything"}], 

72 "CodeSystem": [{"name": "lookup", "definition": f"{_FHIR_OP_BASE}/CodeSystem-lookup"}], 

73 "ValueSet": [{"name": "expand", "definition": f"{_FHIR_OP_BASE}/ValueSet-expand"}], 

74 "ConceptMap": [{"name": "translate", "definition": f"{_FHIR_OP_BASE}/ConceptMap-translate"}], 

75 "Appointment": [ 

76 {"name": "book", "definition": f"{_ISIK_OP_BASE}/ISiKAppointmentBook"}, 

77 ], 

78 "DocumentReference": [ 

79 { 

80 "name": "update-metadata", 

81 "definition": f"{_ISIK_OP_BASE}/ISiKDocumentReferenceUpdateMetadata", 

82 }, 

83 ], 

84} 

85 

86 

87# Operations contributed by registered Cortex plugins, keyed the same way. 

88# Kept apart from the built-in table so unregistering a plugin cannot damage it. 

89_PLUGIN_OPERATIONS: dict[str, list[dict[str, str]]] = {} 

90 

91 

92def operations_for(resource_type: str) -> list[dict[str, str]]: 

93 """Return the CapabilityStatement ``operation`` entries for a resource type.""" 

94 return [ 

95 *CAPABILITY_OPERATIONS["*"], 

96 *CAPABILITY_OPERATIONS.get(resource_type, []), 

97 *_PLUGIN_OPERATIONS.get("*", []), 

98 *_PLUGIN_OPERATIONS.get(resource_type, []), 

99 ] 

100 

101 

102def operation(name: str, *scopes: OpScope) -> Callable[[OperationHandler], OperationHandler]: 

103 """Register ``fn`` as the handler for ``$name`` at each of ``scopes``.""" 

104 

105 def decorator(fn: OperationHandler) -> OperationHandler: 

106 for scope in scopes: 

107 _REGISTRY[(name, scope)] = fn 

108 return fn 

109 

110 return decorator 

111 

112 

113# ── Plugin-contributed operations (Cortex) ─────────────────────────────────── 

114# 

115# The registry above was always dynamic; what was missing was a way in from 

116# outside the core. These two functions are that way in — the plugin host calls 

117# them, nothing else does. 

118 

119 

120def _plugin_scope(scope: str) -> OpScope: 

121 return OpScope(scope) 

122 

123 

124def register_plugin_operation(registered: Any) -> None: 

125 """Expose a registered Cortex operation at its URL and in ``/metadata``. 

126 

127 ``registered`` is a ``lumen.cortex.registry.RegisteredOperation``; it is 

128 typed loosely here so the API layer keeps no import of the plugin host. 

129 """ 

130 import asyncio 

131 

132 from ..cortex.context import build_context 

133 from ..cortex.operation import OperationInvocation 

134 

135 spec = registered.spec 

136 scope = _plugin_scope(spec.scope) 

137 key = (spec.name, scope) 

138 if key in _REGISTRY: 

139 from ..cortex.contract import PluginRegistrationError 

140 

141 raise PluginRegistrationError( 

142 f"${spec.name} is already provided at {scope.value} level; " 

143 f"plugin {registered.plugin_name!r} cannot take it over" 

144 ) 

145 

146 async def handler(ctx: OperationContext) -> Response: 

147 from .deps import resolve_tenant_settings 

148 

149 if spec.resource_type and ctx.resource_type != spec.resource_type: 

150 raise OperationOutcomeError( 

151 404, 

152 "not-supported", 

153 f"Operation ${spec.name} is not supported on {ctx.resource_type}", 

154 ) 

155 

156 tenant_settings = await resolve_tenant_settings(ctx.tenant_id) 

157 plugin_ctx = build_context( 

158 ctx.tenant_id, tenant_settings, registered.plugin_name, registered.plugin_version 

159 ) 

160 invocation = OperationInvocation( 

161 name=spec.name, 

162 scope=scope.value, 

163 resource_type=ctx.resource_type, 

164 resource_id=ctx.resource_id, 

165 tenant_id=ctx.tenant_id, 

166 body=ctx.body, 

167 query={k: ctx.request.query_params.getlist(k) for k in ctx.request.query_params}, 

168 ) 

169 

170 try: 

171 result = await asyncio.wait_for( 

172 registered.callback(plugin_ctx, invocation), 

173 timeout=spec.timeout_ms / 1000, 

174 ) 

175 except TimeoutError as exc: 

176 raise OperationOutcomeError( 

177 504, 

178 "timeout", 

179 f"Operation ${spec.name} exceeded its {spec.timeout_ms} ms budget", 

180 ) from exc 

181 

182 if result is None: 

183 return _operation_outcome( 

184 [{"severity": "information", "code": "informational", "diagnostics": "OK"}] 

185 ) 

186 return Response(content=json.dumps(result), media_type=FHIR_JSON) 

187 

188 _REGISTRY[key] = handler 

189 # definition is 1..1 in the CapabilityStatement, so a plugin that declares 

190 # no canonical gets a stable synthetic one rather than an empty string. 

191 definition = spec.definition or f"https://lumen.health/fhir/OperationDefinition/{spec.name}" 

192 entry = {"name": spec.name, "definition": definition} 

193 _PLUGIN_OPERATIONS.setdefault(spec.resource_type or "*", []).append(entry) 

194 

195 

196def unregister_plugin_operation(registered: Any) -> None: 

197 """Remove a plugin operation from the URL space and from ``/metadata``.""" 

198 spec = registered.spec 

199 _REGISTRY.pop((spec.name, _plugin_scope(spec.scope)), None) 

200 key = spec.resource_type or "*" 

201 entries = _PLUGIN_OPERATIONS.get(key, []) 

202 _PLUGIN_OPERATIONS[key] = [e for e in entries if e["name"] != spec.name] 

203 if not _PLUGIN_OPERATIONS[key]: 

204 del _PLUGIN_OPERATIONS[key] 

205 

206 

207def _operation_outcome(issues: list[dict[str, Any]], status_code: int = 200) -> Response: 

208 body = {"resourceType": "OperationOutcome", "issue": issues} 

209 return Response(content=json.dumps(body), media_type=FHIR_JSON, status_code=status_code) 

210 

211 

212# ── Dispatch ───────────────────────────────────────────────────────────────── 

213 

214 

215async def _dispatch(ctx: OperationContext, name: str) -> Response: 

216 if ctx.resource_type is not None and ctx.resource_type not in known_resource_types(): 

217 raise OperationOutcomeError(404, "not-found", f"Unknown resource type: {ctx.resource_type}") 

218 handler = _REGISTRY.get((name, ctx.scope)) 

219 if handler is None: 

220 raise OperationOutcomeError( 

221 404, 

222 "not-supported", 

223 f"Operation ${name} is not supported at the {ctx.scope.value} level", 

224 ) 

225 return await handler(ctx) 

226 

227 

228async def _system_op(operation: str, request: Request, tenant: TenantDep) -> Response: 

229 body = await request.body() 

230 ctx = OperationContext(request, OpScope.SYSTEM, None, None, body, tenant) 

231 return await _dispatch(ctx, operation) 

232 

233 

234@router.get("/${operation}", operation_id="system_operation_get") 

235async def system_operation_get(operation: str, request: Request, tenant: TenantDep) -> Response: 

236 return await _system_op(operation, request, tenant) 

237 

238 

239@router.post("/${operation}", operation_id="system_operation_post") 

240async def system_operation_post(operation: str, request: Request, tenant: TenantDep) -> Response: 

241 return await _system_op(operation, request, tenant) 

242 

243 

244async def _type_op( 

245 resource_type: str, operation: str, request: Request, tenant: TenantDep 

246) -> Response: 

247 body = await request.body() 

248 ctx = OperationContext(request, OpScope.TYPE, resource_type, None, body, tenant) 

249 return await _dispatch(ctx, operation) 

250 

251 

252@router.get("/{resource_type}/${operation}", operation_id="type_operation_get") 

253async def type_operation_get( 

254 resource_type: str, operation: str, request: Request, tenant: TenantDep 

255) -> Response: 

256 return await _type_op(resource_type, operation, request, tenant) 

257 

258 

259@router.post("/{resource_type}/${operation}", operation_id="type_operation_post") 

260async def type_operation_post( 

261 resource_type: str, operation: str, request: Request, tenant: TenantDep 

262) -> Response: 

263 return await _type_op(resource_type, operation, request, tenant) 

264 

265 

266async def _instance_op( 

267 resource_type: str, resource_id: str, operation: str, request: Request, tenant: TenantDep 

268) -> Response: 

269 body = await request.body() 

270 ctx = OperationContext(request, OpScope.INSTANCE, resource_type, resource_id, body, tenant) 

271 return await _dispatch(ctx, operation) 

272 

273 

274@router.get("/{resource_type}/{resource_id}/${operation}", operation_id="instance_operation_get") 

275async def instance_operation_get( 

276 resource_type: str, resource_id: str, operation: str, request: Request, tenant: TenantDep 

277) -> Response: 

278 return await _instance_op(resource_type, resource_id, operation, request, tenant) 

279 

280 

281@router.post("/{resource_type}/{resource_id}/${operation}", operation_id="instance_operation_post") 

282async def instance_operation_post( 

283 resource_type: str, resource_id: str, operation: str, request: Request, tenant: TenantDep 

284) -> Response: 

285 return await _instance_op(resource_type, resource_id, operation, request, tenant) 

286 

287 

288# ── $validate ──────────────────────────────────────────────────────────────── 

289 

290 

291def _extract_resource(body: bytes) -> dict[str, Any]: 

292 """Return the resource to validate. 

293 

294 Accepts either a bare resource or a ``Parameters`` wrapper with a parameter 

295 named ``resource``. 

296 """ 

297 try: 

298 parsed: Any = json.loads(body) 

299 except json.JSONDecodeError as exc: 

300 raise OperationOutcomeError(400, "invalid", f"Invalid JSON: {exc}") from exc 

301 

302 if not isinstance(parsed, dict): 

303 raise OperationOutcomeError(400, "invalid", "Body must be a JSON object") 

304 resource = cast("dict[str, Any]", parsed) 

305 

306 if resource.get("resourceType") == "Parameters": 

307 for raw in cast("list[Any]", resource.get("parameter", [])): 

308 if not isinstance(raw, dict): 

309 continue 

310 param = cast("dict[str, Any]", raw) 

311 if param.get("name") == "resource" and isinstance(param.get("resource"), dict): 

312 return cast("dict[str, Any]", param["resource"]) 

313 raise OperationOutcomeError( 

314 400, "invalid", "Parameters body must contain a 'resource' parameter" 

315 ) 

316 return resource 

317 

318 

319@operation("validate", OpScope.TYPE, OpScope.INSTANCE) 

320async def op_validate(ctx: OperationContext) -> Response: 

321 resource = _extract_resource(ctx.body) 

322 issues: list[dict[str, Any]] = [] 

323 

324 rt = resource.get("resourceType") 

325 if ctx.resource_type is not None and rt != ctx.resource_type: 

326 issues.append( 

327 { 

328 "severity": "error", 

329 "code": "invalid", 

330 "diagnostics": ( 

331 f"resourceType in body ({rt!r}) does not match URL ({ctx.resource_type!r})" 

332 ), 

333 } 

334 ) 

335 

336 for err in validate_resource(resource): 

337 issues.append({"severity": "error", "code": "structure", "diagnostics": err}) 

338 for err in validate_code_bindings(resource): 

339 issues.append({"severity": "error", "code": "code-invalid", "diagnostics": err}) 

340 for violation in validate_invariants(resource): 

341 issues.append( 

342 { 

343 "severity": violation.severity, 

344 "code": "invariant", 

345 "diagnostics": f"{violation.key}: {violation.human}", 

346 } 

347 ) 

348 

349 if not issues: 

350 issues.append( 

351 { 

352 "severity": "information", 

353 "code": "informational", 

354 "diagnostics": "Validation successful", 

355 } 

356 ) 

357 # $validate always returns 200; the OperationOutcome carries the verdict. 

358 return _operation_outcome(issues) 

359 

360 

361# ── $everything (Patient / Encounter) ───────────────────────────────────────── 

362 

363# R4B defines $everything on these compartment resource types. 

364_EVERYTHING_TYPES = ("Patient", "Encounter") 

365 

366 

367async def _fetch_compartment_entries( 

368 repo: ResourceRepository, 

369 ctx: OperationContext, 

370 base_url: str, 

371 compartment: str, 

372 root: Any, 

373) -> list[dict[str, Any]]: 

374 assert ctx.resource_id is not None 

375 entries: list[dict[str, Any]] = [_entry(base_url, compartment, ctx.resource_id, root)] 

376 members = await repo.compartment_members( 

377 ctx.tenant_id, compartment, ctx.resource_id, compartment_params(compartment) 

378 ) 

379 for member_type, member_id in members: 

380 member = await repo.read(ctx.tenant_id, member_type, member_id) 

381 if member is not None: 

382 entries.append(_entry(base_url, member_type, member_id, member)) 

383 return entries 

384 

385 

386@operation("everything", OpScope.INSTANCE) 

387async def op_everything(ctx: OperationContext) -> Response: 

388 compartment = ctx.resource_type 

389 if compartment not in _EVERYTHING_TYPES: 

390 raise OperationOutcomeError( 

391 400, 

392 "not-supported", 

393 f"$everything is only supported on {' and '.join(_EVERYTHING_TYPES)} instances", 

394 ) 

395 assert ctx.resource_id is not None # instance scope guarantees this 

396 

397 base_url = str(ctx.request.base_url).rstrip("/") 

398 pool: asyncpg.Pool[asyncpg.Record] = get_pool() 

399 async with pool.acquire() as conn: 

400 repo = ResourceRepository(conn, get_registry()) 

401 root = await repo.read(ctx.tenant_id, compartment, ctx.resource_id) 

402 if root is None: 

403 raise OperationOutcomeError( 

404 404, "not-found", f"{compartment}/{ctx.resource_id} not found" 

405 ) 

406 entries = await _fetch_compartment_entries(repo, ctx, base_url, compartment, root) 

407 

408 bundle = { 

409 "resourceType": "Bundle", 

410 "type": "searchset", 

411 "total": len(entries), 

412 "entry": entries, 

413 } 

414 return Response(content=json.dumps(bundle), media_type=FHIR_JSON) 

415 

416 

417def _entry(base_url: str, resource_type: str, resource_id: str, read_result: Any) -> dict[str, Any]: 

418 body = _inject_meta( 

419 read_result["resource"], read_result["version_id"], read_result["last_updated"] 

420 ) 

421 return { 

422 "fullUrl": f"{base_url}/{resource_type}/{resource_id}", 

423 "resource": body, 

424 "search": {"mode": "match"}, 

425 } 

426 

427 

428# ── $meta / $meta-add / $meta-delete ─────────────────────────────────────────── 

429 

430# Meta elements these operations add to / delete from. tag/security are lists of 

431# Coding; profile is a list of canonical URLs (strings). 

432_META_LIST_FIELDS = ("profile", "security", "tag") 

433 

434 

435def _meta_return(meta: dict[str, Any]) -> Response: 

436 body = { 

437 "resourceType": "Parameters", 

438 "parameter": [{"name": "return", "valueMeta": meta}], 

439 } 

440 return Response(content=json.dumps(body), media_type=FHIR_JSON) 

441 

442 

443def _meta_arg(ctx: OperationContext) -> dict[str, Any]: 

444 """Extract the ``meta`` parameter (a Meta) from a Parameters body.""" 

445 try: 

446 parsed: Any = json.loads(ctx.body) 

447 except json.JSONDecodeError as exc: 

448 raise OperationOutcomeError(400, "invalid", f"Invalid JSON: {exc}") from exc 

449 if isinstance(parsed, dict): 

450 for raw in cast("list[Any]", cast("dict[str, Any]", parsed).get("parameter", [])): 

451 if not isinstance(raw, dict): 

452 continue 

453 param = cast("dict[str, Any]", raw) 

454 if param.get("name") == "meta" and isinstance(param.get("valueMeta"), dict): 

455 return cast("dict[str, Any]", param["valueMeta"]) 

456 raise OperationOutcomeError(400, "invalid", "Body must contain a 'meta' parameter") 

457 

458 

459async def _read_or_404(repo: ResourceRepository, ctx: OperationContext) -> Any: 

460 assert ctx.resource_type is not None 

461 assert ctx.resource_id is not None 

462 result = await repo.read(ctx.tenant_id, ctx.resource_type, ctx.resource_id) 

463 if result is None: 

464 raise OperationOutcomeError( 

465 404, "not-found", f"{ctx.resource_type}/{ctx.resource_id} not found" 

466 ) 

467 return result 

468 

469 

470@operation("meta", OpScope.INSTANCE) 

471async def op_meta(ctx: OperationContext) -> Response: 

472 pool: asyncpg.Pool[asyncpg.Record] = get_pool() 

473 async with pool.acquire() as conn: 

474 result = await _read_or_404(ResourceRepository(conn, get_registry()), ctx) 

475 body = _inject_meta(result["resource"], result["version_id"], result["last_updated"]) 

476 return _meta_return(body.get("meta", {})) 

477 

478 

479@operation("meta-add", OpScope.INSTANCE) 

480async def op_meta_add(ctx: OperationContext) -> Response: 

481 delta = _meta_arg(ctx) 

482 return await _apply_meta_change(ctx, delta, add=True) 

483 

484 

485@operation("meta-delete", OpScope.INSTANCE) 

486async def op_meta_delete(ctx: OperationContext) -> Response: 

487 delta = _meta_arg(ctx) 

488 return await _apply_meta_change(ctx, delta, add=False) 

489 

490 

491async def _apply_meta_change( 

492 ctx: OperationContext, delta: dict[str, Any], *, add: bool 

493) -> Response: 

494 assert ctx.resource_type is not None 

495 assert ctx.resource_id is not None 

496 pool: asyncpg.Pool[asyncpg.Record] = get_pool() 

497 async with pool.acquire() as conn: 

498 repo = ResourceRepository(conn, get_registry()) 

499 result = await _read_or_404(repo, ctx) 

500 resource = result["resource"] 

501 meta = dict(cast("dict[str, Any]", resource.get("meta") or {})) 

502 

503 for field in _META_LIST_FIELDS: 

504 requested = cast("list[Any]", delta.get(field) or []) 

505 if not requested: 

506 continue 

507 current = cast("list[Any]", meta.get(field) or []) 

508 if add: 

509 merged = list(current) 

510 merged.extend(item for item in requested if item not in merged) 

511 meta[field] = merged 

512 else: 

513 remaining = [item for item in current if item not in requested] 

514 if remaining: 

515 meta[field] = remaining 

516 else: 

517 meta.pop(field, None) 

518 

519 resource = {**resource, "meta": meta} 

520 written = await repo.update(ctx.tenant_id, ctx.resource_type, ctx.resource_id, resource) 

521 

522 body = _inject_meta(written["resource"], written["version_id"], written["last_updated"]) 

523 return _meta_return(body.get("meta", {})) 

524 

525 

526# ── $book (ISiK Terminplanung) ──────────────────────────────────────────────── 

527 

528 

529def _extract_appointment(ctx: OperationContext) -> tuple[dict[str, Any], str | None]: 

530 """Extract the Appointment (and optional schedule reference) from the $book body. 

531 

532 Accepts a bare Appointment or a Parameters wrapper with an ``appt-resource`` 

533 parameter; a Parameters wrapper may also carry a ``schedule`` valueReference, 

534 used to resolve a free slot when the Appointment itself names none. 

535 """ 

536 try: 

537 parsed: Any = json.loads(ctx.body) if ctx.body else {} 

538 except json.JSONDecodeError as exc: 

539 raise OperationOutcomeError(400, "invalid", f"Invalid JSON: {exc}") from exc 

540 

541 if not isinstance(parsed, dict): 

542 raise OperationOutcomeError(400, "invalid", "Body must be a JSON object") 

543 

544 resource = cast("dict[str, Any]", parsed) 

545 if resource.get("resourceType") == "Parameters": 

546 appointment: dict[str, Any] | None = None 

547 schedule_ref: str | None = None 

548 for raw in cast("list[Any]", resource.get("parameter", [])): 

549 if not isinstance(raw, dict): 

550 continue 

551 param = cast("dict[str, Any]", raw) 

552 if param.get("name") == "appt-resource" and isinstance(param.get("resource"), dict): 

553 appointment = cast("dict[str, Any]", param["resource"]) 

554 elif param.get("name") == "schedule" and isinstance(param.get("valueReference"), dict): 

555 ref = cast("dict[str, Any]", param["valueReference"]).get("reference") 

556 if isinstance(ref, str): 

557 schedule_ref = ref 

558 if appointment is None: 

559 raise OperationOutcomeError( 

560 400, "invalid", "Parameters body must contain an 'appt-resource' parameter" 

561 ) 

562 return appointment, schedule_ref 

563 if resource.get("resourceType") == "Appointment": 

564 return resource, None 

565 raise OperationOutcomeError( 

566 400, 

567 "invalid", 

568 "$book requires an Appointment resource (bare or as 'appt-resource' parameter)", 

569 ) 

570 

571 

572async def _find_free_slot( 

573 conn: Any, 

574 tenant_id: UUID, 

575 schedule_ref: str, 

576 start: str | None, 

577 end: str | None, 

578) -> str | None: 

579 """Return a free Slot (``Slot/{id}``) on the schedule covering [start, end], or None.""" 

580 from ..search.parser import parse_query 

581 from ..search.query import execute_search 

582 

583 schedule_id = schedule_ref.split("/")[-1] 

584 pq = parse_query({"schedule": [schedule_id], "status": ["free"]}) 

585 params = await get_registry().get_params(conn, tenant_id, "Slot") 

586 result = await execute_search(conn, tenant_id, "Slot", pq, params) 

587 for slot in result.resources: 

588 s_start = cast(str | None, slot.get("start")) 

589 s_end = cast(str | None, slot.get("end")) 

590 # A slot fits when it starts no later and ends no earlier than the request 

591 # window; missing request bounds impose no constraint on that side. 

592 if start is not None and s_start is not None and s_start > start: 

593 continue 

594 if end is not None and s_end is not None and s_end < end: 

595 continue 

596 slot_id = slot.get("id") 

597 if isinstance(slot_id, str): 

598 return f"Slot/{slot_id}" 

599 return None 

600 

601 

602async def _ensure_slots_free(conn: Any, tenant_id: UUID, appointment: dict[str, Any]) -> None: 

603 """Reject booking (409) if any referenced Slot is not free. 

604 

605 Part of the ISiK $book contract: an Appointment may only book slots that are 

606 actually available. Missing/unresolvable slots are left to resource 

607 validation (a booked Appointment with no slot fails its own invariants). 

608 """ 

609 repo = ResourceRepository(conn, get_registry()) 

610 for raw in cast("list[Any]", appointment.get("slot") or []): 

611 if not isinstance(raw, dict): 

612 continue 

613 ref = cast("dict[str, Any]", raw).get("reference") 

614 if not isinstance(ref, str) or "/" not in ref: 

615 continue 

616 slot_id = ref.split("/")[-1] 

617 record = await repo.read(tenant_id, "Slot", slot_id) 

618 if record is None: 

619 continue 

620 status = record["resource"].get("status") 

621 if status != "free": 

622 raise OperationOutcomeError( 

623 409, 

624 "conflict", 

625 f"Slot/{slot_id} is not free (status={status!r}); it cannot be booked.", 

626 ) 

627 

628 

629@operation("book", OpScope.TYPE) 

630async def op_book(ctx: OperationContext) -> Response: 

631 """ISiK $book — create or update an Appointment with status='booked'.""" 

632 from ..db.repository import WriteResult 

633 

634 appointment, schedule_ref = _extract_appointment(ctx) 

635 

636 pool: asyncpg.Pool[asyncpg.Record] = get_pool() 

637 async with pool.acquire() as conn: 

638 # Schedule-driven booking: no slot named, but a schedule is given — 

639 # resolve a free slot on that schedule covering the requested window. 

640 if not appointment.get("slot") and schedule_ref: 

641 slot_ref = await _find_free_slot( 

642 conn, 

643 ctx.tenant_id, 

644 schedule_ref, 

645 cast(str | None, appointment.get("start")), 

646 cast(str | None, appointment.get("end")), 

647 ) 

648 if slot_ref: 

649 appointment["slot"] = [{"reference": slot_ref}] 

650 

651 # Reject booking a slot that is not free (ISiK $book contract). 

652 await _ensure_slots_free(conn, ctx.tenant_id, appointment) 

653 

654 appointment["status"] = "booked" 

655 

656 # Validate the booked Appointment (structure + FHIR invariants such as 

657 # app-3: a booked appointment must carry start and end). An incomplete 

658 # request fails here with 422, matching the normal write path. 

659 issues: list[str] = list(validate_resource(appointment)) 

660 issues.extend(v.human for v in validate_invariants(appointment) if v.severity == "error") 

661 if issues: 

662 raise OperationOutcomeError(422, "invalid", "; ".join(issues)) 

663 

664 repo = ResourceRepository(conn, get_registry()) 

665 appt_id = appointment.get("id") 

666 written: WriteResult 

667 if appt_id: 

668 written = await repo.update(ctx.tenant_id, "Appointment", str(appt_id), appointment) 

669 else: 

670 resource_id = _new_id() 

671 appointment["id"] = resource_id 

672 written = await repo.create(ctx.tenant_id, "Appointment", resource_id, appointment) 

673 

674 body = _inject_meta(written["resource"], written["version_id"], written["last_updated"]) 

675 return Response(content=json.dumps(body), media_type=FHIR_JSON) 

676 

677 

678# ── $update-metadata (ISiK Dokumentenaustausch) ─────────────────────────────── 

679 

680# Fields transferred from the input DocumentReference; document content is preserved. 

681_DOCREF_METADATA_FIELDS = ( 

682 "status", 

683 "docStatus", 

684 "type", 

685 "category", 

686 "subject", 

687 "date", 

688 "author", 

689 "authenticator", 

690 "custodian", 

691 "relatesTo", 

692 "description", 

693 "securityLabel", 

694 "context", 

695 "extension", 

696 "meta", 

697) 

698 

699 

700@operation("update-metadata", OpScope.INSTANCE) 

701async def op_update_metadata(ctx: OperationContext) -> Response: 

702 """ISiK $update-metadata — update DocumentReference metadata without replacing document content.""" # noqa: E501 

703 if ctx.resource_type != "DocumentReference": 

704 raise OperationOutcomeError( 

705 400, 

706 "not-supported", 

707 "$update-metadata is only supported on DocumentReference instances", 

708 ) 

709 assert ctx.resource_id is not None 

710 

711 try: 

712 parsed: Any = json.loads(ctx.body) if ctx.body else {} 

713 except json.JSONDecodeError as exc: 

714 raise OperationOutcomeError(400, "invalid", f"Invalid JSON: {exc}") from exc 

715 

716 doc_ref: dict[str, Any] | None = None 

717 if isinstance(parsed, dict): 

718 body_dict = cast("dict[str, Any]", parsed) 

719 if body_dict.get("resourceType") == "Parameters": 

720 for raw in cast("list[Any]", body_dict.get("parameter", [])):