cola/git.py

   1 import os
   2 import sys
   3 import subprocess
   4 import threading
   5 from os.path import join
   6
   7 from cola import core
   8 from cola.decorators import memoize
   9 from cola.interaction import Interaction
  10
  11
  12 INDEX_LOCK = threading.Lock()
  13 GIT_COLA_TRACE = core.getenv('GIT_COLA_TRACE', '')
  14 STATUS = 0
  15 STDOUT = 1
  16 STDERR = 2
  17
  18
  19 def dashify(s):
  20     return s.replace('_', '-')
  21
  22
  23 def is_git_dir(d):
  24     """From git's setup.c:is_git_directory()."""
  25     if (core.isdir(d) and core.isdir(join(d, 'objects')) and
  26             core.isdir(join(d, 'refs'))):
  27         headref = join(d, 'HEAD')
  28         return (core.isfile(headref) or
  29                 (core.islink(headref) and
  30                     core.readlink(headref).startswith('refs')))
  31
  32     return is_git_file(d)
  33
  34
  35 def is_git_file(f):
  36     return core.isfile(f) and '.git' == os.path.basename(f)
  37
  38
  39 def is_git_worktree(d):
  40     return is_git_dir(join(d, '.git'))
  41
  42
  43 def read_git_file(path):
  44     if path is None:
  45         return None
  46     if is_git_file(path):
  47         data = core.read(path).strip()
  48         if data.startswith('gitdir: '):
  49             return data[len('gitdir: '):]
  50     return None
  51
  52
  53 class Git(object):
  54     """
  55     The Git class manages communication with the Git binary
  56     """
  57     def __init__(self):
  58         self._git_cwd = None #: The working directory used by execute()
  59         self._worktree = None
  60         self._git_file_path = None
  61         self.set_worktree(core.getcwd())
  62
  63     def set_worktree(self, path):
  64         self._git_dir = path
  65         self._git_file_path = None
  66         self._worktree = None
  67         self.worktree()
  68
  69     def worktree(self):
  70         if self._worktree:
  71             return self._worktree
  72         self.git_dir()
  73         if self._git_dir:
  74             curdir = self._git_dir
  75         else:
  76             curdir = core.getcwd()
  77
  78         if is_git_dir(join(curdir, '.git')):
  79             return curdir
  80
  81         # Handle bare repositories
  82         if (len(os.path.basename(curdir)) > 4
  83                 and curdir.endswith('.git')):
  84             return curdir
  85         if 'GIT_WORK_TREE' in os.environ:
  86             self._worktree = core.getenv('GIT_WORK_TREE')
  87         if not self._worktree or not core.isdir(self._worktree):
  88             if self._git_dir:
  89                 gitparent = join(core.abspath(self._git_dir), '..')
  90                 self._worktree = core.abspath(gitparent)
  91                 self.set_cwd(self._worktree)
  92         return self._worktree
  93
  94     def is_valid(self):
  95         return self._git_dir and is_git_dir(self._git_dir)
  96
  97     def git_path(self, *paths):
  98         if self._git_file_path is None:
  99             return join(self.git_dir(), *paths)
 100         else:
 101             return join(self._git_file_path, *paths)
 102
 103     def git_dir(self):
 104         if self.is_valid():
 105             return self._git_dir
 106         if 'GIT_DIR' in os.environ:
 107             self._git_dir = core.getenv('GIT_DIR')
 108         if self._git_dir:
 109             curpath = core.abspath(self._git_dir)
 110         else:
 111             curpath = core.abspath(core.getcwd())
 112         # Search for a .git directory
 113         while curpath:
 114             if is_git_dir(curpath):
 115                 self._git_dir = curpath
 116                 break
 117             gitpath = join(curpath, '.git')
 118             if is_git_dir(gitpath):
 119                 self._git_dir = gitpath
 120                 break
 121             curpath, dummy = os.path.split(curpath)
 122             if not dummy:
 123                 break
 124         self._git_file_path = read_git_file(self._git_dir)
 125         return self._git_dir
 126
 127     def set_cwd(self, path):
 128         """Sets the current directory."""
 129         self._git_cwd = path
 130
 131     def __getattr__(self, name):
 132         if name[:1] == '_':
 133             raise AttributeError(name)
 134         return lambda *args, **kwargs: self._call_process(name, *args, **kwargs)
 135
 136     @staticmethod
 137     def execute(command,
 138                 _cwd=None,
 139                 _decode=True,
 140                 _encoding=None,
 141                 _raw=False,
 142                 _stdin=None,
 143                 _stderr=subprocess.PIPE,
 144                 _stdout=subprocess.PIPE):
 145         """
 146         Execute a command and returns its output
 147
 148         :param command: argument list to execute.
 149         :param _cwd: working directory, defaults to the current directory.
 150         :param _decode: whether to decode output, defaults to True.
 151         :param _encoding: default encoding, defaults to None (utf-8).
 152         :param _raw: do not strip trailing whitespace.
 153         :param _stdin: optional stdin filehandle.
 154         :returns (status, out, err): exit status, stdout, stderr
 155
 156         """
 157         # Allow the user to have the command executed in their working dir.
 158         if not _cwd:
 159             _cwd = core.getcwd()
 160
 161         extra = {}
 162         if sys.platform == 'win32':
 163             command = map(replace_carot, command)
 164             extra['shell'] = True
 165
 166         # Start the process
 167         # Guard against thread-unsafe .git/index.lock files
 168         INDEX_LOCK.acquire()
 169         status, out, err = core.run_command(command,
 170                                             cwd=_cwd,
 171                                             encoding=_encoding,
 172                                             stdin=_stdin, stdout=_stdout, stderr=_stderr,
 173                                             **extra)
 174         # Let the next thread in
 175         INDEX_LOCK.release()
 176         if not _raw and out is not None:
 177             out = out.rstrip('\n')
 178
 179         cola_trace = GIT_COLA_TRACE
 180         if cola_trace == 'trace':
 181             msg = 'trace: ' + subprocess.list2cmdline(command)
 182             Interaction.log_status(status, msg, '')
 183         elif cola_trace == 'full':
 184             if out:
 185                 core.stderr("%s -> %d: '%s' '%s'" %
 186                             (' '.join(command), status, out, err))
 187             else:
 188                 core.stderr("%s -> %d" % (' '.join(command), status))
 189         elif cola_trace:
 190             core.stderr(' '.join(command))
 191
 192         # Allow access to the command's status code
 193         return (status, out, err)
 194
 195     def transform_kwargs(self, **kwargs):
 196         """Transform kwargs into git command line options"""
 197         args = []
 198         for k, v in kwargs.items():
 199             if len(k) == 1:
 200                 if v is True:
 201                     args.append("-%s" % k)
 202                 elif type(v) is not bool:
 203                     args.append("-%s%s" % (k, v))
 204             else:
 205                 if v is True:
 206                     args.append("--%s" % dashify(k))
 207                 elif type(v) is not bool:
 208                     args.append("--%s=%s" % (dashify(k), v))
 209         return args
 210
 211     def _call_process(self, cmd, *args, **kwargs):
 212         # Handle optional arguments prior to calling transform_kwargs
 213         # otherwise they'll end up in args, which is bad.
 214         _kwargs = dict(_cwd=self._git_cwd)
 215         execute_kwargs = ('_cwd', '_decode', '_encoding',
 216                 '_stdin', '_stdout', '_stderr', '_raw')
 217         for kwarg in execute_kwargs:
 218             if kwarg in kwargs:
 219                 _kwargs[kwarg] = kwargs.pop(kwarg)
 220
 221         # Prepare the argument list
 222         opt_args = self.transform_kwargs(**kwargs)
 223         call = ['git', dashify(cmd)] + opt_args
 224         call.extend(args)
 225         return self.execute(call, **_kwargs)
 226
 227
 228 def replace_carot(cmd_arg):
 229     """
 230     Guard against the windows command shell.
 231
 232     In the Windows shell, a carat character (^) may be used for
 233     line continuation.  To guard against this, escape the carat
 234     by using two of them.
 235
 236     http://technet.microsoft.com/en-us/library/cc723564.aspx
 237
 238     """
 239     return cmd_arg.replace('^', '^^')
 240
 241
 242 @memoize
 243 def instance():
 244     """Return the Git singleton"""
 245     return Git()
 246
 247
 248 git = instance()
 249 """
 250 Git command singleton
 251
 252 >>> from cola.git import git
 253 >>> from cola.git import STDOUT
 254 >>> 'git' == git.version()[STDOUT][:3].lower()
 255 True
 256
 257 """