lib/hg.py

   1 # -*- coding: utf-8 -*-
   2 import os
   3 from mercurial import localrepo, ui, encoding, util
   4 import mercurial.merge, mercurial.error
   5
   6 encoding.encoding = 'utf-8'
   7
   8
   9 def clearpath(path):
  10     return unicode(path).encode("utf-8")
  11
  12 class Repository(object):
  13     """Abstrakcja repozytorium Mercurial. Działa z Mercurial w wersji 1.3.1."""
  14
  15     def __init__(self, path, create=False):
  16         self.ui = ui.ui()
  17         self.ui.config('ui', 'quiet', 'true')
  18         self.ui.config('ui', 'interactive', 'false')
  19
  20         self.real_path = os.path.realpath(path)
  21         self.repo = self.open_repository(self.real_path, create)
  22         self._pending_files = []
  23
  24     def open_repository(self, path, create=False):
  25         if os.path.isdir(path):
  26             try:
  27                 return localrepo.localrepository(self.ui, path)
  28             except mercurial.error.RepoError:
  29                 # dir is not an hg repo, we must init it
  30                 if create:
  31                     return localrepo.localrepository(self.ui, path, create=1)
  32         elif create:
  33             os.makedirs(path)
  34             return localrepo.localrepository(self.ui, path, create=1)
  35         raise RepositoryDoesNotExist("Repository %s does not exist." % path)
  36
  37     def file_list(self, branch):
  38         return self.in_branch(lambda: self._file_list(), branch)
  39
  40     def _file_list(self):
  41         return list(self.repo[None])
  42
  43     def get_file(self, path, branch):
  44         return self.in_branch(lambda: self._get_file(path), branch)
  45
  46     def _get_file(self, path):
  47         path = clearpath(path)
  48
  49         if not self._file_exists(path):
  50             raise RepositoryException("File not availble in this branch.")
  51
  52         return self.repo.wread(path)
  53
  54     def file_exists(self, path, branch):
  55         return self.in_branch(lambda: self._file_exists(path), branch)
  56
  57     def _file_exists(self, path):
  58         path = clearpath(path)
  59         return self.repo.dirstate[path] != "?"
  60
  61     def write_file(self, path, value, branch):
  62         return self.in_branch(lambda: self._write_file(path, value), branch)
  63
  64     def _write_file(self, path, value):
  65         path = clearpath(path)
  66         return self.repo.wwrite(path, value, [])
  67
  68     def add_file(self, path, value, branch):
  69         return self.in_branch(lambda: self._add_file(path, value), branch)
  70
  71     def _add_file(self, path, value):
  72         path = clearpath(path)
  73         self._write_file(path, value)
  74         return self.repo.add( [path] )
  75
  76     def _commit(self, message, user=None):
  77         return self.repo.commit(text=message, user=user)
  78
  79     def commit(self, message, branch, user=None):
  80         return self.in_branch(lambda: self._commit(message, key=key, user=user), branch)
  81
  82     def in_branch(self, action, bname):
  83         wlock = self.repo.wlock()
  84         try:
  85             old = self._switch_to_branch(bname)
  86             try:
  87                 # do some stuff
  88                 return action()
  89             finally:
  90                 self._switch_to_branch(old)
  91         finally:
  92             wlock.release()
  93
  94     def _switch_to_branch(self, bname, create=True):
  95         wlock = self.repo.wlock()
  96         try:
  97             current = self.repo[None].branch()
  98             if current == bname:
  99                 return current
 100             try:
 101                 tip = self.repo.branchtags()[bname]
 102             except KeyError, ke:
 103                 if not create: raise ke
 104
 105                 # create the branch on the fly
 106
 107                 # first switch to default branch
 108                 default_tip = self.repo.branchtags()['default']
 109                 mercurial.merge.update(self.repo, default_tip, False, True, None)
 110
 111                 # set the dirstate to new branch
 112                 self.repo.dirstate.setbranch(bname)
 113                 self._commit('Initial commit for automatic branch "%s".' % bname, user="django-admin")
 114
 115                 # collect the new tip
 116                 tip = self.repo.branchtags()[bname]
 117
 118             upstats = mercurial.merge.update(self.repo, tip, False, True, None)
 119             return current
 120         except KeyError, ke:
 121             raise RepositoryException("Can't switch to branch '%s': no such branch." % bname , ke)
 122         except util.Abort, ae:
 123             raise RepositoryException("Can't switch to branch '%s': %s"  % (bname, ae.message), ae)
 124         finally:
 125             wlock.release()
 126
 127     def write_lock(self):
 128         """Returns w write lock to the repository."""
 129         return self.repo.wlock()
 130
 131
 132 class RepositoryException(Exception):
 133
 134     def __init__(self, msg, cause=None):
 135         Exception.__init__(self, msg)
 136         self.cause = cause
 137
 138 class RepositoryDoesNotExist(RepositoryException):
 139     pass