1"""
2GitLab API client implementation.
3
4Implements :any:`/specs/spec_api_client` §1-5, covering:
5
6* Endpoint responsibilities (§1)
7* Configuration and authentication (§2)
8* Error mapping (§3)
9* Retry strategy (§4)
10* Resource management (§5)
11"""
12
13from __future__ import annotations
14
15from typing import Optional, Any
16from urllib.parse import quote
17import time
18import httpx # type: ignore
19
20from ..utils.logging import get_logger
21from ..config.settings import Settings
22from ..api.errors import (
23 AuthenticationError,
24 NotFoundError,
25 RateLimitError,
26 GitLabAPIError,
27)
28from ..models import Group, Project, Issue
29from ..models import mapper as _mapper
30
31logger = get_logger(__name__)
32
33
[docs]
34class GitLabClient:
35 """Minimal GitLab API Client (sync, httpx)."""
36
[docs]
37 def __init__(self) -> None:
38 self.settings = Settings.current()
39 self._client = httpx.Client(
40 base_url=self.settings.gitlab_host,
41 headers={"Accept": "application/json"},
42 timeout=30,
43 )
44 # Insert first token as default; can be overridden per-request
45 if self.settings.group_api_key:
46 self._client.headers["Authorization"] = (
47 f"Bearer {self.settings.group_api_key[0]}"
48 )
49
50 # Retry settings – configurable via Settings (defaults to 3 attempts)
51 self._max_retries: int = getattr(self.settings, "api_retry_count", 3)
52 self._retry_backoff_base: float = getattr(
53 self.settings, "api_retry_backoff", 1.0
54 )
55
56 # ----------------------- Helper ---------------------------------
57 def _raise_for_status(self, resp: httpx.Response) -> None: # noqa: D401
58 match resp.status_code:
59 case 401 | 403:
60 raise AuthenticationError(resp.text)
61 case 404:
62 raise NotFoundError(resp.text)
63 case 429:
64 raise RateLimitError(resp.text)
65 case code if code >= 400:
66 raise GitLabAPIError(f"Unexpected status {code}: {resp.text}")
67
68 # ----------------------- Endpoints ------------------------------
[docs]
69 def list_groups(self) -> list[Group]: # noqa: D401
70 """List all groups with pagination support.
71
72 Collates all pages and returns a complete list as required by specification.
73 """
74 all_groups = []
75 page = 1
76 per_page = 100 # GitLab default
77 while True:
78 params = {"page": page, "per_page": per_page}
79 resp = self._request("GET", "/api/v4/groups", params=params)
80 groups = resp.json()
81 if not groups:
82 break
83 all_groups.extend([_mapper.group_from_json(item) for item in groups])
84 page += 1
85 if len(groups) < per_page:
86 break
87 return all_groups
88
[docs]
89 def list_projects(
90 self, group_id: int, with_shared: bool = False
91 ) -> list[Project]: # noqa: D401
92 """List all projects for a group with pagination support.
93
94 If with_shared is True, also include projects shared with the group.
95 Collates all pages and returns a complete list as required by specification.
96 """
97 all_projects = []
98 page = 1
99 per_page = 100 # GitLab default
100 while True:
101 params = {"page": str(page), "per_page": str(per_page)}
102 if with_shared:
103 params["with_shared"] = "true"
104 resp = self._request(
105 "GET", f"/api/v4/groups/{group_id}/projects", params=params
106 )
107 projects = resp.json()
108 if not projects:
109 break
110 all_projects.extend([_mapper.project_from_json(item) for item in projects])
111 page += 1
112 if len(projects) < per_page:
113 break
114 return all_projects
115
[docs]
116 def fetch_file(
117 self, project_id: int, path: str, ref: str = "main"
118 ) -> str: # noqa: D401
119 encoded_path = quote(path, safe="")
120 url = f"/api/v4/projects/{project_id}/repository/files/{encoded_path}/raw?ref={ref}"
121 resp = self._request("GET", url)
122 return resp.text
123
[docs]
124 def list_issues(self, project_id: int) -> list[Issue]: # noqa: D401
125 resp = self._request("GET", f"/api/v4/projects/{project_id}/issues")
126 return [_mapper.issue_from_json(item) for item in resp.json()]
127
128 # ----------------------- Context mgmt ---------------------------
[docs]
129 def close(self): # noqa: D401
130 self._client.close()
131
132 def __enter__(self):
133 return self
134
135 def __exit__(self, exc_type, exc, tb): # noqa: D401
136 self.close()
137 return False
138
139 # ----------------------- Internal helpers -------------------------------
140
[docs]
141 def _request(
142 self, method: str, url: str, **kwargs: Any
143 ) -> httpx.Response: # noqa: D401
144 """Perform *method* request with simple exponential back-off retry.
145
146 Retries are triggered for the following cases:
147 * Network errors (connection errors, timeouts, etc.)
148 * 429 (rate limit)
149 * 5xx server errors (502/503/504 or any >=500)
150 """
151 backoff = self._retry_backoff_base
152 for attempt in range(1, self._max_retries + 1):
153 try:
154 resp = self._client.request(method, url, **kwargs)
155 # Success – return immediately
156 if resp.status_code < 400:
157 return resp
158 # Error handling -------------------------------------------------
159 should_retry = (
160 resp.status_code in {429, 502, 503, 504} or resp.status_code >= 500
161 )
162 if should_retry and attempt < self._max_retries:
163 logger.warning(
164 "Request %s %s failed with status %s (attempt %s/%s). Retrying in %.1fs.",
165 method,
166 url,
167 resp.status_code,
168 attempt,
169 self._max_retries,
170 backoff,
171 )
172 time.sleep(backoff)
173 backoff *= 2
174 continue
175 # No retry (or last attempt) – raise specific API error
176 self._raise_for_status(resp)
177 # _raise_for_status should always raise; added safeguard
178 return resp # pragma: no cover
179 except (
180 httpx.ConnectError,
181 httpx.TimeoutException,
182 httpx.NetworkError,
183 ) as e:
184 # Network errors - retry if we have attempts left
185 if attempt < self._max_retries:
186 logger.warning(
187 "Request %s %s failed with network error %s (attempt %s/%s). Retrying in %.1fs.",
188 method,
189 url,
190 type(e).__name__,
191 attempt,
192 self._max_retries,
193 backoff,
194 )
195 time.sleep(backoff)
196 backoff *= 2
197 continue
198 raise
199 # Should be unreachable
200 raise RuntimeError("Unreachable code in _request") # pragma: no cover