Coverage for brokers / korea_investment / korea_invest_url_provider.py: 100%

1# core/korea_invest_url_provider.py 

2from __future__ import annotations 

3from typing import Mapping, Iterable, Optional, Callable, Union 

4from enum import Enum 

5from urllib.parse import urljoin 

6from config.config_loader import load_configs, load_config, KIS_CONFIG_PATH 

7from brokers.korea_investment.korea_invest_env import KoreaInvestApiEnv 

8 

9 

10class KoreaInvestUrlProvider: 

11 """ 

12 kis_config.yaml(paths) + base_url을 바탕으로 엔드포인트 절대 URL을 만들어주는 경량 Provider. 

13 - 기본적으로 kis_config.yaml 만 읽어 경로 테이블(paths)을 구성 

14 - base_url은 생성자 인자로 주입 (권장)하거나, load_configs() 병합 결과에 존재하면 자동 사용 

15 """ 

16 def __init__(self, get_base_url: Callable[[], str], paths: Mapping[str, str]) -> None: 

17 self._get_base_url = get_base_url 

18 self._paths = dict(paths or {}) 

19 

20 @classmethod 

21 def from_env_and_kis_config( 

22 cls, env: KoreaInvestApiEnv, 

23 kis_config_override: Optional[dict] = None, 

24 get_base_url_override: Optional[Callable[[], str]] = None, 

25 ) -> "KoreaInvestUrlProvider": 

26 kis_conf = kis_config_override if kis_config_override is not None else load_config(KIS_CONFIG_PATH) 

27 paths = kis_conf.get("paths") 

28 if not isinstance(paths, dict) or not paths: 

29 raise ValueError("kis_config.yaml의 paths가 없거나 비었습니다.") 

30 get_base_url = get_base_url_override or env.get_base_url 

31 return cls(get_base_url=get_base_url, paths=paths) 

32 

33 

34 # ---------- 조회 ---------- 

35 def has(self, key: str) -> bool: 

36 return key in self._paths 

37 

38 def keys(self) -> Iterable[str]: 

39 return self._paths.keys() 

40 

41 def path(self, key: str) -> str: 

42 if key not in self._paths: 

43 raise KeyError(f"kis_config.paths에 '{key}'가 없습니다.") 

44 return self._paths[key] 

45 

46 def url(self, key_or_path: Union[str, Enum]) -> str: 

47 base = (self._get_base_url() or "").rstrip("/") 

48 if not base: 

49 raise ValueError("env.get_base_url()이 빈 값입니다. env.set_trading_mode(...)로 활성화했는지 확인하세요.") 

50 # Enum이면 .value 사용 

51 key_str = key_or_path.value if isinstance(key_or_path, Enum) else str(key_or_path) 

52 rel = self.path(key_str) if self.has(key_str) else key_str 

53 return urljoin(base + "/", rel.lstrip("/"))