Coverage for src/lumen/api/root.py: 97%

1""" 

2This router handles requests to the root URL and the base FHIR endpoint. 

3It serves metadata info and acts as the entry point for batch/transaction bundles. 

4""" 

5 

6from __future__ import annotations 

7 

8import json 

9from typing import Any, cast 

10 

11from fastapi import APIRouter, Request 

12from fastapi.responses import JSONResponse, Response 

13 

14from ..core.exceptions import FHIR_JSON, OperationOutcomeError 

15from ..db.pool import get_pool 

16from .bundle import process_bundle 

17from .deps import TenantDep, TenantSettingsDep 

18 

19router = APIRouter() 

20 

21 

22@router.get("/") 

23async def root_get(request: Request) -> Response: 

24 base = str(request.base_url).rstrip("/") 

25 return JSONResponse( 

26 { 

27 "name": "Lumen FHIR Server", 

28 "fhirVersion": "4.3.0", 

29 "docsUrl": f"{base}/docs", 

30 "metadataUrl": f"{base}/metadata", 

31 } 

32 ) 

33 

34 

35@router.post("/") 

36async def root_post( 

37 request: Request, 

38 tenant: TenantDep, 

39 tenant_settings: TenantSettingsDep, 

40) -> Response: 

41 body_bytes = await request.body() 

42 try: 

43 parsed = json.loads(body_bytes) 

44 except json.JSONDecodeError as exc: 

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

46 

47 if not isinstance(parsed, dict): 

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

49 

50 bundle = cast(dict[str, Any], parsed) 

51 if bundle.get("resourceType") != "Bundle": 

52 raise OperationOutcomeError(400, "invalid", "resourceType must be 'Bundle' for POST /") 

53 

54 pool = get_pool() 

55 base_url = str(request.base_url).rstrip("/") 

56 result = await process_bundle(bundle, pool, tenant, tenant_settings, base_url) 

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