Coverage for src/mafw/devtools/gitlab/registry.py: 99%
173 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-09-01 19:49 +0000
1# Copyright 2026 European Union
2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
3# SPDX-License-Identifier: EUPL-1.2
4"""
5GitLab Generic Package Registry operations.
7This module centralizes the HTTP helpers for uploading, downloading, listing,
8and deleting files in the GitLab Generic Package Registry.
10:author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
11"""
13from __future__ import annotations
15import re
16from pathlib import Path
17from typing import Any, Final, cast
19from mafw.devtools import ensure_devtools_available
21ensure_devtools_available()
23import requests # noqa: E402
24from packaging.version import InvalidVersion, Version # noqa: E402
26from mafw.devtools.gitlab.api import ( # noqa: E402
27 GitlabAPIConfiguration,
28 build_gitlab_auth_headers,
29)
31_PYLOCK_RE: Final[re.Pattern[str]] = re.compile(r'^pylock\.py(?P<python_version>3_\d+)_ref\.toml$')
32"""Pattern for MAFw reference dependency lock files.
34The Python version uses an underscore instead of a dot (e.g. ``3_14``) so that the
35filename matches the ``uv`` lockfile naming constraint ``^pylock\\.([^.]+\\.)?toml$``.
36"""
38_MAFW_VERSION_RE: Final[re.Pattern[str]] = re.compile(r'^(?:v)?(?P<version>\d+\.\d+\.\d+)$')
39"""Pattern for accepted MAFw version overrides."""
41_CONTENT_TYPE_MAP: Final[dict[str, str]] = {
42 '.toml': 'application/toml',
43 '.zip': 'application/zip',
44 '.json': 'application/json',
45 '.tar': 'application/x-tar',
46 '.gz': 'application/gzip',
47}
48"""Mapping of file extensions to Content-Type values for registry uploads."""
51def pylock_version_tag(python_version: str) -> str:
52 """Return the filename-safe version tag for a Python ``major.minor`` version.
54 ``uv`` (>= 0.12) only accepts lock files matching ``^pylock\\.([^.]+\\.)?toml$``,
55 which forbids an extra dot in the filename. To comply, the dot in the Python
56 version is replaced with an underscore (e.g. ``3.14`` becomes ``3_14``).
58 :param python_version: Python version in ``major.minor`` form (e.g. ``3.14``).
59 :type python_version: str
60 :return: The version tag with the dot replaced by an underscore (e.g. ``3_14``).
61 :rtype: str
62 """
63 return python_version.replace('.', '_')
66def build_pylock_reference_filename(python_version: str) -> str:
67 """Build the reference lock filename for a given Python version.
69 :param python_version: Python version in ``major.minor`` form (e.g. ``3.14``).
70 :type python_version: str
71 :return: The reference lock filename (e.g. ``pylock.py3_14_ref.toml``).
72 :rtype: str
73 """
74 return f'pylock.py{pylock_version_tag(python_version)}_ref.toml'
77def build_pylock_compiled_filename(python_version: str) -> str:
78 """Build the compiled lock filename for a given Python version.
80 :param python_version: Python version in ``major.minor`` form (e.g. ``3.14``).
81 :type python_version: str
82 :return: The compiled lock filename (e.g. ``pylock.py3_14.toml``).
83 :rtype: str
84 """
85 return f'pylock.py{pylock_version_tag(python_version)}.toml'
88def parse_pylock_reference_filename(file_name: str) -> tuple[str, str] | None:
89 """Parse a reference dependency filename into the Python version and normalized file name.
91 :param file_name: The filename to parse.
92 :type file_name: str
93 :return: A tuple of (python_version, filename) if matched, else None.
94 :rtype: tuple[str, str] | None
95 """
96 base = Path(file_name).name
97 match = _PYLOCK_RE.fullmatch(base)
98 if match is None:
99 return None
100 # The pattern captures the filename-safe tag (e.g. ``3_14``); restore the dotted
101 # ``major.minor`` form expected by the rest of the tooling.
102 python_version = match.group('python_version').replace('_', '.')
103 return python_version, base
106def normalize_dependency_registry_item(item: str) -> tuple[str, str]:
107 """Normalize a registry item into a Python version and dependency file name.
109 :param item: A filename or a version string.
110 :type item: str
111 :return: A tuple of (python_version, reference_filename).
112 :rtype: tuple[str, str]
113 :raises ValueError: If the item is invalid or unsupported.
114 """
115 item = item.strip()
116 parsed = parse_pylock_reference_filename(item)
117 if parsed is not None:
118 return parsed
120 try:
121 version = Version(item)
122 except InvalidVersion as exc:
123 raise ValueError(f'Invalid version or lock filename: {item}') from exc
125 if version.is_prerelease or version.is_devrelease:
126 raise ValueError(f'Pre-release/dev versions are not supported here: {item}')
128 python_version = f'{version.major}.{version.minor}'
129 return python_version, build_pylock_reference_filename(python_version)
132def normalize_mafw_version(version_text: str) -> str:
133 """Normalize a MAFw version override and strip an optional leading ``v``.
135 :param version_text: The version string to normalize.
136 :type version_text: str
137 :return: The normalized version string.
138 :rtype: str
139 :raises ValueError: If the version string is invalid.
140 """
141 match = _MAFW_VERSION_RE.fullmatch(version_text.strip())
142 if match is None:
143 raise ValueError(f'Invalid MAFw version: {version_text}')
144 return match.group('version')
147def iter_local_pylock_reference_files(directory: Path) -> list[tuple[str, Path]]:
148 """List local reference dependency files in a directory.
150 :param directory: The directory to search for reference files.
151 :type directory: Path
152 :return: A list of (python_version, file_path) tuples, sorted by version.
153 :rtype: list[tuple[str, Path]]
154 """
155 directory = Path(directory).resolve()
156 if not directory.exists():
157 return []
158 items: list[tuple[str, Path]] = []
159 for fp in directory.iterdir():
160 if not fp.is_file():
161 continue
162 parsed = parse_pylock_reference_filename(fp.name)
163 if parsed is None:
164 continue
165 python_version, _ = parsed
166 items.append((python_version, fp))
167 items.sort(key=lambda item: tuple(int(part) for part in item[0].split('.')))
168 return items
171def list_generic_packages(api_config: GitlabAPIConfiguration, package_name: str) -> list[dict[str, Any]]:
172 """List generic packages for a given name in the GitLab Package Registry.
174 :param api_config: The GitLab API configuration.
175 :type api_config: gitlab.GitlabAPIConfiguration
176 :param package_name: The name of the package to list.
177 :type package_name: str
178 :return: A list of package dictionaries from the API.
179 :rtype: list[dict[str, Any]]
180 :raises RuntimeError: If the API request fails.
181 """
182 url = f'{api_config.api_url.rstrip("/")}/projects/{api_config.project_id}/packages'
183 headers = build_gitlab_auth_headers(api_config)
184 out: list[dict[str, Any]] = []
185 page = 1
186 while True:
187 resp = requests.get(
188 url,
189 headers=headers,
190 params=cast(
191 dict[str, str | int],
192 {
193 'package_type': 'generic',
194 'package_name': package_name,
195 'per_page': 100,
196 'page': page,
197 'order_by': 'version',
198 'sort': 'asc',
199 },
200 ),
201 timeout=60.0,
202 )
203 if not (200 <= resp.status_code < 300):
204 body_preview = (resp.text or '')[:500].replace('\n', ' ')
205 raise RuntimeError(f'GitLab list packages failed ({resp.status_code}): {body_preview}')
206 data = resp.json()
207 if not data:
208 break
209 for pkg in data:
210 if pkg.get('package_type') == 'generic' and pkg.get('name') == package_name:
211 out.append(pkg)
212 if len(data) < 100:
213 break
214 page += 1
215 return out
218def resolve_package_ids_by_version(api_config: GitlabAPIConfiguration, package_name: str) -> dict[str, int]:
219 """Resolve package IDs for a package name by version.
221 :param api_config: The GitLab API configuration.
222 :type api_config: gitlab.GitlabAPIConfiguration
223 :param package_name: The name of the package.
224 :type package_name: str
225 :return: A mapping of version strings to package IDs.
226 :rtype: dict[str, int]
227 """
228 mapping: dict[str, int] = {}
229 for pkg in list_generic_packages(api_config, package_name):
230 version = pkg.get('version')
231 pkg_id = pkg.get('id')
232 if isinstance(version, str) and isinstance(pkg_id, int):
233 mapping[version] = pkg_id
234 return mapping
237def _infer_content_type(file_path: Path) -> str:
238 """Infer a Content-Type from the file extension.
240 :param file_path: Path to the file.
241 :type file_path: Path
242 :return: A MIME type string.
243 :rtype: str
244 """
245 return _CONTENT_TYPE_MAP.get(file_path.suffix.lower(), 'application/octet-stream')
248def upload_generic_file(
249 api_config: GitlabAPIConfiguration,
250 package_name: str,
251 package_version: str,
252 file_path: Path,
253 *,
254 replace_existing: bool = False,
255 package_id: int | None = None,
256 timeout_s: float = 60.0,
257) -> bool:
258 """Upload a file to the GitLab Generic Package Registry.
260 :param api_config: The GitLab API configuration.
261 :type api_config: gitlab.GitlabAPIConfiguration
262 :param package_name: The name of the generic package.
263 :type package_name: str
264 :param package_version: The version of the generic package.
265 :type package_version: str
266 :param file_path: The local path to the file to upload.
267 :type file_path: Path
268 :param replace_existing: Whether to overwrite if the file already exists.
269 :type replace_existing: bool
270 :param package_id: Optional numeric package ID to avoid extra API calls.
271 :type package_id: int | None
272 :param timeout_s: Request timeout in seconds.
273 :type timeout_s: float
274 :return: True if the file was uploaded, False if skipped.
275 :rtype: bool
276 :raises FileNotFoundError: If the file_path does not exist.
277 :raises RuntimeError: If the API request fails.
278 """
279 file_path = Path(file_path).resolve()
280 if not file_path.exists():
281 raise FileNotFoundError(f'File not found: {file_path}')
283 if package_id is None:
284 mapping = resolve_package_ids_by_version(api_config, package_name)
285 package_id = mapping.get(package_version)
287 if package_id is not None:
288 existing_files = list_generic_package_files(api_config, package_id, timeout_s=timeout_s)
289 matching = [f for f in existing_files if f.get('file_name') == file_path.name and isinstance(f.get('id'), int)]
290 if matching:
291 if not replace_existing:
292 return False
293 for f in matching:
294 delete_generic_package_file(api_config, package_id, f['id'], timeout_s=timeout_s)
296 url = (
297 f'{api_config.api_url.rstrip("/")}/projects/{api_config.project_id}/packages/generic/'
298 f'{package_name}/{package_version}/{file_path.name}'
299 )
300 headers = build_gitlab_auth_headers(api_config)
301 with file_path.open('rb') as fp:
302 put_headers = dict(headers)
303 put_headers['Content-Type'] = _infer_content_type(file_path)
304 resp = requests.put(url, headers=put_headers, data=fp, timeout=timeout_s)
305 if not (200 <= resp.status_code < 300):
306 body_preview = (resp.text or '')[:500].replace('\n', ' ')
307 raise RuntimeError(f'GitLab upload failed ({resp.status_code}): {body_preview}')
308 return True
311def download_generic_file(
312 api_config: GitlabAPIConfiguration,
313 package_name: str,
314 package_version: str,
315 file_name: str,
316 download_dir: Path,
317 *,
318 timeout_s: float = 60.0,
319) -> Path | None:
320 """Download a file from the GitLab Generic Package Registry.
322 :param api_config: The GitLab API configuration.
323 :type api_config: gitlab.GitlabAPIConfiguration
324 :param package_name: The name of the generic package.
325 :type package_name: str
326 :param package_version: The version of the generic package.
327 :type package_version: str
328 :param file_name: The name of the file to download.
329 :type file_name: str
330 :param download_dir: The local directory where to save the file.
331 :type download_dir: Path
332 :param timeout_s: Request timeout in seconds.
333 :type timeout_s: float
334 :return: The Path to the downloaded file, or None if not found.
335 :rtype: Path | None
336 :raises RuntimeError: If the API request fails.
337 """
338 download_dir = Path(download_dir).resolve()
339 download_dir.mkdir(parents=True, exist_ok=True)
341 url = (
342 f'{api_config.api_url.rstrip("/")}/projects/{api_config.project_id}/packages/generic/'
343 f'{package_name}/{package_version}/{file_name}'
344 )
345 headers = build_gitlab_auth_headers(api_config)
346 head_resp = requests.head(url, headers=headers, timeout=timeout_s)
347 if head_resp.status_code == 404:
348 return None
349 if head_resp.status_code != 200:
350 body_preview = (head_resp.text or '')[:200].replace('\n', ' ')
351 raise RuntimeError(f'GitLab existence check failed ({head_resp.status_code}): {body_preview}')
353 dest = download_dir / file_name
354 resp = requests.get(url, headers=headers, stream=True, timeout=timeout_s)
355 if not (200 <= resp.status_code < 300):
356 body_preview = (resp.text or '')[:200].replace('\n', ' ')
357 raise RuntimeError(f'GitLab download failed ({resp.status_code}): {body_preview}')
359 with dest.open('wb') as fp:
360 for chunk in resp.iter_content(chunk_size=1024 * 1024):
361 if chunk:
362 fp.write(chunk)
363 return dest
366def delete_generic_package_version(
367 api_config: GitlabAPIConfiguration, package_id: int, package_name: str, package_version: str
368) -> bool:
369 """Delete a package version from the GitLab Package Registry.
371 :param api_config: The GitLab API configuration.
372 :type api_config: gitlab.GitlabAPIConfiguration
373 :param package_id: The numeric ID of the package version.
374 :type package_id: int
375 :param package_name: The name of the package (for error reporting).
376 :type package_name: str
377 :param package_version: The version of the package (for error reporting).
378 :type package_version: str
379 :return: True if deleted, False if not found or forbidden.
380 :rtype: bool
381 :raises RuntimeError: If the API request fails.
382 """
383 url = f'{api_config.api_url.rstrip("/")}/projects/{api_config.project_id}/packages/{package_id}'
384 resp = requests.delete(url, headers=build_gitlab_auth_headers(api_config), timeout=60.0)
385 if resp.status_code == 204:
386 return True
387 if resp.status_code in (403, 404):
388 return False
389 body_preview = (resp.text or '')[:500].replace('\n', ' ')
390 raise RuntimeError(
391 f'GitLab delete failed for {package_name}/{package_version} (id={package_id}): {resp.status_code} {body_preview}'
392 )
395def list_generic_package_files(
396 api_config: GitlabAPIConfiguration,
397 package_id: int,
398 *,
399 timeout_s: float = 60.0,
400) -> list[dict[str, Any]]:
401 """List files belonging to a specific package version.
403 :param api_config: The GitLab API configuration.
404 :type api_config: gitlab.GitlabAPIConfiguration
405 :param package_id: The numeric ID of the package version.
406 :type package_id: int
407 :param timeout_s: Request timeout in seconds.
408 :type timeout_s: float
409 :return: A list of file metadata dictionaries.
410 :rtype: list[dict[str, Any]]
411 :raises RuntimeError: If the API request fails.
412 """
413 url = f'{api_config.api_url.rstrip("/")}/projects/{api_config.project_id}/packages/{package_id}/package_files'
414 headers = build_gitlab_auth_headers(api_config)
415 out: list[dict[str, Any]] = []
416 page = 1
417 while True:
418 resp = requests.get(
419 url,
420 headers=headers,
421 params={'per_page': 100, 'page': page},
422 timeout=timeout_s,
423 )
424 if not (200 <= resp.status_code < 300):
425 body_preview = (resp.text or '')[:500].replace('\n', ' ')
426 raise RuntimeError(f'GitLab list package files failed ({resp.status_code}): {body_preview}')
427 data = resp.json()
428 if not data: 428 ↛ 429line 428 didn't jump to line 429 because the condition on line 428 was never true
429 break
430 out.extend(data)
431 if len(data) < 100:
432 break
433 page += 1
434 return out
437def delete_generic_package_file(
438 api_config: GitlabAPIConfiguration,
439 package_id: int,
440 file_id: int,
441 *,
442 timeout_s: float = 60.0,
443) -> bool:
444 """Delete a single file from a package version.
446 :param api_config: The GitLab API configuration.
447 :type api_config: gitlab.GitlabAPIConfiguration
448 :param package_id: The numeric ID of the package version.
449 :type package_id: int
450 :param file_id: The numeric ID of the package file to delete.
451 :type file_id: int
452 :param timeout_s: Request timeout in seconds.
453 :type timeout_s: float
454 :return: True if deleted, False if not found or forbidden.
455 :rtype: bool
456 :raises RuntimeError: If the API request fails.
457 """
458 url = (
459 f'{api_config.api_url.rstrip("/")}/projects/{api_config.project_id}'
460 f'/packages/{package_id}/package_files/{file_id}'
461 )
462 resp = requests.delete(url, headers=build_gitlab_auth_headers(api_config), timeout=timeout_s)
463 if resp.status_code == 204:
464 return True
465 if resp.status_code in (403, 404):
466 return False
467 body_preview = (resp.text or '')[:500].replace('\n', ' ')
468 raise RuntimeError(
469 f'GitLab delete file failed (package_id={package_id}, file_id={file_id}): {resp.status_code} {body_preview}'
470 )