Source code for planemo.git

"""Utilities for interacting with git using planemo abstractions."""

import os
import subprocess
from typing import (
    Dict,
    List,
    Optional,
    TYPE_CHECKING,
)

from galaxy.util import unicodify

from planemo import io

if TYPE_CHECKING:
    from planemo.cli import PlanemoCliContext


[docs]def git_env_for(path: str) -> Dict[str, str]: """Setup env dictionary to target specified git repo with git commands.""" env = os.environ.copy() env.update({"GIT_WORK_DIR": path, "GIT_DIR": os.path.join(path, ".git")}) return env
[docs]def ls_remote(ctx: "PlanemoCliContext", remote_repo: str) -> Dict[str, str]: """Return a dictionary with refs as key and commits as value.""" commits_and_refs = io.communicate( ["git", "ls-remote", remote_repo], stdout=subprocess.PIPE, )[0] return dict(line.split()[::-1] for line in commits_and_refs.decode("utf-8").splitlines())
[docs]def init(ctx: "PlanemoCliContext", repo_path: str) -> None: env = git_env_for(repo_path) io.communicate(["git", "init"], env=env)
[docs]def add(ctx: "PlanemoCliContext", repo_path: str, file_path: str) -> None: env = git_env_for(repo_path) io.communicate(["git", "add", os.path.relpath(file_path, repo_path)], env=env, cwd=repo_path)
[docs]def commit(ctx: "PlanemoCliContext", repo_path: str, message: str = "") -> None: env = git_env_for(repo_path) io.communicate(["git", "commit", "-m", message], env=env)
[docs]def push( ctx: "PlanemoCliContext", repo_path: str, to: Optional[str] = None, branch: Optional[str] = None, force: bool = False, ) -> None: env = git_env_for(repo_path) cmd = ["git", "push"] if force: cmd += ["--force"] if to and branch: cmd += ["-u", to, branch] io.communicate(cmd, env=env, cwd=repo_path)
[docs]def branch(ctx, repo_path, branch, from_branch=None): env = git_env_for(repo_path) cmd = ["git", "checkout", "-b", branch] if from_branch is not None: cmd.append(from_branch) io.communicate(cmd, env=env)
[docs]def checkout(ctx, remote_repo, local_path, branch=None, remote="origin", from_branch="master"): """Checkout a new branch from a remote repository.""" env = git_env_for(local_path) if not os.path.exists(local_path): io.communicate(command_clone(ctx, remote_repo, local_path)) else: io.communicate(["git", "fetch", remote], env=env) if branch: io.communicate(["git", "checkout", f"{remote}/{from_branch}", "-b", branch], env=env) else: io.communicate(["git", "merge", "--ff-only", f"{remote}/{from_branch}"], env=env)
[docs]def command_clone( ctx: "PlanemoCliContext", src: str, dest: str, mirror: bool = False, branch: Optional[str] = None ) -> List[str]: """Produce a command-line string to clone a repository. Take in ``ctx`` to allow more configurability down the road. """ cmd = ["git", "clone"] if mirror: cmd.append("--mirror") if branch is not None: cmd.extend(["--branch", branch]) cmd.extend([src, dest]) return cmd
[docs]def diff(ctx: "PlanemoCliContext", directory: str, range: str) -> List[str]: """Produce a list of diff-ed files for commit range.""" cmd = f"cd '{directory}' && git diff --name-only '{range}' --" stdout, _ = io.communicate(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True) return [line.strip() for line in unicodify(stdout).splitlines() if line]
[docs]def clone(*args, **kwds) -> None: """Clone a git repository. See :func:`command_clone` for description of arguments. """ command = command_clone(*args, **kwds) io.communicate(command)
[docs]def rev(ctx: "PlanemoCliContext", directory: str) -> str: """Raw revision for git directory specified. Throws ``RuntimeError`` if not a git directory. """ cmd = f"cd '{directory}' && git rev-parse HEAD" stdout, _ = io.communicate(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) return unicodify(stdout).strip()
[docs]def is_rev_dirty(ctx: "PlanemoCliContext", directory: str) -> bool: """Check if specified git repository has uncommitted changes.""" return io.shell(["git", "diff", "--quiet"], cwd=directory) != 0
[docs]def rev_if_git(ctx: "PlanemoCliContext", directory: str) -> Optional[str]: """Determine git revision (or ``None``).""" try: the_rev = rev(ctx, directory) is_dirty = is_rev_dirty(ctx, directory) if is_dirty: the_rev += "-dirty" return the_rev except RuntimeError: return None