Coverage for cli / config.py: 97%
102 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-30 21:47 +0000
1"""
2CLI Configuration management for GCO.
4Handles configuration loading, caching, and validation for the CLI.
5Supports both file-based configuration and environment variables.
7Configuration is loaded in this order (later sources override earlier):
81. Default values
92. cdk.json (if present in current directory)
103. ~/.gco/config.yaml or config.json
114. Environment variables (GCO_*)
12"""
14from __future__ import annotations
16import json
17import logging
18import os
19from dataclasses import dataclass, field
20from pathlib import Path
21from typing import Any
23import yaml
25logger = logging.getLogger(__name__)
28def _load_cdk_json() -> dict[str, Any]:
29 """Load deployment_regions from cdk.json if present."""
30 cdk_json_path = Path.cwd() / "cdk.json"
31 if cdk_json_path.exists():
32 try:
33 with open(cdk_json_path, encoding="utf-8") as f:
34 data = json.load(f)
35 result = data.get("context", {}).get("deployment_regions", {})
36 if isinstance(result, dict):
37 return result
38 except Exception as e:
39 logger.debug("Failed to load cdk.json: %s", e)
40 return {}
43@dataclass
44class GCOConfig:
45 """Configuration for GCO CLI."""
47 # Project settings
48 project_name: str = "gco"
50 # AWS settings - defaults can be overridden by cdk.json or env vars
51 default_region: str = "us-east-1"
52 api_gateway_region: str = "us-east-2"
53 global_region: str = "us-east-2"
54 monitoring_region: str = "us-east-2"
56 # Stack naming
57 global_stack_name: str = "gco-global"
58 api_gateway_stack_name: str = "gco-api-gateway"
59 regional_stack_prefix: str = "gco"
61 # Default namespaces
62 default_namespace: str = "gco-jobs"
63 allowed_namespaces: list[str] = field(default_factory=lambda: ["default", "gco-jobs"])
65 # Capacity checking
66 spot_price_history_days: int = 7
67 capacity_check_timeout: int = 30
69 # File system settings
70 efs_mount_path: str = "/mnt/gco"
71 fsx_mount_path: str = "/mnt/fsx"
73 # Output settings
74 output_format: str = "table" # table, json, yaml
75 verbose: bool = False
77 # Cache settings
78 cache_dir: str = field(default_factory=lambda: str(Path.home() / ".gco" / "cache"))
79 cache_ttl_seconds: int = 300 # 5 minutes
81 # API access mode
82 use_regional_api: bool = False # Use regional APIs for private access
84 @classmethod
85 def from_file(cls, config_path: str | None = None) -> GCOConfig:
86 """Load configuration from file."""
87 if config_path is None:
88 # Check default locations
89 default_paths = [
90 Path.cwd() / ".gco.yaml",
91 Path.cwd() / ".gco.json",
92 Path.home() / ".gco" / "config.yaml",
93 Path.home() / ".gco" / "config.json",
94 ]
95 for path in default_paths:
96 if path.exists():
97 config_path = str(path)
98 break
100 if config_path and Path(config_path).exists():
101 with open(config_path, encoding="utf-8") as f:
102 data = json.load(f) if config_path.endswith(".json") else yaml.safe_load(f)
103 return cls(**{k: v for k, v in data.items() if hasattr(cls, k)})
105 return cls()
107 @classmethod
108 def from_env(cls) -> GCOConfig:
109 """Load configuration from environment variables."""
110 config = cls()
112 env_mappings = {
113 "GCO_PROJECT_NAME": "project_name",
114 "GCO_DEFAULT_REGION": "default_region",
115 "GCO_API_GATEWAY_REGION": "api_gateway_region",
116 "GCO_GLOBAL_REGION": "global_region",
117 "GCO_MONITORING_REGION": "monitoring_region",
118 "GCO_DEFAULT_NAMESPACE": "default_namespace",
119 "GCO_OUTPUT_FORMAT": "output_format",
120 "GCO_VERBOSE": "verbose",
121 "GCO_CACHE_DIR": "cache_dir",
122 }
124 for env_var, attr in env_mappings.items():
125 value: Any = os.environ.get(env_var)
126 if value is not None:
127 if attr == "verbose":
128 setattr(config, attr, value.lower() in ("true", "1", "yes"))
129 elif attr == "allowed_namespaces": 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true
130 setattr(config, attr, value.split(","))
131 else:
132 setattr(config, attr, value)
134 return config
136 def to_dict(self) -> dict[str, Any]:
137 """Convert configuration to dictionary."""
138 return {
139 "project_name": self.project_name,
140 "default_region": self.default_region,
141 "api_gateway_region": self.api_gateway_region,
142 "global_region": self.global_region,
143 "monitoring_region": self.monitoring_region,
144 "global_stack_name": self.global_stack_name,
145 "api_gateway_stack_name": self.api_gateway_stack_name,
146 "regional_stack_prefix": self.regional_stack_prefix,
147 "default_namespace": self.default_namespace,
148 "allowed_namespaces": self.allowed_namespaces,
149 "spot_price_history_days": self.spot_price_history_days,
150 "capacity_check_timeout": self.capacity_check_timeout,
151 "efs_mount_path": self.efs_mount_path,
152 "fsx_mount_path": self.fsx_mount_path,
153 "output_format": self.output_format,
154 "verbose": self.verbose,
155 "cache_dir": self.cache_dir,
156 "cache_ttl_seconds": self.cache_ttl_seconds,
157 "use_regional_api": self.use_regional_api,
158 }
160 def save(self, config_path: str | None = None) -> None:
161 """Save configuration to file."""
162 if config_path is None:
163 config_dir = Path.home() / ".gco"
164 config_dir.mkdir(parents=True, exist_ok=True)
165 config_path = str(config_dir / "config.yaml")
167 with open(config_path, "w", encoding="utf-8") as f:
168 yaml.dump(self.to_dict(), f, default_flow_style=False)
171def get_config() -> GCOConfig:
172 """Get merged configuration from cdk.json, file, and environment.
174 Configuration is loaded in this order (later sources override earlier):
175 1. Default values
176 2. cdk.json deployment_regions (if present)
177 3. ~/.gco/config.yaml or config.json
178 4. Environment variables (GCO_*)
179 """
180 # Start with defaults
181 config = GCOConfig()
183 # Load from cdk.json if present
184 cdk_regions = _load_cdk_json()
185 if cdk_regions:
186 if "api_gateway" in cdk_regions: 186 ↛ 188line 186 didn't jump to line 188 because the condition on line 186 was always true
187 config.api_gateway_region = cdk_regions["api_gateway"]
188 if "global" in cdk_regions:
189 config.global_region = cdk_regions["global"]
190 if "monitoring" in cdk_regions:
191 config.monitoring_region = cdk_regions["monitoring"]
192 if cdk_regions.get("regional"): 192 ↛ 196line 192 didn't jump to line 196 because the condition on line 192 was always true
193 config.default_region = cdk_regions["regional"][0]
195 # Override with file config
196 file_config = GCOConfig.from_file()
197 for attr in [
198 "project_name",
199 "default_region",
200 "api_gateway_region",
201 "global_region",
202 "monitoring_region",
203 "default_namespace",
204 "output_format",
205 "verbose",
206 "cache_dir",
207 ]:
208 file_value = getattr(file_config, attr)
209 default_value = getattr(GCOConfig(), attr)
210 if file_value != default_value:
211 setattr(config, attr, file_value)
213 # Override with environment variables
214 env_config = GCOConfig.from_env()
215 for attr in [
216 "project_name",
217 "default_region",
218 "api_gateway_region",
219 "global_region",
220 "monitoring_region",
221 "default_namespace",
222 "output_format",
223 "verbose",
224 "cache_dir",
225 ]:
226 env_value = getattr(env_config, attr)
227 default_value = getattr(GCOConfig(), attr)
228 if env_value != default_value:
229 setattr(config, attr, env_value)
231 return config