""" Nom du script : gitlab_client.py Chemin : /gitlab-bridge/app/clients/gitlab_client.py Description : Client HTTP sécurisé pour l'API GitLab utilisé par les services métiers. Utilise un client httpx persistant (connection pooling), supporte la pagination automatique et le retry sur erreurs transitoires. Options éventuelles : Aucune. Exemples d'utilisation : `GitLabClient().get_project_by_path(...)`. Prérequis : Python 3.11+, httpx. Auteur : Sylvain SCATTOLINI Date de création / modification : 2026-03-25 Version : 1.2 """ from __future__ import annotations import logging import time from urllib.parse import quote import httpx from app.core.config import settings from app.core.exceptions import GitLabApiError logger = logging.getLogger(__name__) # Client partagé au niveau du module — réutilise les connexions TCP (keep-alive) _shared_client: httpx.Client | None = None _RETRY_STATUS_CODES = {429, 500, 502, 503, 504} _MAX_RETRIES = 3 _RETRY_BACKOFF_BASE = 1.0 # secondes def _get_shared_client() -> httpx.Client: """Retourne le client httpx partagé, en le (re)créant si nécessaire.""" global _shared_client if _shared_client is None or _shared_client.is_closed: _shared_client = httpx.Client( timeout=settings.gitlab_timeout, verify=settings.gitlab_verify_ssl, headers={ 'PRIVATE-TOKEN': settings.gitlab_api_token, 'Accept': 'application/json', 'Content-Type': 'application/json', }, limits=httpx.Limits( max_connections=20, max_keepalive_connections=10, keepalive_expiry=30, ), ) return _shared_client class GitLabClient: """Client vers l'API GitLab v4 avec connection pooling, pagination et retry.""" def __init__(self) -> None: self._base_url = str(settings.gitlab_base_url).rstrip('/') def _request_raw( self, method: str, path: str, *, params: dict | None = None, json_body: dict | None = None, ) -> httpx.Response: """Exécute une requête HTTP avec retry automatique sur erreurs transitoires.""" url = f'{self._base_url}/api/v4{path}' client = _get_shared_client() last_exc: Exception | None = None for attempt in range(_MAX_RETRIES): try: response = client.request( method=method, url=url, params=params, json=json_body, ) if response.status_code in _RETRY_STATUS_CODES and attempt < _MAX_RETRIES - 1: wait = _RETRY_BACKOFF_BASE * (2 ** attempt) logger.warning( "GitLab HTTP %s sur %s — retry %d/%d dans %.1fs", response.status_code, path, attempt + 1, _MAX_RETRIES - 1, wait, ) time.sleep(wait) continue response.raise_for_status() return response except httpx.HTTPStatusError: raise except httpx.HTTPError as exc: last_exc = exc if attempt < _MAX_RETRIES - 1: wait = _RETRY_BACKOFF_BASE * (2 ** attempt) logger.warning( "Erreur réseau GitLab sur %s — retry %d/%d dans %.1fs : %s", path, attempt + 1, _MAX_RETRIES - 1, wait, exc, ) time.sleep(wait) raise GitLabApiError(f'Erreur réseau GitLab sur {path} après {_MAX_RETRIES} tentatives') from last_exc def _request( self, method: str, path: str, *, params: dict | None = None, json_body: dict | None = None, ) -> dict | list: """Exécute une requête et retourne le corps JSON parsé.""" try: response = self._request_raw(method, path, params=params, json_body=json_body) return response.json() except httpx.HTTPStatusError as exc: raise GitLabApiError( f'GitLab HTTP {exc.response.status_code} sur {path}', http_status=exc.response.status_code, ) from exc except ValueError as exc: raise GitLabApiError('Réponse JSON GitLab invalide') from exc def _list_all_pages(self, path: str, params: dict | None = None) -> list[dict]: """ Récupère toutes les pages d'une ressource GitLab (pagination automatique). Utilise l'en-tête X-Total-Pages retourné par GitLab. """ base_params = {**(params or {}), 'per_page': 100, 'page': 1} results: list[dict] = [] while True: try: response = self._request_raw('GET', path, params=base_params) except httpx.HTTPStatusError as exc: raise GitLabApiError(f'GitLab HTTP {exc.response.status_code} sur {path}') from exc page_data = response.json() if not isinstance(page_data, list): break results.extend(page_data) total_pages = int(response.headers.get('X-Total-Pages', '1')) current_page = int(response.headers.get('X-Page', base_params['page'])) if current_page >= total_pages: break base_params = {**base_params, 'page': current_page + 1} return results # ------------------------------------------------------------------ # Méthodes métier # ------------------------------------------------------------------ def search_users(self, *, email: str | None = None, username: str | None = None) -> list[dict]: params: dict[str, str] = {} if email: params['search'] = email elif username: params['username'] = username return self._request('GET', '/users', params=params) # type: ignore[return-value] def get_project_by_path(self, project_path: str) -> dict: encoded_path = quote(project_path, safe='') return self._request('GET', f'/projects/{encoded_path}') # type: ignore[return-value] def get_project_member(self, project_id: int, user_id: int) -> dict: return self._request('GET', f'/projects/{project_id}/members/all/{user_id}') # type: ignore[return-value] def list_open_issues(self, project_id: int) -> list[dict]: return self._list_all_pages( f'/projects/{project_id}/issues', params={'state': 'opened', 'order_by': 'updated_at', 'sort': 'desc'}, ) def list_closed_issues_since(self, project_id: int, closed_after_iso: str) -> list[dict]: return self._list_all_pages( f'/projects/{project_id}/issues', params={ 'state': 'closed', 'updated_after': closed_after_iso, 'order_by': 'updated_at', 'sort': 'desc', }, ) def list_all_issues(self, project_id: int, state: str = 'all') -> list[dict]: """Retourne toutes les issues (toutes pages) dans un état donné.""" return self._list_all_pages( f'/projects/{project_id}/issues', params={'state': state, 'order_by': 'updated_at', 'sort': 'desc'}, ) def list_groups(self, *, search: str | None = None, all_available: bool = True) -> list[dict]: """Retourne tous les groupes accessibles (toutes pages).""" params: dict = {'all_available': all_available} if search: params['search'] = search return self._list_all_pages('/groups', params=params) def list_group_projects(self, group_id: int) -> list[dict]: """Retourne tous les projets d'un groupe (toutes pages).""" return self._list_all_pages(f'/groups/{group_id}/projects', params={'simple': True}) def list_milestones(self, project_id: int, state: str = 'active') -> list[dict]: """Retourne tous les milestones d'un projet.""" return self._list_all_pages( f'/projects/{project_id}/milestones', params={'state': state}, ) def create_issue(self, project_id: int, title: str, description: str, labels: list[str]) -> dict: payload = { 'title': title, 'description': description, 'labels': ','.join(labels), } return self._request('POST', f'/projects/{project_id}/issues', json_body=payload) # type: ignore[return-value] def check_version(self) -> dict: """Vérifie la connectivité à GitLab en récupérant la version.""" return self._request('GET', '/version') # type: ignore[return-value]