Tool to upload a stack of commits to GitHub.

git-push-stack

#!/usr/bin/python3
"""Tool to upload a stack of commits to GitHub.

This expects a `.git/branch-mapping.yml` file to exist in the repository this
is being run. Example file:

```yaml
commits:
  - subject: My cool commit
    branch: my-cool-commit
  - subject: This depends on my cool commit
    branch: my-cool-commit-2
    parent: my-cool-commit
  - subject: This is an independent change
    branch: refactor-thingy
```

For every commit in the current branch that is present in the branch mapping,
it will create a local branch to match the name, checked out to the specified
parent (or `origin/master`) if no parent was specified with the commit
cherry-picked on top. That operation is performed on a temporary worktree, so
the current checkout and the state of the repository is preserved.

In dry-run mode (`--dry-run`), it still creates the branches, but does not push
the commits.
"""

import argparse
import contextlib
import dataclasses
import logging
import os
import os.path
import shlex
import subprocess
import tempfile

from typing import Any, Dict, Iterator, List, NamedTuple, Optional

import yaml


@dataclasses.dataclass
class CommitConfig:
    subject: str
    branch: str
    parent: Optional[str] = None


@dataclasses.dataclass
class BranchMappingFile:
    commit_configs: List[CommitConfig]

    def get_commit(self, subject: str) -> Optional[CommitConfig]:
        for branch in self.commit_configs:
            if branch.subject == subject:
                return branch
        return None

    @staticmethod
    def from_dict(d: Dict[str, Any]) -> 'BranchMappingFile':
        return BranchMappingFile(
            commit_configs=[CommitConfig(**x) for x in d['commits']])


@dataclasses.dataclass
class Commit:
    oid: str
    parent: str
    subject: str


def _get_default_mapping_path() -> str:
    return os.path.join(
        subprocess.check_output([
            'git',
            'rev-parse',
            '--absolute-git-dir',
        ],
                                universal_newlines=True).strip(),
        'branch-mapping.yml')


def _get_merge_base(remote: str, target_branch: str) -> str:
    return subprocess.check_output([
        'git', 'merge-base',
        os.path.join('refs/remotes', remote, target_branch), 'HEAD'
    ],
                                   universal_newlines=True).strip()


def _commits(merge_base: str) -> List[Commit]:
    commits: List[Commit] = []
    for line in subprocess.check_output(
        [
            'git',
            'log',
            '--format=%H\t%P\t%s',
            f'{merge_base}...',
        ],
            universal_newlines=True).strip().split('\n'):
        commits.append(Commit(*line.strip().split('\t')))
    return commits


@contextlib.contextmanager
def _worktree() -> Iterator[str]:
    root = subprocess.check_output(
        [
            'git',
            'rev-parse',
            '--show-superproject-working-tree',
            '--show-toplevel',
        ],
        universal_newlines=True).strip().split('\n')[0]
    with tempfile.TemporaryDirectory(prefix='git-push-branches',
                                     dir=os.path.dirname(root)) as d:
        subprocess.check_call(['git', 'worktree', 'add', d])
        try:
            yield d
        finally:
            subprocess.check_call(['git', 'worktree', 'remove', '--force', d])
            subprocess.check_call(['git', 'branch', '-D', os.path.basename(d)])


def _main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument('--mapping', default=_get_default_mapping_path()),
    parser.add_argument(
        '--remote',
        default='origin',
        help='Name of the remote that the commits will be merged to')
    parser.add_argument(
        '--upstream-target-branch',
        default='master',
        help='Upstream branch that the commits will be merged to')
    parser.add_argument('--dry-run', action='store_true')
    args = parser.parse_args()

    config: Dict[str, Optional[str]] = {}
    with open(args.mapping) as f:
        mapping = BranchMappingFile.from_dict(yaml.load(f,
                                                        Loader=yaml.CLoader))

    merge_base = _get_merge_base(args.remote, args.upstream_target_branch)
    push_refspecs: List[str] = []
    branch_name_mapping: Dict[str, str] = {}

    with _worktree() as worktree:
        for commit in _commits(merge_base):
            commit_config = mapping.get_commit(commit.subject)
            if commit_config is None:
                logging.warn('Skipping %r', commit.subject)
                continue
            if commit_config.parent is not None:
                expected_parent = branch_name_mapping[commit_config.parent]
            else:
                expected_parent = merge_base
            if commit.parent == expected_parent:
                commit_oid = commit.oid
                subprocess.check_call([
                    'git', 'branch', '--force', commit_config.branch,
                    commit_oid
                ],
                                      cwd=worktree)
            else:
                subprocess.check_call([
                    'git', 'branch', '--force', commit_config.branch,
                    expected_parent
                ],
                                      cwd=worktree)
                subprocess.check_call(['git', 'switch', commit_config.branch],
                                      cwd=worktree)
                subprocess.check_call(['git', 'cherry-pick', commit.oid],
                                      cwd=worktree)
                commit_oid = subprocess.check_output(
                    ['git', 'rev-parse', commit_config.branch],
                    cwd=worktree,
                    universal_newlines=True).strip()
            branch_name_mapping[commit_config.branch] = commit_oid
            push_refspecs.append(
                f'{commit_oid}:refs/heads/{commit_config.branch}')

    push_args = ['git', 'push', '-f', args.remote] + push_refspecs
    if args.dry_run:
        print(shlex.join(push_args))
    else:
        subprocess.check_call(push_args)


if __name__ == '__main__':
    _main()