Coverage for netrun_errors / base.py: 86%

1""" 

2Base exception class for Netrun Systems unified error handling. 

3 

4Provides structured error responses with correlation IDs, timestamps, 

5and machine-readable error codes for FastAPI applications. 

6 

7v1.1.0: Added netrun-logging integration for structured exception logging. 

8""" 

9 

10from datetime import datetime, timezone 

11from typing import Any, Dict, Optional, TYPE_CHECKING 

12from uuid import uuid4 

13 

14from fastapi import HTTPException, status 

15 

16# Optional netrun-logging integration (soft dependency) 

17_netrun_logging_available = False 

18_get_correlation_id = None 

19 

20try: 

21 from netrun_logging import get_correlation_id as _logging_get_correlation_id 

22 _netrun_logging_available = True 

23 _get_correlation_id = _logging_get_correlation_id 

24except ImportError: 

25 pass 

26 

27 

28class NetrunException(HTTPException): 

29 """ 

30 Base exception class for all Netrun Systems exceptions. 

31 

32 Extends FastAPI's HTTPException with additional metadata: 

33 - error_code: Machine-readable error identifier 

34 - correlation_id: Request tracking identifier 

35 - timestamp: ISO 8601 formatted timestamp 

36 - details: Additional context dictionary 

37 

38 All Netrun exceptions inherit from this class and return structured 

39 JSON responses compatible with frontend error handling. 

40 """ 

41 

42 def __init__( 

43 self, 

44 status_code: int, 

45 error_code: str, 

46 message: str, 

47 details: Optional[Dict[str, Any]] = None, 

48 correlation_id: Optional[str] = None, 

49 ): 

50 """ 

51 Initialize NetrunException with structured error data. 

52 

53 Args: 

54 status_code: HTTP status code (401, 403, 404, etc.) 

55 error_code: Machine-readable error code (e.g., "AUTH_INVALID_CREDENTIALS") 

56 message: Human-readable error message 

57 details: Additional context dictionary (optional) 

58 correlation_id: Request tracking ID (auto-generated if not provided) 

59 """ 

60 self.error_code = error_code 

61 self.message = message 

62 self.correlation_id = correlation_id or self._get_or_generate_correlation_id() 

63 self.timestamp = datetime.now(timezone.utc).isoformat() 

64 self.details = details or {} 

65 

66 # Store message as detail for HTTPException (plain string) 

67 super().__init__(status_code=status_code, detail=message) 

68 

69 @staticmethod 

70 def _get_or_generate_correlation_id() -> str: 

71 """ 

72 Get correlation ID from netrun-logging context or generate a new one. 

73 

74 If netrun-logging is available and has an active correlation ID, 

75 uses that for consistent request tracking. Otherwise generates a new ID. 

76 

77 Returns: 

78 Correlation ID string 

79 """ 

80 # Try to get correlation ID from netrun-logging context 

81 if _netrun_logging_available and _get_correlation_id is not None: 

82 try: 

83 existing_id = _get_correlation_id() 

84 if existing_id: 

85 return existing_id 

86 except Exception: 

87 pass # Fall through to generate new ID 

88 

89 return NetrunException._generate_correlation_id() 

90 

91 @staticmethod 

92 def _generate_correlation_id() -> str: 

93 """ 

94 Generate a unique correlation ID for request tracking. 

95 

96 Format: req-YYYYMMDD-HHMMSS-uuid4_prefix 

97 Example: req-20251125-143210-a8f3c9 

98 

99 Returns: 

100 Correlation ID string 

101 """ 

102 timestamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") 

103 uuid_prefix = str(uuid4())[:6] 

104 return f"req-{timestamp}-{uuid_prefix}" 

105 

106 def to_dict(self) -> Dict[str, Any]: 

107 """ 

108 Convert exception to dictionary representation. 

109 

110 Returns: 

111 Dictionary with error structure 

112 """ 

113 return { 

114 "error": { 

115 "code": self.error_code, 

116 "message": self.message, 

117 "details": self.details, 

118 "correlation_id": self.correlation_id, 

119 "timestamp": self.timestamp, 

120 } 

121 }