feat(script): parse version bump commits

Add helper to iterate over commits in a repository
where the version (calculated by Strategy callback)
has changed.

Gemini used to help write the test cases.

Author: scottzach1

src/scottzach1/semantic_release/githelper.py renamed to src/scottzach1/semantic_release/message_parser.py

@@ -0,0 +1,145 @@
+import logging
+from collections.abc import Iterator
+from os import PathLike
+from pathlib import Path
+from typing import Protocol
+
+import toml
+from git import Commit, Repo
+from pydantic import BaseModel, ConfigDict
+
+
+class VersionCommit(BaseModel):
+    """
+    Model abstraction
+    """
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    commit: Commit
+    version: str
+
+
+log = logging.getLogger("scottzach1").getChild("semantic_release").getChild("version_parser")
+
+
+class VersionStrategy(Protocol):
+    watch_path: PathLike
+
+    @staticmethod
+    def get_version(commit: Commit) -> str | None: ...
+
+
+class Pep261:
+    """
+    Strategy to extract version from pyproject.toml following PEP 261 standard.
+
+    ```toml
+    [project]
+    name = "project-name"
+    version = "0.0.1"
+    ```
+
+    https://peps.python.org/pep-0621/
+    """
+
+    watch_path = "pyproject.toml"
+
+    @staticmethod
+    def get_version(commit: Commit) -> str | None:
+        try:
+            toml_content = (commit.tree / "pyproject.toml").data_stream.read().decode("utf-8")
+        except KeyError:
+            return None  # File missing for commit
+
+        try:
+            toml_dict = toml.loads(toml_content)
+        except toml.TomlDecodeError:
+            return None  # Malformed pyproject.toml file
+
+        try:
+            version = toml_dict["project"]["version"]
+
+            if not isinstance(version, str):
+                raise TypeError(f"Version must be a string (found {type(version)})")
+
+            return version
+        except (KeyError, TypeError):
+            return None  # TOML data missing key(s) or malformed (e.g. is None or str)
+
+
+class Uv(Pep261):
+    """
+    Callback strategy to extract version for an astral-sh/uv project (follows PEP 261).
+
+    https://github.com/astral-sh/uv
+    """
+
+
+class PoetryV2(Pep261):
+    """
+    Callback strategy to extract version for a python-poetry/poetry (version 2) project (follows PEP 261).
+
+    https://github.com/python-poetry/poetry
+    """
+
+
+class PoetryV1(Pep261):
+    """
+    Callback strategy to extract version for a python-poetry/poetry (version < 2) project.
+
+    https://github.com/python-poetry/poetry/tree/1.8 (latest minor version as of 25/05/2025).
+    """
+
+    watch_path = "pyproject.toml"
+
+    @staticmethod
+    def get_version(commit: Commit) -> str | None:
+        try:
+            toml_content = (commit.tree / "pyproject.toml").data_stream.read().decode("utf-8")
+        except KeyError:
+            return None  # File missing for commit
+
+        try:
+            toml_dict = toml.loads(toml_content)
+        except toml.TomlDecodeError:
+            return None  # Malformed pyproject.toml file
+
+        try:
+            version = toml_dict["tool"]["poetry"]["version"]
+
+            if not isinstance(version, str):
+                raise TypeError(f"Version must be a string (found {type(version)})")
+
+            return version
+        except (KeyError, TypeError):
+            return None  # TOML data missing key(s) or malformed (e.g. is None or str)
+
+
+def iter_version_commits(repo: Repo, *, strategy: VersionStrategy) -> Iterator[VersionCommit]:
+    """
+    Extract commits containing version bumps based on user provided strategy.
+
+    Args:
+        repo: the repo to parse commits from.
+        strategy: callback to extract version string from commit.
+
+    Returns:
+        iterator of commits where the version string is present and has changed (newest to oldest).
+    """
+
+    nxt_version = None  # version for previous iteration (we are recursing newest to oldest)
+    nxt_commit = None
+
+    for cur_commit in repo.iter_commits(paths=Path(repo.working_dir, strategy.watch_path)):
+        cur_version = None
+        try:
+            cur_version = strategy.get_version(cur_commit)
+        except Exception as e:
+            log.warning("unexpected %s while parsing commit %s: %s", type(e).__name__, repr(cur_commit), str(e))
+
+        if cur_version != nxt_version and nxt_commit is not None and nxt_version is not None:
+            yield VersionCommit(commit=nxt_commit, version=nxt_version)
+
+        nxt_commit = cur_commit
+        nxt_version = cur_version

src/scottzach1/semantic_release/version_parser.py

@@ -4,7 +4,7 @@
 import pytest
 from pydantic import BaseModel
 
-from scottzach1.semantic_release.githelper import (
+from scottzach1.semantic_release.message_parser import (
     LegacyMessage,
     SemanticMessage,
     parse_commit_msg,

tests/test_git.py renamed to tests/test_message_parser.py

@@ -0,0 +1,79 @@
+import hashlib
+import logging
+from collections.abc import Iterator
+from pathlib import Path
+from unittest.mock import MagicMock, call
+
+import git
+from git import Commit
+from gitdb.util import hex_to_bin
+
+from scottzach1.semantic_release.version_parser import Pep261, VersionCommit, VersionStrategy, iter_version_commits
+
+PROJECT_ROOT = Path(__file__).parent.parent
+
+
+def test_version_bumps():
+    repo = git.Repo(PROJECT_ROOT)
+
+    for bump in iter_version_commits(repo, strategy=Pep261):
+        print(bump.commit.hexsha, bump.version, bump.commit.message)
+
+
+class MockCommit(Commit):
+    def __init__(self, repo, seed: str, **kwargs):
+        hex_sha = hashlib.sha1(seed.encode("ascii")).hexdigest()
+        bin_sha = hex_to_bin(hex_sha)
+
+        self.seed = seed
+        super().__init__(repo, bin_sha, **kwargs)
+
+    def __repr__(self) -> str:
+        return f"<MockCommit {self.seed}>"
+
+
+def test_version_commits_logic(caplog):
+    """
+    Verity the toplevel behavior of iter_version_commits() with mocked repo and strategy.
+
+    Validates expected behavior for None, str, and Exception results.
+    """
+    repo = MagicMock()  # spec=git.Repo)
+
+    versions = [
+        None,
+        "1.1.1",
+        "1.1.1",
+        None,
+        "1.1.0",
+        "1.1.0",
+        "1.1.0",
+        "1.0.9",
+        None,
+        None,
+        ValueError("MEEP"),
+        "1.0.8",
+    ]
+    commits = [MockCommit(repo, f"commits[{i}]") for i, _ in enumerate(versions)]
+
+    repo.working_dir = Path("/tmp/bogus")
+    repo.iter_commits.return_value = commits
+
+    strategy = MagicMock(spec=VersionStrategy)
+    strategy.watch_path = "meep.txt"
+    strategy.get_version.side_effect = versions
+
+    # noinspection PyTypeChecker
+    iterator = iter_version_commits(repo, strategy=strategy)
+    assert isinstance(iterator, Iterator)
+
+    with caplog.at_level(logging.WARNING):
+        result = list(iterator)
+
+    assert result == [
+        VersionCommit(version="1.1.1", commit=commits[2]),
+        VersionCommit(version="1.1.0", commit=commits[6]),
+        VersionCommit(version="1.0.9", commit=commits[7]),
+    ]
+    assert "unexpected ValueError while parsing commit <MockCommit commits[10]>: MEEP" in caplog.text
+    assert repo.iter_commits.mock_calls == [call(paths=Path("/tmp/bogus/meep.txt"))]