Coverage for config / config_loader.py: 100%

1# core/config_loader.py 

2import yaml 

3import os 

4import json 

5from typing import Dict, Any, Optional, List 

6from pydantic import BaseModel, Field, ValidationError, field_validator 

7 

8# config.yaml 및 tr_ids_config.yaml 파일 경로 설정 

9BASE_DIR = os.path.dirname(os.path.abspath(__file__)) 

10MAIN_CONFIG_PATH = os.path.join(BASE_DIR, 'config.yaml') 

11TR_IDS_CONFIG_PATH = os.path.join(BASE_DIR, 'tr_ids_config.yaml') 

12KIS_CONFIG_PATH = os.path.join(BASE_DIR, 'kis_config.yaml') 

13 

14 

15class WebConfig(BaseModel): 

16 host: str 

17 port: int = Field(..., ge=1, le=65535, description="웹 서버 포트 (1~65535)") 

18 

19class CacheConfig(BaseModel): 

20 base_dir: str = ".cache" 

21 enabled_methods: List[str] = Field(default_factory=list) 

22 deserializable_classes: List[str] = Field(default_factory=list) 

23 memory_cache_enabled: bool = True 

24 file_cache_enabled: bool = True 

25 

26class AppConfig(BaseModel): 

27 # Core API keys 

28 api_key: Optional[str] = None 

29 api_secret_key: Optional[str] = None 

30 base_url: Optional[str] = None 

31 websocket_url: Optional[str] = None 

32 

33 # Account info 

34 custtype: str = "P" 

35 stock_account_number: Optional[str] = None 

36 htsid: Optional[str] = None 

37 

38 # Flags 

39 is_paper_trading: bool = False 

40 

41 # Sub-configs 

42 web: WebConfig 

43 cache: CacheConfig = Field(default_factory=CacheConfig) 

44 

45 # Dynamic/Merged configs 

46 tr_ids: Dict[str, Any] = Field(default_factory=dict) 

47 paths: Dict[str, str] = Field(default_factory=dict) 

48 

49 # ✅ 필드 추가 (기본값 False 설정) 

50 performance_logging: bool = False 

51 performance_threshold: float = 0.1 

52 

53 # Extra fields for anything else in config.yaml 

54 model_config = {"extra": "allow"} 

55 

56 @field_validator('base_url') 

57 @classmethod 

58 def validate_base_url(cls, v: Optional[str]) -> Optional[str]: 

59 if v and not (v.startswith("http://") or v.startswith("https://")): 

60 raise ValueError("base_url은 'http://' 또는 'https://'로 시작해야 합니다.") 

61 return v 

62 

63 def __getitem__(self, item): 

64 return getattr(self, item) 

65 

66 def get(self, item, default=None): 

67 return getattr(self, item, default) 

68 

69def load_configs() -> AppConfig: 

70 main_config_data = load_config(MAIN_CONFIG_PATH) or {} 

71 tr_ids_data = load_config(TR_IDS_CONFIG_PATH) or {} 

72 kis_config_data = load_config(KIS_CONFIG_PATH) or {} 

73 

74 config_data = {} 

75 config_data.update(main_config_data) 

76 config_data.update(tr_ids_data) 

77 config_data.update(kis_config_data) 

78 

79 try: 

80 return AppConfig(**config_data) 

81 except ValidationError as e: 

82 raise ValueError(f"설정 파일 유효성 검사 실패: {e}") 

83 

84 

85def load_config(file_path): 

86 """지정된 경로에서 YAML 또는 JSON 설정 파일을 로드합니다.""" 

87 try: 

88 with open(file_path, 'r', encoding='utf-8') as f: 

89 try: 

90 return yaml.safe_load(f) 

91 except ImportError: 

92 f.seek(0) 

93 return json.load(f) 

94 except FileNotFoundError: 

95 raise FileNotFoundError(f"파일을 찾을 수 없습니다: {file_path}") 

96 except (json.JSONDecodeError, yaml.YAMLError) as e: 

97 raise ValueError(f"설정 파일 형식이 올바르지 않습니다 ({file_path}): {e}")