Coverage for src / mysingle / core / middleware.py: 0%

1""" 

2FastAPI 미들웨어 for Correlation ID and Request Logging 

3""" 

4 

5import time 

6import uuid 

7from typing import Callable, Optional 

8 

9from fastapi import Request, Response 

10from starlette.middleware.base import BaseHTTPMiddleware 

11 

12from .logging import ( 

13 clear_logging_context, 

14 get_structured_logger, 

15 set_correlation_id, 

16 set_request_id, 

17 set_user_id, 

18) 

19 

20 

21class LoggingMiddleware(BaseHTTPMiddleware): 

22 """요청/응답 로깅 및 Correlation ID 관리 미들웨어""" 

23 

24 def __init__(self, app, service_name: str = "unknown"): 

25 super().__init__(app) 

26 self.service_name = service_name 

27 self.logger = get_structured_logger(__name__) 

28 

29 async def dispatch(self, request: Request, call_next: Callable) -> Response: 

30 # 컨텍스트 초기화 

31 clear_logging_context() 

32 

33 # Request ID 생성 

34 request_id = str(uuid.uuid4()) 

35 set_request_id(request_id) 

36 

37 # Correlation ID 설정 

38 correlation_id = request.headers.get("correlation-id") or str(uuid.uuid4()) 

39 set_correlation_id(correlation_id) 

40 

41 # User ID 설정 (X-User-Id 헤더에서) 

42 user_id = request.headers.get("x-user-id", "") 

43 if user_id: 

44 set_user_id(user_id) 

45 

46 # 요청 시작 시간 

47 start_time = time.time() 

48 

49 # 요청 로깅 

50 self.logger.info( 

51 "Request started", 

52 extra={ 

53 "method": request.method, 

54 "url": str(request.url), 

55 "path": request.url.path, 

56 "query_params": ( 

57 str(request.query_params) if request.query_params else None 

58 ), 

59 "user_agent": request.headers.get("user-agent"), 

60 "client_ip": self._get_client_ip(request), 

61 }, 

62 ) 

63 

64 # 요청 처리 

65 try: 

66 response = await call_next(request) 

67 duration = time.time() - start_time 

68 

69 # 성공 응답 로깅 

70 self.logger.info( 

71 "Request completed", 

72 extra={ 

73 "method": request.method, 

74 "url": str(request.url), 

75 "status_code": response.status_code, 

76 "duration_ms": round(duration * 1000, 2), 

77 }, 

78 ) 

79 

80 except Exception as exc: 

81 duration = time.time() - start_time 

82 

83 # 에러 응답 로깅 

84 self.logger.error( 

85 "Request failed", 

86 extra={ 

87 "method": request.method, 

88 "url": str(request.url), 

89 "duration_ms": round(duration * 1000, 2), 

90 "error": str(exc), 

91 "error_type": type(exc).__name__, 

92 }, 

93 ) 

94 raise 

95 

96 # Correlation ID를 응답 헤더에 추가 

97 response.headers["correlation-id"] = correlation_id 

98 response.headers["request-id"] = request_id 

99 

100 return response 

101 

102 def _get_client_ip(self, request: Request) -> str: 

103 """클라이언트 IP 주소 추출""" 

104 # X-Forwarded-For 헤더 확인 (프록시 환경) 

105 forwarded_for = request.headers.get("x-forwarded-for") 

106 if forwarded_for: 

107 # 첫 번째 IP가 실제 클라이언트 IP 

108 return forwarded_for.split(",")[0].strip() 

109 

110 # X-Real-IP 헤더 확인 

111 real_ip = request.headers.get("x-real-ip") 

112 if real_ip: 

113 return real_ip 

114 

115 # 직접 연결된 클라이언트 IP 

116 return request.client.host if request.client else "unknown" 

117 

118 

119class HealthCheckLoggingFilter: 

120 """헬스체크 요청 로깅 필터""" 

121 

122 def __init__(self, health_paths: Optional[list[str]] = None): 

123 self.health_paths = health_paths or ["/health", "/ready", "/alive"] 

124 

125 def should_log_request(self, request: Request) -> bool: 

126 """요청을 로깅할지 결정""" 

127 path = request.url.path 

128 

129 # 헬스체크 경로는 로깅하지 않음 

130 if path in self.health_paths: 

131 return False 

132 

133 # OPTIONS 요청은 로깅하지 않음 (CORS preflight) 

134 if request.method == "OPTIONS": 

135 return False 

136 

137 return True 

138 

139 

140class TimingLogMiddleware(BaseHTTPMiddleware): 

141 """간단한 타이밍 로그 미들웨어 (디버그용)""" 

142 

143 def __init__(self, app, enable_timing_logs: bool = False): 

144 super().__init__(app) 

145 self.enable_timing_logs = enable_timing_logs 

146 self.logger = get_structured_logger(__name__) 

147 

148 async def dispatch(self, request: Request, call_next: Callable) -> Response: 

149 if not self.enable_timing_logs: 

150 return await call_next(request) 

151 

152 start_time = time.time() 

153 response = await call_next(request) 

154 duration = time.time() - start_time 

155 

156 # 느린 요청만 로깅 (>1초) 

157 if duration > 1.0: 

158 self.logger.warning( 

159 "Slow request detected", 

160 extra={ 

161 "method": request.method, 

162 "path": request.url.path, 

163 "duration_ms": round(duration * 1000, 2), 

164 }, 

165 ) 

166 

167 return response 

168 

169 

170def add_logging_middleware(app, service_name: str, enable_timing_logs: bool = False): 

171 """로깅 미들웨어를 FastAPI 앱에 추가""" 

172 

173 # 타이밍 로그 미들웨어 (선택적) 

174 if enable_timing_logs: 

175 app.add_middleware(TimingLogMiddleware, enable_timing_logs=True) 

176 

177 # 메인 로깅 미들웨어 

178 app.add_middleware(LoggingMiddleware, service_name=service_name) 

179 

180 

181def setup_request_id_dependency(): 

182 """Request ID 의존성 함수 (FastAPI dependency)""" 

183 from typing import Optional 

184 

185 from fastapi import Depends, Header 

186 

187 def get_request_context( 

188 correlation_id: Optional[str] = Header(None, alias="correlation-id"), 

189 user_id: Optional[str] = Header(None, alias="x-user-id"), 

190 request_id: Optional[str] = Header(None, alias="request-id"), 

191 ): 

192 """요청 컨텍스트 정보 반환""" 

193 return { 

194 "correlation_id": correlation_id, 

195 "user_id": user_id, 

196 "request_id": request_id, 

197 } 

198 

199 return Depends(get_request_context)