Mercurial > repos > shellac > sam_consensus_v3

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

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author shellac
date Mon, 22 Mar 2021 18:12:50 +0000
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:4f3585e2f14b
1 """Utilities for interacting with git using planemo abstractions."""
2 from __future__ import absolute_import
3
4 import os
5 import subprocess
6
7 from galaxy.util import unicodify
8
9 from planemo import io
10
11
12 def git_env_for(path):
13 """Setup env dictionary to target specified git repo with git commands."""
14 env = os.environ.copy()
15 env.update({
16 "GIT_WORK_DIR": path,
17 "GIT_DIR": os.path.join(path, ".git")
18 })
19 return env
20
21
22 def ls_remote(ctx, remote_repo):
23 """Return a dictionary with refs as key and commits as value."""
24 commits_and_refs = io.communicate(
25 ["git", "ls-remote", remote_repo],
26 stdout=subprocess.PIPE,
27 )[0]
28 return dict(line.split()[::-1] for line in commits_and_refs.decode('utf-8').splitlines())
29
30
31 def init(ctx, repo_path):
32 env = git_env_for(repo_path)
33 io.communicate(["git", "init"], env=env)
34
35
36 def add(ctx, repo_path, file_path):
37 env = git_env_for(repo_path)
38 io.communicate(["git", "add", os.path.relpath(file_path, repo_path)], env=env, cwd=repo_path)
39
40
41 def commit(ctx, repo_path, message=""):
42 env = git_env_for(repo_path)
43 io.communicate(["git", "commit", "-m", message], env=env)
44
45
46 def push(ctx, repo_path, to=None, branch=None, force=False):
47 env = git_env_for(repo_path)
48 cmd = ["git", "push"]
49 if force:
50 cmd += ["--force"]
51 if to and branch:
52 cmd += ['-u', to, branch]
53 io.communicate(cmd, env=env, cwd=repo_path)
54
55
56 def branch(ctx, repo_path, branch, from_branch=None):
57 env = git_env_for(repo_path)
58 cmd = ["git", "checkout", "-b", branch]
59 if from_branch is not None:
60 cmd.append(from_branch)
61 io.communicate(cmd, env=env)
62
63
64 def checkout(ctx, remote_repo, local_path, branch=None, remote="origin", from_branch="master"):
65 """Checkout a new branch from a remote repository."""
66 env = git_env_for(local_path)
67 if not os.path.exists(local_path):
68 io.communicate(command_clone(ctx, remote_repo, local_path))
69 else:
70 io.communicate(["git", "fetch", remote], env=env)
71
72 if branch:
73 io.communicate(["git", "checkout", "%s/%s" % (remote, from_branch), "-b", branch], env=env)
74 else:
75 io.communicate(["git", "merge", "--ff-only", "%s/%s" % (remote, from_branch)], env=env)
76
77
78 def command_clone(ctx, src, dest, mirror=False, branch=None):
79 """Produce a command-line string to clone a repository.
80
81 Take in ``ctx`` to allow more configurability down the road.
82 """
83 cmd = ['git', 'clone']
84 if mirror:
85 cmd.append("--mirror")
86 if branch is not None:
87 cmd.extend(["--branch", branch])
88 cmd.extend([src, dest])
89 return cmd
90
91
92 def diff(ctx, directory, range):
93 """Produce a list of diff-ed files for commit range."""
94 cmd_template = "cd '%s' && git diff --name-only '%s' --"
95 cmd = cmd_template % (directory, range)
96 stdout, _ = io.communicate(
97 cmd,
98 stdout=subprocess.PIPE, stderr=subprocess.PIPE,
99 universal_newlines=True
100 )
101 return [line.strip() for line in unicodify(stdout).splitlines() if line]
102
103
104 def clone(*args, **kwds):
105 """Clone a git repository.
106
107 See :func:`command_clone` for description of arguments.
108 """
109 command = command_clone(*args, **kwds)
110 return io.communicate(command)
111
112
113 def rev(ctx, directory):
114 """Raw revision for git directory specified.
115
116 Throws ``RuntimeError`` if not a git directory.
117 """
118 cmd_template = "cd '%s' && git rev-parse HEAD"
119 cmd = cmd_template % directory
120 stdout, _ = io.communicate(
121 cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
122 )
123 return unicodify(stdout).strip()
124
125
126 def is_rev_dirty(ctx, directory):
127 """Check if specified git repository has uncommitted changes."""
128 return io.shell(['git', 'diff', '--quiet'], cwd=directory) != 0
129
130
131 def rev_if_git(ctx, directory):
132 """Determine git revision (or ``None``)."""
133 try:
134 the_rev = rev(ctx, directory)
135 is_dirty = is_rev_dirty(ctx, directory)
136 if is_dirty:
137 the_rev += "-dirty"
138 return the_rev
139 except RuntimeError:
140 return None