# Copyright Kevin Deldycke <kevin@deldycke.com> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
"""Git operations for GitHub Actions workflows.
This module provides utilities for common Git operations in CI/CD contexts,
with idempotent behavior to allow safe re-runs of failed workflows.
All operations follow a "belt-and-suspenders" approach: combine workflow
timing guarantees (e.g. `workflow_run` ensures tags exist) with idempotent
guards (e.g. `skip_existing` on tag creation). This ensures correctness
in the face of race conditions, API eventual consistency, and partial failures
that are common in GitHub Actions.
```{warning} Tag push requires `REPOMATIC_PAT`
Tags pushed with the default `GITHUB_TOKEN` do not trigger downstream
`on.push.tags` workflows. The custom PAT is required so that tagging
a release commit actually fires the publish and release creation jobs.
```
"""
from __future__ import annotations
import logging
import re
import subprocess
from typing import NamedTuple
from packaging.version import Version
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Sequence
from pathlib import Path
COMMIT_IDENTITY_EMAIL = "41898282+github-actions[bot]@users.noreply.github.com"
"""Commit author email for automated commits: GitHub's own Actions bot user.
The `41898282+` prefix is the bot's stable user ID, which makes GitHub link
the commit to the verified `github-actions[bot]` account.
"""
COMMIT_IDENTITY_NAME = "github-actions[bot]"
"""Commit author name for automated commits."""
SHORT_SHA_LENGTH = 7
"""Default SHA length hard-coded to `7`.
```{caution}
The [default is subject to change](https://stackoverflow.com/a/21015031) and
depends on the size of the repository.
```
"""
GITHUB_REMOTE_PATTERN = re.compile(r"github\.com[:/](?P<slug>[^/]+/[^/]+?)(?:\.git)?$")
"""Extracts an `owner/repo` slug from a GitHub remote URL.
Handles both HTTPS (`https://github.com/owner/repo.git`) and SSH
(`git@github.com:owner/repo.git`) formats.
"""
RELEASE_COMMIT_PATTERN = re.compile(
r"^\[changelog\] Release v(?P<version>[0-9]+\.[0-9]+\.[0-9]+)$"
)
"""Pre-compiled regex for release commit messages.
Matches the full message and captures the version number. Use `fullmatch`
to validate a commit is a release commit, or `match`/`search` with
`.group("version")` to extract the version string.
A rebase merge preserves the original commit messages, so release commits
match this pattern. A squash merge replaces them with the PR title
(e.g. ``Release `v1.2.3` (#42)``), which does **not** match. This mismatch
is the mechanism by which squash merges are safely skipped: the `create-tag`
job only processes commits matching this pattern, so no tag, PyPI publish, or
GitHub release is created from a squash merge. The `detect-squash-merge`
job in `release.yaml` detects this and opens an issue to notify the
maintainer.
"""
GIT_LOG_FORMAT = "%H%x00%B"
"""`git log` pretty-format placeholders for a single commit: full SHA, then a
`NUL`, then the raw body.
Paired with `git log -z` (which terminates each commit's output with a `NUL`),
this frames the stream as alternating `(hash, message)` tokens. Commit messages
may contain newlines but never `NUL` bytes, so splitting on `NUL` recovers the
fields unambiguously even for multi-line messages.
"""
[docs]
class Commit(NamedTuple):
"""A minimal git commit.
Only the hash and message are ever consumed downstream, so a full git
library object (with diffs, modified-file analysis, and complexity metrics)
is unnecessary: the `git` CLI feeds these two fields directly.
"""
hash: str
"""The commit's full 40-character SHA-1 hash."""
msg: str
"""The commit message, stripped of surrounding whitespace."""
def _git(*args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
"""Run a `git` command and capture its output.
Decodes output as UTF-8 so non-ASCII commit metadata (accented author
names, emoji in messages) survives on platforms whose default encoding is
not UTF-8 (Windows `cp1252`).
"""
return subprocess.run(
["git", *args],
capture_output=True,
text=True,
encoding="UTF-8",
check=check,
)
def _parse_commit_log(output: str) -> tuple[Commit, ...]:
"""Parse `NUL`-framed `git log --format=GIT_LOG_FORMAT` output into commits.
See {data}`GIT_LOG_FORMAT` for the framing. `git log -z` leaves a trailing
empty token after the final commit; a plain `git log` does not. Either way,
tokens pair up as `(hash, message)`.
"""
tokens = output.split("\x00")
if tokens and not tokens[-1]:
tokens.pop()
return tuple(
Commit(hash=tokens[i], msg=tokens[i + 1].strip())
for i in range(0, len(tokens) - 1, 2)
)
[docs]
def get_commit(ref: str = "HEAD") -> Commit:
"""Return the commit at *ref*.
:raises subprocess.CalledProcessError: if *ref* does not resolve to a
commit present in the repository.
"""
result = _git("log", "-1", "-z", f"--format={GIT_LOG_FORMAT}", ref)
return _parse_commit_log(result.stdout)[0]
[docs]
def list_commits(start: str, end: str) -> tuple[Commit, ...]:
"""Return the commits in the `start..end` range, oldest first.
Follows git range semantics: *start* is excluded, *end* is included. Both
endpoints must already exist locally, so deepen a shallow clone before
calling if necessary.
"""
result = _git(
"log", "--reverse", "-z", f"--format={GIT_LOG_FORMAT}", f"{start}..{end}"
)
return _parse_commit_log(result.stdout)
[docs]
def commit_exists(ref: str) -> bool:
"""Return `True` if *ref* resolves to a commit object present locally."""
return _git("cat-file", "-e", f"{ref}^{{commit}}", check=False).returncode == 0
[docs]
def count_commits(ref: str = "HEAD") -> int:
"""Return the number of commits reachable from *ref*."""
return int(_git("rev-list", "--count", ref).stdout.strip())
[docs]
def head_sha() -> str:
"""Return the full SHA of the current `HEAD` commit."""
return _git("rev-parse", "HEAD").stdout.strip()
[docs]
def current_branch() -> str | None:
"""Return the checked-out branch name, or `None` when `HEAD` is detached."""
result = _git("symbolic-ref", "--short", "--quiet", "HEAD", check=False)
return result.stdout.strip() or None
[docs]
def checkout(ref: str) -> None:
"""Check out *ref* (a branch name or commit SHA)."""
_git("checkout", ref)
[docs]
def stash() -> None:
"""Stash the working tree's local changes."""
_git("stash")
[docs]
def stash_pop() -> None:
"""Restore the most recently stashed local changes."""
_git("stash", "pop")
[docs]
def stash_count() -> int:
"""Return the number of entries on the stash reflog."""
result = _git(
"rev-list", "--walk-reflogs", "--ignore-missing", "--count", "refs/stash"
)
return int(result.stdout.strip())
[docs]
def fetch_deepen(depth: int) -> None:
"""Deepen a shallow clone by fetching *depth* more commits.
:raises subprocess.CalledProcessError: if the fetch fails.
"""
_git("fetch", f"--deepen={depth}")
[docs]
def diff_names(start: str, end: str) -> tuple[str, ...]:
"""Return the paths that differ between *start* and *end*.
:raises subprocess.CalledProcessError: if either ref is unknown.
"""
output = _git("diff", "--name-only", start, end).stdout.strip()
return tuple(output.splitlines()) if output else ()
[docs]
def get_repo_slug_from_remote(remote: str = "origin") -> str | None:
"""Extract the `owner/repo` slug from a git remote URL.
Parses both HTTPS and SSH GitHub remote formats. Returns `None` if the
remote is not set, not a GitHub URL, or git is unavailable.
"""
try:
result = subprocess.run(
["git", "remote", "get-url", remote],
capture_output=True,
text=True,
check=False,
)
except FileNotFoundError:
return None
if result.returncode:
return None
match = GITHUB_REMOTE_PATTERN.search(result.stdout.strip())
return match.group("slug") if match else None
[docs]
def get_latest_tag_version() -> Version | None:
"""Returns the latest release version from Git tags.
Looks for tags matching the pattern `vX.Y.Z` and returns the highest version.
Returns `None` if no matching tags are found.
"""
# Get all tags matching the version pattern.
tags = _git("tag", "--list", "v[0-9]*.[0-9]*.[0-9]*").stdout.splitlines()
if not tags:
logging.debug("No version tags found in repository.")
return None
# Parse and find the highest version.
versions = []
for tag in tags:
# Strip the 'v' prefix and parse.
version = Version(tag.lstrip("v"))
versions.append(version)
latest = max(versions)
logging.debug(f"Latest tag version: {latest}")
return latest
[docs]
def get_release_version_from_commits(max_count: int = 10) -> Version | None:
"""Extract release version from recent commit messages.
Searches recent commits for messages matching the pattern
`[changelog] Release vX.Y.Z` and returns the version from the most recent match.
This provides a fallback when tags haven't been pushed yet due to race conditions
between workflows. The release commit message contains the version information
before the tag is created.
:param max_count: Maximum number of commits to search.
:return: The version from the most recent release commit, or `None` if not found.
"""
if max_count <= 0:
return None
result = _git(
"log", "-n", str(max_count), "-z", f"--format={GIT_LOG_FORMAT}", "HEAD"
)
for commit in _parse_commit_log(result.stdout):
match = RELEASE_COMMIT_PATTERN.fullmatch(commit.msg)
if match:
version = Version(match.group("version"))
logging.debug(f"Found release version {version} in commit {commit.hash}")
return version
logging.debug("No release commit found in recent history.")
return None
[docs]
def get_tag_date(tag: str) -> str | None:
"""Get the date of a Git tag in `YYYY-MM-DD` format.
Uses `creatordate` which resolves to the tagger date for annotated
tags and the commit date for lightweight tags.
:param tag: The tag name to look up.
:return: Date string in `YYYY-MM-DD` format, or `None` if the
tag does not exist.
"""
result = subprocess.run(
["git", "tag", "-l", "--format=%(creatordate:short)", tag],
capture_output=True,
text=True,
check=False,
)
date = result.stdout.strip()
if not date:
return None
return date
[docs]
def tag_exists(tag: str) -> bool:
"""Check if a Git tag already exists locally.
:param tag: The tag name to check.
:return: True if the tag exists, False otherwise.
"""
result = subprocess.run(
["git", "show-ref", "--tags", tag, "--quiet"],
capture_output=True,
check=False,
)
return result.returncode == 0
[docs]
def create_tag(tag: str, commit: str | None = None) -> None:
"""Create a local Git tag.
:param tag: The tag name to create.
:param commit: The commit to tag. Defaults to HEAD.
:raises subprocess.CalledProcessError: If tag creation fails.
"""
cmd = ["git", "tag", tag]
if commit:
cmd.append(commit)
logging.debug(f"Creating tag: {' '.join(cmd)}")
subprocess.run(cmd, check=True, capture_output=True, text=True)
[docs]
def push_tag(tag: str, remote: str = "origin") -> None:
"""Push a Git tag to a remote repository.
:param tag: The tag name to push.
:param remote: The remote name. Defaults to "origin".
:raises subprocess.CalledProcessError: If push fails.
"""
cmd = ["git", "push", remote, tag]
logging.debug(f"Pushing tag: {' '.join(cmd)}")
subprocess.run(cmd, check=True, capture_output=True, text=True)
[docs]
def commit_and_push_files(
paths: Sequence[Path | str],
message: str,
remote: str = "origin",
branch: str = "main",
attempts: int = 3,
) -> bool:
"""Commit the given files and push, rebasing and retrying on rejection.
Designed for CI jobs that append to tracked files (scan records, the
binaries page) and publish the result on the default branch. The commit
is authored as {data}`COMMIT_IDENTITY_NAME` via per-command `-c` config,
since CI checkouts carry no git identity.
Idempotent: when the files are unchanged, no commit is created and the
function returns `False`. A rejected push (another job or the maintainer
pushed meanwhile) is retried after fetching and rebasing onto the fresh
remote tip. Works from a detached `HEAD`: the push targets
``HEAD:{branch}`` explicitly.
:param paths: Files to stage and commit.
:param message: Commit message.
:param remote: Remote to push to.
:param branch: Remote branch to push to.
:param attempts: Maximum push attempts before giving up.
:return: `True` when a commit was pushed, `False` when there was nothing
to commit.
:raises RuntimeError: When the rebase hits a conflict (the local change
overlaps a concurrent push) or every push attempt is rejected.
:raises subprocess.CalledProcessError: When a git command fails outright.
"""
_git("add", "--", *(str(path) for path in paths))
if _git("diff", "--cached", "--quiet", check=False).returncode == 0:
logging.info("No changes to commit.")
return False
_git(
"-c",
f"user.name={COMMIT_IDENTITY_NAME}",
"-c",
f"user.email={COMMIT_IDENTITY_EMAIL}",
"commit",
"--message",
message,
)
logging.info(f"Committed: {message}")
for attempt in range(1, attempts + 1):
push = _git("push", remote, f"HEAD:{branch}", check=False)
if push.returncode == 0:
logging.info(f"Pushed to {remote}/{branch}.")
return True
logging.warning(
f"Push attempt {attempt}/{attempts} rejected: "
f"{push.stderr.strip()}\nRebasing onto fresh {remote}/{branch}."
)
_git("fetch", remote, branch)
# Replaying the commit needs a committer identity too.
rebase = _git(
"-c",
f"user.name={COMMIT_IDENTITY_NAME}",
"-c",
f"user.email={COMMIT_IDENTITY_EMAIL}",
"rebase",
"FETCH_HEAD",
check=False,
)
if rebase.returncode:
_git("rebase", "--abort", check=False)
raise RuntimeError(
f"Rebase onto {remote}/{branch} conflicted, aborted. "
"A concurrent push touched the same files; re-run the job "
"once it settles."
)
raise RuntimeError(f"Push to {remote}/{branch} failed after {attempts} attempts.")
[docs]
def create_and_push_tag(
tag: str,
commit: str | None = None,
push: bool = True,
skip_existing: bool = True,
) -> bool:
"""Create and optionally push a Git tag.
This function is idempotent: if the tag already exists and `skip_existing`
is True, it returns False without failing. This allows safe re-runs of
workflows that were interrupted after tag creation but before other steps.
:param tag: The tag name to create.
:param commit: The commit to tag. Defaults to HEAD.
:param push: Whether to push the tag to the remote. Defaults to True.
:param skip_existing: If True, skip silently when tag exists.
If False, raise an error. Defaults to True.
:return: True if the tag was created, False if it already existed.
:raises ValueError: If tag exists and skip_existing is False.
:raises subprocess.CalledProcessError: If Git operations fail.
"""
if tag_exists(tag):
if skip_existing:
logging.info(f"Tag {tag!r} already exists, skipping.")
return False
msg = f"Tag {tag!r} already exists."
raise ValueError(msg)
create_tag(tag, commit)
logging.info(f"Created tag {tag!r}")
if push:
push_tag(tag)
logging.info(f"Pushed tag {tag!r} to remote.")
return True