Coverage for src/lumen/core/middleware.py: 95%

1"""FHIR content negotiation middleware. 

2 

3Enforces Accept / Content-Type rules and handles JSON ↔ XML conversion 

4so that every route handler only ever speaks application/fhir+json internally. 

5""" 

6 

7from __future__ import annotations 

8 

9import contextlib 

10import json 

11 

12from starlette.datastructures import MutableHeaders 

13from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint 

14from starlette.requests import Request 

15from starlette.responses import JSONResponse, Response 

16 

17from ..fhir.xml import json_to_xml, xml_to_json_bytes 

18from .exceptions import FHIR_JSON, FHIR_XML 

19 

20_FHIR_PATHS_EXCLUDED = {"/docs", "/openapi.json", "/redoc", "/health"} 

21_FHIR_XML_MEDIA_TYPE = "application/fhir+xml" 

22_ACCEPTED_TYPES = { 

23 "application/fhir+json", 

24 _FHIR_XML_MEDIA_TYPE, 

25 "application/json", 

26 "*/*", 

27 "application/*", 

28} 

29 

30_406_BODY = { 

31 "resourceType": "OperationOutcome", 

32 "issue": [ 

33 { 

34 "severity": "error", 

35 "code": "not-supported", 

36 "diagnostics": ( 

37 "This server supports application/fhir+json and application/fhir+xml responses." 

38 ), 

39 } 

40 ], 

41} 

42 

43 

44def _is_binary_read(request: Request) -> bool: 

45 """True for ``GET /Binary/{id}`` — FHIR allows requesting the resource's 

46 native content type here (e.g. ``Accept: application/pdf``), so the 

47 generic FHIR-format Accept check is skipped and the route handler 

48 negotiates the response itself. 

49 """ 

50 if request.method != "GET": 

51 return False 

52 parts = request.url.path.strip("/").split("/") 

53 return len(parts) == 2 and parts[0] == "Binary" 

54 

55 

56def _wants_xml(accept: str) -> bool: 

57 """Return True if the highest-priority accepted type is an XML variant.""" 

58 best_type = "" 

59 best_q = -1.0 

60 for part in accept.split(","): 

61 segments = part.strip().split(";") 

62 media_type = segments[0].strip().lower() 

63 q = 1.0 

64 for seg in segments[1:]: 

65 seg = seg.strip() 

66 if seg.startswith("q="): 

67 with contextlib.suppress(ValueError): 

68 q = float(seg[2:]) 

69 if q > best_q: 

70 best_q = q 

71 best_type = media_type 

72 return best_type == _FHIR_XML_MEDIA_TYPE 

73 

74 

75class FhirContentNegotiationMiddleware(BaseHTTPMiddleware): 

76 """Enforce FHIR MIME type rules and transparently convert XML ↔ JSON.""" 

77 

78 async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response: 

79 if request.url.path in _FHIR_PATHS_EXCLUDED or request.url.path.startswith("/docs"): 

80 return await call_next(request) 

81 

82 # --- Inbound XML body → JSON conversion --- 

83 content_type = request.headers.get("content-type", "") 

84 if _FHIR_XML_MEDIA_TYPE in content_type: 

85 xml_body = await request.body() 

86 try: 

87 json_bytes = xml_to_json_bytes(xml_body) 

88 except Exception: 

89 return JSONResponse( 

90 status_code=400, 

91 content={ 

92 "resourceType": "OperationOutcome", 

93 "issue": [ 

94 { 

95 "severity": "error", 

96 "code": "invalid", 

97 "diagnostics": "Invalid FHIR XML request body.", 

98 } 

99 ], 

100 }, 

101 media_type=FHIR_JSON, 

102 ) 

103 request._body = json_bytes # type: ignore[attr-defined] 

104 new_headers = MutableHeaders(scope=request.scope) 

105 new_headers["content-type"] = FHIR_JSON 

106 new_headers["content-length"] = str(len(json_bytes)) 

107 

108 # --- Accept header validation --- 

109 accept = request.headers.get("accept", "*/*") 

110 media_types = {part.split(";")[0].strip().lower() for part in accept.split(",")} 

111 if not (media_types & _ACCEPTED_TYPES) and not _is_binary_read(request): 

112 return JSONResponse(status_code=406, content=_406_BODY, media_type=FHIR_JSON) 

113 

114 response = await call_next(request) 

115 

116 # --- Outbound JSON → XML conversion when client prefers XML --- 

117 if _wants_xml(accept): 

118 resp_ct = response.headers.get("content-type", "") 

119 if "application/fhir+json" in resp_ct or "application/json" in resp_ct: 

120 body = b"".join([chunk async for chunk in response.body_iterator]) # type: ignore[attr-defined] 

121 try: 

122 xml_bytes = json_to_xml(json.loads(body)) 

123 safe_headers = { 

124 k: v 

125 for k, v in response.headers.items() 

126 if k.lower() not in ("content-type", "content-length", "transfer-encoding") 

127 } 

128 return Response( 

129 content=xml_bytes, 

130 media_type=FHIR_XML, 

131 status_code=response.status_code, 

132 headers=safe_headers, 

133 ) 

134 except Exception: 

135 # Conversion failed — return original JSON rather than 500 

136 return Response( 

137 content=body, 

138 status_code=response.status_code, 

139 headers=dict(response.headers), 

140 ) 

141 

142 return response