feat(config): add YAML config and graceful pagination timeout
fixes #17, fixes #18 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
16
CHANGELOG.md
16
CHANGELOG.md
@@ -6,6 +6,22 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/).
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
### Added
|
||||
|
||||
- Support de fichier de configuration YAML (`.gitea-dashboard.yml` ou `~/.config/gitea-dashboard/config.yml`)
|
||||
- Option `--config` pour specifier un fichier de configuration alternatif
|
||||
- Resolution des variables `${VAR}` dans les fichiers de configuration
|
||||
- Priorite de configuration : CLI > variables d'environnement > fichier config > defauts
|
||||
|
||||
### Fixed
|
||||
|
||||
- Degradation gracieuse sur timeout reseau pendant la pagination (retourne les donnees partielles au lieu de crasher)
|
||||
|
||||
### Technical
|
||||
|
||||
- Nouveau module `config.py` pour la gestion de configuration YAML (ADR-013)
|
||||
- Nouvelle dependance PyYAML >= 6.0
|
||||
|
||||
## [1.3.0] - 2026-03-12
|
||||
|
||||
### Added
|
||||
|
||||
@@ -10,6 +10,7 @@ requires-python = ">=3.10"
|
||||
dependencies = [
|
||||
"requests>=2.31",
|
||||
"rich>=13.0",
|
||||
"pyyaml>=6.0",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
|
||||
@@ -11,6 +11,7 @@ from rich.console import Console
|
||||
|
||||
from gitea_dashboard.client import GiteaClient
|
||||
from gitea_dashboard.collector import collect_all
|
||||
from gitea_dashboard.config import load_config, merge_config
|
||||
from gitea_dashboard.display import render_dashboard, sort_repos
|
||||
from gitea_dashboard.exporter import export_json
|
||||
|
||||
@@ -71,9 +72,63 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
default=False,
|
||||
help="Masque la colonne Description dans le tableau.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--config",
|
||||
default=None,
|
||||
help="Chemin vers un fichier de configuration YAML alternatif.",
|
||||
)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def _resolve_config(args: argparse.Namespace) -> argparse.Namespace:
|
||||
"""Resout la configuration en appliquant la priorite CLI > env > config > defaults.
|
||||
|
||||
1. Charge le fichier config (args.config ou chemin par defaut)
|
||||
2. Lit les variables d'environnement pertinentes
|
||||
3. Fusionne avec les valeurs CLI
|
||||
4. Retourne un Namespace enrichi
|
||||
"""
|
||||
file_config = load_config(args.config)
|
||||
|
||||
env_vars: dict = {}
|
||||
gitea_url_env = os.environ.get("GITEA_URL")
|
||||
if gitea_url_env:
|
||||
env_vars["url"] = gitea_url_env
|
||||
gitea_auth_env = os.environ.get("GITEA_TOKEN")
|
||||
if gitea_auth_env:
|
||||
env_vars["auth"] = gitea_auth_env
|
||||
|
||||
cli_args: dict = {}
|
||||
if args.sort != "name":
|
||||
cli_args["sort"] = args.sort
|
||||
if args.repo is not None:
|
||||
cli_args["include"] = args.repo
|
||||
if args.exclude is not None:
|
||||
cli_args["exclude"] = args.exclude
|
||||
if args.no_desc:
|
||||
cli_args["no_desc"] = True
|
||||
|
||||
defaults = {
|
||||
"url": _DEFAULT_URL,
|
||||
"sort": "name",
|
||||
"no_desc": False,
|
||||
}
|
||||
|
||||
merged = merge_config(cli_args, env_vars, file_config, defaults)
|
||||
|
||||
args.resolved_url = merged.get("url", _DEFAULT_URL)
|
||||
args.resolved_auth = merged.get("auth")
|
||||
args.sort = merged.get("sort", args.sort)
|
||||
if merged.get("include") and args.repo is None:
|
||||
args.repo = merged["include"]
|
||||
if merged.get("exclude") and args.exclude is None:
|
||||
args.exclude = merged["exclude"]
|
||||
if merged.get("no_desc") and not args.no_desc:
|
||||
args.no_desc = merged["no_desc"]
|
||||
|
||||
return args
|
||||
|
||||
|
||||
def _run_health_check(client: GiteaClient, console: Console) -> None:
|
||||
"""Execute le health check et affiche les resultats.
|
||||
|
||||
@@ -104,16 +159,28 @@ def main(argv: list[str] | None = None) -> None:
|
||||
args = parse_args(argv)
|
||||
console = Console(stderr=True)
|
||||
|
||||
token = os.environ.get("GITEA_TOKEN")
|
||||
if not token:
|
||||
try:
|
||||
args = _resolve_config(args)
|
||||
except (FileNotFoundError, ValueError) as exc:
|
||||
console.print(f"[red]Erreur config : {exc}[/red]")
|
||||
sys.exit(1)
|
||||
|
||||
auth = args.resolved_auth if hasattr(args, "resolved_auth") and args.resolved_auth else None
|
||||
if not auth:
|
||||
auth = os.environ.get("GITEA_TOKEN")
|
||||
if not auth:
|
||||
console.print(
|
||||
"[red]Erreur : GITEA_TOKEN non defini. Exportez la variable d'environnement.[/red]"
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
url = os.environ.get("GITEA_URL", _DEFAULT_URL)
|
||||
url = (
|
||||
args.resolved_url
|
||||
if hasattr(args, "resolved_url")
|
||||
else os.environ.get("GITEA_URL", _DEFAULT_URL)
|
||||
)
|
||||
|
||||
client = GiteaClient(url, token)
|
||||
client = GiteaClient(url, auth)
|
||||
|
||||
try:
|
||||
if args.health:
|
||||
@@ -132,8 +199,8 @@ def main(argv: list[str] | None = None) -> None:
|
||||
except requests.RequestException as exc:
|
||||
# Ne jamais afficher le token dans les messages d'erreur
|
||||
msg = str(exc)
|
||||
if token in msg:
|
||||
msg = msg.replace(token, "***")
|
||||
if auth in msg:
|
||||
msg = msg.replace(auth, "***")
|
||||
console.print(f"[red]Erreur API : {msg}[/red]")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
import warnings
|
||||
|
||||
import requests
|
||||
|
||||
@@ -86,6 +87,11 @@ class GiteaClient:
|
||||
|
||||
Boucle tant que len(page) == limit (50).
|
||||
Utilise _get_with_retry pour la resilience aux timeouts.
|
||||
|
||||
Si _get_with_retry leve une exception Timeout sur une page
|
||||
intermediaire (page > 1), retourne les donnees collectees
|
||||
jusque-la et emet un warning via warnings.warn().
|
||||
Si la premiere page echoue, l'exception remonte normalement.
|
||||
"""
|
||||
all_items: list[dict] = []
|
||||
page = 1
|
||||
@@ -95,7 +101,16 @@ class GiteaClient:
|
||||
merged_params["limit"] = self._PAGE_LIMIT
|
||||
merged_params["page"] = page
|
||||
url = f"{self.base_url}{endpoint}"
|
||||
resp = self._get_with_retry(url, params=merged_params)
|
||||
try:
|
||||
resp = self._get_with_retry(url, params=merged_params)
|
||||
except (requests.exceptions.ReadTimeout, requests.exceptions.ConnectTimeout):
|
||||
if page == 1:
|
||||
raise
|
||||
warnings.warn(
|
||||
f"Partial data: timeout on page {page} of {endpoint}",
|
||||
stacklevel=2,
|
||||
)
|
||||
return all_items
|
||||
resp.raise_for_status()
|
||||
items = resp.json()
|
||||
all_items.extend(items)
|
||||
|
||||
112
src/gitea_dashboard/config.py
Normal file
112
src/gitea_dashboard/config.py
Normal file
@@ -0,0 +1,112 @@
|
||||
"""Configuration YAML pour gitea-dashboard."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
_DEFAULT_CONFIG_PATHS = [
|
||||
Path(".gitea-dashboard.yml"),
|
||||
Path.home() / ".config" / "gitea-dashboard" / "config.yml",
|
||||
]
|
||||
|
||||
_ENV_VAR_RE = re.compile(r"\$\{([^}]+)\}")
|
||||
|
||||
|
||||
def resolve_env_vars(value: str) -> str:
|
||||
"""Resout les references ${VAR} dans une valeur string.
|
||||
|
||||
Remplace ${VAR} par os.environ[VAR].
|
||||
Si VAR n'est pas defini, laisse la reference telle quelle.
|
||||
Ne resout pas les references imbriquees.
|
||||
"""
|
||||
|
||||
def _replace(match: re.Match) -> str:
|
||||
var_name = match.group(1)
|
||||
return os.environ.get(var_name, match.group(0))
|
||||
|
||||
return _ENV_VAR_RE.sub(_replace, value)
|
||||
|
||||
|
||||
def _resolve_values(data: Any) -> Any:
|
||||
"""Resout recursivement les ${VAR} dans les valeurs string et listes."""
|
||||
if isinstance(data, str):
|
||||
return resolve_env_vars(data)
|
||||
if isinstance(data, list):
|
||||
return [_resolve_values(item) for item in data]
|
||||
if isinstance(data, dict):
|
||||
return {k: _resolve_values(v) for k, v in data.items()}
|
||||
return data
|
||||
|
||||
|
||||
def load_config(config_path: str | None = None) -> dict[str, Any]:
|
||||
"""Charge la configuration depuis un fichier YAML.
|
||||
|
||||
Ordre de recherche si config_path est None :
|
||||
1. .gitea-dashboard.yml (repertoire courant)
|
||||
2. ~/.config/gitea-dashboard/config.yml
|
||||
|
||||
Retourne un dict vide si aucun fichier trouve.
|
||||
Leve FileNotFoundError si config_path est fourni mais n'existe pas.
|
||||
Leve ValueError si le YAML est syntaxiquement invalide.
|
||||
"""
|
||||
if config_path is not None:
|
||||
path = Path(config_path)
|
||||
if not path.exists():
|
||||
msg = f"Config file not found: {config_path}"
|
||||
raise FileNotFoundError(msg)
|
||||
return _load_yaml_file(path)
|
||||
|
||||
for path in _DEFAULT_CONFIG_PATHS:
|
||||
if path.exists():
|
||||
return _load_yaml_file(path)
|
||||
|
||||
return {}
|
||||
|
||||
|
||||
def _load_yaml_file(path: Path) -> dict[str, Any]:
|
||||
"""Charge et parse un fichier YAML avec resolution des variables."""
|
||||
try:
|
||||
raw = yaml.safe_load(path.read_text(encoding="utf-8"))
|
||||
except yaml.YAMLError as exc:
|
||||
msg = f"Invalid YAML in {path}: {exc}"
|
||||
raise ValueError(msg) from exc
|
||||
|
||||
if raw is None:
|
||||
return {}
|
||||
if not isinstance(raw, dict):
|
||||
msg = f"Invalid config format in {path}: expected a mapping, got {type(raw).__name__}"
|
||||
raise ValueError(msg)
|
||||
|
||||
return _resolve_values(raw)
|
||||
|
||||
|
||||
def merge_config(
|
||||
cli_args: dict[str, Any],
|
||||
env_vars: dict[str, Any],
|
||||
file_config: dict[str, Any],
|
||||
defaults: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
"""Fusionne les sources de configuration par priorite.
|
||||
|
||||
Priorite : cli_args > env_vars > file_config > defaults.
|
||||
Une valeur None dans une source de priorite superieure ne masque pas
|
||||
la valeur d'une source de priorite inferieure.
|
||||
"""
|
||||
all_keys = set()
|
||||
for source in (cli_args, env_vars, file_config, defaults):
|
||||
all_keys.update(source.keys())
|
||||
|
||||
result: dict[str, Any] = {}
|
||||
for key in all_keys:
|
||||
for source in (cli_args, env_vars, file_config, defaults):
|
||||
value = source.get(key)
|
||||
if value is not None:
|
||||
result[key] = value
|
||||
break
|
||||
|
||||
return result
|
||||
Reference in New Issue
Block a user