Coverage for src/mafw/devtools/dependencies/freeze.py: 83%
170 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-26 09:13 +0000
1# Copyright 2026 European Union
2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
3# SPDX-License-Identifier: EUPL-1.2
4"""
5Dependency freezing and unfreezing utilities for MAFw.
7This module provides functions for adding and removing computed upper-bound
8constraints in ``pyproject.toml`` dependency declarations. It is used by
9the release workflow to pin dependencies during release and unpin them
10afterwards.
11"""
13from __future__ import annotations
15from collections.abc import Callable
16from typing import Any, Final
18import tomlkit
20from mafw.devtools import ensure_devtools_available
22ensure_devtools_available()
24from packaging.requirements import Requirement # noqa: E402
25from packaging.specifiers import Specifier, SpecifierSet # noqa: E402
26from packaging.version import Version # noqa: E402
28from mafw.devtools import DevtoolsError # noqa: E402
29from mafw.devtools.dependencies.compile import ( # noqa: E402
30 PYPROJECT_FILE,
31 collect_compiled_dependency_versions,
32 load_pyproject_doc,
33 project_python_versions_from_doc,
34)
35from mafw.devtools.documentation.requirements import REQUIREMENTS_GROUPS # noqa: E402
36from mafw.tools.shell_tools import CONSOLE # noqa: E402
37from mafw.tools.shell_tools import run as cmd # noqa: E402
39_FROZEN_OPERATORS: Final[set[str]] = {'<', '<=', '~=', '==', '==='}
40"""Operators that already constrain the maximum compatible version and should not be auto-frozen."""
43def compute_upper_bound(lower_bound: str) -> str:
44 """
45 Compute an upper bound for a dependency based on PEP 440 compatible-release philosophy.
47 The rule implemented here is purposely conservative and mirrors the intent of compatible
48 release clauses while remaining explicit:
50 - For major-versioned releases (``X.*`` with ``X > 0``), freeze to ``<(X + 1)``.
51 - For ``0.*`` releases, freeze to ``<0.(minor + 1)`` (rolling compatibility during pre-1.0).
53 :param lower_bound: Version string used as the starting point for the freeze rule.
54 :type lower_bound: str
55 :return: Upper-bound version string without operator.
56 :rtype: str
57 :raises DevtoolsError: If the version cannot be parsed.
58 """
59 try:
60 version = Version(lower_bound)
61 except Exception as exc: # pragma: no cover
62 raise DevtoolsError(f'Unable to parse version "{lower_bound}" while computing dependency upper bound.') from exc
64 if version.major > 0:
65 return str(version.major + 1)
66 return f'0.{version.minor + 1}'
69def format_requirement(requirement: Requirement) -> str:
70 """
71 Serialize a packaging requirement object back to a PEP 508 compatible string.
73 :param requirement: Parsed requirement instance.
74 :type requirement: Requirement
75 :return: PEP 508 requirement string.
76 :rtype: str
77 """
78 name = str(requirement.name)
79 extras = sorted(str(extra) for extra in getattr(requirement, 'extras', set()) or set())
80 if extras:
81 name = f'{name}[{",".join(extras)}]'
83 if requirement.url: 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true
84 rendered = f'{name} @ {requirement.url}'
85 else:
86 spec = str(requirement.specifier).strip()
87 rendered = f'{name}{spec}' if spec else name
89 if requirement.marker:
90 rendered = f'{rendered} ; {requirement.marker}'
91 return rendered
94def iter_specifiers(requirement: Requirement) -> list[Specifier]:
95 """
96 Return a concrete list of specifiers for the given requirement.
98 :param requirement: Parsed requirement instance.
99 :type requirement: Requirement
100 :return: List of specifier objects.
101 :rtype: list[Specifier]
102 """
103 return list(requirement.specifier) if requirement.specifier else []
106def has_frozen_upper_bound(requirement: Requirement) -> bool:
107 """
108 Determine whether a requirement already contains an upper bound constraint.
110 :param requirement: Parsed requirement instance.
111 :type requirement: Requirement
112 :return: ``True`` if the requirement is already frozen.
113 :rtype: bool
114 """
115 return any(getattr(spec, 'operator', '') in _FROZEN_OPERATORS for spec in iter_specifiers(requirement))
118def highest_lower_bound(requirement: Requirement) -> str | None:
119 """
120 Extract the highest lower-bound version from ``>=`` and ``>`` specifiers.
122 :param requirement: Parsed requirement instance.
123 :type requirement: Requirement
124 :return: Highest lower bound version string, or ``None`` if missing.
125 :rtype: str | None
126 :raises DevtoolsError: If version parsing fails.
127 """
128 best: tuple[Version, str] | None = None
129 for spec in iter_specifiers(requirement):
130 if spec.operator not in {'>=', '>'}:
131 continue
132 try:
133 parsed = Version(spec.version)
134 except Exception as exc: # pragma: no cover
135 raise DevtoolsError(
136 f'Unable to parse lower bound "{spec.version}" in requirement "{requirement}".'
137 ) from exc
138 if best is None or parsed > best[0]: 138 ↛ 129line 138 didn't jump to line 129 because the condition on line 138 was always true
139 best = (parsed, spec.version)
140 return best[1] if best else None
143def freeze_requirement(requirement_text: str, *, resolved_version: str | None = None) -> tuple[str, list[str]]:
144 """
145 Add a computed upper bound to a requirement string, if eligible.
147 The function is intentionally conservative:
149 - URL-based requirements are skipped (cannot be version constrained).
150 - Requirements already containing ``<``, ``<=``, ``~=``, ``==`` or ``===`` are skipped.
151 - Requirements without a lower bound emit a warning and are left unchanged.
152 - When a resolved version is provided, the upper bound is computed from that
153 version instead of the declared lower bound.
155 :param requirement_text: Raw PEP 508 requirement string.
156 :type requirement_text: str
157 :return: Updated requirement text plus warnings.
158 :rtype: tuple[str, list[str]]
159 :raises DevtoolsError: If parsing fails.
160 """
161 warnings: list[str] = []
162 stripped = requirement_text.strip()
163 if not stripped: 163 ↛ 164line 163 didn't jump to line 164 because the condition on line 163 was never true
164 return requirement_text, warnings
166 try:
167 requirement = Requirement(stripped)
168 except Exception as exc:
169 raise DevtoolsError(f'Unable to parse dependency requirement "{requirement_text}".') from exc
171 if requirement.url:
172 warnings.append(f'Skipping URL requirement (cannot freeze): {requirement_text}')
173 return requirement_text, warnings
175 if has_frozen_upper_bound(requirement):
176 return requirement_text, warnings
178 lower = highest_lower_bound(requirement)
179 if lower is None:
180 warnings.append(f'Dependency has no lower bound and will not be frozen: {requirement_text}')
181 return requirement_text, warnings
183 upper = compute_upper_bound(resolved_version or lower)
184 combined = SpecifierSet(f'{requirement.specifier},<{upper}' if str(requirement.specifier).strip() else f'<{upper}')
185 requirement.specifier = combined
186 return format_requirement(requirement), warnings
189def unfreeze_requirement(requirement_text: str) -> tuple[str, list[str]]:
190 """
191 Remove a computed upper bound from a requirement string, if it matches the computed rule.
193 Only the auto-generated upper bound ``<upper`` is removed; existing manual upper bounds are preserved.
195 :param requirement_text: Raw PEP 508 requirement string.
196 :type requirement_text: str
197 :return: Updated requirement text plus warnings.
198 :rtype: tuple[str, list[str]]
199 :raises DevtoolsError: If parsing fails.
200 """
201 warnings: list[str] = []
202 stripped = requirement_text.strip()
203 if not stripped: 203 ↛ 204line 203 didn't jump to line 204 because the condition on line 203 was never true
204 return requirement_text, warnings
206 try:
207 requirement = Requirement(stripped)
208 except Exception as exc:
209 raise DevtoolsError(f'Unable to parse dependency requirement "{requirement_text}".') from exc
211 if requirement.url: 211 ↛ 212line 211 didn't jump to line 212 because the condition on line 211 was never true
212 return requirement_text, warnings
214 # Do not try to unfreeze already pinned/compatible requirements, or requirements that already
215 # had an upper bound before freezing (we cannot safely distinguish manual bounds).
216 specifiers = iter_specifiers(requirement)
217 if any(spec.operator in {'~=', '==', '==='} for spec in specifiers):
218 return requirement_text, warnings
219 lower = highest_lower_bound(requirement)
220 if lower is None: 220 ↛ 221line 220 didn't jump to line 221 because the condition on line 220 was never true
221 return requirement_text, warnings
223 if any(spec.operator == '<=' for spec in specifiers): 223 ↛ 224line 223 didn't jump to line 224 because the condition on line 223 was never true
224 return requirement_text, warnings
226 remaining: list[Specifier] = []
227 removed = False
228 for spec in specifiers:
229 if spec.operator == '<':
230 removed = True
231 continue
232 remaining.append(spec)
234 if not removed: 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true
235 return requirement_text, warnings
237 requirement.specifier = SpecifierSet(','.join(str(spec) for spec in remaining))
238 return format_requirement(requirement), warnings
241def update_dependency_list(
242 dependencies: Any,
243 *,
244 transformer: Callable[[str], tuple[str, list[str]]],
245 warnings: list[str],
246 context: str,
247) -> None:
248 """
249 Update a TOML list of dependency strings in-place.
251 :param dependencies: TOML array that contains dependency strings.
252 :type dependencies: Any
253 :param transformer: Callable applied to each dependency string.
254 :type transformer: Any
255 :param warnings: List of warnings to append to.
256 :type warnings: list[str]
257 :param context: Human-readable location for warnings.
258 :type context: str
259 """
260 for idx, item in enumerate(list(dependencies)):
261 if not isinstance(item, str): 261 ↛ 262line 261 didn't jump to line 262 because the condition on line 261 was never true
262 warnings.append(f'Skipping non-string dependency entry in {context}: {item!r}')
263 continue
264 updated, item_warnings = transformer(item)
265 warnings.extend(item_warnings)
266 dependencies[idx] = updated
269def freeze_pyproject_toml(
270 toml_text: str,
271 *,
272 resolved_versions: dict[str, Version] | None = None,
273 doc: tomlkit.TOMLDocument | None = None,
274) -> tuple[str, list[str]]:
275 """
276 Freeze dependencies in a ``pyproject.toml`` payload by adding upper bounds.
278 The TOML structure is preserved via tomlkit; only dependency strings may be normalized.
280 When ``resolved_versions`` is provided, the function uses those compiled
281 versions as the basis for upper-bound computation. Otherwise the declared
282 lower bounds are used as a fallback.
284 :param toml_text: Raw TOML file content.
285 :type toml_text: str
286 :param resolved_versions: Optional mapping of dependency names to resolved versions.
287 :type resolved_versions: dict[str, Version] | None
288 :param doc: Optional pre-parsed TOML document to reuse when available.
289 :type doc: tomlkit.TOMLDocument | None
290 :return: Updated TOML plus warnings.
291 :rtype: tuple[str, list[str]]
292 :raises DevtoolsError: If TOML parsing fails.
293 """
294 warnings: list[str] = []
295 if doc is None:
296 doc = load_pyproject_doc(toml_text)
298 project = doc.get('project')
299 if project is None: 299 ↛ 300line 299 didn't jump to line 300 because the condition on line 299 was never true
300 raise DevtoolsError(f'Missing [project] table in {PYPROJECT_FILE}.')
302 def freeze_for_requirement(requirement_text: str) -> tuple[str, list[str]]:
303 try:
304 requirement = Requirement(requirement_text.strip())
305 except Exception as exc:
306 raise DevtoolsError(f'Unable to parse dependency requirement "{requirement_text}".') from exc
308 resolved_version = None
309 if resolved_versions is not None: 309 ↛ 313line 309 didn't jump to line 313 because the condition on line 309 was always true
310 resolved = resolved_versions.get(requirement.name.lower())
311 if resolved is not None:
312 resolved_version = str(resolved)
313 return freeze_requirement(requirement_text, resolved_version=resolved_version)
315 dependencies = project.get('dependencies')
316 if dependencies is not None: 316 ↛ 324line 316 didn't jump to line 324 because the condition on line 316 was always true
317 update_dependency_list(
318 dependencies,
319 transformer=freeze_for_requirement,
320 warnings=warnings,
321 context='project.dependencies',
322 )
324 optional = project.get('optional-dependencies')
325 if optional is not None:
326 for group, group_deps in optional.items():
327 update_dependency_list(
328 group_deps,
329 transformer=freeze_for_requirement,
330 warnings=warnings,
331 context=f'project.optional-dependencies.{group}',
332 )
334 return tomlkit.dumps(doc), warnings
337def unfreeze_pyproject_toml(
338 toml_text: str,
339 *,
340 baseline_toml_text: str | None = None,
341 doc: tomlkit.TOMLDocument | None = None,
342 baseline_doc: tomlkit.TOMLDocument | None = None,
343) -> tuple[str, list[str]]:
344 """
345 Unfreeze dependencies in a ``pyproject.toml`` payload by removing computed upper bounds.
347 Only upper bounds matching the computed rule are removed; existing manual constraints remain.
349 :param toml_text: Raw TOML file content.
350 :type toml_text: str
351 :param baseline_toml_text: Optional original TOML text captured before freezing.
352 When provided, unfreezing is computed against the baseline to avoid
353 altering dependencies that were already frozen before release.
354 :type baseline_toml_text: str | None
355 :param doc: Optional pre-parsed TOML document to reuse when available.
356 :type doc: tomlkit.TOMLDocument | None
357 :param baseline_doc: Optional pre-parsed baseline TOML document to reuse when available.
358 :type baseline_doc: tomlkit.TOMLDocument | None
359 :return: Updated TOML plus warnings.
360 :rtype: tuple[str, list[str]]
361 :raises DevtoolsError: If TOML parsing fails.
362 """
363 warnings: list[str] = []
364 if doc is None:
365 doc = load_pyproject_doc(toml_text)
366 if baseline_toml_text is not None:
367 if baseline_doc is None:
368 baseline_doc = load_pyproject_doc(baseline_toml_text)
370 current_project = doc.get('project')
371 baseline_project = baseline_doc.get('project')
372 if current_project is None or baseline_project is None: 372 ↛ 373line 372 didn't jump to line 373 because the condition on line 372 was never true
373 raise DevtoolsError(f'Missing [project] table in {PYPROJECT_FILE}.')
375 current_project['dependencies'] = baseline_project.get('dependencies', current_project.get('dependencies'))
376 current_optional = current_project.get('optional-dependencies')
377 baseline_optional = baseline_project.get('optional-dependencies')
378 if current_optional is not None and baseline_optional is not None:
379 for group in current_optional:
380 if group in baseline_optional: 380 ↛ 379line 380 didn't jump to line 379 because the condition on line 380 was always true
381 current_optional[group] = baseline_optional[group]
382 return tomlkit.dumps(doc), warnings
384 project = doc.get('project')
385 if project is None: 385 ↛ 386line 385 didn't jump to line 386 because the condition on line 385 was never true
386 raise DevtoolsError(f'Missing [project] table in {PYPROJECT_FILE}.')
388 dependencies = project.get('dependencies')
389 if dependencies is not None: 389 ↛ 397line 389 didn't jump to line 397 because the condition on line 389 was always true
390 update_dependency_list(
391 dependencies,
392 transformer=unfreeze_requirement,
393 warnings=warnings,
394 context='project.dependencies',
395 )
397 optional = project.get('optional-dependencies')
398 if optional is not None: 398 ↛ 399line 398 didn't jump to line 399 because the condition on line 398 was never true
399 for group, group_deps in optional.items():
400 update_dependency_list(
401 group_deps,
402 transformer=unfreeze_requirement,
403 warnings=warnings,
404 context=f'project.optional-dependencies.{group}',
405 )
407 return tomlkit.dumps(doc), warnings
410def summarize_freeze_changes(before: str, after: str) -> str:
411 """
412 Build a short summary for dry-run output.
414 :param before: Original TOML content.
415 :type before: str
416 :param after: Updated TOML content.
417 :type after: str
418 :return: Human-readable summary line.
419 :rtype: str
420 """
421 if before == after: 421 ↛ 423line 421 didn't jump to line 423 because the condition on line 421 was always true
422 return 'No dependency constraints would be updated.'
423 before_lines = before.splitlines()
424 after_lines = after.splitlines()
425 return f'pyproject.toml would be updated ({len(before_lines)} -> {len(after_lines)} lines).'
428def freeze_dependencies(*, dry_run: bool) -> str: # pragma: no cover
429 """
430 Freeze dependencies by adding computed upper bounds in ``pyproject.toml``.
432 :param dry_run: Whether command execution is disabled.
433 :type dry_run: bool
434 :return: Original ``pyproject.toml`` content captured before freeze.
435 :rtype: str
436 """
437 CONSOLE.print('Freezing dependencies (adding upper bounds) ...')
438 before = PYPROJECT_FILE.read_text(encoding='utf-8')
439 doc = load_pyproject_doc(before)
440 supported_python_versions = project_python_versions_from_doc(doc)
441 resolved_versions = collect_compiled_dependency_versions(supported_python_versions)
442 after, warnings = freeze_pyproject_toml(before, resolved_versions=resolved_versions, doc=doc)
443 for warning in warnings:
444 CONSOLE.print(f'WARNING: {warning}')
446 if dry_run:
447 CONSOLE.print(summarize_freeze_changes(before, after))
448 return before
449 if before != after:
450 PYPROJECT_FILE.write_text(after, encoding='utf-8')
451 return before
454def unfreeze_dependencies(original_pyproject_toml: str, *, dry_run: bool) -> None: # pragma: no cover
455 """
456 Unfreeze dependencies by removing computed upper bounds in ``pyproject.toml``.
458 :param dry_run: Whether command execution is disabled.
459 :type dry_run: bool
460 """
461 CONSOLE.print('Unfreezing dependencies (removing computed upper bounds) ...')
462 before = PYPROJECT_FILE.read_text(encoding='utf-8')
463 doc = load_pyproject_doc(before)
464 baseline_doc = load_pyproject_doc(original_pyproject_toml)
465 after, warnings = unfreeze_pyproject_toml(
466 before,
467 baseline_toml_text=original_pyproject_toml,
468 doc=doc,
469 baseline_doc=baseline_doc,
470 )
471 for warning in warnings:
472 CONSOLE.print(f'WARNING: {warning}')
474 if dry_run:
475 CONSOLE.print(summarize_freeze_changes(before, after))
476 return
477 if before != after:
478 PYPROJECT_FILE.write_text(after, encoding='utf-8')
481def update_requirements_and_readme(*, dry_run: bool) -> None: # pragma: no cover
482 """
483 Update the requirements RST files and README.rst.
485 :param dry_run: Whether command execution is disabled.
486 :type dry_run: bool
487 """
488 CONSOLE.print('Updating requirements and README.rst...')
489 cmd(['hatch', 'run', 'dev:multidoc', 'requirements', '--update-readme', *REQUIREMENTS_GROUPS], dry_run=dry_run)