Coverage for src/mafw/devtools/cli/dependencies/registry.py: 100%
42 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"""GitLab registry operations for dependency reference files."""
6from __future__ import annotations
8import re
9from pathlib import Path
10from typing import Any, Final
12import click
13from rich.prompt import Confirm
15from mafw.__about__ import __version__ as MAFW_VERSION
16from mafw.devtools.gitlab import (
17 GitlabAPIConfiguration,
18 build_gitlab_api_configuration,
19 delete_generic_package_file,
20 delete_generic_package_version,
21 download_generic_file,
22 iter_local_pylock_reference_files,
23 list_generic_package_files,
24 normalize_dependency_registry_item,
25 normalize_mafw_version,
26 parse_pylock_reference_filename,
27 resolve_package_ids_by_version,
28 upload_generic_file,
29)
30from mafw.tools.click_extensions import AbbreviateGroup
31from mafw.tools.shell_tools import CONSOLE
33DEPS_REGISTRY_PACKAGE_NAME: Final[str] = 'mafw-deps'
34"""Generic package name used for dependency reference uploads."""
36PYLOCK_REFERENCE_PATTERN = re.compile(r'^pylock\.py(?P<python_version>3_\d+)_ref\.toml$')
37"""Regular expression used to validate dependency reference lock filenames.
39The Python version uses an underscore instead of a dot (e.g. ``3_14``) so that the
40filename matches the ``uv`` lockfile naming constraint ``^pylock\\.([^.]+\\.)?toml$``.
41"""
44def _registry_api_config(ctx: click.Context) -> GitlabAPIConfiguration:
45 obj = ctx.obj or {}
46 try:
47 return build_gitlab_api_configuration(
48 obj.get('gitlab_api_url'),
49 obj.get('gitlab_project_id'),
50 obj.get('gitlab_token'),
51 )
52 except ValueError as exc:
53 raise click.ClickException(str(exc)) from exc
56def _normalize_pylock_input(item: str) -> tuple[str, str]:
57 """Normalize a dependency registry item into its Python version and file name."""
58 try:
59 return normalize_dependency_registry_item(item)
60 except ValueError as exc:
61 raise click.ClickException(str(exc)) from exc
64def _normalize_mafw_version_option(value: str | None) -> str:
65 """Return the registry label version, defaulting to the current MAFw version."""
66 if value is None:
67 return normalize_mafw_version(MAFW_VERSION)
68 try:
69 return normalize_mafw_version(value)
70 except ValueError as exc:
71 raise click.ClickException(str(exc)) from exc
74@click.group(
75 context_settings={'help_option_names': ['-h', '--help']},
76 help='GitLab registry operations for dependency reference files.',
77 cls=AbbreviateGroup,
78)
79@click.option('--gitlab-api-url', default=None, envvar='CI_API_V4_URL', help='GitLab API v4 base URL.')
80@click.option(
81 '--gitlab-project-id',
82 default=None,
83 type=int,
84 envvar='CI_PROJECT_ID',
85 help='GitLab project numeric id (env: CI_PROJECT_ID).',
86)
87@click.option('--gitlab-token', default=None, envvar='CI_JOB_TOKEN', help='GitLab API token.')
88@click.pass_context
89def registry(
90 ctx: click.Context,
91 gitlab_api_url: str | None,
92 gitlab_project_id: int | None,
93 gitlab_token: str | None,
94) -> None:
95 """GitLab registry command group for dependency reference files."""
96 ctx.obj = ctx.obj or {}
97 ctx.obj['gitlab_api_url'] = gitlab_api_url
98 ctx.obj['gitlab_project_id'] = gitlab_project_id
99 ctx.obj['gitlab_token'] = gitlab_token
102@registry.command()
103@click.argument(
104 'items',
105 nargs=-1,
106 required=False,
107 help='Reference filenames or Python versions to upload (e.g. 3.14 or pylock.py3_14_ref.toml)',
108)
109@click.option('-a', '--all', 'all_entries', is_flag=True, default=False, help='Upload all matching reference files.')
110@click.option(
111 '-f',
112 '--force',
113 is_flag=True,
114 default=False,
115 help='Force upload by removing existing remote files before uploading. '
116 'Without this flag, files already present in the registry are skipped.',
117)
118@click.option(
119 '--mafw-version',
120 default=None,
121 help='MAFw package version label to use in the registry. Accepts X.Y.Z or vX.Y.Z.',
122)
123@click.pass_context
124def upload( # pragma: no cover — thin CLI delegation to tested gitlab.registry
125 ctx: click.Context, items: tuple[str, ...], all_entries: bool, force: bool, mafw_version: str | None
126) -> None:
127 """Upload dependency reference files to the GitLab generic registry."""
128 api_config = _registry_api_config(ctx)
129 package_version = _normalize_mafw_version_option(mafw_version)
130 candidates: list[tuple[str, Path]] = []
132 if all_entries:
133 if items:
134 raise click.ClickException('Do not pass filenames when using --all.')
135 candidates = iter_local_pylock_reference_files(Path.cwd())
136 else:
137 if not items:
138 raise click.ClickException('Provide one or more filenames or Python versions, or use --all.')
139 for item in items:
140 python_version, normalized_name = _normalize_pylock_input(item)
141 fp = Path(item)
142 if not fp.is_absolute():
143 cwd_fp = Path.cwd() / fp
144 if cwd_fp.exists():
145 fp = cwd_fp
146 elif item == python_version:
147 fp = Path.cwd() / normalized_name
148 if not fp.exists():
149 raise click.ClickException(f'File not found: {fp}')
150 if not PYLOCK_REFERENCE_PATTERN.fullmatch(fp.name):
151 raise click.ClickException(f'File does not match pylock.py3_xx_ref.toml: {fp.name}')
152 candidates.append((python_version, fp))
154 if not candidates:
155 CONSOLE.print('No matching reference files found.')
156 return
158 for python_version, fp in candidates:
159 uploaded = upload_generic_file(
160 api_config,
161 DEPS_REGISTRY_PACKAGE_NAME,
162 package_version,
163 fp,
164 replace_existing=force,
165 )
166 if uploaded:
167 CONSOLE.print(
168 f'Uploaded {fp.name} for Python {python_version} as {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}'
169 )
170 else:
171 CONSOLE.print(
172 f'Skipped {fp.name} (already exists in '
173 f'{DEPS_REGISTRY_PACKAGE_NAME}/{package_version}). Use --force to replace.'
174 )
177@registry.command()
178@click.argument(
179 'items',
180 nargs=-1,
181 required=False,
182 help='Reference filenames or Python versions to download (e.g. 3.14 or pylock.py3_14_ref.toml)',
183)
184@click.option('--output-dir', default='.', type=click.Path(file_okay=False, dir_okay=True, path_type=Path))
185@click.option('-a', '--all', 'all_entries', is_flag=True, default=False, help='Download all registry files.')
186@click.option(
187 '--mafw-version',
188 default=None,
189 help='MAFw package version label to use in the registry. Accepts X.Y.Z or vX.Y.Z.',
190)
191@click.pass_context
192def download( # pragma: no cover — thin CLI delegation to tested gitlab.registry
193 ctx: click.Context,
194 items: tuple[str, ...],
195 output_dir: Path,
196 all_entries: bool,
197 mafw_version: str | None,
198) -> None:
199 """Download dependency reference files from the GitLab generic registry."""
200 api_config = _registry_api_config(ctx)
201 package_version = _normalize_mafw_version_option(mafw_version)
202 output_dir = Path(output_dir).resolve()
203 output_dir.mkdir(parents=True, exist_ok=True)
205 if all_entries and items:
206 raise click.ClickException('Do not pass filenames when using --all.')
208 mapping = resolve_package_ids_by_version(api_config, DEPS_REGISTRY_PACKAGE_NAME)
209 pkg_id = mapping.get(package_version)
210 if pkg_id is None:
211 CONSOLE.print(f'Not found in registry: {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}')
212 return
214 targets: list[tuple[str, str]] = []
215 if all_entries:
216 package_files = list_generic_package_files(api_config, pkg_id)
217 for pf in package_files:
218 fname = pf.get('file_name')
219 if isinstance(fname, str):
220 parsed = parse_pylock_reference_filename(fname)
221 if parsed is not None:
222 python_version, normalized_name = parsed
223 targets.append((python_version, normalized_name))
224 targets.sort(key=lambda item: tuple(int(part) for part in item[0].split('.')))
225 else:
226 if not items:
227 raise click.ClickException('Provide one or more filenames or Python versions, or use --all.')
228 for item in items:
229 python_version, file_name = _normalize_pylock_input(item)
230 targets.append((python_version, file_name))
232 if not targets:
233 CONSOLE.print(f'No reference files found in {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}.')
234 return
236 for python_version, file_name in targets:
237 dest = download_generic_file(
238 api_config,
239 DEPS_REGISTRY_PACKAGE_NAME,
240 package_version,
241 file_name,
242 output_dir,
243 )
244 if dest is not None:
245 CONSOLE.print(f'Downloaded {python_version}: {dest}')
246 else:
247 CONSOLE.print(f'Not found in registry: {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}/{file_name}')
250@registry.command()
251@click.argument(
252 'items',
253 nargs=-1,
254 required=False,
255 help='Reference filenames or Python versions to delete (e.g. 3.14 or pylock.py3_14_ref.toml)',
256)
257@click.option('-a', '--all', 'all_entries', is_flag=True, default=False, help='Delete all registry files.')
258@click.option(
259 '-y',
260 '--yes',
261 is_flag=True,
262 default=False,
263 help='Skip confirmation prompt and proceed with deletion. Useful for CI/CD automation.',
264)
265@click.option(
266 '--mafw-version',
267 default=None,
268 help='MAFw package version label to use in the registry. Accepts X.Y.Z or vX.Y.Z.',
269)
270@click.pass_context
271def delete(
272 ctx: click.Context, items: tuple[str, ...], all_entries: bool, yes: bool, mafw_version: str | None
273) -> None: # pragma: no cover — thin CLI delegation to tested gitlab.registry
274 """Delete dependency reference files from the GitLab generic registry."""
275 api_config = _registry_api_config(ctx)
276 package_version = _normalize_mafw_version_option(mafw_version)
278 mapping = resolve_package_ids_by_version(api_config, DEPS_REGISTRY_PACKAGE_NAME)
279 pkg_id = mapping.get(package_version)
280 if pkg_id is None:
281 CONSOLE.print(f'Not found in registry: {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}')
282 return
284 if all_entries and items:
285 raise click.ClickException('Do not pass filenames when using --all.')
287 if all_entries:
288 if not yes:
289 confirmed = Confirm.ask(
290 f'Delete all reference files from {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}?',
291 default=False,
292 console=CONSOLE,
293 )
294 if not confirmed:
295 CONSOLE.print('Aborted.')
296 return
298 deleted = delete_generic_package_version(api_config, pkg_id, DEPS_REGISTRY_PACKAGE_NAME, package_version)
299 if deleted:
300 CONSOLE.print(f'Deleted all reference files from {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}')
301 else:
302 CONSOLE.print(f'Not found in registry: {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}')
303 return
305 if not items:
306 raise click.ClickException('Provide one or more filenames or Python versions, or use --all.')
308 targets: list[tuple[str, str]] = []
309 for item in items:
310 python_version, file_name = _normalize_pylock_input(item)
311 targets.append((python_version, file_name))
313 if not yes:
314 file_list = ', '.join(file_name for _, file_name in targets)
315 confirmed = Confirm.ask(
316 f'Delete {file_list} from {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}?',
317 default=False,
318 console=CONSOLE,
319 )
320 if not confirmed:
321 CONSOLE.print('Aborted.')
322 return
324 package_files = list_generic_package_files(api_config, pkg_id)
325 file_name_to_id: dict[str, int] = {}
326 for pf in package_files:
327 fname = pf.get('file_name')
328 fid = pf.get('id')
329 if isinstance(fname, str) and isinstance(fid, int):
330 file_name_to_id[fname] = fid
332 for _python_version, file_name in targets:
333 file_id = file_name_to_id.get(file_name)
334 if file_id is None:
335 CONSOLE.print(f'Not found in registry: {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}/{file_name}')
336 continue
337 deleted = delete_generic_package_file(api_config, pkg_id, file_id)
338 if deleted:
339 CONSOLE.print(f'Deleted {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}/{file_name}')
340 else:
341 CONSOLE.print(f'Not found in registry: {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}/{file_name}')
344@registry.command()
345@click.option(
346 '--keep-newest/--keep-oldest',
347 default=True,
348 show_default=True,
349 help='When duplicates are found, keep the newest (default) or the oldest file.',
350)
351@click.option(
352 '-f',
353 '--force',
354 is_flag=True,
355 default=False,
356 help='Skip confirmation prompt and proceed with pruning. Useful for CI/CD automation.',
357)
358@click.option(
359 '--mafw-version',
360 default=None,
361 help='MAFw package version label to use in the registry. Accepts X.Y.Z or vX.Y.Z.',
362)
363@click.pass_context
364def prune(
365 ctx: click.Context, keep_newest: bool, force: bool, mafw_version: str | None
366) -> None: # pragma: no cover — thin CLI delegation to tested gitlab.registry
367 """Remove duplicated reference files from a registry package version."""
368 api_config = _registry_api_config(ctx)
369 package_version = _normalize_mafw_version_option(mafw_version)
371 mapping = resolve_package_ids_by_version(api_config, DEPS_REGISTRY_PACKAGE_NAME)
372 pkg_id = mapping.get(package_version)
373 if pkg_id is None:
374 CONSOLE.print(f'Not found in registry: {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}')
375 return
377 package_files = list_generic_package_files(api_config, pkg_id)
379 files_by_name: dict[str, list[dict[str, Any]]] = {}
380 for pf in package_files:
381 fname = pf.get('file_name')
382 if isinstance(fname, str):
383 files_by_name.setdefault(fname, []).append(pf)
385 to_delete: list[tuple[str, int, str]] = []
386 for fname, entries in files_by_name.items():
387 if len(entries) <= 1:
388 continue
389 sorted_entries = sorted(entries, key=lambda e: e.get('created_at', ''))
390 if keep_newest:
391 duplicates = sorted_entries[:-1]
392 else:
393 duplicates = sorted_entries[1:]
394 for entry in duplicates:
395 fid = entry.get('id')
396 created_at = entry.get('created_at', 'unknown')
397 if isinstance(fid, int):
398 to_delete.append((fname, fid, str(created_at)))
400 if not to_delete:
401 CONSOLE.print(f'No duplicates found in {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}.')
402 return
404 keep_label = 'newest' if keep_newest else 'oldest'
405 CONSOLE.print(
406 f'Found {len(to_delete)} duplicate(s) to remove '
407 f'(keeping {keep_label}) in {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}:'
408 )
409 for fname, _fid, created_at in to_delete:
410 CONSOLE.print(f' - {fname} (created: {created_at})')
412 if not force:
413 confirmed = Confirm.ask(
414 'Proceed with deletion?',
415 default=False,
416 console=CONSOLE,
417 )
418 if not confirmed:
419 CONSOLE.print('Aborted.')
420 return
422 deleted_count = 0
423 for fname, fid, created_at in to_delete:
424 deleted = delete_generic_package_file(api_config, pkg_id, fid)
425 if deleted:
426 CONSOLE.print(f'Deleted {fname} (created: {created_at})')
427 deleted_count += 1
428 else:
429 CONSOLE.print(f'Failed to delete {fname} (id={fid})')
431 CONSOLE.print(f'Pruning complete: {deleted_count} duplicate(s) removed.')