Coverage for core / cache / cache_wrapper.py: 85%

1# core/cache_wrapper.py 

2import asyncio 

3from typing import TypeVar, Callable, Optional, Any 

4 

5from common.types import ResCommonResponse, ErrorCode 

6from core.cache.cache_store import CacheStore 

7from core.cache.cache_config import load_cache_config 

8from datetime import datetime 

9 

10T = TypeVar("T") 

11 

12 

13class ClientWithCache: 

14 def __init__( 

15 self, 

16 client, 

17 logger, 

18 market_clock, 

19 mode_fn: Callable[[], str], 

20 cache_store: Optional[CacheStore] = None, 

21 config: Optional[dict] = None, 

22 market_calendar_service: Optional[Any] = None # [추가] MarketCalendarService 주입 

23 ): 

24 self._client = client 

25 self._logger = logger 

26 self._market_clock = market_clock 

27 self._mcs = market_calendar_service # [추가] 

28 self._mode_fn = mode_fn # 동적으로 모드 가져오기 

29 

30 # ✅ 설정에서 읽기 

31 if config is None: 

32 config = load_cache_config() 

33 

34 cache_cfg = config.get("cache", {}) 

35 

36 self._file_enabled = bool(cache_cfg.get("file_cache_enabled", True)) 

37 self._memory_enabled = bool(cache_cfg.get("memory_cache_enabled", True)) 

38 self._caching_enabled = self._file_enabled or self._memory_enabled 

39 

40 self._cache = cache_store if cache_store else CacheStore(config) 

41 self._cache.set_logger(self._logger) 

42 self.cached_methods = set(config["cache"]["enabled_methods"]) 

43 

44 # market_close datetime 파싱 결과를 거래일 문자열 단위로 캐시 (hot-path strptime 절감) 

45 self._market_close_cache: dict = {} 

46 # in-flight 요청 중복 방지 (Thundering Herd 방어) 

47 self._in_flight: dict = {} 

48 

49 def __getattr__(self, name: str): 

50 # ✅ 무한 루프 방지 

51 if name.startswith("_"): # ✅ 내부 속성은 직접 접근 

52 return object.__getattribute__(self, name) 

53 

54 orig_attr = getattr(self._client, name) 

55 

56 if not callable(orig_attr) or name not in self.cached_methods: 

57 self._logger.debug(f"Bypass - {name} 캐시 건너뜀") 

58 return orig_attr 

59 

60 def _build_cache_key(mode: str, func_name: str, args: tuple, kwargs: dict) -> str: 

61 arg_str = "_".join(map(str, args)) if args else "" 

62 kwarg_str = "_".join(f"{k}={v}" for k, v in sorted(kwargs.items())) if kwargs else "" 

63 parts = [p for p in [mode, func_name, arg_str, kwarg_str] if p] 

64 return "_".join(parts) 

65 

66 async def wrapped(*args, **kwargs): 

67 # _skip_cache 플래그가 있으면 캐시를 완전히 우회 (무한 재귀 방지용) 

68 skip_cache = kwargs.pop("_skip_cache", False) 

69 if skip_cache: 

70 self._logger.debug(f"🔓 _skip_cache=True → 캐시 우회: {name}") 

71 return await orig_attr(*args, **kwargs) 

72 

73 mode = self._mode_fn() or "unknown" 

74 key = _build_cache_key(mode, name, args, kwargs) 

75 

76 # 캐시 전체 비활성화면 즉시 API 호출 

77 if not self._caching_enabled: 

78 self._logger.debug(f"🧺 Caching disabled → direct API call: {key}") 

79 return await orig_attr(*args, **kwargs) 

80 

81 # ✅ 1. 메모리 or 파일 캐시 조회 

82 # [수정] is_market_open_now는 async 메서드이므로 await 필요 

83 is_open = False 

84 if self._mcs: 

85 is_open = await self._mcs.is_market_open_now() 

86 

87 if is_open: 

88 self._logger.debug(f"⏳ 시장 개장 중 → 캐시 우회: {key}") 

89 else: 

90 raw = self._cache.get_raw(key) 

91 wrapper, cache_type = raw if raw is not None else (None, None) 

92 

93 if wrapper: 

94 cache_time = self._parse_timestamp(wrapper.get("timestamp")) 

95 

96 is_valid = False 

97 # [수정] next_open_time도 async 메서드 

98 next_open_time = await self._mcs.get_next_open_time() if self._mcs else None 

99 

100 if cache_time and next_open_time and cache_time < next_open_time: 

101 # [수정] MarketCalendarService가 있으면 실제 거래일 기준으로 검증 

102 if self._mcs: 102 ↛ 130line 102 didn't jump to line 130 because the condition on line 102 was always true

103 latest_trading_date_str = await self._mcs.get_latest_trading_date() 

104 if latest_trading_date_str: 

105 cache_date_str = cache_time.strftime("%Y%m%d") 

106 if cache_date_str > latest_trading_date_str: 106 ↛ 108line 106 didn't jump to line 108 because the condition on line 106 was never true

107 # 캐시가 최근 거래일 이후에 저장됨 → 유효 

108 is_valid = True 

109 elif cache_date_str == latest_trading_date_str: 

110 # 캐시 날짜 == 최근 거래일: 장 마감(15:40) 이후에 저장된 경우만 유효 

111 # 장 마감 전에 저장된 캐시는 전일 데이터이므로 무효 

