Source code for plugins.gitrepo

from plugins.git.repo import Repo, Head, Commit
from plugins.git.errors import *

class GitCommit (Commit) :
	def get_age (self) :
		"""return relative time since the commit date.
		note: If you want the absolute time, use self.committed_date instead"""
		age = self.repo.git.show('--format="%ar"', self.sha)
		return age.splitlines()[0].strip('"')

	def get_tag(self) :
		"""Check if commit correspond to a tag.
		return Tag or None"""
		for tag in self.repo.tags :
			if tag.commit.sha == self.sha :
				return tag
		return None

	def get_branch(self) :
		"""Check if commit correspond to a head of branch
		return Branch or None"""
		for branch in self.repo.branches :
			if branch.commit.sha == self.sha :
				return branch
		for remote in self.repo.remotes :
			for branch in remote.refs :
				if branch.remote_head == 'HEAD' :
					continue
				if branch.commit.sha == self.sha :
					return branch
		return None

	def get_parent_branch(self) :
		"""Return branch if commit is on head or a parent branch"""
		branch = self.get_branch ()
		if branch :
			return branch
		for branch in self.repo.branches :
			for commit in self.repo.iter_commits (branch) :
				if commit.sha == self.sha :
					return branch
		for remote in self.repo.remotes :
			for branch in remote.refs :
				if branch.remote_head == 'HEAD' :
					continue
				for commit in self.repo.iter_commits (branch) :
					if commit.sha == self.sha :
						return branch
		return None

class GitRepoError(Exception) :
	"""
	Base Class for GitRepo Exception
	"""

class GitNoBranchError(GitRepoError) :
	"""
	Thrown when head is in strange state (Not in any branch ?).
	(Can be happened when rebasing for example)
	"""

