Coverage for src/lumen/api/bundle.py: 69%

1""" 

2This module processes FHIR Bundles of type batch and transaction. 

3It parses individual bundle entries and executes them sequentially or inside a database transaction. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import Any, cast 

9from urllib.parse import parse_qs 

10from uuid import UUID 

11 

12import asyncpg 

13import asyncpg.pool 

14import structlog 

15 

16from ..core.exceptions import OperationOutcomeError 

17from ..core.write_events import collect_write_events 

18from ..cortex.dispatch import dispatch_write_events 

19from ..db.repository import ResourceRepository 

20from ..fhir.binding_validator import validate_code_bindings 

21from ..fhir.profile_validator import validate_profiles 

22from ..fhir.resource_types import known_resource_types 

23from ..fhir.validator import validate_resource 

24from ..search.bundle import build_searchset_bundle 

25from ..search.parser import parse_query 

26from ..search.query import execute_search 

27from ..search.registry import get_registry 

28from ._helpers import etag, inject_meta, last_modified_instant, new_id, parse_if_match 

29from .deps import ResolvedTenantSettings 

30 

31logger = structlog.get_logger(__name__) 

32 

33# ── URL parsing ─────────────────────────────────────────────────────────────── 

34 

35 

36def _parse_entry_url(url: str) -> tuple[str, str | None, str | None]: 

37 """Parse bundle entry request.url → (resource_type, resource_id | None, query | None).""" 

38 path, _, query = url.partition("?") 

39 parts = [p for p in path.split("/") if p] 

40 resource_type = parts[0] if parts else "" 

41 resource_id = parts[1] if len(parts) > 1 else None 

42 return resource_type, resource_id, query or None 

43 

44 

45async def _conditional_search_ids_with_conn( 

46 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

47 resource_type: str, 

48 query_string: str, 

49 tenant_id: UUID, 

50 tenant_settings: ResolvedTenantSettings, 

51) -> list[str]: 

52 """Execute a conditional search query inside the bundle transaction/connection.""" 

53 raw = parse_qs(query_string, keep_blank_values=True) 

54 multi: dict[str, list[str]] = {k: v for k, v in raw.items()} 

55 pq = parse_query(multi) 

56 registry = get_registry() 

57 ignore_unknown = tenant_settings.ignore_unknown_search_parameters 

58 params = await registry.get_params(conn, tenant_id, resource_type) 

59 result = await execute_search(conn, tenant_id, resource_type, pq, params, ignore_unknown) 

60 matching_ids = [r["id"] for r in result.resources if "id" in r] 

61 return matching_ids 

62 

63 

64# ── urn:uuid resolution ─────────────────────────────────────────────────────── 

65 

66 

67async def _resolve_entry_urn_put_id( 

68 resource_type: str, 

69 query: str, 

70 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

71 tenant_id: UUID, 

72 tenant_settings: ResolvedTenantSettings, 

73) -> str | None: 

74 matching_ids = await _conditional_search_ids_with_conn( 

75 conn, resource_type, query, tenant_id, tenant_settings 

76 ) 

77 if len(matching_ids) == 1: 

78 return matching_ids[0] 

79 if len(matching_ids) > 1: 

80 raise OperationOutcomeError( 

81 412, 

82 "multiple-matches", 

83 f"Conditional update matched {len(matching_ids)} resources; must match exactly 0 or 1", 

84 ) 

85 return None 

86 

87 

88async def _resolve_entry_urn_post_id( 

89 resource_type: str, 

90 req: dict[str, Any], 

91 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

92 tenant_id: UUID, 

93 tenant_settings: ResolvedTenantSettings, 

94) -> str | None: 

95 if_none_exist = req.get("ifNoneExist") 

96 if if_none_exist: 

97 matching_ids = await _conditional_search_ids_with_conn( 

98 conn, resource_type, str(if_none_exist), tenant_id, tenant_settings 

99 ) 

100 if len(matching_ids) == 1: 

101 return matching_ids[0] 

102 if len(matching_ids) > 1: 

103 raise OperationOutcomeError( 

104 412, 

105 "duplicate", 

106 f"If-None-Exist matched {len(matching_ids)} resources; expected 0 or 1", 

107 ) 

108 return None 

109 

110 

111async def _resolve_entry_urn_id( 

112 entry: dict[str, Any], 

113 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

114 tenant_id: UUID, 

115 tenant_settings: ResolvedTenantSettings, 

116) -> tuple[str, str, str] | None: 

117 full_url = cast(str, entry.get("fullUrl") or "") 

118 if not full_url.startswith("urn:uuid:"): 

119 return None 

120 resource_type = cast( 

121 str, cast(dict[str, Any], entry.get("resource") or {}).get("resourceType") or "" 

122 ) 

123 if not resource_type: 

124 return None 

125 

126 req = cast(dict[str, Any], entry.get("request") or {}) 

127 method = cast(str, req.get("method") or "").upper() 

128 url = cast(str, req.get("url") or "") 

129 _, direct_id, query = _parse_entry_url(url) 

130 

131 resolved_id = None 

132 if method == "PUT" and direct_id and not query: 

133 # Direct PUT by id (e.g. url="Patient/some-id") — the id is already 

134 # known, no DB lookup needed. Distinct from the conditional-PUT case 

135 # below (url="Patient?identifier=..."), which must resolve via search. 

136 resolved_id = direct_id 

137 elif method == "PUT" and query: 

138 resolved_id = await _resolve_entry_urn_put_id( 

139 resource_type, query, conn, tenant_id, tenant_settings 

140 ) 

141 elif method == "POST": 

142 resolved_id = await _resolve_entry_urn_post_id( 

143 resource_type, req, conn, tenant_id, tenant_settings 

144 ) 

145 

146 if not resolved_id: 

147 resolved_id = new_id() 

148 

149 return full_url, resource_type, resolved_id 

150 

151 

152async def _resolve_by_identifier( 

153 resource_type: str, 

154 resource: dict[str, Any], 

155 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

156 tenant_id: UUID, 

157 tenant_settings: ResolvedTenantSettings, 

158) -> str | None: 

159 identifiers: Any = resource.get("identifier") 

160 if not identifiers: 

161 raise OperationOutcomeError( 

162 422, 

163 "value", 

164 f"Document bundle entry {resource_type} has no identifier and cannot be resolved.", 

165 ) 

166 if not isinstance(identifiers, list): 

167 identifiers = [identifiers] 

168 identifiers_list = cast(list[Any], identifiers) 

169 

170 tokens: list[str] = [] 

171 for id_obj in identifiers_list: 

172 if not isinstance(id_obj, dict): 

173 continue 

174 id_dict = cast(dict[str, Any], id_obj) 

175 val = cast(str | None, id_dict.get("value")) 

176 sys = cast(str | None, id_dict.get("system")) 

177 if val: 

178 if sys: 

179 tokens.append(f"{sys}|{val}") 

180 else: 

181 tokens.append(val) 

182 

183 if not tokens: 

184 raise OperationOutcomeError( 

185 422, 

186 "value", 

187 f"Document bundle entry {resource_type} has no valid identifier " 

188 "value and cannot be resolved.", 

189 ) 

190 

191 query_string = f"identifier={','.join(tokens)}" 

192 matching_ids = await _conditional_search_ids_with_conn( 

193 conn, resource_type, query_string, tenant_id, tenant_settings 

194 ) 

195 

196 if not matching_ids: 

197 return None 

198 if len(matching_ids) > 1: 

199 logger.warning( 

200 "bundle.resolve_identifier.multiple_matches", 

201 resource_type=resource_type, 

202 matching_ids=matching_ids, 

203 ) 

204 status = resource.get("status") 

205 if status: 

206 repo = ResourceRepository(conn, get_registry()) 

207 filtered_ids: list[str] = [] 

208 for mid in matching_ids: 

209 try: 

210 existing = await repo.read(tenant_id, resource_type, mid) 

211 if existing and existing["resource"].get("status") == status: 

212 filtered_ids.append(mid) 

213 except Exception: 

214 pass 

215 if filtered_ids: 

216 matching_ids = filtered_ids 

217 

218 return matching_ids[0] 

219 

220 

221async def _build_urn_map( 

222 entries: list[dict[str, Any]], 

223 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

224 tenant_id: UUID, 

225 tenant_settings: ResolvedTenantSettings, 

226 is_document: bool = False, 

227) -> tuple[dict[str, str], dict[str, str], set[str]]: 

228 """Pre-assign real IDs for entries with urn:uuid fullUrls. 

