release.py

#!/usr/bin/env python
"""A script to publish a release of pgcli to PyPI."""

import io
import re
import subprocess
import sys
from optparse import OptionParser

import click

DEBUG = False
CONFIRM_STEPS = False
DRY_RUN = False


def skip_step():
    """
    Asks for user's response whether to run a step. Default is yes.
    :return: boolean
    """
    global CONFIRM_STEPS

    if CONFIRM_STEPS:
        return not click.confirm("--- Run this step?", default=True)
    return False


def run_step(*args):
    """
    Prints out the command and asks if it should be run.
    If yes (default), runs it.
    :param args: list of strings (command and args)
    """
    global DRY_RUN

    cmd = args
    print(" ".join(cmd))
    if skip_step():
        print("--- Skipping...")
    elif DRY_RUN:
        print("--- Pretending to run...")
    else:
        subprocess.check_output(cmd)


def version(version_file):
    _version_re = re.compile(r'__version__\s+=\s+(?P<quote>[\'"])(?P<version>.*)(?P=quote)')

    with io.open(version_file, encoding="utf-8") as f:
        ver = _version_re.search(f.read()).group("version")

    return ver


def get_merged_prs_since_last_tag():
    """Get list of PR numbers and titles merged since the last tag."""
    try:
        previous_tag = (
            subprocess
            .check_output(
                ["git", "describe", "--abbrev=0", "--tags"],
                stderr=subprocess.DEVNULL,
            )
            .decode()
            .strip()
        )
    except subprocess.CalledProcessError:
        return []

    log = subprocess.check_output(["git", "log", "--merges", "--oneline", "{}..HEAD".format(previous_tag)]).decode()

    prs = re.findall(r"(.+\(#(\d+)\))", log)
    seen = set()
    result = []
    for line, num in prs:
        if num not in seen:
            seen.add(num)
            result.append(num)
    return result


def commit_for_release(version_file, ver):
    pr_numbers = get_merged_prs_since_last_tag()
    pr_list = ""
    if pr_numbers:
        pr_list = "\n\n" + "\n".join("- #{}".format(n) for n in pr_numbers)

    message = "Releasing version {}{}".format(ver, pr_list)
    run_step("git", "reset")
    run_step("git", "add", "-u")
    run_step("git", "commit", "--message", message)


def create_git_tag(tag_name):
    run_step("git", "tag", tag_name)


def create_distribution_files():
    run_step("rm", "-rf", "dist/")
    run_step("python", "-m", "build")


def upload_distribution_files():
    run_step("twine", "upload", "dist/*")


def push_to_github():
    run_step("git", "push", "origin", "main")


def push_tags_to_github():
    run_step("git", "push", "--tags", "origin")


def check_tag(ver):
    """Verify that HEAD is on the expected tag."""
    tag = "v{}".format(ver)
    try:
        current_tag = (
            subprocess
            .check_output(
                ["git", "describe", "--exact-match", "--tags", "HEAD"],
                stderr=subprocess.DEVNULL,
            )
            .decode()
            .strip()
        )
    except subprocess.CalledProcessError:
        print("ERROR: HEAD is not on any tag. Expected tag '{}'.".format(tag))
        sys.exit(1)
    if current_tag != tag:
        print("ERROR: HEAD is on tag '{}', expected '{}'.".format(current_tag, tag))
        sys.exit(1)
    print("OK: on tag '{}'".format(tag))


def comment_on_released_prs(ver):
    """Post a comment on all PRs included in this release."""
    tag = "v{}".format(ver)
    try:
        previous_tag = (
            subprocess
            .check_output(
                ["git", "describe", "--abbrev=0", "--tags", "{}^".format(tag)],
                stderr=subprocess.DEVNULL,
            )
            .decode()
            .strip()
        )
    except subprocess.CalledProcessError:
        print("WARNING: Could not find previous tag. Skipping PR comments.")
        return

    log = subprocess.check_output(["git", "log", "--merges", "--oneline", "{}..{}".format(previous_tag, tag)]).decode()

    pr_numbers = re.findall(r"#(\d+)", log)
    pr_numbers = list(set(pr_numbers))

    if not pr_numbers:
        print("No PRs found between {} and {}.".format(previous_tag, tag))
        return

    print("Found PRs: {}".format(", ".join("#" + n for n in pr_numbers)))
    message = "Released as part of {}.".format(ver)

    print("gh pr comment --body '{}' {}".format(message, " ".join(pr_numbers)))
    if skip_step():
        print("--- Skipping...")
    elif DRY_RUN:
        print("--- Pretending to run...")
    else:
        for pr in pr_numbers:
            subprocess.check_output(["gh", "pr", "comment", pr, "--body", message])


def checklist(questions):
    for question in questions:
        if not click.confirm("--- {}".format(question), default=False):
            sys.exit(1)


if __name__ == "__main__":
    if DEBUG:
        subprocess.check_output = lambda x: x

    # checks = [
    #     "Have you updated the AUTHORS file?",
    #     "Have you updated the `Usage` section of the README?",
    # ]
    # checklist(checks)

    ver = version("pgcli/__init__.py")
    print("Releasing Version:", ver)

    parser = OptionParser()
    parser.add_option(
        "-c",
        "--confirm-steps",
        action="store_true",
        dest="confirm_steps",
        default=False,
        help=("Confirm every step. If the step is not confirmed, it will be skipped."),
    )
    parser.add_option(
        "-d",
        "--dry-run",
        action="store_true",
        dest="dry_run",
        default=False,
        help="Print out, but not actually run any steps.",
    )

    popts, pargs = parser.parse_args()
    CONFIRM_STEPS = popts.confirm_steps
    DRY_RUN = popts.dry_run

    if not click.confirm("Are you sure?", default=False):
        sys.exit(1)

    commit_for_release("pgcli/__init__.py", ver)
    create_git_tag("v{}".format(ver))
    push_to_github()
    push_tags_to_github()
    check_tag(ver)
    create_distribution_files()
    upload_distribution_files()
    comment_on_released_prs(ver)