01f88a0eca
- client: add raise_for_status() in get_latest_release() for non-404 errors (FINDING-001) - client: add timeout parameter (default 30s) passed to all session.get() calls (FINDING-004/SEC-002) - cli: replace return with sys.exit(1) in all except blocks (FINDING-002) - test_cli: remove duplicate test_exits_when_token_missing, assert GITEA_TOKEN in stderr (FINDING-006) - test_cli: update connection error tests to expect SystemExit(1) after exit code fix - test_cli: rework token masking test to inject token into exception message (FINDING-007) - test_client: add test_raises_on_server_error for HTTP 500 path (FINDING-001) - test_client: add tests for default and custom timeout values (FINDING-004) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
"""Client API Gitea en lecture seule."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import requests
|
|
|
|
|
|
class GiteaClient:
|
|
"""Client API Gitea en lecture seule.
|
|
|
|
Utilise requests.Session pour reutiliser les connexions HTTP.
|
|
Authentification via header Authorization: token <TOKEN>.
|
|
"""
|
|
|
|
_PAGE_LIMIT = 50
|
|
|
|
def __init__(self, base_url: str, token: str, timeout: int = 30) -> None:
|
|
"""Initialise le client avec l'URL de base et le token API.
|
|
|
|
Args:
|
|
base_url: URL de base de l'instance Gitea.
|
|
token: Token API pour l'authentification.
|
|
timeout: Delai maximum en secondes pour chaque requete (defaut: 30).
|
|
"""
|
|
self.base_url = base_url.rstrip("/")
|
|
self.timeout = timeout
|
|
self.session = requests.Session()
|
|
self.session.headers["Authorization"] = f"token {token}"
|
|
|
|
def _get_paginated(self, endpoint: str, params: dict | None = None) -> list[dict]:
|
|
"""Requete GET avec pagination automatique.
|
|
|
|
Boucle tant que len(page) == limit (50).
|
|
"""
|
|
all_items: list[dict] = []
|
|
page = 1
|
|
merged_params = dict(params) if params else {}
|
|
|
|
while True:
|
|
merged_params["limit"] = self._PAGE_LIMIT
|
|
merged_params["page"] = page
|
|
url = f"{self.base_url}{endpoint}"
|
|
resp = self.session.get(url, params=merged_params, timeout=self.timeout)
|
|
resp.raise_for_status()
|
|
items = resp.json()
|
|
all_items.extend(items)
|
|
if len(items) < self._PAGE_LIMIT:
|
|
break
|
|
page += 1
|
|
|
|
return all_items
|
|
|
|
def get_repos(self) -> list[dict]:
|
|
"""Retourne tous les repos de l'utilisateur (pagination automatique).
|
|
|
|
Endpoint: GET /api/v1/user/repos
|
|
"""
|
|
return self._get_paginated("/api/v1/user/repos")
|
|
|
|
def get_latest_release(self, owner: str, repo: str) -> dict | None:
|
|
"""Retourne la derniere release du repo, ou None si aucune.
|
|
|
|
Endpoint: GET /api/v1/repos/{owner}/{repo}/releases/latest
|
|
Gere HTTP 404 en retournant None.
|
|
"""
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/releases/latest"
|
|
resp = self.session.get(url, timeout=self.timeout)
|
|
if resp.status_code == 404:
|
|
return None
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
def get_milestones(self, owner: str, repo: str) -> list[dict]:
|
|
"""Retourne les milestones ouvertes du repo.
|
|
|
|
Endpoint: GET /api/v1/repos/{owner}/{repo}/milestones?state=open
|
|
"""
|
|
return self._get_paginated(
|
|
f"/api/v1/repos/{owner}/{repo}/milestones",
|
|
params={"state": "open"},
|
|
)
|