229 Resolves conditional creates/updates first to ensure correct reference mapping. 

230 For document bundles, Patient and Encounter entries are resolved by identifier. 

231 """ 

232 ref_map: dict[str, str] = {} 

233 id_map: dict[str, str] = {} 

234 resolved_pre_existing: set[str] = set() 

235 

236 for entry in entries: 

237 full_url = cast(str, entry.get("fullUrl") or "") 

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

239 resource_type = cast(str, resource.get("resourceType") or "") 

240 

241 if is_document and resource_type in ("Patient", "Encounter"): 

242 matched_id = await _resolve_by_identifier( 

243 resource_type, resource, conn, tenant_id, tenant_settings 

244 ) 

245 if not matched_id: 

246 raise OperationOutcomeError( 

247 422, 

248 "processing", 

249 f"Document bundle entry {resource_type} with identifier " 

250 "could not be resolved to an existing resource.", 

251 ) 

252 ref_map[full_url] = f"{resource_type}/{matched_id}" 

253 id_map[full_url] = matched_id 

254 resolved_pre_existing.add(full_url) 

255 else: 

256 resolved = await _resolve_entry_urn_id(entry, conn, tenant_id, tenant_settings) 

257 if resolved: 

258 full_url, resource_type, resolved_id = resolved 

259 ref_map[full_url] = f"{resource_type}/{resolved_id}" 

260 id_map[full_url] = resolved_id 

261 

262 return ref_map, id_map, resolved_pre_existing 

263 

264 

265def _rewrite_refs(obj: Any, ref_map: dict[str, str]) -> Any: 

266 """Recursively replace urn:uuid references in a resource.""" 

267 if isinstance(obj, dict): 

268 d: dict[str, Any] = cast(dict[str, Any], obj) 

269 return { 

270 k: ( 

271 ref_map[v] 

272 if k == "reference" and isinstance(v, str) and v in ref_map 

273 else _rewrite_refs(v, ref_map) 

274 ) 

275 for k, v in d.items() 

276 } 

277 if isinstance(obj, list): 

278 lst: list[Any] = cast(list[Any], obj) 

279 return [_rewrite_refs(item, ref_map) for item in lst] 

280 return obj 

281 

282 

283# ── Validation ──────────────────────────────────────────────────────────────── 

284 

285 

286async def _validate_dict( 

287 resource: dict[str, Any], 

288 expected_type: str, 

289 tenant_id: UUID, 

290 tenant_settings: ResolvedTenantSettings, 

291 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

292) -> None: 

293 rt = resource.get("resourceType") 

294 if rt != expected_type: 

295 raise OperationOutcomeError( 

296 400, 

297 "invalid", 

298 f"resourceType in body ({rt!r}) does not match URL ({expected_type!r})", 

299 ) 

300 errors = validate_resource(resource) 

301 if errors: 

302 raise OperationOutcomeError(422, "invalid", f"Resource validation failed: {errors[0]}") 

303 binding_errors = validate_code_bindings(resource) 

304 if binding_errors: 

305 raise OperationOutcomeError(422, "code-invalid", binding_errors[0]) 

306 

307 # Bundle-processed resources previously skipped profile validation 

308 # entirely (unlike the single-resource POST/PUT/PATCH routes in 

309 # resources.py) — an ISiK-invalid Composition/Patient/etc. written via a 

310 # batch/transaction/document Bundle would silently persist. 

311 profile_errors = await validate_profiles( 

312 resource, 

313 tenant_id, 

314 conn, 

315 tenant_settings.profile_validation_mode, 

316 tenant_settings.enforced_profiles, 

317 ) 

318 if profile_errors: 

319 raise OperationOutcomeError( 

320 422, 

321 "invalid", 

322 f"Profile validation failed: {profile_errors[0]}", 

323 ) 

324 

325 

326# ── Entry dispatcher ────────────────────────────────────────────────────────── 

327 

328 

329async def _dispatch( 

330 entry: dict[str, Any], 

331 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

332 tenant_id: UUID, 

333 tenant_settings: ResolvedTenantSettings, 

334 base_url: str, 

335 id_map: dict[str, str], 

336 ref_map: dict[str, str], 

337 transaction_refs: set[str] | None = None, 

338) -> dict[str, Any]: 

339 """Process one bundle entry. Returns a response entry dict. Raises on failure.""" 

340 req: dict[str, Any] = cast(dict[str, Any], entry.get("request") or {}) 

341 method: str = cast(str, req.get("method") or "").upper() 

342 url: str = cast(str, req.get("url") or "") 

343 resource_type, resource_id, query = _parse_entry_url(url) 

344 repo = ResourceRepository(conn, get_registry()) 

345 

346 if method == "GET": 

347 return await _dispatch_get( 

348 resource_type, resource_id, query, repo, tenant_id, tenant_settings, base_url, url, conn 

349 ) 

350 if method == "POST": 

351 return await _dispatch_post( 

352 entry, 

353 req, 

354 resource_type, 

355 repo, 

356 tenant_id, 

357 tenant_settings, 

358 base_url, 

359 id_map, 

360 ref_map, 

361 conn, 

362 transaction_refs, 

363 ) 

364 if method == "PUT": 

365 return await _dispatch_put( 

366 entry, 

367 resource_type, 

368 resource_id, 

369 query, 

370 req, 

371 repo, 

372 tenant_id, 

373 tenant_settings, 

374 base_url, 

375 id_map, 

376 ref_map, 

377 conn, 

378 transaction_refs, 

379 ) 

380 if method == "DELETE": 

381 return await _dispatch_delete( 

382 resource_type, resource_id, query, repo, tenant_id, tenant_settings, conn 

383 ) 

384 raise OperationOutcomeError(400, "invalid", f"Unsupported method in bundle entry: {method!r}") 

385 

386 

387async def _dispatch_get( 

388 resource_type: str, 

389 resource_id: str | None, 

390 query: str | None, 

391 repo: ResourceRepository, 

392 tenant_id: UUID, 

393 tenant_settings: ResolvedTenantSettings, 

394 base_url: str, 

395 raw_url: str, 

396 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

397) -> dict[str, Any]: 

398 if resource_id: 

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

400 if result is None: 

401 if await repo.is_deleted(tenant_id, resource_type, resource_id): 

402 raise OperationOutcomeError( 

403 410, "gone", f"{resource_type}/{resource_id} has been deleted" 

404 ) 

405 raise OperationOutcomeError( 

406 404, "not-found", f"{resource_type}/{resource_id} not found" 

407 ) 

408 body = inject_meta(result["resource"], result["version_id"], result["last_updated"]) 

409 return { 

410 "resource": body, 

411 "response": { 

412 "status": "200 OK", 

413 "etag": etag(result["version_id"]), 

414 "lastModified": last_modified_instant(result["last_updated"]), 

415 }, 

416 } 

417 

418 if resource_type not in known_resource_types(): 

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

420 raw: dict[str, list[str]] = {} 

421 if query: 

422 for k, vs in parse_qs(query, keep_blank_values=True).items(): 

423 raw[k] = vs 

424 pq = parse_query(raw) 

425 registry = get_registry() 

426 params = await registry.get_params(conn, tenant_id, resource_type) 

427 ignore_unknown = tenant_settings.ignore_unknown_search_parameters 

428 search_result = await execute_search(conn, tenant_id, resource_type, pq, params, ignore_unknown) 

429 bundle_body = build_searchset_bundle( 

430 base_url, resource_type, search_result, pq, f"{base_url}/{raw_url}" 

431 ) 

432 return {"resource": bundle_body, "response": {"status": "200 OK"}} 

433 

434 

435async def _dispatch_post( 

436 entry: dict[str, Any], 

437 req: dict[str, Any], 

438 resource_type: str, 

439 repo: ResourceRepository, 

440 tenant_id: UUID, 

441 tenant_settings: ResolvedTenantSettings, 

442 base_url: str, 

443 id_map: dict[str, str], 

444 ref_map: dict[str, str], 

445 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

446 transaction_refs: set[str] | None = None, 

447) -> dict[str, Any]: 

448 if resource_type not in known_resource_types(): 

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

450 raw_resource: dict[str, Any] = dict(cast(dict[str, Any], entry.get("resource") or {})) 

451 if ref_map: 

452 raw_resource = cast(dict[str, Any], _rewrite_refs(raw_resource, ref_map)) 

453 await _validate_dict(raw_resource, resource_type, tenant_id, tenant_settings, conn) 

454 

455 if_none_exist = req.get("ifNoneExist") 

456 if if_none_exist: 

457 matching_ids = await _conditional_search_ids_with_conn( 

458 conn, resource_type, str(if_none_exist), tenant_id, tenant_settings 

459 ) 

460 if len(matching_ids) == 1: 

461 existing = await repo.read(tenant_id, resource_type, matching_ids[0]) 

462 if existing: 

463 body = inject_meta( 

464 existing["resource"], existing["version_id"], existing["last_updated"] 

465 ) 

466 return { 

467 "resource": body, 

468 "response": { 

469 "status": "200 OK", 

470 "location": f"{base_url}/{resource_type}/{matching_ids[0]}", 

471 "etag": etag(existing["version_id"]), 

472 "lastModified": last_modified_instant(existing["last_updated"]), 

473 }, 

474 } 

475 elif len(matching_ids) > 1: 

476 raise OperationOutcomeError( 

477 412, 

478 "duplicate", 

479 f"If-None-Exist matched {len(matching_ids)} resources; expected 0 or 1", 

480 ) 

481 full_url = cast(str, entry.get("fullUrl") or "") 

482 rid = id_map.get(full_url) or new_id() 

483 resource = {**raw_resource, "id": rid} 

484 write = await repo.create(tenant_id, resource_type, rid, resource, transaction_refs) 

485 body = inject_meta(write["resource"], write["version_id"], write["last_updated"]) 

486 return { 

487 "resource": body, 

488 "response": { 

489 "status": "201 Created", 

490 "location": f"{base_url}/{resource_type}/{rid}", 

491 "etag": etag(write["version_id"]), 

492 "lastModified": last_modified_instant(write["last_updated"]), 

493 }, 

494 } 

495 

496 

497async def _resolve_put_resource_id( 

498 resource_id: str | None, 

499 query: str | None, 

500 resource_type: str, 

501 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

502 tenant_id: UUID, 

503 tenant_settings: ResolvedTenantSettings, 

504 entry: dict[str, Any], 

505 id_map: dict[str, str], 

506 raw_resource: dict[str, Any], 

507) -> str: 

508 if not resource_id and query: 

509 matching_ids = await _conditional_search_ids_with_conn( 

510 conn, resource_type, query, tenant_id, tenant_settings 

511 ) 

512 if len(matching_ids) > 1: 

513 raise OperationOutcomeError( 

514 412, 

515 "multiple-matches", 

516 f"Conditional update matched {len(matching_ids)} resources; " 

517 "must match exactly 0 or 1", 

518 ) 

519 resource_id = matching_ids[0] if matching_ids else None 

520 

521 if not resource_id: 

522 # 0 matches -> create 

523 full_url = cast(str, entry.get("fullUrl") or "") 

524 rid = id_map.get(full_url) or raw_resource.get("id") or new_id() 

525 assert isinstance(rid, str) 

526 resource_id = rid 

527 return resource_id 

528 

529 

530async def _dispatch_put( 

531 entry: dict[str, Any], 

532 resource_type: str, 

533 resource_id: str | None, 

534 query: str | None, 

535 req: dict[str, Any], 

536 repo: ResourceRepository, 

537 tenant_id: UUID, 

538 tenant_settings: ResolvedTenantSettings, 

539 base_url: str, 

540 id_map: dict[str, str], 

541 ref_map: dict[str, str], 

542 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

543 transaction_refs: set[str] | None = None, 

544) -> dict[str, Any]: 

545 if not resource_id and not query: 

546 raise OperationOutcomeError( 

547 400, "invalid", "PUT entry requires resource id in URL or query parameters" 

548 ) 

549 if resource_type not in known_resource_types(): 

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

551 

552 raw_resource: dict[str, Any] = dict(cast(dict[str, Any], entry.get("resource") or {})) 

553 if ref_map: 

554 raw_resource = cast(dict[str, Any], _rewrite_refs(raw_resource, ref_map)) 

555 await _validate_dict(raw_resource, resource_type, tenant_id, tenant_settings, conn) 

556 

557 resource_id = await _resolve_put_resource_id( 

558 resource_id, 

559 query, 

560 resource_type, 

561 conn, 

562 tenant_id, 

563 tenant_settings, 

564 entry, 

565 id_map, 

566 raw_resource, 

567 ) 

568 assert isinstance(resource_id, str) 

569 resource = {**raw_resource, "id": resource_id} 

570 

571 if_match_str: str | None = cast(str | None, req.get("ifMatch")) 

572 if_match = parse_if_match(if_match_str) if if_match_str else None 

573 

574 existing = await repo.read(tenant_id, resource_type, resource_id) 

575 if existing is None: 

576 write = await repo.create(tenant_id, resource_type, resource_id, resource, transaction_refs) 

577 body = inject_meta(write["resource"], write["version_id"], write["last_updated"]) 

578 return { 

579 "resource": body, 

580 "response": { 

581 "status": "201 Created", 

582 "location": f"{base_url}/{resource_type}/{resource_id}", 

583 "etag": etag(write["version_id"]), 

584 }, 

585 } 

586 

587 if if_match is not None and if_match != existing["version_id"]: 

588 raise OperationOutcomeError(412, "conflict", "Version mismatch") 

589 

590 write = await repo.update(tenant_id, resource_type, resource_id, resource, transaction_refs) 

591 body = inject_meta(write["resource"], write["version_id"], write["last_updated"]) 

592 return { 

593 "resource": body, 

594 "response": {"status": "200 OK", "etag": etag(write["version_id"])}, 

595 } 

596 

597 

598async def _dispatch_delete( 

599 resource_type: str, 

600 resource_id: str | None, 

601 query: str | None, 

602 repo: ResourceRepository, 

603 tenant_id: UUID, 

604 tenant_settings: ResolvedTenantSettings, 

605 conn: asyncpg.Connection[asyncpg.Record] | asyncpg.pool.PoolConnectionProxy[asyncpg.Record], 

606) -> dict[str, Any]: 

607 if not resource_id and not query: 

608 raise OperationOutcomeError( 

609 400, "invalid", "DELETE entry requires resource id in URL or query parameters" 

610 ) 

611 

612 if not resource_id and query: 

613 matching_ids = await _conditional_search_ids_with_conn( 

614 conn, resource_type, query, tenant_id, tenant_settings 

615 ) 

616 if not matching_ids: 

617 return {"response": {"status": "204 No Content"}} 

618 for rid in matching_ids: 

619 await repo.delete(tenant_id, resource_type, rid) 

620 return {"response": {"status": "204 No Content"}} 

621 

622 if resource_id: 

623 if await repo.is_deleted(tenant_id, resource_type, resource_id): 

624 return {"response": {"status": "204 No Content"}} 

625 deleted = await repo.delete(tenant_id, resource_type, resource_id) 

626 if not deleted: 

627 raise OperationOutcomeError( 

628 404, "not-found", f"{resource_type}/{resource_id} not found" 

629 ) 

630 return {"response": {"status": "204 No Content"}} 

631 return {"response": {"status": "204 No Content"}} 

632 

633 

634# ── Batch ───────────────────────────────────────────────────────────────────── 

635 

636 

637async def _process_batch( 

638 entries: list[dict[str, Any]], 

639 pool: asyncpg.Pool[asyncpg.Record], 

640 tenant_id: UUID, 

641 tenant_settings: ResolvedTenantSettings, 

642 base_url: str, 

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

644 empty: dict[str, str] = {} 

645 response_entries: list[dict[str, Any]] = [] 

646 

647 transaction_refs: set[str] = set() 

648 for entry in entries: 

649 req = cast(dict[str, Any], entry.get("request") or {}) 

650 method = cast(str, req.get("method") or "").upper() 

651 url = cast(str, req.get("url") or "") 

652 res_type, res_id, _ = _parse_entry_url(url) 

653 if method in ("POST", "PUT", "PATCH") and res_type and res_id: 

654 transaction_refs.add(f"{res_type}/{res_id}") 

655 

656 logger.info("bundle.started", type="batch", entries=len(entries)) 

657 

658 for entry in entries: 

659 async with pool.acquire() as conn: 

660 try: 

661 resp_entry = await _dispatch( 

662 entry, 

663 conn, 

664 tenant_id, 

665 tenant_settings, 

666 base_url, 

667 empty, 

668 empty, 

669 transaction_refs, 

670 ) 

671 except OperationOutcomeError as exc: 

672 logger.warning( 

673 "bundle.entry_failed", 

674 code=exc.code, 

675 diagnostics=exc.diagnostics, 

676 ) 

677 resp_entry = { 

678 "response": { 

679 "status": f"{exc.status_code} Error", 

680 "outcome": { 

681 "resourceType": "OperationOutcome", 

682 "issue": [ 

683 { 

684 "severity": "error", 

685 "code": exc.code, 

686 "diagnostics": exc.diagnostics, 

687 } 

688 ], 

689 }, 

690 } 

691 } 

692 response_entries.append(resp_entry) 

693 

694 logger.info("bundle.finished", type="batch", entries=len(entries)) 

695 return response_entries 

696 

697 

698# ── Transaction ─────────────────────────────────────────────────────────────── 

699 

700 

701async def _process_transaction( 

702 entries: list[dict[str, Any]], 

703 pool: asyncpg.Pool[asyncpg.Record], 

704 tenant_id: UUID, 

705 tenant_settings: ResolvedTenantSettings, 

706 base_url: str, 

707 is_document: bool = False, 

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

709 logger.info("bundle.started", type="transaction", entries=len(entries)) 

710 

711 async with pool.acquire() as conn, conn.transaction(): 

712 ref_map, id_map, resolved_pre_existing = await _build_urn_map( 

713 entries, conn, tenant_id, tenant_settings, is_document 

714 ) 

715 transaction_refs = set(ref_map.values()) 

716 for entry in entries: 

717 full_url = cast(str, entry.get("fullUrl") or "") 

718 if full_url in resolved_pre_existing: 

719 continue 

720 req = cast(dict[str, Any], entry.get("request") or {}) 

721 method = cast(str, req.get("method") or "").upper() 

722 url = cast(str, req.get("url") or "")