feat(client): add GiteaClient with auth and pagination (fixes #1)
- GiteaClient with requests.Session and token auth header - _get_paginated for automatic pagination (limit=50) - get_repos, get_latest_release (None on 404), get_milestones - 9 unit tests with mocked requests.Session - Fix setuptools build backend in pyproject.toml Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
73
src/gitea_dashboard/client.py
Normal file
73
src/gitea_dashboard/client.py
Normal file
@@ -0,0 +1,73 @@
|
||||
"""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) -> None:
|
||||
"""Initialise le client avec l'URL de base et le token API."""
|
||||
self.base_url = base_url.rstrip("/")
|
||||
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)
|
||||
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)
|
||||
if resp.status_code == 404:
|
||||
return None
|
||||
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"},
|
||||
)
|
||||
Reference in New Issue
Block a user