# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Structure-aware ``pyproject.toml`` synchronization service.
This module implements :class:`PyprojectSyncService`, the component that bridges
the pure Python version policy core and the on-disk ``pyproject.toml``. It uses
:mod:`tomlkit` so that comments, whitespace, array formatting and key ordering
are preserved, mutates **only** the managed derived locations in place (never by
whole-file string replacement), writes atomically so a failure leaves the file
byte-for-byte unchanged, and offers a shared ``--dry-run`` driver that renders a
unified diff without touching the file.
The managed derived locations and their canonical sources are:
================================================ ==============================
TOML location Canonical source
================================================ ==============================
``project.classifiers`` ``supported-python``
``tool.hatch.envs.hatch-test.matrix`` (standard) ``supported-python``
``tool.hatch.envs.hatch-test.matrix`` (free-thr) ``free-threaded-python``
``tool.hatch.envs.dev.matrix`` ``supported-python``
``tool.hatch.envs.types.matrix`` ``supported-python``
``tool.hatch.envs.nogil.python`` ``free-threaded-python``
``tool.hatch.envs.gil.python`` base of ``free-threaded-python``
``tool.hatch.envs.experimental.matrix`` ``experimental-python``
``tool.hatch.envs.default.pre-push`` (default) ``default-python``
``tool.hatch.envs.default.pre-push`` (free-thr) ``free-threaded-python``
================================================ ==============================
.. versionadded:: 2.4
:author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
"""
from __future__ import annotations
import difflib
import os
import tempfile
from collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
import tomlkit
from tomlkit import TOMLDocument
from mafw.devtools import DevtoolsError
from mafw.devtools.pyversions.config import PythonConfig
from mafw.devtools.pyversions.derivations import (
PrePushClassification,
classify_pre_push,
desired_classifiers,
desired_dev_matrix,
desired_experimental_matrix,
desired_free_threaded_matrix,
desired_gil_python,
desired_nogil_python,
desired_standard_matrix,
desired_types_matrix,
retarget_default_command,
retarget_free_threaded_command,
)
_CLASSIFIER_PREFIX = 'Programming Language :: Python :: '
"""Prefix of the per-version Python classifiers."""
[docs]
@dataclass(frozen=True, slots=True)
class DerivedChange:
"""A single managed derived-location modification, used for reporting.
:ivar location: Human-readable TOML key path that changed.
:ivar previous: The previous value, rendered as a string.
:ivar new: The new value, rendered as a string.
"""
location: str
previous: str
new: str
[docs]
@dataclass(slots=True)
class SyncReport:
"""The outcome of a synchronization pass.
:ivar changes: The managed locations that were modified.
:ivar skipped: Locations that could not be classified and were left untouched.
:ivar notices: Informational messages for the operator about structural
changes that are not simple value edits (for example the removal of an
environment table).
"""
changes: list[DerivedChange] = field(default_factory=list)
skipped: list[str] = field(default_factory=list)
notices: list[str] = field(default_factory=list)
@property
def modified(self) -> bool:
"""Whether any managed location was modified."""
return bool(self.changes)
[docs]
@dataclass(slots=True)
class SyncOutcome:
"""The result of running a full operation through the service.
:ivar report: The per-location change report.
:ivar diff: The unified diff of ``pyproject.toml`` (empty if nothing changed).
:ivar written: Whether the file was actually written (``False`` for a
dry-run or a no-op).
"""
report: SyncReport
diff: str
written: bool
# A transform maps the loaded canonical configuration to the desired one. ``sync``
# uses the identity transform; the other commands use a VersionManager operation.
Transform = Callable[[PythonConfig], PythonConfig]
[docs]
class PyprojectSyncService:
"""Load, transform, synchronize and persist ``pyproject.toml``.
:param path: Path to the ``pyproject.toml`` file to manage.
:type path: Path
"""
def __init__(self, path: Path) -> None:
self._path = path
# ------------------------------------------------------------------
# Loading / rendering / diffing
# ------------------------------------------------------------------
[docs]
def load(self) -> tuple[TOMLDocument, PythonConfig]:
"""Parse the document and read the canonical configuration.
:return: The parsed tomlkit document and the (unvalidated) configuration.
:rtype: tuple[tomlkit.TOMLDocument, PythonConfig]
:raises DevtoolsError: If the file is missing or cannot be parsed.
"""
if not self._path.exists():
raise DevtoolsError(f'pyproject.toml not found at {self._path}.')
try:
doc = tomlkit.parse(self._path.read_text(encoding='utf-8'))
except Exception as exc: # tomlkit raises a variety of parse errors
raise DevtoolsError(f'Failed to parse {self._path}: {exc}') from exc
cfg = PythonConfig.from_document(doc)
return doc, cfg
[docs]
@staticmethod
def render(doc: TOMLDocument) -> str:
"""Serialize a tomlkit document back to text, preserving formatting."""
return tomlkit.dumps(doc)
[docs]
def diff(self, before: str, after: str) -> str:
"""Return a unified diff between two renderings of the document."""
before_lines = before.splitlines(keepends=True)
after_lines = after.splitlines(keepends=True)
name = self._path.name
return ''.join(difflib.unified_diff(before_lines, after_lines, fromfile=f'a/{name}', tofile=f'b/{name}'))
# ------------------------------------------------------------------
# Atomic write
# ------------------------------------------------------------------
[docs]
def write_atomic(self, text: str) -> None:
"""Write *text* to ``pyproject.toml`` atomically.
The content is first written to a temporary file in the *same* directory
and then moved into place with ``os.replace``, which is atomic on the
same filesystem. Any failure before the replacement leaves the original
file byte-for-byte unchanged.
:param text: The full new file content.
:raises DevtoolsError: If writing or replacing the file fails.
"""
directory = self._path.parent
tmp_path: str | None = None
try:
with tempfile.NamedTemporaryFile(
mode='w', encoding='utf-8', dir=directory, prefix='.pyproject-', suffix='.toml.tmp', delete=False
) as handle:
tmp_path = handle.name
handle.write(text)
os.replace(tmp_path, self._path)
tmp_path = None
except OSError as exc:
raise DevtoolsError(f'Failed to write {self._path}: {exc}') from exc
finally:
# Clean up the temporary file if the replace did not consume it.
if tmp_path is not None and os.path.exists(tmp_path):
os.unlink(tmp_path)
# ------------------------------------------------------------------
# Synchronization
# ------------------------------------------------------------------
[docs]
def synchronize(self, doc: TOMLDocument, cfg: PythonConfig, *, previous: PythonConfig | None = None) -> SyncReport:
"""Update every managed derived location in *doc* to match *cfg*.
Mutates the document in place using structured tomlkit assignment so
that unrelated content, comments and formatting survive. Returns a report
of the per-location previous → new changes and of any pre-push command
that could not be classified (and was therefore left untouched).
:param doc: The tomlkit document to mutate.
:param cfg: The validated canonical configuration to project.
:param previous: The configuration *before* the transform, used to detect
which stable selector managed pre-push commands currently carry. When
omitted, *cfg* is used (appropriate for a pure ``sync``).
:return: The change report.
"""
report = SyncReport()
previous = previous if previous is not None else cfg
self._sync_classifiers(doc, cfg, report)
self._sync_hatch_test_matrix(doc, cfg, report)
self._sync_simple_matrix(doc, cfg, report, env='dev', desired=desired_dev_matrix(cfg))
self._sync_simple_matrix(doc, cfg, report, env='types', desired=desired_types_matrix(cfg))
self._sync_env_python(doc, report, env='nogil', desired=desired_nogil_python(cfg))
self._sync_env_python(doc, report, env='gil', desired=desired_gil_python(cfg))
self._sync_experimental_matrix(doc, cfg, report)
self._sync_pre_push(doc, cfg, report, previous_default=previous.default)
return report
[docs]
def run_operation(self, transform: Transform, *, dry_run: bool) -> SyncOutcome:
"""Load, transform, synchronize and (optionally) persist the document.
This is the shared driver behind every modifying command. Because dry-run
and real runs share the same load → transform → validate → synchronize
pipeline, a dry-run reports exactly the same validation errors and the
same diff a real run would produce — it simply does not write.
:param transform: Maps the current canonical configuration to the desired
one. For ``sync`` this is the identity; for the other commands it is a
:class:`~mafw.devtools.pyversions.manager.VersionManager` operation.
:param dry_run: If ``True``, compute and return the diff without writing.
:return: The operation outcome.
:raises DevtoolsError: On parse errors, invalid canonical configuration,
rejected transforms, or write failures.
"""
before_text = self._path.read_text(encoding='utf-8') if self._path.exists() else ''
doc, cfg = self.load()
# The canonical configuration must be valid before any change is applied;
# an invalid source is never silently repaired.
cfg.validate()
desired = transform(cfg)
desired.validate()
# Write the desired canonical values back before projecting derived ones.
self._write_canonical(doc, desired)
report = self.synchronize(doc, desired, previous=cfg)
after_text = self.render(doc)
diff = self.diff(before_text, after_text)
if not diff:
# Nothing changed: idempotent no-op, never write.
return SyncOutcome(report=report, diff='', written=False)
if dry_run:
return SyncOutcome(report=report, diff=diff, written=False)
self.write_atomic(after_text)
return SyncOutcome(report=report, diff=diff, written=True)
[docs]
def check(self) -> SyncReport:
"""Compute the inconsistencies between canonical config and derived state.
This is the read-only dual of :meth:`synchronize`: it loads the document,
validates the canonical configuration, then computes what a
synchronization *would* change. Every would-be change is an existing
inconsistency. The document on disk is never modified.
:return: A report whose ``changes`` list enumerates the inconsistencies
(empty when the configuration is fully consistent).
:raises DevtoolsError: If the file is missing, unparsable, or the
canonical configuration is invalid.
"""
doc, cfg = self.load()
cfg.validate()
# Synchronize a working copy; the report captures the drift without
# persisting anything.
return self.synchronize(doc, cfg, previous=cfg)
# ------------------------------------------------------------------
# Canonical writers
# ------------------------------------------------------------------
[docs]
@staticmethod
def _write_canonical(doc: TOMLDocument, cfg: PythonConfig) -> None:
"""Write the canonical ``[tool.mafw]`` keys from *cfg* into *doc*."""
tool_mafw = doc['tool']['mafw']
tool_mafw['supported-python'] = list(cfg.supported)
tool_mafw['experimental-python'] = list(cfg.experimental)
tool_mafw['default-python'] = cfg.default
tool_mafw['free-threaded-python'] = cfg.free_threaded
# ------------------------------------------------------------------
# Derived-location writers
# ------------------------------------------------------------------
[docs]
@staticmethod
def _record(report: SyncReport, location: str, previous: Any, new: Any) -> None:
"""Record a change only when the rendered value actually differs."""
prev_str = repr(previous)
new_str = repr(new)
if prev_str != new_str:
report.changes.append(DerivedChange(location=location, previous=prev_str, new=new_str))
[docs]
@staticmethod
def _assign_array(container: Any, key: str, values: list[str]) -> None:
"""Replace the string items of a TOML array while preserving its style.
Reusing the existing :class:`tomlkit.items.Array` (rather than assigning a
plain Python list) keeps the array's multiline-or-single-line layout and
surrounding whitespace intact, which is what keeps the diff minimal.
"""
existing = container[key]
if isinstance(existing, tomlkit.items.Array):
was_multiline = '\n' in existing.as_string()
existing.clear()
for value in values:
existing.append(value)
if was_multiline:
existing.multiline(True)
else:
container[key] = values
[docs]
def _sync_classifiers(self, doc: TOMLDocument, cfg: PythonConfig, report: SyncReport) -> None:
"""Rewrite the per-version Python classifiers, keeping all other entries."""
project = doc.get('project')
if project is None or 'classifiers' not in project:
return
classifiers = project['classifiers']
existing = [str(c) for c in classifiers]
# Partition into managed (per-version) and everything else, preserving order.
non_version = [c for c in existing if not self._is_version_classifier(c)]
desired_versions = desired_classifiers(cfg)
# Re-insert the version classifiers at the position of the first existing one
# (or appended after the generic "Programming Language :: Python" entry).
rebuilt = self._rebuild_classifiers(existing, non_version, desired_versions)
if rebuilt != existing:
self._record(report, 'project.classifiers (python versions)', existing, rebuilt)
self._assign_array(project, 'classifiers', rebuilt)
[docs]
@staticmethod
def _is_version_classifier(classifier: str) -> bool:
"""Whether *classifier* is a per-minor-version Python classifier."""
if not classifier.startswith(_CLASSIFIER_PREFIX):
return False
tail = classifier[len(_CLASSIFIER_PREFIX) :]
# Managed entries look like "3.14"; "Implementation :: CPython" is not managed.
return bool(tail) and tail[0].isdigit()
[docs]
def _rebuild_classifiers(
self, existing: list[str], non_version: list[str], desired_versions: list[str]
) -> list[str]:
"""Rebuild the classifier list with managed version entries resynchronised."""
# Find where the first managed version classifier currently sits so the
# rewritten block keeps the same position within the array.
insert_at = next((i for i, c in enumerate(existing) if self._is_version_classifier(c)), None)
if insert_at is None:
# No managed classifiers yet: append after the generic Python entry if present.
result = list(non_version)
anchor = next((i for i, c in enumerate(result) if c.strip() == 'Programming Language :: Python'), None)
position = anchor + 1 if anchor is not None else len(result)
return result[:position] + desired_versions + result[position:]
# Map the insertion point into the non_version list to preserve ordering.
preceding = [c for c in existing[:insert_at] if not self._is_version_classifier(c)]
following = [c for c in existing[insert_at:] if not self._is_version_classifier(c)]
return preceding + desired_versions + following
[docs]
def _sync_hatch_test_matrix(self, doc: TOMLDocument, cfg: PythonConfig, report: SyncReport) -> None:
"""Synchronize the standard and free-threaded ``hatch-test`` matrix rows."""
matrix = self._matrix_rows(doc, 'hatch-test')
if matrix is None:
return
for index, row in enumerate(matrix):
if 'python' not in row:
continue
current = [str(v) for v in row['python']]
if self._is_free_threaded_row(current):
desired = desired_free_threaded_matrix(cfg)
location = 'tool.hatch.envs.hatch-test.matrix (free-threaded).python'
else:
desired = desired_standard_matrix(cfg)
location = 'tool.hatch.envs.hatch-test.matrix (standard).python'
if current != desired:
self._record(report, location, current, desired)
self._assign_array(row, 'python', desired)
[docs]
@staticmethod
def _is_free_threaded_row(versions: list[str]) -> bool:
"""Whether a matrix row is the dedicated free-threaded row (``X.Yt`` only)."""
return bool(versions) and all(v.endswith('t') for v in versions)
[docs]
def _sync_simple_matrix(
self, doc: TOMLDocument, cfg: PythonConfig, report: SyncReport, *, env: str, desired: list[str]
) -> None:
"""Synchronize the single ``python`` matrix row of a simple environment."""
matrix = self._matrix_rows(doc, env)
if matrix is None:
return
for row in matrix:
if 'python' not in row:
continue
current = [str(v) for v in row['python']]
if current != desired:
self._record(report, f'tool.hatch.envs.{env}.matrix.python', current, desired)
self._assign_array(row, 'python', desired)
[docs]
def _sync_experimental_matrix(self, doc: TOMLDocument, cfg: PythonConfig, report: SyncReport) -> None:
"""Create or update the dedicated ``experimental`` environment matrix.
The environment is created on demand so that experimental versions have a
home that never expands the mandatory matrices.
"""
envs = doc.get('tool', {}).get('hatch', {}).get('envs')
if envs is None:
return
desired = desired_experimental_matrix(cfg)
experimental_env = envs.get('experimental')
# No experimental versions: the environment must not survive with an empty
# matrix (``python = []`` crashes Hatch). Remove the whole environment table
# when it exists, and tell the operator it is being dropped entirely.
if not desired:
if experimental_env is not None:
del envs['experimental']
self._record(report, 'tool.hatch.envs.experimental.matrix.python', ['<...>'], [])
report.notices.append(
'No experimental versions remain; the [tool.hatch.envs.experimental] environment '
'has been removed entirely from pyproject.toml.'
)
return
if experimental_env is None:
# Build a new environment table with a single matrix row. tomlkit does
# not manage blank lines around a freshly inserted array-of-tables, so
# a trailing newline is appended to keep the following table visually
# separated and the file style consistent.
row = tomlkit.table()
row['python'] = desired
matrix_array = tomlkit.aot()
matrix_array.append(row)
env_table = tomlkit.table()
env_table['matrix'] = matrix_array
envs['experimental'] = env_table
envs['experimental'].add(tomlkit.nl())
self._record(report, 'tool.hatch.envs.experimental.matrix.python', [], desired)
return
matrix = experimental_env.get('matrix')
if matrix is None:
row = tomlkit.table()
row['python'] = desired
matrix_array = tomlkit.aot()
matrix_array.append(row)
experimental_env['matrix'] = matrix_array
self._record(report, 'tool.hatch.envs.experimental.matrix.python', [], desired)
return
for row in matrix:
if 'python' not in row:
continue
current = [str(v) for v in row['python']]
if current != desired:
self._record(report, 'tool.hatch.envs.experimental.matrix.python', current, desired)
self._assign_array(row, 'python', desired)
[docs]
@staticmethod
def _sync_env_python(doc: TOMLDocument, report: SyncReport, *, env: str, desired: str) -> None:
"""Synchronize a scalar ``python`` key of an environment (``nogil`` / ``gil``)."""
envs = doc.get('tool', {}).get('hatch', {}).get('envs')
if envs is None or env not in envs:
return
env_table = envs[env]
if 'python' not in env_table:
return
current = str(env_table['python'])
if current != desired:
report.changes.append(
DerivedChange(location=f'tool.hatch.envs.{env}.python', previous=repr(current), new=repr(desired))
)
env_table['python'] = desired
[docs]
def _sync_pre_push(
self, doc: TOMLDocument, cfg: PythonConfig, report: SyncReport, *, previous_default: str
) -> None:
"""Rewrite managed pre-push commands, leaving pins and unknowns untouched."""
envs = doc.get('tool', {}).get('hatch', {}).get('envs')
if envs is None:
return
default_env = envs.get('default')
if default_env is None:
return
# The ``pre-push`` array may live directly under ``[tool.hatch.envs.default]``
# or under its ``extra-scripts`` subtable, depending on where it is written
# in the file. Locate whichever container actually holds it.
container = None
if 'pre-push' in default_env:
container = default_env
elif 'extra-scripts' in default_env and 'pre-push' in default_env['extra-scripts']:
container = default_env['extra-scripts']
if container is None:
return
pre_push = container['pre-push']
# Classify each command against the *previous* configuration, so that a
# command still carrying the old default selector is recognised as managed
# even though the canonical default has already been updated in the document.
previous_cfg = PythonConfig(
supported=cfg.supported,
experimental=cfg.experimental,
default=previous_default,
free_threaded=cfg.free_threaded,
)
for index, item in enumerate(pre_push):
command = str(item)
classification = classify_pre_push(command, previous_cfg)
if classification is PrePushClassification.DEFAULT_MANAGED:
rewritten = retarget_default_command(command, previous_default, cfg.default)
elif classification is PrePushClassification.FREE_THREADED_MANAGED:
rewritten = retarget_free_threaded_command(command, cfg.free_threaded)
else:
# PINNED or UNCLASSIFIED commands are left exactly as they are.
continue
if rewritten != command:
self._record(report, f'tool.hatch.envs.default.pre-push[{index}]', command, rewritten)
pre_push[index] = rewritten
# ------------------------------------------------------------------
# Navigation helpers
# ------------------------------------------------------------------
[docs]
@staticmethod
def _matrix_rows(doc: TOMLDocument, env: str) -> Any | None:
"""Return the ``matrix`` array-of-tables for an environment, or ``None``."""
envs = doc.get('tool', {}).get('hatch', {}).get('envs')
if envs is None or env not in envs:
return None
env_table = envs[env]
return env_table.get('matrix')