# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Documentation-specific GitLab Generic Package Registry operations.
This module wraps the generic GitLab registry helpers with
MAFw-docs-specific defaults (package name, filename conventions).
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
import requests
from mafw.devtools.gitlab.api import (
GitlabAPIConfiguration,
build_gitlab_auth_headers,
)
[docs]
def list_mafw_docs_generic_packages(
api_config: GitlabAPIConfiguration,
package_name: str = 'mafw-docs',
) -> list[dict[str, Any]]:
"""List generic packages for mafw-docs in the GitLab Package Registry.
Uses the Packages API:
``GET /projects/:id/packages``
and filters for ``package_type=generic`` and exact ``name == package_name``.
:param api_config: GitLab API configuration
:type api_config: gitlab.GitlabAPIConfiguration
:param package_name: Package name, defaults to ``mafw-docs``
:type package_name: str
:return: List of package dictionaries from the API
:rtype: list[dict[str, Any]]
"""
base_url = api_config.api_url.rstrip('/')
url = f'{base_url}/projects/{api_config.project_id}/packages'
headers = build_gitlab_auth_headers(api_config)
per_page = 100
page = 1
out: list[dict[str, Any]] = []
while True:
params: dict[str, str | int] = {
'package_type': 'generic',
'package_name': package_name,
'per_page': per_page,
'page': page,
'order_by': 'version',
'sort': 'asc',
}
resp = requests.get(url, headers=headers, params=params, timeout=60.0)
if not (200 <= resp.status_code < 300):
body_preview = (resp.text or '')[:500].replace('\n', ' ')
raise RuntimeError(f'GitLab list packages failed ({resp.status_code}): {body_preview}')
data = resp.json()
if not data:
break
for pkg in data:
if pkg.get('package_type') != 'generic':
continue
if pkg.get('name') != package_name:
continue
out.append(pkg)
if len(data) < per_page:
break
page += 1
return out
[docs]
def resolve_mafw_docs_package_ids_by_version(
api_config: GitlabAPIConfiguration, package_name: str = 'mafw-docs'
) -> dict[str, int]:
"""Resolve package IDs for mafw-docs generic packages by version.
:param api_config: GitLab API configuration
:type api_config: gitlab.GitlabAPIConfiguration
:param package_name: Package name, defaults to ``mafw-docs``
:type package_name: str
:return: Mapping from version string to package id
:rtype: dict[str, int]
"""
pkgs = list_mafw_docs_generic_packages(api_config, package_name=package_name)
mapping: dict[str, int] = {}
for pkg in pkgs:
version = pkg.get('version')
pkg_id = pkg.get('id')
if isinstance(version, str) and isinstance(pkg_id, int):
mapping[version] = pkg_id
return mapping
[docs]
def upload_docs_zip_to_gitlab_generic_registry(
api_config: GitlabAPIConfiguration,
package_version: str,
zip_path: Path,
package_name: str = 'mafw-docs',
timeout_s: float = 60.0,
) -> bool:
"""Upload a documentation zip archive to the GitLab Generic Package Registry.
The equivalent GitLab API endpoint is:
``PUT /projects/:id/packages/generic/:package_name/:package_version/:file_name``
Authentication headers:
- ``JOB-TOKEN`` when running on CI (``api_config.on_ci`` is True)
- ``PRIVATE-TOKEN`` for local execution
:param api_config: GitLab API configuration
:type api_config: gitlab.GitlabAPIConfiguration
:param package_version: Package version (typically the git tag, e.g. ``v2.1.0``)
:type package_version: str
:param zip_path: Path to the zip file to upload
:type zip_path: Path
:param package_name: Generic package name, defaults to ``mafw-docs``
:type package_name: str
:param timeout_s: Request timeout in seconds, defaults to 60.0
:type timeout_s: float
:return: True if the file was uploaded, False if the upload was skipped because the target already exists
:rtype: bool
:raises FileNotFoundError: If ``zip_path`` does not exist
:raises RuntimeError: If the upload fails (non-2xx response)
"""
zip_path = Path(zip_path).resolve()
if not zip_path.exists():
raise FileNotFoundError(f'Zip file not found: {zip_path}')
base_url = api_config.api_url.rstrip('/')
file_name = zip_path.name
url = f'{base_url}/projects/{api_config.project_id}/packages/generic/{package_name}/{package_version}/{file_name}'
headers = build_gitlab_auth_headers(api_config)
head_resp = requests.head(url, headers=headers, timeout=timeout_s)
if head_resp.status_code == 200:
print(f'ℹ️ Zip already present on GitLab, skipping upload: {url}')
return False
if head_resp.status_code != 404:
body_preview = (head_resp.text or '')[:500].replace('\n', ' ')
raise RuntimeError(f'GitLab existence check failed ({head_resp.status_code}): {body_preview}')
with open(zip_path, 'rb') as f:
put_headers = dict(headers)
put_headers['Content-Type'] = 'application/zip'
resp = requests.put(url, headers=put_headers, data=f, timeout=timeout_s)
if not (200 <= resp.status_code < 300):
body_preview = (resp.text or '')[:500].replace('\n', ' ')
raise RuntimeError(f'GitLab upload failed ({resp.status_code}): {body_preview}')
return True
[docs]
def download_docs_zip_from_gitlab_generic_registry(
api_config: GitlabAPIConfiguration,
package_version: str,
download_dir: Path,
package_name: str = 'mafw-docs',
file_name: str | None = None,
timeout_s: float = 60.0,
) -> Path | None:
"""Download a documentation zip archive from the GitLab Generic Package Registry.
The equivalent GitLab API endpoint is:
``GET /projects/:id/packages/generic/:package_name/:package_version/:file_name``
The function first issues a HEAD request to determine if the file exists:
- 404: the package file does not exist (cache miss) and ``None`` is returned.
- 200: the file exists and is downloaded.
- otherwise: a warning is printed and ``None`` is returned.
Authentication headers:
- ``JOB-TOKEN`` when running on CI (``api_config.on_ci`` is True)
- ``PRIVATE-TOKEN`` for local execution
:param api_config: GitLab API configuration
:type api_config: gitlab.GitlabAPIConfiguration
:param package_version: Package version (typically the git tag, e.g. ``v2.1.0``)
:type package_version: str
:param download_dir: Directory where the downloaded zip is stored
:type download_dir: Path
:param package_name: Generic package name, defaults to ``mafw-docs``
:type package_name: str
:param file_name: File name to retrieve, defaults to ``mafw-docs-<version>.zip``
:type file_name: str | None
:param timeout_s: Request timeout in seconds, defaults to 60.0
:type timeout_s: float
:return: Path to the downloaded zip archive, or None if it does not exist or cannot be retrieved
:rtype: Path | None
"""
download_dir = Path(download_dir).resolve()
download_dir.mkdir(parents=True, exist_ok=True)
base_url = api_config.api_url.rstrip('/')
target_file_name = file_name or f'mafw-docs-{package_version}.zip'
url = (
f'{base_url}/projects/{api_config.project_id}/packages/generic/'
f'{package_name}/{package_version}/{target_file_name}'
)
headers = build_gitlab_auth_headers(api_config)
try:
head_resp = requests.head(url, headers=headers, timeout=timeout_s)
except Exception as e: # pragma: no cover
print(f'⚠️ Warning: cache HEAD request failed for {package_version}: {e}')
return None
if head_resp.status_code == 404:
return None
if head_resp.status_code != 200:
body_preview = (head_resp.text or '')[:200].replace('\n', ' ')
print(
f'⚠️ Warning: cache existence check returned {head_resp.status_code} for {package_version}: {body_preview}'
)
return None
dest = download_dir / target_file_name
try:
resp = requests.get(url, headers=headers, stream=True, timeout=timeout_s)
except Exception as e: # pragma: no cover
print(f'⚠️ Warning: cache download request failed for {package_version}: {e}')
return None
if not (200 <= resp.status_code < 300):
body_preview = (resp.text or '')[:200].replace('\n', ' ')
print(f'⚠️ Warning: cache download failed for {package_version} ({resp.status_code}): {body_preview}')
return None
try:
with open(dest, 'wb') as f:
for chunk in resp.iter_content(chunk_size=1024 * 1024):
if chunk:
f.write(chunk)
except OSError as e:
print(f'⚠️ Warning: could not write downloaded zip for {package_version} to {dest}: {e}')
return None
return dest