Coverage for netrun_errors \ handlers.py: 76%

1""" 

2Global exception handlers for FastAPI applications. 

3 

4Provides centralized exception handling with structured JSON responses, 

5correlation ID injection, and logging integration. 

6""" 

7 

8import logging 

9from typing import Union 

10 

11from fastapi import FastAPI, Request, status 

12from fastapi.exceptions import RequestValidationError 

13from fastapi.responses import JSONResponse 

14from starlette.exceptions import HTTPException as StarletteHTTPException 

15 

16from .base import NetrunException 

17 

18logger = logging.getLogger(__name__) 

19 

20 

21async def netrun_exception_handler( 

22 request: Request, exc: NetrunException 

23) -> JSONResponse: 

24 """ 

25 Handle NetrunException instances with structured JSON responses. 

26 

27 Args: 

28 request: FastAPI request object 

29 exc: NetrunException instance 

30 

31 Returns: 

32 JSONResponse with structured error format 

33 """ 

34 # Add request path to error details 

35 error_dict = exc.to_dict() 

36 error_dict["error"]["path"] = str(request.url.path) 

37 

38 # Log error with correlation ID 

39 logger.error( 

40 f"NetrunException: {exc.error_code} - {exc.message}", 

41 extra={ 

42 "correlation_id": exc.correlation_id, 

43 "error_code": exc.error_code, 

44 "path": request.url.path, 

45 "method": request.method, 

46 }, 

47 ) 

48 

49 return JSONResponse( 

50 status_code=exc.status_code, 

51 content=error_dict, 

52 ) 

53 

54 

55async def validation_exception_handler( 

56 request: Request, exc: RequestValidationError 

57) -> JSONResponse: 

58 """ 

59 Handle FastAPI request validation errors with structured format. 

60 

61 Args: 

62 request: FastAPI request object 

63 exc: RequestValidationError instance 

64 

65 Returns: 

66 JSONResponse with validation error details 

67 """ 

68 from datetime import datetime, timezone 

69 

70 correlation_id = NetrunException._generate_correlation_id() 

71 

72 error_response = { 

73 "error": { 

74 "code": "VALIDATION_ERROR", 

75 "message": "Request validation failed", 

76 "details": { 

77 "validation_errors": exc.errors(), 

78 }, 

79 "correlation_id": correlation_id, 

80 "timestamp": datetime.now(timezone.utc).isoformat(), 

81 "path": str(request.url.path), 

82 } 

83 } 

84 

85 logger.warning( 

86 f"Validation error: {request.method} {request.url.path}", 

87 extra={ 

88 "correlation_id": correlation_id, 

89 "validation_errors": exc.errors(), 

90 }, 

91 ) 

92 

93 return JSONResponse( 

94 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, 

95 content=error_response, 

96 ) 

97 

98 

99async def http_exception_handler( 

100 request: Request, exc: Union[StarletteHTTPException, Exception] 

101) -> JSONResponse: 

102 """ 

103 Handle generic HTTP exceptions with structured format. 

104 

105 Args: 

106 request: FastAPI request object 

107 exc: HTTPException or generic Exception 

108 

109 Returns: 

110 JSONResponse with structured error format 

111 """ 

112 from datetime import datetime, timezone 

113 

114 correlation_id = NetrunException._generate_correlation_id() 

115 

116 # Determine status code 

117 if isinstance(exc, StarletteHTTPException): 

118 status_code = exc.status_code 

119 message = exc.detail if isinstance(exc.detail, str) else str(exc.detail) 

120 else: 

121 status_code = status.HTTP_500_INTERNAL_SERVER_ERROR 

122 message = "An unexpected error occurred" 

123 

124 error_response = { 

125 "error": { 

126 "code": "HTTP_ERROR", 

127 "message": message, 

128 "details": {}, 

129 "correlation_id": correlation_id, 

130 "timestamp": datetime.now(timezone.utc).isoformat(), 

131 "path": str(request.url.path), 

132 } 

133 } 

134 

135 logger.error( 

136 f"HTTP exception: {status_code} - {message}", 

137 extra={ 

138 "correlation_id": correlation_id, 

139 "status_code": status_code, 

140 "path": request.url.path, 

141 "method": request.method, 

142 }, 

143 exc_info=True if status_code >= 500 else False, 

144 ) 

145 

146 return JSONResponse( 

147 status_code=status_code, 

148 content=error_response, 

149 ) 

150 

151 

152async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse: 

153 """ 

154 Handle unhandled exceptions with structured format. 

155 

156 Args: 

157 request: FastAPI request object 

158 exc: Exception instance 

159 

160 Returns: 

161 JSONResponse with generic error format 

162 """ 

163 from datetime import datetime, timezone 

164 

165 correlation_id = NetrunException._generate_correlation_id() 

166 

167 error_response = { 

168 "error": { 

169 "code": "INTERNAL_SERVER_ERROR", 

170 "message": "An unexpected error occurred. Please try again later.", 

171 "details": {}, 

172 "correlation_id": correlation_id, 

173 "timestamp": datetime.now(timezone.utc).isoformat(), 

174 "path": str(request.url.path), 

175 } 

176 } 

177 

178 logger.exception( 

179 f"Unhandled exception: {type(exc).__name__}", 

180 extra={ 

181 "correlation_id": correlation_id, 

182 "path": request.url.path, 

183 "method": request.method, 

184 "exception_type": type(exc).__name__, 

185 }, 

186 ) 

187 

188 return JSONResponse( 

189 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, 

190 content=error_response, 

191 ) 

192 

193 

194def install_exception_handlers(app: FastAPI) -> None: 

195 """ 

196 Install all Netrun exception handlers on a FastAPI application. 

197 

198 Usage: 

199 from fastapi import FastAPI 

200 from netrun_errors import install_exception_handlers 

201 

202 app = FastAPI() 

203 install_exception_handlers(app) 

204 

205 Args: 

206 app: FastAPI application instance 

207 """ 

208 app.add_exception_handler(NetrunException, netrun_exception_handler) 

209 app.add_exception_handler(RequestValidationError, validation_exception_handler) 

210 app.add_exception_handler(StarletteHTTPException, http_exception_handler) 

211 app.add_exception_handler(Exception, unhandled_exception_handler) 

212 

213 logger.info("Netrun exception handlers installed successfully")