112 # 최근 거래일 날짜 기준 장 마감 시간과 비교 (다음날 접근 시에도 정확히 비교) 

113 if latest_trading_date_str not in self._market_close_cache: 

114 latest_trading_date_dt = datetime.strptime(latest_trading_date_str, "%Y%m%d") 

115 self._market_close_cache[latest_trading_date_str] = \ 

116 self._market_clock.get_market_close_time(target_dt=latest_trading_date_dt) 

117 market_close = self._market_close_cache[latest_trading_date_str] 

118 if cache_time >= market_close: 

119 is_valid = True 

120 else: 

121 self._logger.debug( 

122 f"📉 캐시 만료 (거래일 {latest_trading_date_str} 장 마감 전 저장: {cache_time})" 

123 ) 

124 else: 

125 self._logger.debug(f"📉 캐시 만료 (최근 거래일 {latest_trading_date_str} > 캐시 데이터 {cache_date_str})") 

126 else: 

127 # mcs이 거래일을 확인할 수 없는 경우 캐시를 유효한 것으로 간주 

128 is_valid = True 

129 

130 if is_valid: 

131 if cache_type == "memory": 131 ↛ 134line 131 didn't jump to line 134 because the condition on line 131 was always true

132 if self._cache.memory_cache and self._cache.memory_cache.has(key): 132 ↛ 137line 132 didn't jump to line 137 because the condition on line 132 was always true

133 self._logger.debug(f"🧠 Memory Cache HIT (유효): {key}") 

134 elif cache_type == "file": 

135 if self._cache.file_cache and self._cache.file_cache.exists(key): 

136 self._logger.debug(f"📂 File Cache HIT (유효): {key}") 

137 cached_result = wrapper.get("data") 

138 try: 

139 if cached_result is not None: 139 ↛ 143line 139 didn't jump to line 143 because the condition on line 139 was always true

140 cached_result._cache_hit = True 

141 except AttributeError: 

142 pass # dict 등 속성 설정 불가 타입 

143 return cached_result 

144 else: 

145 if self._cache.file_cache and self._cache.file_cache.exists(key): 

146 self._logger.debug(f"📂 File Cache 무시 (만료됨): {key} / 저장 시각: {cache_time}") 

147 self._cache.file_cache.delete(key) 

148 if self._cache.memory_cache: 

149 self._cache.memory_cache.delete(key) 

150 

151 # ✅ 2. In-flight 중복 방지 (같은 키에 동시 요청 시 첫 번째 결과 공유) 

152 if key in self._in_flight: 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true

153 self._logger.debug(f"⏳ In-flight HIT (중복 요청 방지): {key}") 

154 return await asyncio.shield(self._in_flight[key]) 

155 

156 loop = asyncio.get_event_loop() 

157 future = loop.create_future() 

158 self._in_flight[key] = future 

159 

160 # ✅ 3. API 호출 

161 self._logger.debug(f"🌐 실시간 API 호출: {key}") 

162 try: 

163 result = await orig_attr(*args, **kwargs) 

164 if not future.done(): 164 ↛ 171line 164 didn't jump to line 171 because the condition on line 164 was always true

165 future.set_result(result) 

166 except Exception as exc: 

167 if not future.done(): 

168 future.set_exception(exc) 

169 raise 

170 finally: 

171 self._in_flight.pop(key, None) 

172 

173 if isinstance(result, ResCommonResponse) and result.rt_cd == ErrorCode.SUCCESS.value: 

174 # ✅ 4. 캐싱 데이터 저장 

175 self._cache.set(key, { 

176 "data": result, 

177 "timestamp": datetime.now().isoformat() 

178 }, save_to_file=True) 

179 else: 

180 self._logger.debug(f"응답 실패로 🧠📂 Cache Update 무시 : {key}") 

181 

182 return result 

183 

184 return wrapped 

185 

186 def __dir__(self): 

187 # 포함해야 할 속성 목록: 

188 # 1. self._client의 속성 

189 # 2. self 객체의 __dict__ 속성 

190 # 3. 클래스 자체의 속성 

191 return list(set( 

192 dir(self._client) + 

193 list(self.__dict__.keys()) + 

194 dir(type(self)) 

195 )) 

196 

197 def _parse_timestamp(self, timestamp_str: str) -> Optional[datetime]: 

198 if not timestamp_str: 

199 return None 

200 try: 

201 dt = datetime.fromisoformat(timestamp_str) 

202 if dt.tzinfo is None: 202 ↛ 204line 202 didn't jump to line 204 because the condition on line 202 was always true

203 return self._market_clock.market_timezone.localize(dt) 

204 return dt 

205 except Exception as e: 

206 if self._logger: 206 ↛ 208line 206 didn't jump to line 208 because the condition on line 206 was always true

207 self._logger.warning(f"[CacheWrapper] 잘못된 timestamp 포맷: {timestamp_str} ({e})") 

208 return None 

209 

210 

211def cache_wrap_client( 

212 api_client: T, 

213 logger, 

214 market_clock, 

215 mode_getter: Callable[[], str], 

216 config: Optional[dict] = None, 

217 cache_store: Optional[CacheStore] = None, 

218 market_calendar_service: Optional[Any] = None 

219) -> T: 

220 return ClientWithCache( 

221 client=api_client, 

222 logger=logger, 

223 market_clock=market_clock, 

224 mode_fn=mode_getter, 

225 cache_store=cache_store, 

226 config=config, 

227 market_calendar_service=market_calendar_service 

228 )