Coverage for cli/config.py: 98.90%
135 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +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 dataclasses import fields as dataclass_fields
21from pathlib import Path
22from typing import Any
24import yaml
26logger = logging.getLogger(__name__)
29def _load_cdk_json() -> dict[str, Any]:
30 """Load deployment_regions from cdk.json if present."""
31 cdk_json_path = Path.cwd() / "cdk.json"
32 if cdk_json_path.exists():
33 try:
34 with open(cdk_json_path, encoding="utf-8") as f:
35 data = json.load(f)
36 result = data.get("context", {}).get("deployment_regions", {})
37 if isinstance(result, dict):
38 return result
39 except Exception as e:
40 logger.debug("Failed to load cdk.json: %s", e)
41 return {}
44def _load_cdk_project_name() -> str | None:
45 """Load ``context.project_name`` from cdk.json if present (#139).
47 The CDK reads the deployment's identity from ``context.project_name``
48 (see ``gco/config/config_loader.ConfigLoader.get_project_name``). The CLI
49 must resolve the same value so it addresses the right project-scoped
50 resources — otherwise a non-``gco`` deployment's stacks, EKS clusters, and
51 DynamoDB tables are unreachable from the CLI.
52 """
53 cdk_json_path = Path.cwd() / "cdk.json"
54 if cdk_json_path.exists():
55 try:
56 with open(cdk_json_path, encoding="utf-8") as f:
57 data = json.load(f)
58 value = data.get("context", {}).get("project_name")
59 if isinstance(value, str) and value:
60 return value
61 except Exception as e:
62 logger.debug("Failed to load project_name from cdk.json: %s", e)
63 return None
66@dataclass
67class GCOConfig:
68 """Configuration for GCO CLI."""
70 # Project settings
71 project_name: str = "gco"
73 # AWS settings - defaults can be overridden by cdk.json or env vars
74 default_region: str = "us-east-1"
75 api_gateway_region: str = "us-east-2"
76 global_region: str = "us-east-2"
77 monitoring_region: str = "us-east-2"
79 # Stack naming
80 global_stack_name: str = "gco-global"
81 api_gateway_stack_name: str = "gco-api-gateway"
82 regional_stack_prefix: str = "gco"
84 # Default namespace for namespaced workload resources
85 default_namespace: str = "gco-jobs"
87 # Capacity checking
88 spot_price_history_days: int = 7
89 capacity_check_timeout: int = 30
91 # File system settings
92 efs_mount_path: str = "/mnt/gco"
93 fsx_mount_path: str = "/mnt/fsx"
95 # Output settings
96 output_format: str = "table" # table, json, yaml
97 verbose: bool = False
99 # Cache settings
100 cache_dir: str = field(default_factory=lambda: str(Path.home() / ".gco" / "cache"))
101 cache_ttl_seconds: int = 300 # 5 minutes
103 # API access mode
104 use_regional_api: bool = False # Use regional APIs for private access
106 # Tracks fields explicitly supplied by a file/environment source so a
107 # value equal to the dataclass default can still override an earlier source.
108 _specified_fields: set[str] = field(default_factory=set, init=False, repr=False)
110 def __post_init__(self) -> None:
111 # The stack names and regional prefix always derive from
112 # project_name (#139) so a non-"gco" deployment addresses its own
113 # stacks/clusters. get_config() re-applies this after merging the
114 # cdk.json / file / env overrides.
115 self._apply_project_scoped_names()
117 def _apply_project_scoped_names(self) -> None:
118 """Derive project-scoped stack names from ``project_name`` (#139).
120 The global/api-gateway stack names and the regional-stack prefix are
121 not independent knobs — they are always ``<project_name>-global``,
122 ``<project_name>-api-gateway``, and ``<project_name>`` respectively, to
123 match what the CDK deploys. For the default ``gco`` this yields the
124 identical ``gco-*`` names.
125 """
126 self.global_stack_name = f"{self.project_name}-global"
127 self.api_gateway_stack_name = f"{self.project_name}-api-gateway"
128 self.regional_stack_prefix = self.project_name
130 @classmethod
131 def from_file(cls, config_path: str | None = None) -> GCOConfig:
132 """Load configuration from a file, or defaults if no default file exists."""
133 explicit_path = config_path is not None
134 if config_path is None:
135 default_paths = [
136 Path.cwd() / ".gco.yaml",
137 Path.cwd() / ".gco.json",
138 Path.home() / ".gco" / "config.yaml",
139 Path.home() / ".gco" / "config.json",
140 ]
141 config_path = next((str(path) for path in default_paths if path.exists()), None)
143 if config_path is None:
144 return cls()
146 path = Path(config_path).expanduser()
147 if not path.exists():
148 if explicit_path:
149 raise FileNotFoundError(f"Configuration file not found: {path}")
150 return cls()
152 with open(path, encoding="utf-8") as f:
153 data = json.load(f) if path.suffix.lower() == ".json" else yaml.safe_load(f)
155 # ``yaml.safe_load`` returns None for an empty document.
156 if data is None:
157 data = {}
158 if not isinstance(data, dict): 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true
159 raise ValueError(f"Configuration file must contain a mapping: {path}")
161 valid_fields = {
162 item.name
163 for item in dataclass_fields(cls)
164 if item.init and not item.name.startswith("_")
165 }
166 values = {key: value for key, value in data.items() if key in valid_fields}
167 config = cls(**values)
168 config._specified_fields = set(values)
169 return config
171 @classmethod
172 def from_env(cls) -> GCOConfig:
173 """Load configuration from environment variables."""
174 config = cls()
176 env_mappings = {
177 "GCO_PROJECT_NAME": "project_name",
178 "GCO_DEFAULT_REGION": "default_region",
179 "GCO_API_GATEWAY_REGION": "api_gateway_region",
180 "GCO_GLOBAL_REGION": "global_region",
181 "GCO_MONITORING_REGION": "monitoring_region",
182 "GCO_DEFAULT_NAMESPACE": "default_namespace",
183 "GCO_OUTPUT_FORMAT": "output_format",
184 "GCO_VERBOSE": "verbose",
185 "GCO_CACHE_DIR": "cache_dir",
186 "GCO_REGIONAL_API": "use_regional_api",
187 }
189 for env_var, attr in env_mappings.items():
190 value: Any = os.environ.get(env_var)
191 if value is not None:
192 if attr in {"verbose", "use_regional_api"}:
193 setattr(config, attr, value.lower() in ("true", "1", "yes"))
194 else:
195 setattr(config, attr, value)
196 config._specified_fields.add(attr)
198 return config
200 def to_dict(self) -> dict[str, Any]:
201 """Convert configuration to dictionary."""
202 return {
203 "project_name": self.project_name,
204 "default_region": self.default_region,
205 "api_gateway_region": self.api_gateway_region,
206 "global_region": self.global_region,
207 "monitoring_region": self.monitoring_region,
208 "global_stack_name": self.global_stack_name,
209 "api_gateway_stack_name": self.api_gateway_stack_name,
210 "regional_stack_prefix": self.regional_stack_prefix,
211 "default_namespace": self.default_namespace,
212 "spot_price_history_days": self.spot_price_history_days,
213 "capacity_check_timeout": self.capacity_check_timeout,
214 "efs_mount_path": self.efs_mount_path,
215 "fsx_mount_path": self.fsx_mount_path,
216 "output_format": self.output_format,
217 "verbose": self.verbose,
218 "cache_dir": self.cache_dir,
219 "cache_ttl_seconds": self.cache_ttl_seconds,
220 "use_regional_api": self.use_regional_api,
221 }
223 def save(self, config_path: str | None = None) -> None:
224 """Save configuration to file."""
225 if config_path is None:
226 config_dir = Path.home() / ".gco"
227 config_dir.mkdir(parents=True, exist_ok=True)
228 config_path = str(config_dir / "config.yaml")
230 with open(config_path, "w", encoding="utf-8") as f:
231 yaml.dump(self.to_dict(), f, default_flow_style=False)
234def get_config(config_path: str | None = None) -> GCOConfig:
235 """Get merged configuration from cdk.json, file, and environment.
237 Configuration is loaded in this order (later sources override earlier):
238 1. Default values
239 2. cdk.json deployment regions and project name (if present)
240 3. ``config_path`` when supplied, otherwise the first default config file
241 4. Environment variables (GCO_*)
242 """
243 # Start with defaults
244 config = GCOConfig()
246 # cdk.json is the CDK's source of truth for project_name (#139). Load it
247 # first so file/env can still override per the documented precedence.
248 cdk_project = _load_cdk_project_name()
249 if cdk_project:
250 config.project_name = cdk_project
252 # Load from cdk.json if present
253 cdk_regions = _load_cdk_json()
254 if cdk_regions:
255 if "api_gateway" in cdk_regions:
256 config.api_gateway_region = cdk_regions["api_gateway"]
257 if "global" in cdk_regions:
258 config.global_region = cdk_regions["global"]
259 if "monitoring" in cdk_regions:
260 config.monitoring_region = cdk_regions["monitoring"]
261 if cdk_regions.get("regional"):
262 config.default_region = cdk_regions["regional"][0]
264 # Merge only fields explicitly supplied by real file/env loaders. For
265 # callers/tests that construct a GCOConfig directly, retain the historical
266 # non-default inference as a compatibility fallback.
267 derived_fields = {"global_stack_name", "api_gateway_stack_name", "regional_stack_prefix"}
268 mergeable_fields = [
269 item.name
270 for item in dataclass_fields(GCOConfig)
271 if item.init and item.name not in derived_fields and not item.name.startswith("_")
272 ]
273 defaults = GCOConfig()
275 def apply_overrides(source: GCOConfig) -> None:
276 specified = set(source._specified_fields)
277 if not specified:
278 specified = {
279 attr
280 for attr in mergeable_fields
281 if getattr(source, attr) != getattr(defaults, attr)
282 }
283 for attr in mergeable_fields:
284 if attr in specified:
285 setattr(config, attr, getattr(source, attr))
287 apply_overrides(GCOConfig.from_file(config_path))
288 apply_overrides(GCOConfig.from_env())
290 # Stack names / regional prefix always track the final project_name (#139),
291 # regardless of which source set it.
292 config._apply_project_scoped_names()
294 return config