view env/lib/python3.9/site-packages/planemo/ @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author shellac
date Mon, 22 Mar 2021 18:12:50 +0000
line wrap: on
line source

"""Utilities for interacting with Github."""
from __future__ import absolute_import

import io
import os
import stat
import tarfile
import tempfile
from distutils.dir_util import copy_tree
from pathlib import Path

import requests

from planemo import git
from import (

    import github
    has_github_lib = True
except ImportError:
    github = None
    has_github_lib = False

GH_VERSION = "1.5.0"

NO_GITHUB_DEP_ERROR = ("Cannot use github functionality - "
                       "PyGithub library not available.")
FAILED_TO_DOWNLOAD_GH = "No gh executable available and it could not be installed."
DEFAULT_REMOTE_NAME = 'planemo-remote'

def get_github_config(ctx, allow_anonymous=False):
    """Return a :class:`planemo.github_util.GithubConfig` for given configuration."""
    global_github_config = _get_raw_github_config(ctx)
    return GithubConfig(global_github_config, allow_anonymous=allow_anonymous)

def clone_fork_branch(ctx, target, path, remote_name=DEFAULT_REMOTE_NAME, **kwds):
    """Clone, fork, and branch a repository ahead of building a pull request."""
        branch=kwds.get("branch", None),
    if kwds.get("fork"):
            fork(ctx, path, remote_name=remote_name, **kwds)
        except Exception:
    if 'GITHUB_USER' in os.environ:
        # On CI systems fork doesn't add a local remote under circumstances I don't quite understand,
        # but that's probably linked to
        cmd = ['git', 'remote', 'add', remote_name, f"{os.environ['GITHUB_USER']}/{os.path.basename(target)}"]
            communicate(cmd, cwd=path)
        except RuntimeError:
            # Can add the remote only once
    return remote_name

def fork(ctx, path, remote_name=DEFAULT_REMOTE_NAME, **kwds):
    """Fork the target repository using ``gh``."""
    gh_path = ensure_gh(ctx, **kwds)
    gh_env = get_gh_env(ctx, path, **kwds)
    cmd = [gh_path, "repo", "fork", '--remote=true', '--remote-name', remote_name]
    communicate(cmd, cwd=path, env=gh_env)
    return remote_name

def get_or_create_repository(ctx, owner, repo, dry_run=True, **kwds):
    """Clones or creates a repository and returns path on disk"""
    target = os.path.realpath(tempfile.mkdtemp())
    remote_repo = "{owner}/{repo}".format(owner=owner, repo=repo)
        ctx.log('Cloning {}'.format(remote_repo))
        git.clone(ctx, src=remote_repo, dest=target)
    except Exception:
        ctx.log('Creating repository {}'.format(remote_repo))
        target = create_repository(ctx, owner=owner, repo=repo, dest=target, dry_run=dry_run)
    return target

def create_repository(ctx, owner, repo, dest, dry_run, **kwds):
    gh_path = ensure_gh(ctx, **kwds)
    gh_env = get_gh_env(ctx, dry_run=dry_run, **kwds)
    cmd = [gh_path, 'repo', 'create', '-y', '--public', "{owner}/{repo}".format(owner=owner, repo=repo)]
    if dry_run:
        "Would run command '{}'".format(" ".join(cmd))
        git.init(ctx, dest)
        return dest
    communicate(cmd, env=gh_env, cwd=dest)
    return os.path.join(dest, repo)

def rm_dir_contents(directory, ignore_dirs=(".git")):
    directory = Path(directory)
    for item in directory.iterdir():
        if not in ignore_dirs:
            if item.is_dir():

def add_dir_contents_to_repo(ctx, from_dir, target_dir, target_repository_path, version, dry_run, notes=""):
    ctx.log("From {} to {}".format(from_dir, target_repository_path))
    copy_tree(from_dir, target_repository_path)
    git.add(ctx, target_repository_path, target_repository_path)
    message = "Update for version {version}".format(version=version)
    if notes:
        message += "\n{notes}".format(notes=notes)
    git.commit(ctx, repo_path=target_repository_path, message=message)
    if not dry_run:
        git.push(ctx, target_repository_path)

def assert_new_version(ctx, version, owner, repo):
    remote_repo = "{owner}/{repo}".format(owner=owner, repo=repo)
        tags_and_versions = git.ls_remote(ctx, remote_repo=remote_repo)
        if "refs/tags/v{}".format(version) in tags_and_versions or "refs/tags/{}".format(version) in tags_and_versions:
            raise Exception("Version '{}' for {}/{} exists already. Please change the version.".format(version, owner, repo))
    except RuntimeError:
        # repo doesn't exist

def changelog_in_repo(target_repository_path):
    changelog = []
    for path in os.listdir(target_repository_path):
        if '' in path.lower():
            header_seen = False
            header_chars = ('---', '===', '~~~')
            with(open(os.path.join(target_repository_path, path))) as changelog_fh:
                for line in changelog_fh:
                    if line.startswith(header_chars):
                        if header_seen:
                            return "\n".join(changelog[:-1])
                            header_seen = True
    return "\n".join(changelog)

def create_release(ctx, from_dir, target_dir, owner, repo, version, dry_run, notes="", **kwds):
    assert_new_version(ctx, version, owner=owner, repo=repo)
    target_repository_path = get_or_create_repository(ctx, owner=owner, repo=repo, dry_run=dry_run)
    add_dir_contents_to_repo(ctx, from_dir, target_dir, target_repository_path, version=version, dry_run=dry_run, notes=notes)
    gh_path = ensure_gh(ctx, **kwds)
    gh_env = get_gh_env(ctx, dry_run=dry_run, **kwds)
    cmd = [
        "{}/{}".format(owner, repo),
    cmd.extend(['--notes', notes or changelog_in_repo(target_repository_path)])
    if not dry_run:
        communicate(cmd, env=gh_env)
        ctx.log("Would run command '{}'".format(" ".join(cmd)))

def pull_request(ctx, path, message=None, repo=None, **kwds):
    """Create a pull request against the origin of the path using ``gh``."""
    gh_path = ensure_gh(ctx, **kwds)
    gh_env = get_gh_env(ctx, path, **kwds)
    cmd = [gh_path, "pr", "create"]
    if message is None:
        lines = message.splitlines()
        cmd.extend(['--title', lines[0]])
        if len(lines) > 1:
            cmd.extend(["--body", "\n".join(lines[1:])])
    if repo:
        cmd.extend(['--repo', repo])
    communicate(cmd, env=gh_env)

def get_gh_env(ctx, path=None, dry_run=False, **kwds):
    """Return a environment dictionary to run gh with given user and repository target."""
    if path is None:
        env = {}
        env = git.git_env_for(path).copy()
    if not dry_run:
        github_config = _get_raw_github_config(ctx)
        if github_config is not None:
            if "access_token" in github_config:
                env["GITHUB_TOKEN"] = github_config["access_token"]

    return env

def ensure_gh(ctx, **kwds):
    """Ensure gh is available for planemo

    This method will ensure ``gh`` is installed at the correct version.

    For more information on ``gh`` checkout
    planemo_gh_path = os.path.join(ctx.workspace, f"gh-{GH_VERSION}")
    if not os.path.exists(planemo_gh_path):

    if not os.path.exists(planemo_gh_path):
        raise Exception(FAILED_TO_DOWNLOAD_GH)

    return planemo_gh_path

def _try_download_gh(planemo_gh_path):
    link = _gh_link()
    path = Path(planemo_gh_path)
    resp = requests.get(link)
    with as tf,'wb') as outfile:
        for member in tf.getmembers():
    path.chmod(path.stat().st_mode | stat.S_IEXEC)

def _get_raw_github_config(ctx):
    """Return a :class:`planemo.github_util.GithubConfig` for given configuration."""
    if "github" not in ctx.global_config:
        if "GITHUB_TOKEN" in os.environ:
            return {
                "access_token": os.environ["GITHUB_TOKEN"],
    if "github" not in ctx.global_config:
        raise Exception("github account not found in planemo config and GITHUB_TOKEN environment variables unset")
    return ctx.global_config["github"]

class GithubConfig(object):
    """Abstraction around a Github account.

    Required to use ``github`` module methods that require authorization.

    def __init__(self, config, allow_anonymous=False):
        if not has_github_lib:
            raise Exception(NO_GITHUB_DEP_ERROR)
        if "access_token" not in config:
            if not allow_anonymous:
                raise Exception("github authentication unavailable")
            github_object = github.Github()
            github_object = github.Github(config["access_token"])
        self._github = github_object

def _gh_link():
    if IS_OS_X:
        template_link = ""
        template_link = ""
    return template_link % (GH_VERSION, GH_VERSION)

def publish_as_gist_file(ctx, path, name="index"):
    """Publish a gist.

    More information on gists at
    github_config = get_github_config(ctx, allow_anonymous=False)
    user = github_config._github.get_user()
    with open(path, "r") as fh:
        content =
    content_file = github.InputFileContent(content)
    gist = user.create_gist(False, {name: content_file})
    return gist.files[name].raw_url

def get_repository_object(ctx, name):
    github_object = get_github_config(ctx, allow_anonymous=True)
    return github_object._github.get_repo(name)

__all__ = (