refs.py
No OneTemporary
Actions

Size

24 KB

Subscribers

None

refs.py
View Options

	# refs.py -- For dealing with git refs
	# Copyright (C) 2008-2013 Jelmer Vernooij <jelmer@samba.org>
	#
	# This program is free software; you can redistribute it and/or
	# modify it under the terms of the GNU General Public License
	# as published by the Free Software Foundation; version 2
	# of the License or (at your option) any later version of
	# the License.
	#
	# This program is distributed in the hope that it will be useful,
	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	# GNU General Public License for more details.
	#
	# You should have received a copy of the GNU General Public License
	# along with this program; if not, write to the Free Software
	# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
	# MA 02110-1301, USA.


	"""Ref handling.

	"""
	import errno
	import os

	from dulwich.errors import (
	PackedRefsException,
	RefFormatError,
	)
	from dulwich.objects import (
	hex_to_sha,
	git_line,
	)
	from dulwich.file import (
	GitFile,
	ensure_dir_exists,
	)
	from dulwich._py3_compat import (
	items,
	)


	SYMREF = b'ref: '
	LOCAL_BRANCH_PREFIX = b'refs/heads/'
	BAD_REF_CHARS = set(b'\177 ~^:?*[')


	def check_ref_format(refname):
	"""Check if a refname is correctly formatted.

	Implements all the same rules as git-check-ref-format[1].

	[1] http://www.kernel.org/pub/software/scm/git/docs/git-check-ref-format.html

	:param refname: The refname to check
	:return: True if refname is valid, False otherwise
	"""
	# These could be combined into one big expression, but are listed separately
	# to parallel [1].
	if b'/.' in refname or refname.startswith(b'.'):
	return False
	if b'/' not in refname:
	return False
	if b'..' in refname:
	return False
	for i, c in enumerate(refname):
	if ord(refname[i:i+1]) < 0o40 or c in BAD_REF_CHARS:
	return False
	if refname[-1] in b'/.':
	return False
	if refname.endswith(b'.lock'):
	return False
	if b'@{' in refname:
	return False
	if b'\\' in refname:
	return False
	return True


	class RefsContainer(object):
	"""A container for refs."""

	def set_symbolic_ref(self, name, other):
	"""Make a ref point at another ref.

	:param name: Name of the ref to set
	:param other: Name of the ref to point at
	"""
	raise NotImplementedError(self.set_symbolic_ref)

	def get_packed_refs(self):
	"""Get contents of the packed-refs file.

	:return: Dictionary mapping ref names to SHA1s

	:note: Will return an empty dictionary when no packed-refs file is
	present.
	"""
	raise NotImplementedError(self.get_packed_refs)

	def get_peeled(self, name):
	"""Return the cached peeled value of a ref, if available.

	:param name: Name of the ref to peel
	:return: The peeled value of the ref. If the ref is known not point to a
	tag, this will be the SHA the ref refers to. If the ref may point to
	a tag, but no cached information is available, None is returned.
	"""
	return None

	def import_refs(self, base, other):
	for name, value in items(other):
	self[b'/'.join((base, name))] = value

	def allkeys(self):
	"""All refs present in this container."""
	raise NotImplementedError(self.allkeys)

	def keys(self, base=None):
	"""Refs present in this container.

	:param base: An optional base to return refs under.
	:return: An unsorted set of valid refs in this container, including
	packed refs.
	"""
	if base is not None:
	return self.subkeys(base)
	else:
	return self.allkeys()

	def subkeys(self, base):
	"""Refs present in this container under a base.

	:param base: The base to return refs under.
	:return: A set of valid refs in this container under the base; the base
	prefix is stripped from the ref names returned.
	"""
	keys = set()
	base_len = len(base) + 1
	for refname in self.allkeys():
	if refname.startswith(base):
	keys.add(refname[base_len:])
	return keys

	def as_dict(self, base=None):
	"""Return the contents of this container as a dictionary.

	"""
	ret = {}
	keys = self.keys(base)
	if base is None:
	base = b''
	for key in keys:
	try:
	ret[key] = self[(base + b'/' + key).strip(b'/')]
	except KeyError:
	continue # Unable to resolve

	return ret

	def _check_refname(self, name):
	"""Ensure a refname is valid and lives in refs or is HEAD.

	HEAD is not a valid refname according to git-check-ref-format, but this
	class needs to be able to touch HEAD. Also, check_ref_format expects
	refnames without the leading 'refs/', but this class requires that
	so it cannot touch anything outside the refs dir (or HEAD).

	:param name: The name of the reference.
	:raises KeyError: if a refname is not HEAD or is otherwise not valid.
	"""
	if name in (b'HEAD', b'refs/stash'):
	return
	if not name.startswith(b'refs/') or not check_ref_format(name[5:]):
	raise RefFormatError(name)

	def read_ref(self, refname):
	"""Read a reference without following any references.

	:param refname: The name of the reference
	:return: The contents of the ref file, or None if it does
	not exist.
	"""
	contents = self.read_loose_ref(refname)
	if not contents:
	contents = self.get_packed_refs().get(refname, None)
	return contents

	def read_loose_ref(self, name):
	"""Read a loose reference and return its contents.

	:param name: the refname to read
	:return: The contents of the ref file, or None if it does
	not exist.
	"""
	raise NotImplementedError(self.read_loose_ref)

	def _follow(self, name):
	"""Follow a reference name.

	:return: a tuple of (refname, sha), where refname is the name of the
	last reference in the symbolic reference chain
	"""
	contents = SYMREF + name
	depth = 0
	while contents.startswith(SYMREF):
	refname = contents[len(SYMREF):]
	contents = self.read_ref(refname)
	if not contents:
	break
	depth += 1
	if depth > 5:
	raise KeyError(name)
	return refname, contents

	def __contains__(self, refname):
	if self.read_ref(refname):
	return True
	return False

	def __getitem__(self, name):
	"""Get the SHA1 for a reference name.

	This method follows all symbolic references.
	"""
	_, sha = self._follow(name)
	if sha is None:
	raise KeyError(name)
	return sha

	def set_if_equals(self, name, old_ref, new_ref):
	"""Set a refname to new_ref only if it currently equals old_ref.

	This method follows all symbolic references if applicable for the
	subclass, and can be used to perform an atomic compare-and-swap
	operation.

	:param name: The refname to set.
	:param old_ref: The old sha the refname must refer to, or None to set
	unconditionally.
	:param new_ref: The new sha the refname will refer to.
	:return: True if the set was successful, False otherwise.
	"""
	raise NotImplementedError(self.set_if_equals)

	def add_if_new(self, name, ref):
	"""Add a new reference only if it does not already exist."""
	raise NotImplementedError(self.add_if_new)

	def __setitem__(self, name, ref):
	"""Set a reference name to point to the given SHA1.

	This method follows all symbolic references if applicable for the
	subclass.

	:note: This method unconditionally overwrites the contents of a
	reference. To update atomically only if the reference has not
	changed, use set_if_equals().
	:param name: The refname to set.
	:param ref: The new sha the refname will refer to.
	"""
	self.set_if_equals(name, None, ref)

	def remove_if_equals(self, name, old_ref):
	"""Remove a refname only if it currently equals old_ref.

	This method does not follow symbolic references, even if applicable for
	the subclass. It can be used to perform an atomic compare-and-delete
	operation.

	:param name: The refname to delete.
	:param old_ref: The old sha the refname must refer to, or None to delete
	unconditionally.
	:return: True if the delete was successful, False otherwise.
	"""
	raise NotImplementedError(self.remove_if_equals)

	def __delitem__(self, name):
	"""Remove a refname.

	This method does not follow symbolic references, even if applicable for
	the subclass.

	:note: This method unconditionally deletes the contents of a reference.
	To delete atomically only if the reference has not changed, use
	remove_if_equals().

	:param name: The refname to delete.
	"""
	self.remove_if_equals(name, None)


	class DictRefsContainer(RefsContainer):
	"""RefsContainer backed by a simple dict.

	This container does not support symbolic or packed references and is not
	threadsafe.
	"""

	def __init__(self, refs):
	self._refs = refs
	self._peeled = {}

	def allkeys(self):
	return self._refs.keys()

	def read_loose_ref(self, name):
	return self._refs.get(name, None)

	def get_packed_refs(self):
	return {}

	def set_symbolic_ref(self, name, other):
	self._refs[name] = SYMREF + other

	def set_if_equals(self, name, old_ref, new_ref):
	if old_ref is not None and self._refs.get(name, None) != old_ref:
	return False
	realname, _ = self._follow(name)
	self._check_refname(realname)
	self._refs[realname] = new_ref
	return True

	def add_if_new(self, name, ref):
	if name in self._refs:
	return False
	self._refs[name] = ref
	return True

	def remove_if_equals(self, name, old_ref):
	if old_ref is not None and self._refs.get(name, None) != old_ref:
	return False
	del self._refs[name]
	return True

	def get_peeled(self, name):
	return self._peeled.get(name)

	def _update(self, refs):
	"""Update multiple refs; intended only for testing."""
	# TODO(dborowitz): replace this with a public function that uses
	# set_if_equal.
	self._refs.update(refs)

	def _update_peeled(self, peeled):
	"""Update cached peeled refs; intended only for testing."""
	self._peeled.update(peeled)


	class InfoRefsContainer(RefsContainer):
	"""Refs container that reads refs from a info/refs file."""

	def __init__(self, f):
	self._refs = {}
	self._peeled = {}
	for l in f.readlines():
	sha, name = l.rstrip(b'\n').split(b'\t')
	if name.endswith(b'^{}'):
	name = name[:-3]
	if not check_ref_format(name):
	raise ValueError("invalid ref name %r" % name)
	self._peeled[name] = sha
	else:
	if not check_ref_format(name):
	raise ValueError("invalid ref name %r" % name)
	self._refs[name] = sha

	def allkeys(self):
	return self._refs.keys()

	def read_loose_ref(self, name):
	return self._refs.get(name, None)

	def get_packed_refs(self):
	return {}

	def get_peeled(self, name):
	try:
	return self._peeled[name]
	except KeyError:
	return self._refs[name]


	class DiskRefsContainer(RefsContainer):
	"""Refs container that reads refs from disk."""

	def __init__(self, path):
	self.path = path
	self._packed_refs = None
	self._peeled_refs = None

	def __repr__(self):
	return "%s(%r)" % (self.__class__.__name__, self.path)

	def subkeys(self, base):
	subkeys = set()
	path = self.refpath(base)
	for root, dirs, files in os.walk(path):
	dir = root[len(path):].strip(os.path.sep).replace(os.path.sep, "/")
	for filename in files:
	refname = (("%s/%s" % (dir, filename))
	.strip("/").encode('ascii'))
	# check_ref_format requires at least one /, so we prepend the
	# base before calling it.
	if check_ref_format(base + b'/' + refname):
	subkeys.add(refname)
	for key in self.get_packed_refs():
	if key.startswith(base):
	subkeys.add(key[len(base):].strip(b'/'))
	return subkeys

	def allkeys(self):
	allkeys = set()
	if os.path.exists(self.refpath(b'HEAD')):
	allkeys.add(b'HEAD')
	path = self.refpath(b'')
	for root, dirs, files in os.walk(self.refpath(b'refs')):
	dir = root[len(path):].strip(os.path.sep).replace(os.path.sep, "/")
	for filename in files:
	refname = ("%s/%s" % (dir, filename)).strip("/").encode('ascii')
	if check_ref_format(refname):
	allkeys.add(refname)
	allkeys.update(self.get_packed_refs())
	return allkeys

	def refpath(self, name):
	"""Return the disk path of a ref.

	"""
	name = name.decode('ascii')
	if os.path.sep != "/":
	name = name.replace("/", os.path.sep)
	return os.path.join(self.path, name)

	def get_packed_refs(self):
	"""Get contents of the packed-refs file.

	:return: Dictionary mapping ref names to SHA1s

	:note: Will return an empty dictionary when no packed-refs file is
	present.
	"""
	# TODO: invalidate the cache on repacking
	if self._packed_refs is None:
	# set both to empty because we want _peeled_refs to be
	# None if and only if _packed_refs is also None.
	self._packed_refs = {}
	self._peeled_refs = {}
	path = os.path.join(self.path, 'packed-refs')
	try:
	f = GitFile(path, 'rb')
	except IOError as e:
	if e.errno == errno.ENOENT:
	return {}
	raise
	with f:
	first_line = next(iter(f)).rstrip()
	if (first_line.startswith(b'# pack-refs') and b' peeled' in
	first_line):
	for sha, name, peeled in read_packed_refs_with_peeled(f):
	self._packed_refs[name] = sha
	if peeled:
	self._peeled_refs[name] = peeled
	else:
	f.seek(0)
	for sha, name in read_packed_refs(f):
	self._packed_refs[name] = sha
	return self._packed_refs

	def get_peeled(self, name):
	"""Return the cached peeled value of a ref, if available.

	:param name: Name of the ref to peel
	:return: The peeled value of the ref. If the ref is known not point to a
	tag, this will be the SHA the ref refers to. If the ref may point to
	a tag, but no cached information is available, None is returned.
	"""
	self.get_packed_refs()
	if self._peeled_refs is None or name not in self._packed_refs:
	# No cache: no peeled refs were read, or this ref is loose
	return None
	if name in self._peeled_refs:
	return self._peeled_refs[name]
	else:
	# Known not peelable
	return self[name]

	def read_loose_ref(self, name):
	"""Read a reference file and return its contents.

	If the reference file a symbolic reference, only read the first line of
	the file. Otherwise, only read the first 40 bytes.

	:param name: the refname to read, relative to refpath
	:return: The contents of the ref file, or None if the file does not
	exist.
	:raises IOError: if any other error occurs
	"""
	filename = self.refpath(name)
	try:
	with GitFile(filename, 'rb') as f:
	header = f.read(len(SYMREF))
	if header == SYMREF:
	# Read only the first line
	return header + next(iter(f)).rstrip(b'\r\n')
	else:
	# Read only the first 40 bytes
	return header + f.read(40 - len(SYMREF))
	except IOError as e:
	if e.errno == errno.ENOENT:
	return None
	raise

	def _remove_packed_ref(self, name):
	if self._packed_refs is None:
	return
	filename = os.path.join(self.path, 'packed-refs')
	# reread cached refs from disk, while holding the lock
	f = GitFile(filename, 'wb')
	try:
	self._packed_refs = None
	self.get_packed_refs()

	if name not in self._packed_refs:
	return

	del self._packed_refs[name]
	if name in self._peeled_refs:
	del self._peeled_refs[name]
	write_packed_refs(f, self._packed_refs, self._peeled_refs)
	f.close()
	finally:
	f.abort()

	def set_symbolic_ref(self, name, other):
	"""Make a ref point at another ref.

	:param name: Name of the ref to set
	:param other: Name of the ref to point at
	"""
	self._check_refname(name)
	self._check_refname(other)
	filename = self.refpath(name)
	try:
	f = GitFile(filename, 'wb')
	try:
	f.write(SYMREF + other + b'\n')
	except (IOError, OSError):
	f.abort()
	raise
	finally:
	f.close()

	def set_if_equals(self, name, old_ref, new_ref):
	"""Set a refname to new_ref only if it currently equals old_ref.

	This method follows all symbolic references, and can be used to perform
	an atomic compare-and-swap operation.

	:param name: The refname to set.
	:param old_ref: The old sha the refname must refer to, or None to set
	unconditionally.
	:param new_ref: The new sha the refname will refer to.
	:return: True if the set was successful, False otherwise.
	"""
	self._check_refname(name)
	try:
	realname, _ = self._follow(name)
	except KeyError:
	realname = name
	filename = self.refpath(realname)
	ensure_dir_exists(os.path.dirname(filename))
	with GitFile(filename, 'wb') as f:
	if old_ref is not None:
	try:
	# read again while holding the lock
	orig_ref = self.read_loose_ref(realname)
	if orig_ref is None:
	orig_ref = self.get_packed_refs().get(realname, None)
	if orig_ref != old_ref:
	f.abort()
	return False
	except (OSError, IOError):
	f.abort()
	raise
	try:
	f.write(new_ref + b'\n')
	except (OSError, IOError):
	f.abort()
	raise
	return True

	def add_if_new(self, name, ref):
	"""Add a new reference only if it does not already exist.

	This method follows symrefs, and only ensures that the last ref in the
	chain does not exist.

	:param name: The refname to set.
	:param ref: The new sha the refname will refer to.
	:return: True if the add was successful, False otherwise.
	"""
	try:
	realname, contents = self._follow(name)
	if contents is not None:
	return False
	except KeyError:
	realname = name
	self._check_refname(realname)
	filename = self.refpath(realname)
	ensure_dir_exists(os.path.dirname(filename))
	with GitFile(filename, 'wb') as f:
	if os.path.exists(filename) or name in self.get_packed_refs():
	f.abort()
	return False
	try:
	f.write(ref + b'\n')
	except (OSError, IOError):
	f.abort()
	raise
	return True

	def remove_if_equals(self, name, old_ref):
	"""Remove a refname only if it currently equals old_ref.

	This method does not follow symbolic references. It can be used to
	perform an atomic compare-and-delete operation.

	:param name: The refname to delete.
	:param old_ref: The old sha the refname must refer to, or None to delete
	unconditionally.
	:return: True if the delete was successful, False otherwise.
	"""
	self._check_refname(name)
	filename = self.refpath(name)
	ensure_dir_exists(os.path.dirname(filename))
	f = GitFile(filename, 'wb')
	try:
	if old_ref is not None:
	orig_ref = self.read_loose_ref(name)
	if orig_ref is None:
	orig_ref = self.get_packed_refs().get(name, None)
	if orig_ref != old_ref:
	return False
	# may only be packed
	try:
	os.remove(filename)
	except OSError as e:
	if e.errno != errno.ENOENT:
	raise
	self._remove_packed_ref(name)
	finally:
	# never write, we just wanted the lock
	f.abort()
	return True


	def _split_ref_line(line):
	"""Split a single ref line into a tuple of SHA1 and name."""
	fields = line.rstrip(b'\n').split(b' ')
	if len(fields) != 2:
	raise PackedRefsException("invalid ref line %r" % line)
	sha, name = fields
	try:
	hex_to_sha(sha)
	except (AssertionError, TypeError) as e:
	raise PackedRefsException(e)
	if not check_ref_format(name):
	raise PackedRefsException("invalid ref name %r" % name)
	return (sha, name)


	def read_packed_refs(f):
	"""Read a packed refs file.

	:param f: file-like object to read from
	:return: Iterator over tuples with SHA1s and ref names.
	"""
	for l in f:
	if l.startswith(b'#'):
	# Comment
	continue
	if l.startswith(b'^'):
	raise PackedRefsException(
	"found peeled ref in packed-refs without peeled")
	yield _split_ref_line(l)


	def read_packed_refs_with_peeled(f):
	"""Read a packed refs file including peeled refs.

	Assumes the "# pack-refs with: peeled" line was already read. Yields tuples
	with ref names, SHA1s, and peeled SHA1s (or None).

	:param f: file-like object to read from, seek'ed to the second line
	"""
	last = None
	for l in f:
	if l[0] == b'#':
	continue
	l = l.rstrip(b'\r\n')
	if l.startswith(b'^'):
	if not last:
	raise PackedRefsException("unexpected peeled ref line")
	try:
	hex_to_sha(l[1:])
	except (AssertionError, TypeError) as e:
	raise PackedRefsException(e)
	sha, name = _split_ref_line(last)
	last = None
	yield (sha, name, l[1:])
	else:
	if last:
	sha, name = _split_ref_line(last)
	yield (sha, name, None)
	last = l
	if last:
	sha, name = _split_ref_line(last)
	yield (sha, name, None)


	def write_packed_refs(f, packed_refs, peeled_refs=None):
	"""Write a packed refs file.

	:param f: empty file-like object to write to
	:param packed_refs: dict of refname to sha of packed refs to write
	:param peeled_refs: dict of refname to peeled value of sha
	"""
	if peeled_refs is None:
	peeled_refs = {}
	else:
	f.write(b'# pack-refs with: peeled\n')
	for refname in sorted(packed_refs.keys()):
	f.write(git_line(packed_refs[refname], refname))
	if refname in peeled_refs:
	f.write(b'^' + peeled_refs[refname] + b'\n')


	def read_info_refs(f):
	ret = {}
	for l in f.readlines():
	(sha, name) = l.rstrip("\r\n").split("\t", 1)
	ret[name] = sha
	return ret


	def write_info_refs(refs, store):
	"""Generate info refs."""
	for name, sha in sorted(refs.items()):
	# get_refs() includes HEAD as a special case, but we don't want to
	# advertise it
	if name == b'HEAD':
	continue
	try:
	o = store[sha]
	except KeyError:
	continue
	peeled = store.peel_sha(sha)
	yield o.id + b'\t' + name + b'\n'
	if o.id != peeled.id:
	yield peeled.id + b'\t' + name + b'^{}\n'


	is_local_branch = lambda x: x.startswith(b'refs/heads/')

File Metadata

Mime Type: text/x-python
Expires: Jun 4 2025, 6:48 PM (12 w, 5 d ago)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 3398786

refs.pyNo OneTemporaryActions

refs.pyView Options

File Metadata

Event Timeline

refs.py
No OneTemporary
Actions

refs.py
View Options