superlint/.automation/test/python/python_good_1.py

import json
from os import getenv, path
from pprint import pprint
import sys

import click # pylint: disable=import-error
from dotenv import load_dotenv # pylint: disable=import-error
import requests # pylint: disable=import-error

env = load_dotenv()
api_url = getenv('API_URL', default='https://api.github.com/graphql')
github_token = getenv("GITHUB_TOKEN", default=None)

if github_token is None:
    sys.exit("GitHub Token is not set." +
        "Please set the GITHUB_TOKEN env variable in your system or " +
        "the .env file of your project.")

client_id = getenv('CLIENT_ID', default='copy_labels.py')
headers = {
    'Authorization': 'bearer {github_token}'.format(github_token=github_token),
    'Accept': 'application/vnd.github.bane-preview+json',
    'Content-Type': 'application/json'
}

def create_label(repo_id, label):
    """
    Create label in the supplied repo.

    :param repo_id: Unique ID that represents the repo in GitHub
    :type repo_id: str
    :param label: Object with label information.
    :type label: dict
    :return: GitHub API request response
    """

    query_variables = {
        "createLabelInput": {
            "color": label["color"],
            "description": label["description"],
            "name": label["name"],
            "repositoryId": repo_id
        }
    }

    with open(path.join(path.dirname(__file__), 'queries/create_label.gql'), 'r') as query_file:
        query = "".join(query_file.readlines())

    payload = {"query": query, "variables": query_variables}
    response = requests.post(api_url, data=json.dumps(payload), headers=headers).json()
    print('Created label {label}'.format(label=label["name"]))

    return response

def get_labels(owner, repo):
    """
    Gets a list of labels from the supplied repo.
    :param owner: Repo owner GitHub login.
    :type owner: str
    :param repo: Repository name.
    :type repo: str
    :return: A tuple with the GitHub id for the repository and a list of labels defined in the repository
    """

    query_variables = { "owner": owner, "name": repo, }

    with open(path.join(path.dirname(__file__), 'queries/get_repo_data.gql'), 'r') as query_file:
        query = "".join(query_file.readlines())

    payload = {"query": query, "variables": query_variables}
    response = requests.post(api_url, data=json.dumps(payload), headers=headers)

    status_code = response.status_code
    result = response.json()

    if status_code >= 200 and status_code <= 300:
        repo_id = result["data"]["repository"]["id"]
        labels = result["data"]["repository"]["labels"]["nodes"]

        return repo_id, labels
    else:
        raise Exception(
            '[ERROR] getting issue labels. Status Code: {status_code} - Message: {result}'.format(
                status_code=status_code, result=result["message"]))

def delete_label(label_id):
    """
    Delete the specified label
    :param label_id: Label's node id.
    :type label_id: str
    :return: GitHub API request response.
    """

    query_variables = {
        "deleteLabelInput": {
            "clientMutationId": client_id,
            "id": label_id,
        }
    }

    with open(path.join(path.dirname(__file__), 'queries/delete_label.gql'), 'r') as query_file:
        query = "".join(query_file.readlines())

    payload = {"query": query, "variables": query_variables}
    result = requests.post(api_url, data=json.dumps(payload), headers=headers).json()

    return result

@click.command()
@click.option('--dry', is_flag=True)
@click.argument('source_repo')
@click.argument('target_repo')
def copy_labels(source_repo, target_repo, dry):
    """
    Copy labels from the source repository to the target repository.
    \f
    :param source: The full name of a GitHub repo from where the labels will be copied from. Eg. github/opensourcefriday
    :type source: str
    :param target: The full name of a GitHub repo to where the labels will be copied. Eg. github/opensourcefriday
    :type target: str
    :return:
    """
    source_owner, source_repo_name = source_repo.split("/")
    target_owner, target_repo_name = target_repo.split("/")

    try:
        print('Fetching labels for {source_repo_name} repo.'.format(source_repo_name=source_repo_name))
        _, source_repo_labels = get_labels(source_owner, source_repo_name)
        print('Fetched labels for {source_repo_name}'.format(source_repo_name=source_repo_name))

        print('Fetching labels for {target_repo_name} repo.'.format(target_repo_name=target_repo_name))
        target_repo_id, target_repo_labels = get_labels(target_owner, target_repo_name)
        print('Fetched labels for {target_repo_name}'.format(target_repo_name=target_repo_name))

        filtered_labels = list(filter(lambda x: x not in target_repo_labels, source_repo_labels))

        if dry:
            print('This is just a dry run. No labels will be copied/created.')
            print('{label_count} labels would have been created.'.format(label_count=len(filtered_labels)))
            pprint(filtered_labels, indent=4)
        else:
            print('Preparing to created {label_count} labels in {target_repo}'.format(
                label_count=len(filtered_labels), target_repo=target_repo))

            for label in filtered_labels:
                create_label(target_repo_id, label)
    except Exception as error:
        sys.exit(error)

    print('Done')

if __name__ == "__main__":
    # Pylint doesn't know that @click.command takes care of injecting the
    # function parameters. Disabling Pylint error.
    copy_labels() # pylint: disable=no-value-for-parameter