Coverage for src / augint_library / logging.py: 100%

1"""Structured logging configuration for augint-library. 

2 

3This module provides production-ready structured logging using only the Python 

4standard library. It demonstrates: 

5- JSON formatting for log aggregation systems 

6- Structured logging with contextual data 

7- Service-oriented logging patterns 

8- Easy integration with cloud logging services 

9 

10Common Use Cases: 

11 1. Adding structured logging to libraries 

12 2. Cloud-native applications (AWS CloudWatch, GCP Stackdriver) 

13 3. Local development with readable logs 

14 4. Debugging distributed systems 

15 5. Audit trails and compliance logging 

16 

17Examples: 

18 Basic logging setup for development: 

19 >>> from augint_library.logging import setup_logging 

20 >>> logger = setup_logging("my-service") 

21 >>> logger.info("User logged in", extra={"user_id": 123, "ip": "192.168.1.1"}) 

22 2024-01-01 12:00:00 - my-service - INFO - User logged in - user_id=123 ip=192.168.1.1 

23 

24 JSON formatting for production systems: 

25 >>> logger = setup_logging("payment-service", json_format=True) 

26 >>> logger.info("Payment processed", extra={ 

27 ... "amount": 99.99, 

28 ... "currency": "USD", 

29 ... "transaction_id": "tx-12345" 

30 ... }) 

31 {"timestamp": "2024-01-01T12:00:00Z", "service": "payment-service", "level": "INFO", ...} 

32 

33 Error logging with exception details: 

34 >>> logger = setup_logging("api-service", json_format=True) 

35 >>> try: 

36 ... result = process_order(order_id) 

37 ... except Exception as e: 

38 ... logger.exception("Order processing failed", extra={ 

39 ... "order_id": order_id, 

40 ... "error_type": type(e).__name__ 

41 ... }) 

42 

43 Using custom log levels: 

44 >>> logger = setup_logging("debug-service", level="DEBUG") 

45 >>> logger.debug("Detailed trace", extra={"step": 1, "data": {"x": 10}}) 

46 

47 Integration with AWS Lambda: 

48 >>> import os 

49 >>> # AWS Lambda sets this automatically 

50 >>> os.environ['AWS_LAMBDA_FUNCTION_NAME'] = 'my-function' 

51 >>> logger = setup_logging("lambda-service", json_format=True) 

52 >>> # Logs will include Lambda context automatically 

53 

54Best Practices: 

55 1. Always use structured logging (extra={} parameter) 

56 2. Include correlation IDs for request tracing 

57 3. Use appropriate log levels (DEBUG, INFO, WARNING, ERROR) 

58 4. Never log sensitive data (passwords, tokens, PII) 

59 5. Use JSON format in production for better querying 

60 

61Integration Examples: 

62 With Flask/FastAPI: 

63 >>> @app.before_request 

64 >>> def setup_request_logging(): 

65 ... g.request_id = str(uuid.uuid4()) 

66 ... logger.info("Request started", extra={ 

67 ... "request_id": g.request_id, 

68 ... "method": request.method, 

69 ... "path": request.path 

70 ... }) 

71 

72 With Click CLI: 

73 >>> @click.command() 

74 >>> @click.option('--verbose', '-v', is_flag=True) 

75 >>> def process(verbose): 

76 ... level = "DEBUG" if verbose else "INFO" 

77 ... logger = setup_logging("cli-tool", level=level) 

78 

79Note: 

80 This module uses only the standard library to ensure maximum compatibility. 

81 For advanced features (correlation IDs, sampling, etc.), consider using 

82 specialized libraries like structlog or AWS Powertools. 

83""" 

84 

85import json 

86import logging 

87import sys 

88from datetime import datetime, timezone 

89from typing import Any, Optional 

90 

91 

92class JSONFormatter(logging.Formatter): 

93 """Simple JSON formatter using only standard library. 

94 

95 Formats log records as JSON with consistent field names for 

96 structured logging systems. Handles exceptions gracefully. 

97 """ 

98 

99 def format(self, record: logging.LogRecord) -> str: 

100 """Format log record as JSON string. 

101 

102 Args: 

103 record: Python logging record to format. 

104 

105 Returns: 

106 JSON string representation of the log record. 

107 """ 

108 log_entry: dict[str, Any] = { 

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

110 "level": record.levelname, 

111 "logger": record.name, 

112 "message": record.getMessage(), 

113 } 

114 

115 # Add exception info if present 

116 if record.exc_info: 

117 log_entry["exception"] = self.formatException(record.exc_info) 

118 

119 # Add any extra fields from the log call 

120 for key, value in record.__dict__.items(): 

121 if key not in { 

122 "name", 

123 "msg", 

124 "args", 

125 "levelname", 

126 "levelno", 

127 "pathname", 

128 "filename", 

129 "module", 

130 "exc_info", 

131 "exc_text", 

132 "stack_info", 

133 "lineno", 

134 "funcName", 

135 "created", 

136 "msecs", 

137 "relativeCreated", 

138 "thread", 

139 "threadName", 

140 "processName", 

141 "process", 

142 "message", 

143 }: 

144 log_entry[key] = value 

145 

146 return json.dumps(log_entry, default=str) 

147 

148 

149def setup_logging( 

150 service_name: str, 

151 level: str = "INFO", 

152 json_format: bool = False, 

153 handler: Optional[logging.Handler] = None, 

154) -> logging.Logger: 

155 """Configure structured logging for the service. 

156 

157 Creates a logger with consistent formatting that works well with 

158 both local development and production log aggregation systems. 

159 

160 Args: 

161 service_name: Name of the service/component for log identification. 

162 level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL). 

163 json_format: Whether to use JSON formatting for structured logs. 

164 handler: Optional custom handler (defaults to StreamHandler to stdout). 

165 

166 Returns: 

167 Configured logger instance ready for use. 

168 

169 Example: 

170 >>> # Basic usage for development 

171 >>> logger = setup_logging("user-service") 

172 >>> logger.info("User created", extra={"user_id": 123}) 

173 

174 >>> # JSON format for production 

175 >>> logger = setup_logging("user-service", json_format=True) 

176 >>> logger.error("Validation failed", extra={"errors": ["email required"]}) 

177 """ 

178 logger = logging.getLogger(service_name) 

179 

180 # Clear any existing handlers to avoid duplication 

181 logger.handlers.clear() 

182 

183 # Create handler (default to stdout for container/Lambda compatibility) 

184 if handler is None: 

185 handler = logging.StreamHandler(sys.stdout) 

186 

187 # Set formatter based on preference 

188 formatter: logging.Formatter 

189 if json_format: 

190 formatter = JSONFormatter() 

191 else: 

192 formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") 

193 

194 handler.setFormatter(formatter) 

195 logger.addHandler(handler) 

196 logger.setLevel(getattr(logging, level.upper())) 

197 

198 # Prevent log messages from being handled by root logger 

199 logger.propagate = False 

200 

201 return logger