# objects.py -- Access to base git objects
# Copyright (C) 2007 James Westby <jw+debian@jameswestby.net>
# Copyright (C) 2008-2009 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) a 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,
NotCommitError,
NotTreeError,
)
+from dulwich.file import GitFile
from dulwich.misc import (
make_sha,
)
TAGGER_ID = "tagger"
ENCODING_ID = "encoding"
-S_IFGITLINK = 0160000
+S_IFGITLINK = 0160000
+
def S_ISGITLINK(m):
return (stat.S_IFMT(m) == S_IFGITLINK)
class ShaFile(object):
"""A git SHA file."""
-
+
@classmethod
def _parse_legacy_object(cls, map):
"""Parse a legacy object, creating it and setting object._text"""
def as_legacy_object(self):
text = self.as_raw_string()
return zlib.compress("%s %d\0%s" % (self._type, len(text), text))
-
+
def as_raw_string(self):
if self._needs_serialization:
self.serialize()
self._sha = None
self._needs_parsing = True
self._needs_serialization = False
-
+
@classmethod
def _parse_object(cls, map):
"""Parse a new style object , creating it and setting object._text"""
raw = map[used:]
object.set_raw_string(_decompress(raw))
return object
-
+
@classmethod
def _parse_file(cls, map):
word = (ord(map[0]) << 8) + ord(map[1])
return cls._parse_legacy_object(map)
else:
return cls._parse_object(map)
-
+
def __init__(self):
"""Don't call this directly"""
self._sha = None
-
+
def _parse_text(self):
"""For subclasses to do initialisation time parsing"""
-
+
@classmethod
def from_file(cls, filename):
"""Get the contents of a SHA file on disk"""
size = os.path.getsize(filename)
- f = open(filename, 'rb')
+ f = GitFile(filename, 'rb')
try:
map = mmap.mmap(f.fileno(), size, access=mmap.ACCESS_READ)
shafile = cls._parse_file(map)
return shafile
finally:
f.close()
-
+
@classmethod
def from_raw_string(cls, type, string):
"""Creates an object of the indicated type from the raw string given.
-
+
Type is the numeric type of an object. String is the raw uncompressed
contents.
"""
obj.type = type
obj.set_raw_string(string)
return obj
-
+
+ @classmethod
+ def from_string(cls, string):
+ """Create a blob from a string."""
+ shafile = cls()
+ shafile.set_raw_string(string)
+ return shafile
+
+ def _raw_length(self):
+ """Returns the length of the raw string of this object."""
+ return len(self.as_raw_string())
+
def _header(self):
- return "%s %lu\0" % (self._type, len(self.as_raw_string()))
-
+ return "%s %lu\0" % (self._type, self._raw_length())
+
+ def _make_sha(self):
+ ret = make_sha()
+ ret.update(self._header())
+ ret.update(self.as_raw_string())
+ return ret
+
def sha(self):
"""The SHA1 object that is the name of this object."""
if self._needs_serialization or self._sha is None:
- self._sha = make_sha()
- self._sha.update(self._header())
- self._sha.update(self.as_raw_string())
+ self._sha = self._make_sha()
return self._sha
-
+
@property
def id(self):
return self.sha().hexdigest()
-
+
def get_type(self):
return self._num_type
self._num_type = type
type = property(get_type, set_type)
-
+
def __repr__(self):
return "<%s %s>" % (self.__class__.__name__, self.id)
def __ne__(self, other):
return self.id != other.id
-
+
def __eq__(self, other):
"""Return true id the sha of the two objects match.
-
+
The __le__ etc methods aren't overriden as they make no sense,
certainly at this level.
"""
_type = BLOB_ID
_num_type = 3
- _needs_serialization = False
- _needs_parsing = False
- def get_data(self):
+ def __init__(self):
+ super(Blob, self).__init__()
+ self._chunked = []
+ self._text = ""
+ self._needs_parsing = False
+ self._needs_serialization = False
+
+ def _get_data(self):
+ if self._needs_serialization:
+ self.serialize()
return self._text
- def set_data(self, data):
+ def _set_data(self, data):
self._text = data
+ self._needs_parsing = True
+ self._needs_serialization = False
- data = property(get_data, set_data,
+ data = property(_get_data, _set_data,
"The text contained within the blob object.")
+ def _get_chunked(self):
+ self._ensure_parsed()
+ return self._chunked
+
+ def _set_chunked(self, chunks):
+ self._chunked = chunks
+ self._needs_serialization = True
+
+ chunked = property(_get_chunked, _set_chunked,
+ "The text within the blob object, as chunks (not necessarily lines).")
+
+ def _parse_text(self):
+ self._chunked = [self._text]
+
+ def serialize(self):
+ self._text = "".join(self._chunked)
+
+ def _raw_length(self):
+ ret = 0
+ for chunk in self.chunked:
+ ret += len(chunk)
+ return ret
+
+ def _make_sha(self):
+ ret = make_sha()
+ ret.update(self._header())
+ for chunk in self._chunked:
+ ret.update(chunk)
+ return ret
+
@classmethod
def from_file(cls, filename):
blob = ShaFile.from_file(filename)
raise NotBlobError(filename)
return blob
- @classmethod
- def from_string(cls, string):
- """Create a blob from a string."""
- shafile = cls()
- shafile.set_raw_string(string)
- return shafile
-
class Tag(ShaFile):
"""A Git Tag object."""
f.write("%s %s\n" % (TYPE_ID, num_type_map[self._object_type]._type))
f.write("%s %s\n" % (TAG_ID, self._name))
if self._tagger:
- f.write("%s %s %d %s\n" % (TAGGER_ID, self._tagger, self._tag_time, format_timezone(self._tag_timezone)))
+ if self._tag_time is None:
+ f.write("%s %s\n" % (TAGGER_ID, self._tagger))
+ else:
+ f.write("%s %s %d %s\n" % (TAGGER_ID, self._tagger, self._tag_time, format_timezone(self._tag_timezone)))
f.write("\n") # To close headers
f.write(self._message)
self._text = f.getvalue()
elif field == TAG_ID:
self._name = value
elif field == TAGGER_ID:
- sep = value.index("> ")
- self._tagger = value[0:sep+1]
- (timetext, timezonetext) = value[sep+2:].rsplit(" ", 1)
try:
- self._tag_time = int(timetext)
- except ValueError: #Not a unix timestamp
- self._tag_time = time.strptime(timetext)
- self._tag_timezone = parse_timezone(timezonetext)
+ sep = value.index("> ")
+ except ValueError:
+ self._tagger = value
+ self._tag_time = None
+ self._tag_timezone = None
+ else:
+ self._tagger = value[0:sep+1]
+ (timetext, timezonetext) = value[sep+2:].rsplit(" ", 1)
+ try:
+ self._tag_time = int(timetext)
+ except ValueError: #Not a unix timestamp
+ self._tag_time = time.strptime(timetext)
+ self._tag_timezone = parse_timezone(timezonetext)
else:
raise AssertionError("Unknown field %s" % field)
self._message = f.read()
self._needs_parsing = False
- def get_object(self):
+ def _get_object(self):
"""Returns the object pointed by this tag, represented as a tuple(type, sha)"""
self._ensure_parsed()
return (self._object_type, self._object_sha)
- def set_object(self, value):
+ def _set_object(self, value):
self._ensure_parsed()
(self._object_type, self._object_sha) = value
self._needs_serialization = True
- object = property(get_object, set_object)
+ object = property(_get_object, _set_object)
name = serializable_property("name", "The name of this tag")
- tagger = serializable_property("tagger",
+ tagger = serializable_property("tagger",
"Returns the name of the person who created this tag")
- tag_time = serializable_property("tag_time",
+ tag_time = serializable_property("tag_time",
"The creation timestamp of the tag. As the number of seconds since the epoch")
- tag_timezone = serializable_property("tag_timezone",
+ tag_timezone = serializable_property("tag_timezone",
"The timezone that tag_time is in.")
message = serializable_property("message", "The message attached to this tag")
def parse_tree(text):
+ """Parse a tree text.
+
+ :param text: Serialized text to parse
+ :return: Dictionary with names as keys, (mode, sha) tuples as values
+ """
ret = {}
count = 0
- while count < len(text):
- mode = 0
- chr = text[count]
- while chr != ' ':
- assert chr >= '0' and chr <= '7', "%s is not a valid mode char" % chr
- mode = (mode << 3) + (ord(chr) - ord('0'))
- count += 1
- chr = text[count]
- count += 1
- chr = text[count]
- name = ''
- while chr != '\0':
- name += chr
- count += 1
- chr = text[count]
- count += 1
- chr = text[count]
- sha = text[count:count+20]
- hexsha = sha_to_hex(sha)
- ret[name] = (mode, hexsha)
- count = count + 20
+ l = len(text)
+ while count < l:
+ mode_end = text.index(' ', count)
+ mode = int(text[count:mode_end], 8)
+ name_end = text.index('\0', mode_end)
+ name = text[mode_end+1:name_end]
+ count = name_end+21
+ sha = text[name_end+1:count]
+ ret[name] = (mode, sha_to_hex(sha))
return ret
+def serialize_tree(items):
+ """Serialize the items in a tree to a text.
+
+ :param items: Sorted iterable over (name, mode, sha) tuples
+ :return: Serialized tree text
+ """
+ f = StringIO()
+ for name, mode, hexsha in items:
+ f.write("%04o %s\0%s" % (mode, name, hex_to_sha(hexsha)))
+ return f.getvalue()
+
+
+def sorted_tree_items(entries):
+ """Iterate over a tree entries dictionary in the order in which
+ the items would be serialized.
+
+ :param entries: Dictionary mapping names to (mode, sha) tuples
+ :return: Iterator over (name, mode, sha)
+ """
+ def cmp_entry((name1, value1), (name2, value2)):
+ if stat.S_ISDIR(value1[0]):
+ name1 += "/"
+ if stat.S_ISDIR(value2[0]):
+ name2 += "/"
+ return cmp(name1, name2)
+ for name, entry in sorted(entries.iteritems(), cmp=cmp_entry):
+ yield name, entry[0], entry[1]
+
+
class Tree(ShaFile):
"""A Git tree object"""
def entries(self):
"""Return a list of tuples describing the tree entries"""
self._ensure_parsed()
- # The order of this is different from iteritems() for historical reasons
- return [(mode, name, hexsha) for (name, mode, hexsha) in self.iteritems()]
+ # The order of this is different from iteritems() for historical
+ # reasons
+ return [
+ (mode, name, hexsha) for (name, mode, hexsha) in self.iteritems()]
def iteritems(self):
- def cmp_entry((name1, value1), (name2, value2)):
- if stat.S_ISDIR(value1[0]):
- name1 += "/"
- if stat.S_ISDIR(value2[0]):
- name2 += "/"
- return cmp(name1, name2)
+ """Iterate over all entries in the order in which they would be
+ serialized.
+
+ :return: Iterator over (name, mode, sha) tuples
+ """
self._ensure_parsed()
- for name, entry in sorted(self._entries.iteritems(), cmp=cmp_entry):
- yield name, entry[0], entry[1]
+ return sorted_tree_items(self._entries)
def _parse_text(self):
"""Grab the entries in the tree"""
self._needs_parsing = False
def serialize(self):
- f = StringIO()
- for name, mode, hexsha in self.iteritems():
- f.write("%04o %s\0%s" % (mode, name, hex_to_sha(hexsha)))
- self._text = f.getvalue()
+ self._text = serialize_tree(self.iteritems())
self._needs_serialization = False
def as_pretty_string(self):
self._encoding = None
self._needs_parsing = False
self._needs_serialization = True
+ self._extra = {}
@classmethod
def from_file(cls, filename):
def _parse_text(self):
self._parents = []
+ self._extra = []
self._author = None
f = StringIO(self._text)
for l in f:
elif field == ENCODING_ID:
self._encoding = value
else:
- raise AssertionError("Unknown field %s" % field)
+ self._extra.append((field, value))
self._message = f.read()
self._needs_parsing = False
f.write("%s %s %s %s\n" % (COMMITTER_ID, self._committer, str(self._commit_time), format_timezone(self._commit_timezone)))
if self.encoding:
f.write("%s %s\n" % (ENCODING_ID, self.encoding))
+ for k, v in self.extra:
+ if "\n" in k or "\n" in v:
+ raise AssertionError("newline in extra data: %r -> %r" % (k, v))
+ f.write("%s %s\n" % (k, v))
f.write("\n") # There must be a new line after the headers
f.write(self._message)
self._text = f.getvalue()
tree = serializable_property("tree", "Tree that is the state of this commit")
- def get_parents(self):
+ def _get_parents(self):
"""Return a list of parents of this commit."""
self._ensure_parsed()
return self._parents
- def set_parents(self, value):
- """Return a list of parents of this commit."""
+ def _set_parents(self, value):
+ """Set a list of parents of this commit."""
self._ensure_parsed()
self._needs_serialization = True
self._parents = value
- parents = property(get_parents, set_parents)
+ parents = property(_get_parents, _set_parents)
- author = serializable_property("author",
+ def _get_extra(self):
+ """Return extra settings of this commit."""
+ self._ensure_parsed()
+ return self._extra
+
+ extra = property(_get_extra)
+
+ author = serializable_property("author",
"The name of the author of the commit")
- committer = serializable_property("committer",
+ committer = serializable_property("committer",
"The name of the committer of the commit")
message = serializable_property("message",
commit_timezone = serializable_property("commit_timezone",
"The zone the commit time is in")
- author_time = serializable_property("author_time",
+ author_time = serializable_property("author_time",
"The timestamp the commit was written. as the number of seconds since the epoch.")
- author_timezone = serializable_property("author_timezone",
+ author_timezone = serializable_property("author_timezone",
"Returns the zone the author time is in.")
encoding = serializable_property("encoding",
try:
# Try to import C versions
- from dulwich._objects import parse_tree
+ from dulwich._objects import parse_tree, sorted_tree_items
except ImportError:
pass
-