1"""
2Settings layer implementation.
3
4Implements :any:`/specs/spec_settings` §1-4, covering:
5
6* Sources & precedence (§1)
7* Supported settings keys (§2)
8* Parsing rules (§3)
9* Runtime behavior (§4)
10"""
11
12from __future__ import annotations
13
14import json
15import os
16from functools import lru_cache
17from pathlib import Path
18from typing import List
19import threading
20
21from pydantic import BaseModel, Field, field_validator, model_validator, ValidationError, PrivateAttr # type: ignore
22from pydantic_settings import BaseSettings, SettingsConfigDict, EnvSettingsSource # type: ignore
23from ..utils.logging import get_logger
24import yaml
25
26ROOT_DIR = Path(__file__).resolve().parent.parent.parent
27
28_singleton_instance = None # module-level singleton instance
29_singleton_lock = threading.Lock() # module-level lock
30
31
32def _preprocess_env_var(val):
33 # Accepts either a JSON array or a plain string, returns JSON array string
34 if val is None:
35 return None
36 val = val.strip()
37 if val.startswith("["):
38 return val # already JSON array
39 # treat as single string or comma-separated
40 try:
41 # Try comma-separated
42 arr = [v.strip() for v in val.split(",") if v.strip()]
43 return json.dumps(arr)
44 except Exception:
45 return json.dumps([val])
46
47
[docs]
48class CustomEnvSettingsSource(EnvSettingsSource):
[docs]
49 def prepare_field_value(self, field_name, field, value, value_is_complex):
50 # Only preprocess for our special fields
51 if field_name in ("group_api_key", "todo_keywords") and isinstance(value, str):
52 value = _preprocess_env_var(value)
53 return super().prepare_field_value(field_name, field, value, value_is_complex)
54
55
[docs]
56class Settings(BaseSettings):
57 """Central application settings.
58
59 Priority order – highest first:
60 1. Manual overrides via :func:`override` (used from CLI flags)
61 2. Environment variables
62 3. Defaults defined here
63
64 Environment variable names mirror the legacy ones so existing setups keep
65 working.
66 """
67
68 # ---- Core -----------------------------------------------------------------
69 gitlab_host: str = Field(
70 ..., alias="GITLAB_HOST", description="Base URL of GitLab instance"
71 )
72 group_api_key: List[str] = Field(
73 ..., alias="GROUP_API_KEY", description="Exactly one key or a JSON-List of keys"
74 )
75
76 # ---- Behaviour -----------------------------------------------------------
77 debug: bool = Field(False, alias="DEBUG", description="Enables verbose logging")
78 display_shared: bool = Field(
79 False,
80 alias="DISPLAY_SHARED",
81 description="Includes shared projects in output when true",
82 )
83 todo_keywords: List[str] = Field(
84 default=["status", "todo", "stand", "fortschritt", "offen", "geplant"],
85 alias="TODO_KEYWORDS",
86 description="JSON-List with strings of keywords",
87 )
88
89 # ---- Network -----------------------------------------------------------
90 api_retry_count: int = Field(
91 3, alias="API_RETRY_COUNT", description="Max retries for API client"
92 )
93 api_retry_backoff: float = Field(
94 1.0,
95 alias="API_RETRY_BACKOFF",
96 description="Seconds added per retry attempt (linear)",
97 )
98
99 # ---- Runtime -------------------------------------------------------------
100 table_config: str = Field(
101 "table_config.yaml",
102 alias="TABLE_CONFIG",
103 description="Path to YAML file with column definitions",
104 )
105 log_level: str = Field("INFO", alias="LOG_LEVEL", description="Root logger level")
106
107 model_config = SettingsConfigDict(
108 env_file=".env",
109 extra="ignore",
110 env_nested_delimiter="__",
111 populate_by_name=True,
112 )
113
[docs]
114 @classmethod
115 def settings_customise_sources(
116 cls,
117 settings_cls,
118 init_settings,
119 env_settings,
120 dotenv_settings,
121 file_secret_settings,
122 ):
123 # Use our custom env source for env vars
124 return (
125 init_settings,
126 CustomEnvSettingsSource(settings_cls),
127 dotenv_settings,
128 file_secret_settings,
129 )
130
[docs]
131 @model_validator(mode="after")
132 def validate_required_fields(self):
133 """Validate that required fields are provided."""
134 if (
135 not self.gitlab_host
136 or not isinstance(self.gitlab_host, str)
137 or not self.gitlab_host.strip()
138 ):
139 raise ValueError("gitlab_host is required and must be a non-empty string")
140 if (
141 not self.group_api_key
142 or not isinstance(self.group_api_key, list)
143 or not self.group_api_key
144 ):
145 raise ValueError("group_api_key is required and must be a non-empty list")
146 return self
147
148 # ---------- Validators ----------------------------------------------------
149
[docs]
150 @field_validator("gitlab_host", mode="before")
151 @classmethod
152 def _validate_gitlab_host(cls, v):
153 """Validate gitlab_host is provided and is a valid URL."""
154 if v and isinstance(v, str):
155 return v.strip()
156 return v
157
[docs]
158 @field_validator("group_api_key", mode="before")
159 @classmethod
160 def _validate_group_api_key(cls, v):
161 """Validate group_api_key is provided and is a list of strings."""
162 if isinstance(v, str):
163 # This should have been preprocessed by CustomEnvSettingsSource
164 # but handle it here as well for safety
165 try:
166 parsed = json.loads(v)
167 if isinstance(parsed, list):
168 v = parsed
169 else:
170 v = [parsed]
171 except json.JSONDecodeError:
172 # Treat as single string
173 v = [v]
174 if isinstance(v, list):
175 return [str(item) for item in v if item]
176 return v
177
[docs]
178 @field_validator("debug", "display_shared", mode="before")
179 @classmethod
180 def _parse_boolean(cls, v): # noqa: D401
181 """Parse boolean values from various formats."""
182 if isinstance(v, bool):
183 return v
184 if isinstance(v, str):
185 v_lower = v.lower()
186 if v_lower in ("true", "1", "yes"):
187 return True
188 elif v_lower in ("false", "0", "no"):
189 return False
190 else:
191 raise ValueError(f"Invalid boolean value: {v}")
192 if isinstance(v, int):
193 return bool(v)
194 raise TypeError(f"Invalid type for boolean field: {type(v)}")
195
[docs]
196 @field_validator("api_retry_count", mode="before")
197 @classmethod
198 def _parse_retry_count(cls, v): # noqa: D401
199 """Parse and validate retry count."""
200 if isinstance(v, str):
201 try:
202 v = int(v)
203 except ValueError:
204 raise ValueError(f"Invalid integer value for api_retry_count: {v}")
205 if v < 0:
206 raise ValueError("api_retry_count must be non-negative")
207 return v
208
[docs]
209 @field_validator("api_retry_backoff", mode="before")
210 @classmethod
211 def _parse_retry_backoff(cls, v): # noqa: D401
212 """Parse and validate retry backoff."""
213 if isinstance(v, str):
214 try:
215 v = float(v)
216 except ValueError:
217 raise ValueError(f"Invalid float value for api_retry_backoff: {v}")
218 if v < 0:
219 raise ValueError("api_retry_backoff must be non-negative")
220 return v
221
[docs]
222 @field_validator("table_config", mode="before")
223 @classmethod
224 def _validate_table_config_path(cls, v): # noqa: D401
225 """Validate table_config points to a readable file."""
226 if v and not Path(v).is_file():
227 # Don't raise error for default value, just warn
228 import logging
229
230 logger = logging.getLogger(__name__)
231 logger.warning(f"table_config file not found: {v}")
232 return v
233
234 # ---- Public helpers ------------------------------------------------------
235
[docs]
236 @classmethod
237 def current(cls) -> "Settings":
238 """Return a cached singleton Settings instance.
239
240 A new instance is **only** created when no instance has been created yet (first call)
241 """
242 global _singleton_instance, _singleton_lock
243
244 # Store the snapshot on the class so we can compare later
245 if _singleton_instance is None:
246 with _singleton_lock:
247 # Re-check under the lock to avoid a race when multiple threads
248 # arrive here concurrently.
249 _singleton_instance = cls() # type: ignore[call-arg]
250 return _singleton_instance
251
[docs]
252 @classmethod
253 def set_singleton(cls, instance: Settings):
254 global _singleton_instance, _singleton_lock
255 if _singleton_instance is not None:
256 logger = get_logger("gitlab_overviewer.config.settings")
257 logger.critical(
258 "Attempted to set Settings singleton, but it is already set! This indicates a bug or double initialization."
259 )
260 return
261 with _singleton_lock:
262 _singleton_instance = instance
263
264 # Runtime overrides --------------------------------------------------------
265
[docs]
266 def override(self, ignore_env=False, **kwargs): # noqa: D401
267 """Return a new Settings instance with *kwargs* overriding fields.
268 If ignore_env is True, do not load from environment or .env, only use current values and overrides.
269 """
270 data = self.model_dump()
271 data.update(kwargs)
272 if ignore_env:
273 # Temporarily disable .env and unset env vars
274 env_to_restore = {}
275 for field in Settings.model_fields.keys():
276 alias = Settings.model_fields[field].alias or field.upper()
277 if alias in os.environ:
278 env_to_restore[alias] = os.environ[alias]
279 del os.environ[alias]
280 old_env_file = self.model_config.get("env_file")
281 self.model_config["env_file"] = None
282 try:
283 return Settings(**data)
284 finally:
285 for k, v in env_to_restore.items():
286 os.environ[k] = v
287 if old_env_file is not None:
288 self.model_config["env_file"] = old_env_file
289 else:
290 self.model_config.pop("env_file", None)
291 else:
292 return Settings(**data)
293
294 # ---------------------------------------------------------------------
295 # CLI integration – maps argparse.Namespace (or any dot-access object)
296 # ---------------------------------------------------------------------
297
[docs]
298 @classmethod
299 def from_args(cls, args, ignore_env=False) -> "Settings": # noqa: D401
300 """Create settings instance overriding values contained in *args*.
301
302 Only attributes that are **not None** in *args* are considered – this
303 mimics the behaviour of CLI flags having highest priority.
304
305 Environment variables are temporarily unset when CLI values override them
306 to ensure proper precedence.
307
308 Args:
309 args: Namespace object with CLI arguments
310 ignore_env: If True, ignore environment variables and .env file
311 """
312 import os
313 import json
314
315 # Helper for type conversion
316 def parse_cli_value(field, value):
317 if value is None:
318 return None
319 typ = cls.model_fields[field].annotation
320 # List fields: accept JSON or comma-separated
321 if typ == list or typ == List[str]:
322 if isinstance(value, list):
323 return value
324 if isinstance(value, str):
325 try:
326 parsed = json.loads(value)
327 if isinstance(parsed, list):
328 return parsed
329 except Exception:
330 # Fallback: comma-separated
331 return [v.strip() for v in value.split(",") if v.strip()]
332 # Bool fields
333 if typ == bool:
334 if isinstance(value, bool):
335 return value
336 if isinstance(value, str):
337 v = value.lower()
338 if v in ("true", "1", "yes"):
339 return True
340 if v in ("false", "0", "no"):
341 return False
342 if isinstance(value, int):
343 return bool(value)
344 # Int fields
345 if typ == int:
346 if isinstance(value, int):
347 return value
348 try:
349 return int(value)
350 except Exception:
351 pass
352 # Float fields
353 if typ == float:
354 if isinstance(value, float):
355 return value
356 try:
357 return float(value)
358 except Exception:
359 pass
360 # Default: return as is
361 return value
362
363 # Build merged dict and track env vars to unset
364 merged = {}
365 cli_env_to_unset = {}
366
367 for field in cls.model_fields.keys():
368 cli_val = getattr(args, field, None)
369 alias = cls.model_fields[field].alias or field.upper()
370 if cli_val is not None:
371 # CLI value takes precedence
372 merged[field] = parse_cli_value(field, cli_val)
373 # Unset env var for this field (by alias) to ensure precedence
374 if alias in os.environ:
375 cli_env_to_unset[alias] = os.environ[alias]
376 del os.environ[alias]
377 else:
378 # If no CLI value, do not set merged[field], let Pydantic handle env/default
379 pass
380
381 try:
382 if ignore_env:
383 # Create settings instance with only CLI values and defaults, no env
384 # Temporarily unset all relevant environment variables
385 env_to_restore = {}
386 for field in cls.model_fields.keys():
387 alias = cls.model_fields[field].alias or field.upper()
388 if alias in os.environ:
389 env_to_restore[alias] = os.environ[alias]
390 del os.environ[alias]
391
392 # Temporarily disable .env file loading
393 old_env_file = cls.model_config.get("env_file")
394 cls.model_config["env_file"] = None
395
396 try:
397 return cls(**merged)
398 finally:
399 # Restore environment variables
400 for k, v in env_to_restore.items():
401 os.environ[k] = v
402 # Restore .env file setting
403 if old_env_file is not None:
404 cls.model_config["env_file"] = old_env_file
405 else:
406 cls.model_config.pop("env_file", None)
407 else:
408 # Create settings instance - Pydantic will handle env/default precedence
409 # for fields not in merged
410 return cls(**merged)
411 finally:
412 # Restore any env vars we temporarily removed for CLI precedence
413 for k, v in cli_env_to_unset.items():
414 os.environ[k] = v
415
[docs]
416 def __init__(self, **values):
417 import traceback
418
419 super().__init__(**values)
420 for key, value in values.items():
421 if value is not None and key in self.__dict__:
422 self.__setattr__(key, value)
423 logger = get_logger("gitlab_overviewer.config.settings")
424 if logger.isEnabledFor(20): # INFO or lower
425 logger.info(f"gitlab_host: %r", self.gitlab_host)
426 logger.info(f"display_shared: %r", self.display_shared)
427 logger.info(f"todo_keywords: %r", self.todo_keywords)
428 logger.info(f"table_config path: %r", self.table_config)
429 # Try to load table_config columns and sort order
430 try:
431 with open(self.table_config, "r", encoding="utf-8") as f:
432 cfg = yaml.safe_load(f)
433 columns = [col.get("key") for col in cfg.get("columns", [])]
434 sort_order = cfg.get("default_sort", [])
435 logger.info(f"table_config columns: %s", columns)
436 logger.info(f"table_config sort order: %s", sort_order)
437 except Exception as e:
438 logger.warning(f"Could not load table_config for logging: {e}")