class GitRepo :
	"""Git tools on top of GitPython"""

	def __init__ (self) :
		self._repo = None
		self.refresh ()

	def refresh (self) :
		path = self.get_path () # child class must implement get_path()
		if path is None :
			self._repo = None
		else :
			try :
				self._repo = Repo (path)
			except (NoSuchPathError, InvalidGitRepositoryError) :
				return
			if not len (self._repo.heads) :
				# no branch = only init'ed -> cannot work with it
				self._repo = None

	def get_current_branch (self) :
		if self._repo is None :
			return None
		if self._repo.head.is_detached :
			return None
		return self._repo.active_branch

	def get_branches (self) :
		if self._repo is None :
			return []
		# TODO: return remotes ?
		return self._repo.branches

	def get_tracked_branch (self) :
		if not self._repo.remotes :
			return None
		# TODO : get_tracked_branch (can be different of first one)
		return 'remotes/%s/%s' % (self._repo.remotes[0].name, self.get_current_branch().name)

	def is_tagged (self, branch=None) :
		return len (self.get_head_tags (branch)) > 0

	def get_head_tags (self, branch=None) :
		if self._repo is None :
			return []
		if branch is None :
			head_commit = self._repo.head.commit
		elif isinstance (branch, Head) :
			head_commit = branch.commit
		else :
			head = [ head for head in self._repo.heads if head.name == branch ]
			if not head :
				return []
			head_commit = head[0].commit
		return [ tag for tag in self._repo.tags if tag.commit == head_commit ]

	def get_latest_tags (self, branch=None) :
		if self._repo is None :
			return []
		tagged_commits = [ tag.commit for tag in self._repo.tags ]
		if not tagged_commits :
			return []
		if branch is None :
			branch = self.get_current_branch ()
		if branch is None :
			return self.get_head_tags ()
		for commit in self._repo.iter_commits (branch) :
			if commit in tagged_commits :
				# tags of the latest tagged commit of branch
				return [ tag for tag in self._repo.tags if tag.commit == commit ]
		return []

	def get_latest_tag (self, branch=None) :
		latest_tags = self.get_latest_tags (branch)
		if not latest_tags :
			return None
		return latest_tags [0] # choose the first if many

	def get_tags (self, branches=None) :
		if self._repo is None :
			return []
		if branches is None :
			# all tags
			return self._repo.tags
		commits = []
		for branch in branches :
			commits += self._repo.iter_commits (branch)
		# TODO: tag order
		# tags matching commits of the specified branches
		return [ tag for tag in self._repo.tags if tag.commit in commits ]

	def set_tag (self, tag) :
		if self._repo :
			self._repo.create_tag (tag)

	def get_ref_branch (self, ref) :
		"""return branch which contains ref"""
		if ref.get ('type') == 'branch' :
			return self._repo.branches [ref.get ('name')]
		commit = ref.get ('commit')
		if not commit :
			return None
		commit = GitCommit (self._repo, commit.sha)
		return commit.get_parent_branch ()

	def get_branch_refs (self, ref) :
		"""return ordered tags and branch head for branch which contains ref"""
		branch = self.get_ref_branch (ref)
		if not branch :
			return None
		tags = self.get_tags ([branch])
		tags.sort (key=lambda x: x.name)
		refs = [ {'commit': tag.commit, 'name': tag.name, 'type': 'tag'} for tag in tags ]
		if not self.is_tagged (branch) :
			refs.append ({'commit': branch.commit, 'name': branch.name, 'type': 'branch'})
		return refs

	def start_branch (self, branch) :
		raise NotImplementedError ("TODO: start_branch")

	def get_current_commit (self) :
		"""return SHA for HEAD"""
		if self._repo is None :
			return None
		return GitCommit (self._repo, self._repo.head.commit.sha)

	def get_current_ref (self) :
		"""return dict {commit, name, type} for HEAD"""
		if self._repo is None :
			return {}
		head = self.get_current_commit ()
		if head.get_tag() :
			return {'commit' : GitCommit (self._repo, head.sha), 'name' : head.get_tag().__str__(), 'type' : 'tag'}
		elif head.get_branch() :
			return {'commit' : GitCommit (self._repo, head.sha), 'name' : head.get_branch().__str__(), 'type' : 'branch'}
		# Last commit is inevitably in a branch, or something is wrong..
		raise GitNoBranchError('Cannot recognize head state')

	def has_changes (self) :
		if self._repo is None :
			return False
		return self._repo.is_dirty ()

	def is_updated (self) :
		raise NotImplementedError ("TODO: is_updated")

	def is_published (self) :
		if self._repo is None:
			return False
		current_branch = self.get_current_branch ()
		if current_branch is None :
			# TODO: handle correctly detached state (case of a tag)
			if self.get_current_commit ().get_tag () :
				return True
			return False
		remote_branch = self.get_tracked_branch ()
		if remote_branch is None :
			return False
		if len (self._repo.git.cherry (remote_branch, current_branch)) :
			return False
		return True

	def publish (self) :
		# TODO: handle error correctly
		try :
			self._repo.git.push ()
		except GitCommandError :
			pass

	def checkout (self, ref) :
		#ref.checkout () # this checkout doesn't allow to checkout a tag
		print "checkout to %s" % ref
		# TODO: check if it is a remote branch
		# TODO: check if a local branch track this remote
		# TODO: create local branch
		self._repo.git.checkout (ref)

	def is_writeable (self) :
		if self._repo is None :
			return False
		return not self._repo.head.is_detached

	def commit (self, comment, amend=False) :
		if not self.has_changes () :
			return
		if amend :
			arg = "--amend"
		else :
			arg = "-m%s" % comment
		self._repo.git.commit ('-a', arg)

	def get_commit_message(self, ref=None, short=False) :
		"""Return the last commit message or None if ref doesn't exist.
		If short is enable, just return the summary message"""
		if self._repo is None:
			return None
		commit = (ref.commit if ref else self._repo.head.commit)
		return (commit.summary if short else commit.message)