Migrate labels for an existing github repository. Inspired by https://github.com/jasonbellamy/git-label

labels.json

{"color": "bfd4f2", "name": "Effort: High"}
{"color": "f9d0c4", "name": "Effort: Low"}
{"color": "0e8a16", "name": "Effort: Medium"}
{"color": "1d76db", "name": "Impact: High"}
{"color": "0e8a16", "name": "Impact: Low"}
{"color": "bfdadc", "name": "Impact: Medium"}
{"color": "159818", "name": "Status: Help wanted", "old_name": "help wanted"}
{"color": "ededed", "name": "Status: In progress", "old_name": "in progress"}
{"color": "e6e6e6", "name": "Status: Needs review", "old_name": "needs review"}
{"color": "e5987b", "name": "Status: Needs revision"}
{"color": "e6e6e6", "name": "Status: Next", "old_name": "next"}
{"color": "ededed", "name": "Status: Ready", "old_name": "ready"}
{"color": "e67e22", "name": "Type: Archive"}
{"color": "fc2929", "name": "Type: Bug", "old_name": "bug"}
{"color": "cccccc", "name": "Type: Duplicate", "old_name": "duplicate"}
{"color": "84b6eb", "name": "Type: Enhancement", "old_name": "enhancement"}
{"color": "e6e6e6", "name": "Type: Invalid", "old_name": "invalid"}
{"color": "cc317c", "name": "Type: Question", "old_name": "question"}
{"color": "ffffff", "name": "Type: wontfix", "old_name": "wontfix"}

migrate_labels.py

"""
pip install basescript
pip install requests
python migrate_labels.py run -u <user> -r <repo> -t <api-token> --label-file labels.json
"""

from basescript import BaseScript

import json
import requests
import logging
requests.packages.urllib3.disable_warnings()
logging.getLogger("urllib3").setLevel(logging.WARNING)
logging.getLogger("requests").setLevel(logging.WARNING)

class Github(BaseScript):
    def define_args(self, parser):
        parser.add_argument("-u", "--user", required=True,
            help="github user/organization name",
        )
        parser.add_argument("-r", "--repo", required=True,
            help="github repository name",
        )
        parser.add_argument("-t", "--token", required=True,
            help="github api token",
        )
        parser.add_argument("--label-file", required=True,
            help=(
                'A file containing labels in json format '
                '{"name": "...", "color": "123456", "old_name": "..."} '
                '"name" and "color" are mandatory, but "old_name" is '
                'optional. If specified, it migrates all existing issues '
                'with label "old_name" to "name".'
            ),
        )

    def _not_success(self, code):
        if code < 200 or code > 299:
            return True

        return False

    def create_label(self, name, color):
        """
        creates a label with given name and color
        """

        url = 'https://api.github.com/repos/{}/{}/labels'.format(
            self.args.user, self.args.repo,
        )
        label = { 'name': name, 'color': color }

        self.log.debug("creating label", name=name, color=color, url=url)
        resp = requests.post(url, headers=self.headers, json=label)
        if self._not_success(resp.status_code):
            self.log.error("failed to create label",
                label=name, response=resp,
            )
            raise Exception("failed to create label")

        self.log.info("created label", name=name)

    def get_labels(self):
        """
        returns the labels for the users repository
        """
        url = 'https://api.github.com/repos/{}/{}/labels'.format(
            self.args.user, self.args.repo,
        )

        labels = []
        while True:
            resp = requests.get(url, headers=self.headers)
            if self._not_success(resp.status_code):
                self.log.error("failed to list labels", response=resp)
                raise Exception("failed to list labels")

            labels.extend(resp.json())
            if 'next' in resp.links:
                url = resp.links['next']['url']
            elif 'last' in resp.links:
                url = resp.links['last']['url']
            else:
                break

        return labels

    def migrate_issues(self, src_label, dst_label):
        """
        moves all issues having label src to label dst
        """

        # get all issues for label
        url = 'https://api.github.com/repos/{}/{}/issues'.format(
            self.args.user, self.args.repo,
        )
        issues = []
        params = { 'state': 'all', 'labels': src_label }
        while True:
            resp = requests.get(url, headers=self.headers, params=params)
            if self._not_success(resp.status_code):
                self.log.error("failed to list issues for label",
                    label=src_label, response=resp,
                )
                raise Exception("failed to list issues for label")

            issues.extend(resp.json())
            if 'next' in resp.links:
                url = resp.links['next']['url']
            elif 'last' in resp.links:
                url = resp.links['last']['url']
            else:
                break

        for issue in issues:
            resp = requests.delete('{}/labels/{}'.format(issue['url'], src_label),
                headers=self.headers,
            )
            if self._not_success(resp.status_code):
                self.log.error("failed to delete label for issue",
                    label=src_label, issue=issue['number'], response=resp,
                )
                raise Exception("failed to delete label for issue")

            self.log.info("deleted label for issue",
                label=src_label, issue=issue['number'], response=resp,
            )

            resp = requests.post('{}/labels'.format(issue['url']),
                headers=self.headers, json=[dst_label],
            )
            if self._not_success(resp.status_code):
                self.log.error("failed to assign label to issue",
                    label=dst_label, issue=issue['number'], response=resp,
                )
                raise Exception("failed to assign label to issue")

            self.log.info("assigned new label for issue",
                label=dst_label, issue=issue['number'],
            )

    def delete_label(self, label):
        """
        deletes the given label
        """
        self.log.debug("deleting label", label=label)
        url = 'https://api.github.com/repos/{}/{}/labels/{}'.format(
            self.args.user, self.args.repo, label,
        )

        resp = requests.delete(url, headers=self.headers)
        if self._not_success(resp.status_code):
            self.log.error("failed to delete label",
                label=label, response=resp,
            )
            raise Exception("failed to delete label")

        self.log.info("deleted label", name=label)

    def run(self):
        self.headers = {
            "Authorization": "token {}".format(self.args.token),
        }

        labels = []
        with open(self.args.label_file) as fp:
            labels = [ json.loads(i) for i in fp if i.strip() ]

        existing_labels = self.get_labels()
        existing_label_names = { l['name'] for l in existing_labels }
        for label in labels:
            if label['name'] not in existing_label_names:
                self.create_label(label['name'], label['color'])
                existing_label_names.add(label['name'])

            if 'old_name' in label and label['old_name'] in existing_label_names:
                self.migrate_issues(label['old_name'], label['name'])
                self.delete_label(label['old_name'])
                existing_label_names.discard(label['old_name'])

if __name__ == '__main__':
    Github().start()