Coverage for netrun_errors / handlers.py: 75%

1""" 

2Global exception handlers for FastAPI applications. 

3 

4Provides centralized exception handling with structured JSON responses, 

5correlation ID injection, and logging integration. 

6 

7v1.1.0: Enhanced logging with optional netrun-logging integration for 

8structured key=value logging and correlation ID consistency. 

9""" 

10 

11import logging 

12from typing import Union 

13 

14from fastapi import FastAPI, Request, status 

15from fastapi.exceptions import RequestValidationError 

16from fastapi.responses import JSONResponse 

17from starlette.exceptions import HTTPException as StarletteHTTPException 

18 

19from .base import NetrunException 

20 

21# Try to use netrun-logging for structured logging 

22_use_structlog = False 

23_structlog_logger = None 

24 

25try: 

26 from netrun_logging import get_logger as _get_structlog_logger 

27 _structlog_logger = _get_structlog_logger(__name__) 

28 _use_structlog = True 

29except ImportError: 

30 pass 

31 

32# Fallback to standard logging 

33logger = logging.getLogger(__name__) 

34 

35 

36def _log_error( 

37 message: str, 

38 level: str = "error", 

39 exc_info: bool = False, 

40 **kwargs 

41) -> None: 

42 """Log error using netrun-logging if available, otherwise standard logging.""" 

43 if _use_structlog and _structlog_logger is not None: 

44 log_method = getattr(_structlog_logger, level, _structlog_logger.error) 

45 log_method(message, **kwargs) 

46 else: 

47 log_method = getattr(logger, level, logger.error) 

48 log_method(message, extra=kwargs, exc_info=exc_info) 

49 

50 

51async def netrun_exception_handler( 

52 request: Request, exc: NetrunException 

53) -> JSONResponse: 

54 """ 

55 Handle NetrunException instances with structured JSON responses. 

56 

57 Args: 

58 request: FastAPI request object 

59 exc: NetrunException instance 

60 

61 Returns: 

62 JSONResponse with structured error format 

63 """ 

64 # Add request path to error details 

65 error_dict = exc.to_dict() 

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

67 

68 # Log error with correlation ID (structured logging if netrun-logging available) 

69 _log_error( 

70 f"netrun_exception: {exc.message}", 

71 level="error", 

72 correlation_id=exc.correlation_id, 

73 error_code=exc.error_code, 

74 status_code=exc.status_code, 

75 path=str(request.url.path), 

76 method=request.method, 

77 ) 

78 

79 return JSONResponse( 

80 status_code=exc.status_code, 

81 content=error_dict, 

82 ) 

83 

84 

85async def validation_exception_handler( 

86 request: Request, exc: RequestValidationError 

87) -> JSONResponse: 

88 """ 

89 Handle FastAPI request validation errors with structured format. 

90 

91 Args: 

92 request: FastAPI request object 

93 exc: RequestValidationError instance 

94 

95 Returns: 

96 JSONResponse with validation error details 

97 """ 

98 from datetime import datetime, timezone 

99 

100 correlation_id = NetrunException._generate_correlation_id() 

101 

102 error_response = { 

103 "error": { 

104 "code": "VALIDATION_ERROR", 

105 "message": "Request validation failed", 

106 "details": { 

107 "validation_errors": exc.errors(), 

108 }, 

109 "correlation_id": correlation_id, 

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

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

112 } 

113 } 

114 

115 _log_error( 

116 "validation_error", 

117 level="warning", 

118 correlation_id=correlation_id, 

119 path=str(request.url.path), 

120 method=request.method, 

121 error_count=len(exc.errors()), 

122 ) 

123 

124 return JSONResponse( 

125 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, 

126 content=error_response, 

127 ) 

128 

129 

130async def http_exception_handler( 

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

132) -> JSONResponse: 

133 """ 

134 Handle generic HTTP exceptions with structured format. 

135 

136 Args: 

137 request: FastAPI request object 

138 exc: HTTPException or generic Exception 

139 

140 Returns: 

141 JSONResponse with structured error format 

142 """ 

143 from datetime import datetime, timezone 

144 

145 correlation_id = NetrunException._generate_correlation_id() 

146 

147 # Determine status code 

148 if isinstance(exc, StarletteHTTPException): 

149 status_code = exc.status_code 

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

151 else: 

152 status_code = status.HTTP_500_INTERNAL_SERVER_ERROR 

153 message = "An unexpected error occurred" 

154 

155 error_response = { 

156 "error": { 

157 "code": "HTTP_ERROR", 

158 "message": message, 

159 "details": {}, 

160 "correlation_id": correlation_id, 

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

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

163 } 

164 } 

165 

166 _log_error( 

167 f"http_exception: {message}", 

168 level="error", 

169 exc_info=(status_code >= 500), 

170 correlation_id=correlation_id, 

171 status_code=status_code, 

172 path=str(request.url.path), 

173 method=request.method, 

174 ) 

175 

176 return JSONResponse( 

177 status_code=status_code, 

178 content=error_response, 

179 ) 

180 

181 

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

183 """ 

184 Handle unhandled exceptions with structured format. 

185 

186 Args: 

187 request: FastAPI request object 

188 exc: Exception instance 

189 

190 Returns: 

191 JSONResponse with generic error format 

192 """ 

193 from datetime import datetime, timezone 

194 

195 correlation_id = NetrunException._generate_correlation_id() 

196 

197 error_response = { 

198 "error": { 

199 "code": "INTERNAL_SERVER_ERROR", 

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

201 "details": {}, 

202 "correlation_id": correlation_id, 

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

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

205 } 

206 } 

207 

208 _log_error( 

209 "unhandled_exception", 

210 level="error", 

211 exc_info=True, 

212 correlation_id=correlation_id, 

213 path=str(request.url.path), 

214 method=request.method, 

215 exception_type=type(exc).__name__, 

216 exception_message=str(exc), 

217 ) 

218 

219 return JSONResponse( 

220 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, 

221 content=error_response, 

222 ) 

223 

224 

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

226 """ 

227 Install all Netrun exception handlers on a FastAPI application. 

228 

229 Usage: 

230 from fastapi import FastAPI 

231 from netrun_errors import install_exception_handlers 

232 

233 app = FastAPI() 

234 install_exception_handlers(app) 

235 

236 Args: 

237 app: FastAPI application instance 

238 """ 

239 app.add_exception_handler(NetrunException, netrun_exception_handler) 

240 app.add_exception_handler(RequestValidationError, validation_exception_handler) 

241 app.add_exception_handler(StarletteHTTPException, http_exception_handler) 

242 app.add_exception_handler(Exception, unhandled_exception_handler) 

243 

244 _log_error( 

245 "exception_handlers_installed", 

246 level="info", 

247 handler_count=4, 

248 using_structlog=_use_structlog, 

249 )