Coverage for src/lumen/core/exceptions.py: 100%

1""" 

2This file defines the custom exceptions and exception handlers for the application. 

3It ensures that all server errors are returned as valid FHIR OperationOutcome responses. 

4""" 

5 

6import structlog 

7from fastapi import Request 

8from fastapi.responses import JSONResponse 

9 

10from ..auth.token import AuthError 

11 

12logger = structlog.get_logger(__name__) 

13 

14FHIR_JSON = "application/fhir+json; charset=utf-8" 

15FHIR_XML = "application/fhir+xml; charset=utf-8" 

16 

17 

18class OperationOutcomeError(Exception): 

19 def __init__(self, status_code: int, code: str, diagnostics: str) -> None: 

20 self.status_code = status_code 

21 self.code = code 

22 self.diagnostics = diagnostics 

23 super().__init__(diagnostics) 

24 

25 

26async def operation_outcome_handler(request: Request, exc: Exception) -> JSONResponse: 

27 if not isinstance(exc, OperationOutcomeError): 

28 return JSONResponse( 

29 status_code=500, 

30 content={ 

31 "resourceType": "OperationOutcome", 

32 "issue": [ 

33 { 

34 "severity": "fatal", 

35 "code": "exception", 

36 "diagnostics": "Internal server error", 

37 } 

38 ], 

39 }, 

40 media_type=FHIR_JSON, 

41 ) 

42 return JSONResponse( 

43 status_code=exc.status_code, 

44 content={ 

45 "resourceType": "OperationOutcome", 

46 "issue": [ 

47 { 

48 "severity": "error", 

49 "code": exc.code, 

50 "diagnostics": exc.diagnostics, 

51 } 

52 ], 

53 }, 

54 media_type=FHIR_JSON, 

55 ) 

56 

57 

58async def generic_exception_handler(request: Request, exc: Exception) -> JSONResponse: 

59 """ 

60 - basic exception handler for unhandled exceptions 

61 - ErrorCode 500 

62 """ 

63 await logger.aerror("server.unhandled_exception", exc_info=exc) 

64 return JSONResponse( 

65 status_code=500, 

66 content={ 

67 "resourceType": "OperationOutcome", 

68 "issue": [ 

69 {"severity": "fatal", "code": "exception", "diagnostics": "Internal server error"} 

70 ], 

71 }, 

72 media_type=FHIR_JSON, 

73 ) 

74 

75 

76async def auth_error_handler(request: Request, exc: Exception) -> JSONResponse: 

77 """ 

78 - Exception handler for AuthError exceptions. 

79 - ErrorCode 401 

80 """ 

81 if not isinstance(exc, AuthError): 

82 return await generic_exception_handler(request, exc) 

83 raw = exc.detail 

84 if isinstance(raw, dict): 

85 fhir_code: str = raw.get("fhir_code", "security") # type: ignore[assignment] 

86 diagnostics: str = raw.get("diagnostics", str(raw)) # type: ignore[assignment] 

87 else: 

88 fhir_code = "security" 

89 diagnostics = str(raw) 

90 headers = exc.headers or {} 

91 return JSONResponse( 

92 status_code=exc.status_code, 

93 content={ 

94 "resourceType": "OperationOutcome", 

95 "issue": [{"severity": "error", "code": fhir_code, "diagnostics": diagnostics}], 

96 }, 

97 headers=dict(headers), 

98 media_type=FHIR_JSON, 

99 ) 

100 

101 

102def not_found_handler(request: Request, exc: Exception) -> JSONResponse: 

103 """ 

104 - Exception handler for 404 Not Found errors, e.g ...fhir/People/1234 

105 - ErrorCode 404 

106 """ 

107 return JSONResponse( 

108 status_code=404, 

109 content={ 

110 "resourceType": "OperationOutcome", 

111 "issue": [ 

112 { 

113 "severity": "error", 

114 "code": "not-found", 

115 "diagnostics": f"{request.url.path} not found", 

116 } 

117 ], 

118 }, 

119 media_type=FHIR_JSON, 

120 )