Mercurial and Git clients can push and pull from this alias URL to interact with this repository. You can change to which repository an alias points by going to the Aliases link on the project page.
# repo.py -- For dealing with git repositories.# 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) 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."""Repository access.This module contains the base class for git repositories(BaseRepo) and an implementation which uses a repository onlocal disk (Repo)."""fromcStringIOimportStringIOimporterrnoimportosfromdulwich.errorsimport(NoIndexPresent,NotBlobError,NotCommitError,NotGitRepository,NotTreeError,NotTagError,PackedRefsException,CommitError,RefFormatError,)fromdulwich.fileimport(ensure_dir_exists,GitFile,)fromdulwich.object_storeimport(DiskObjectStore,MemoryObjectStore,)fromdulwich.objectsimport(Blob,Commit,ShaFile,Tag,Tree,hex_to_sha,)importwarningsOBJECTDIR='objects'SYMREF='ref: 'REFSDIR='refs'REFSDIR_TAGS='tags'REFSDIR_HEADS='heads'INDEX_FILENAME="index"BASE_DIRECTORIES=[["branches"],[REFSDIR],[REFSDIR,REFSDIR_TAGS],[REFSDIR,REFSDIR_HEADS],["hooks"],["info"]]defread_info_refs(f):ret={}forlinf.readlines():(sha,name)=l.rstrip("\r\n").split("\t",1)ret[name]=shareturnretdefcheck_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'/.'inrefnameorrefname.startswith('.'):returnFalseif'/'notinrefname:returnFalseif'..'inrefname:returnFalseforcinrefname:iford(c)<040orcin'\177 ~^:?*[':returnFalseifrefname[-1]in'/.':returnFalseifrefname.endswith('.lock'):returnFalseif'@{'inrefname:returnFalseif'\\'inrefname:returnFalsereturnTrueclassRefsContainer(object):"""A container for refs."""defset_ref(self,name,other):warnings.warn("RefsContainer.set_ref() is deprecated.""Use set_symblic_ref instead.",category=DeprecationWarning,stacklevel=2)returnself.set_symbolic_ref(name,other)defset_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 """raiseNotImplementedError(self.set_symbolic_ref)defget_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. """raiseNotImplementedError(self.get_packed_refs)defget_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. """returnNonedefimport_refs(self,base,other):forname,valueinother.iteritems():self["%s/%s"%(base,name)]=valuedefallkeys(self):"""All refs present in this container."""raiseNotImplementedError(self.allkeys)defkeys(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. """ifbaseisnotNone:returnself.subkeys(base)else:returnself.allkeys()defsubkeys(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)+1forrefnameinself.allkeys():ifrefname.startswith(base):keys.add(refname[base_len:])returnkeysdefas_dict(self,base=None):"""Return the contents of this container as a dictionary. """ret={}keys=self.keys(base)ifbaseisNone:base=""forkeyinkeys:try:ret[key]=self[("%s/%s"%(base,key)).strip("/")]exceptKeyError:continue# Unable to resolvereturnretdef_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. """ifnamein('HEAD','refs/stash'):returnifnotname.startswith('refs/')ornotcheck_ref_format(name[5:]):raiseRefFormatError(name)defread_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)ifnotcontents:contents=self.get_packed_refs().get(refname,None)returncontentsdefread_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. """raiseNotImplementedError(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+namedepth=0whilecontents.startswith(SYMREF):refname=contents[len(SYMREF):]contents=self.read_ref(refname)ifnotcontents:breakdepth+=1ifdepth>5:raiseKeyError(name)returnrefname,contentsdef__contains__(self,refname):ifself.read_ref(refname):returnTruereturnFalsedef__getitem__(self,name):"""Get the SHA1 for a reference name. This method follows all symbolic references. """_,sha=self._follow(name)ifshaisNone:raiseKeyError(name)returnshadefset_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. """raiseNotImplementedError(self.set_if_equals)defadd_if_new(self,name,ref):"""Add a new reference only if it does not already exist."""raiseNotImplementedError(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)defremove_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. """raiseNotImplementedError(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)classDictRefsContainer(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=refsself._peeled={}defallkeys(self):returnself._refs.keys()defread_loose_ref(self,name):returnself._refs.get(name,None)defget_packed_refs(self):return{}defset_symbolic_ref(self,name,other):self._refs[name]=SYMREF+otherdefset_if_equals(self,name,old_ref,new_ref):ifold_refisnotNoneandself._refs.get(name,None)!=old_ref:returnFalserealname,_=self._follow(name)self._check_refname(realname)self._refs[realname]=new_refreturnTruedefadd_if_new(self,name,ref):ifnameinself._refs:returnFalseself._refs[name]=refreturnTruedefremove_if_equals(self,name,old_ref):ifold_refisnotNoneandself._refs.get(name,None)!=old_ref:returnFalsedelself._refs[name]returnTruedefget_peeled(self,name):returnself._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)classInfoRefsContainer(RefsContainer):"""Refs container that reads refs from a info/refs file."""def__init__(self,f):self._refs={}self._peeled={}forlinf.readlines():sha,name=l.rstrip("\n").split("\t")ifname.endswith("^{}"):name=name[:-3]ifnotcheck_ref_format(name):raiseValueError("invalid ref name '%s'"%name)self._peeled[name]=shaelse:ifnotcheck_ref_format(name):raiseValueError("invalid ref name '%s'"%name)self._refs[name]=shadefallkeys(self):returnself._refs.keys()defread_loose_ref(self,name):returnself._refs.get(name,None)defget_packed_refs(self):return{}defget_peeled(self,name):try:returnself._peeled[name]exceptKeyError:returnself._refs[name]classDiskRefsContainer(RefsContainer):"""Refs container that reads refs from disk."""def__init__(self,path):self.path=pathself._packed_refs=Noneself._peeled_refs=Nonedef__repr__(self):return"%s(%r)"%(self.__class__.__name__,self.path)defsubkeys(self,base):keys=set()path=self.refpath(base)forroot,dirs,filesinos.walk(path):dir=root[len(path):].strip(os.path.sep).replace(os.path.sep,"/")forfilenameinfiles:refname=("%s/%s"%(dir,filename)).strip("/")# check_ref_format requires at least one /, so we prepend the# base before calling it.ifcheck_ref_format("%s/%s"%(base,refname)):keys.add(refname)forkeyinself.get_packed_refs():ifkey.startswith(base):keys.add(key[len(base):].strip("/"))returnkeysdefallkeys(self):keys=set()ifos.path.exists(self.refpath("HEAD")):keys.add("HEAD")path=self.refpath("")forroot,dirs,filesinos.walk(self.refpath("refs")):dir=root[len(path):].strip(os.path.sep).replace(os.path.sep,"/")forfilenameinfiles:refname=("%s/%s"%(dir,filename)).strip("/")ifcheck_ref_format(refname):keys.add(refname)keys.update(self.get_packed_refs())returnkeysdefrefpath(self,name):"""Return the disk path of a ref. """ifos.path.sep!="/":name=name.replace("/",os.path.sep)returnos.path.join(self.path,name)defget_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 repackingifself._packed_refsisNone:# 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')exceptIOError,e:ife.errno==errno.ENOENT:return{}raisetry:first_line=iter(f).next().rstrip()if(first_line.startswith("# pack-refs")and" peeled"infirst_line):forsha,name,peeledinread_packed_refs_with_peeled(f):self._packed_refs[name]=shaifpeeled:self._peeled_refs[name]=peeledelse:f.seek(0)forsha,nameinread_packed_refs(f):self._packed_refs[name]=shafinally:f.close()returnself._packed_refsdefget_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()ifself._peeled_refsisNoneornamenotinself._packed_refs:# No cache: no peeled refs were read, or this ref is loosereturnNoneifnameinself._peeled_refs:returnself._peeled_refs[name]else:# Known not peelablereturnself[name]defread_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:f=GitFile(filename,'rb')try:header=f.read(len(SYMREF))ifheader==SYMREF:# Read only the first linereturnheader+iter(f).next().rstrip("\r\n")else:# Read only the first 40 bytesreturnheader+f.read(40-len(SYMREF))finally:f.close()exceptIOError,e:ife.errno==errno.ENOENT:returnNoneraisedef_remove_packed_ref(self,name):ifself._packed_refsisNone:returnfilename=os.path.join(self.path,'packed-refs')# reread cached refs from disk, while holding the lockf=GitFile(filename,'wb')try:self._packed_refs=Noneself.get_packed_refs()ifnamenotinself._packed_refs:returndelself._packed_refs[name]ifnameinself._peeled_refs:delself._peeled_refs[name]write_packed_refs(f,self._packed_refs,self._peeled_refs)f.close()finally:f.abort()defset_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+'\n')except(IOError,OSError):f.abort()raisefinally:f.close()defset_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)exceptKeyError:realname=namefilename=self.refpath(realname)ensure_dir_exists(os.path.dirname(filename))f=GitFile(filename,'wb')try:ifold_refisnotNone:try:# read again while holding the lockorig_ref=self.read_loose_ref(realname)iforig_refisNone:orig_ref=self.get_packed_refs().get(realname,None)iforig_ref!=old_ref:f.abort()returnFalseexcept(OSError,IOError):f.abort()raisetry:f.write(new_ref+"\n")except(OSError,IOError):f.abort()raisefinally:f.close()returnTruedefadd_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)ifcontentsisnotNone:returnFalseexceptKeyError:realname=nameself._check_refname(realname)filename=self.refpath(realname)ensure_dir_exists(os.path.dirname(filename))f=GitFile(filename,'wb')try:ifos.path.exists(filename)ornameinself.get_packed_refs():f.abort()returnFalsetry:f.write(ref+"\n")except(OSError,IOError):f.abort()raisefinally:f.close()returnTruedefremove_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:ifold_refisnotNone:orig_ref=self.read_loose_ref(name)iforig_refisNone:orig_ref=self.get_packed_refs().get(name,None)iforig_ref!=old_ref:returnFalse# may only be packedtry:os.remove(filename)exceptOSError,e:ife.errno!=errno.ENOENT:raiseself._remove_packed_ref(name)finally:# never write, we just wanted the lockf.abort()returnTruedef_split_ref_line(line):"""Split a single ref line into a tuple of SHA1 and name."""fields=line.rstrip("\n").split(" ")iflen(fields)!=2:raisePackedRefsException("invalid ref line '%s'"%line)sha,name=fieldstry:hex_to_sha(sha)except(AssertionError,TypeError),e:raisePackedRefsException(e)ifnotcheck_ref_format(name):raisePackedRefsException("invalid ref name '%s'"%name)return(sha,name)defread_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. """forlinf:ifl[0]=="#":# Commentcontinueifl[0]=="^":raisePackedRefsException("found peeled ref in packed-refs without peeled")yield_split_ref_line(l)defread_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=Noneforlinf:ifl[0]=="#":continuel=l.rstrip("\r\n")ifl[0]=="^":ifnotlast:raisePackedRefsException("unexpected peeled ref line")try:hex_to_sha(l[1:])except(AssertionError,TypeError),e:raisePackedRefsException(e)sha,name=_split_ref_line(last)last=Noneyield(sha,name,l[1:])else:iflast:sha,name=_split_ref_line(last)yield(sha,name,None)last=liflast:sha,name=_split_ref_line(last)yield(sha,name,None)defwrite_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 """ifpeeled_refsisNone:peeled_refs={}else:f.write('# pack-refs with: peeled\n')forrefnameinsorted(packed_refs.iterkeys()):f.write('%s%s\n'%(packed_refs[refname],refname))ifrefnameinpeeled_refs:f.write('^%s\n'%peeled_refs[refname])classBaseRepo(object):"""Base class for a git repository. :ivar object_store: Dictionary-like object for accessing the objects :ivar refs: Dictionary-like object with the refs in this repository """def__init__(self,object_store,refs):"""Open a repository. This shouldn't be called directly, but rather through one of the base classes, such as MemoryRepo or Repo. :param object_store: Object store to use :param refs: Refs container to use """self.object_store=object_storeself.refs=refsdef_init_files(self,bare):"""Initialize a default set of named files."""fromdulwich.configimportConfigFileself._put_named_file('description',"Unnamed repository")f=StringIO()cf=ConfigFile()cf.set("core","repositoryformatversion","0")cf.set("core","filemode","true")cf.set("core","bare",str(bare).lower())cf.set("core","logallrefupdates","true")cf.write_to_file(f)self._put_named_file('config',f.getvalue())self._put_named_file(os.path.join('info','exclude'),'')defget_named_file(self,path):"""Get a file from the control dir with a specific name. Although the filename should be interpreted as a filename relative to the control dir in a disk-based Repo, the object returned need not be pointing to a file in that location. :param path: The path to the file, relative to the control dir. :return: An open file object, or None if the file does not exist. """raiseNotImplementedError(self.get_named_file)def_put_named_file(self,path,contents):"""Write a file to the control dir with the given name and contents. :param path: The path to the file, relative to the control dir. :param contents: A string to write to the file. """raiseNotImplementedError(self._put_named_file)defopen_index(self):"""Open the index for this repository. :raise NoIndexPresent: If no index is present :return: The matching `Index` """raiseNotImplementedError(self.open_index)deffetch(self,target,determine_wants=None,progress=None):"""Fetch objects into another repository. :param target: The target repository :param determine_wants: Optional function to determine what refs to fetch. :param progress: Optional progress function """ifdetermine_wantsisNone:determine_wants=lambdaheads:heads.values()target.object_store.add_objects(self.fetch_objects(determine_wants,target.get_graph_walker(),progress))returnself.get_refs()deffetch_objects(self,determine_wants,graph_walker,progress,get_tagged=None):"""Fetch the missing objects required for a set of revisions. :param determine_wants: Function that takes a dictionary with heads and returns the list of heads to fetch. :param graph_walker: Object that can iterate over the list of revisions to fetch and has an "ack" method that will be called to acknowledge that a revision is present. :param progress: Simple progress function that will be called with updated progress strings. :param get_tagged: Function that returns a dict of pointed-to sha -> tag sha for including tags. :return: iterator over objects, with __len__ implemented """wants=determine_wants(self.get_refs())ifwantsisNone:# TODO(dborowitz): find a way to short-circuit that doesn't change# this interface.returnNonehaves=self.object_store.find_common_revisions(graph_walker)returnself.object_store.iter_shas(self.object_store.find_missing_objects(haves,wants,progress,get_tagged))defget_graph_walker(self,heads=None):"""Retrieve a graph walker. A graph walker is used by a remote repository (or proxy) to find out which objects are present in this repository. :param heads: Repository heads to use (optional) :return: A graph walker object """ifheadsisNone:heads=self.refs.as_dict('refs/heads').values()returnself.object_store.get_graph_walker(heads)defref(self,name):"""Return the SHA1 a ref is pointing to. :param name: Name of the ref to look up :raise KeyError: when the ref (or the one it points to) does not exist :return: SHA1 it is pointing at """returnself.refs[name]defget_refs(self):"""Get dictionary with all refs. :return: A ``dict`` mapping ref names to SHA1s """returnself.refs.as_dict()defhead(self):"""Return the SHA1 pointed at by HEAD."""returnself.refs['HEAD']def_get_object(self,sha,cls):assertlen(sha)in(20,40)ret=self.get_object(sha)ifnotisinstance(ret,cls):ifclsisCommit:raiseNotCommitError(ret)elifclsisBlob:raiseNotBlobError(ret)elifclsisTree:raiseNotTreeError(ret)elifclsisTag:raiseNotTagError(ret)else:raiseException("Type invalid: %r != %r"%(ret.type_name,cls.type_name))returnretdefget_object(self,sha):"""Retrieve the object with the specified SHA. :param sha: SHA to retrieve :return: A ShaFile object :raise KeyError: when the object can not be found """returnself.object_store[sha]defget_parents(self,sha):"""Retrieve the parents of a specific commit. :param sha: SHA of the commit for which to retrieve the parents :return: List of parents """returnself.commit(sha).parentsdefget_config(self):"""Retrieve the config object. :return: `ConfigFile` object for the ``.git/config`` file. """fromdulwich.configimportConfigFilepath=os.path.join(self._controldir,'config')try:returnConfigFile.from_path(path)except(IOError,OSError),e:ife.errno!=errno.ENOENT:raiseret=ConfigFile()ret.path=pathreturnretdefget_config_stack(self):"""Return a config stack for this repository. This stack accesses the configuration for both this repository itself (.git/config) and the global configuration, which usually lives in ~/.gitconfig. :return: `Config` instance for this repository """fromdulwich.configimportStackedConfigbackends=[self.get_config()]+StackedConfig.default_backends()returnStackedConfig(backends,writable=backends[0])defcommit(self,sha):"""Retrieve the commit with a particular SHA. :param sha: SHA of the commit to retrieve :raise NotCommitError: If the SHA provided doesn't point at a Commit :raise KeyError: If the SHA provided didn't exist :return: A `Commit` object """warnings.warn("Repo.commit(sha) is deprecated. Use Repo[sha] instead.",category=DeprecationWarning,stacklevel=2)returnself._get_object(sha,Commit)deftree(self,sha):"""Retrieve the tree with a particular SHA. :param sha: SHA of the tree to retrieve :raise NotTreeError: If the SHA provided doesn't point at a Tree :raise KeyError: If the SHA provided didn't exist :return: A `Tree` object """warnings.warn("Repo.tree(sha) is deprecated. Use Repo[sha] instead.",category=DeprecationWarning,stacklevel=2)returnself._get_object(sha,Tree)deftag(self,sha):"""Retrieve the tag with a particular SHA. :param sha: SHA of the tag to retrieve :raise NotTagError: If the SHA provided doesn't point at a Tag :raise KeyError: If the SHA provided didn't exist :return: A `Tag` object """warnings.warn("Repo.tag(sha) is deprecated. Use Repo[sha] instead.",category=DeprecationWarning,stacklevel=2)returnself._get_object(sha,Tag)defget_blob(self,sha):"""Retrieve the blob with a particular SHA. :param sha: SHA of the blob to retrieve :raise NotBlobError: If the SHA provided doesn't point at a Blob :raise KeyError: If the SHA provided didn't exist :return: A `Blob` object """warnings.warn("Repo.get_blob(sha) is deprecated. Use Repo[sha] ""instead.",category=DeprecationWarning,stacklevel=2)returnself._get_object(sha,Blob)defget_peeled(self,ref):"""Get the peeled value of a ref. :param ref: The refname to peel. :return: The fully-peeled SHA1 of a tag object, after peeling all intermediate tags; if the original ref does not point to a tag, this will equal the original SHA1. """cached=self.refs.get_peeled(ref)ifcachedisnotNone:returncachedreturnself.object_store.peel_sha(self.refs[ref]).iddefget_walker(self,include=None,*args,**kwargs):"""Obtain a walker for this repository. :param include: Iterable of SHAs of commits to include along with their ancestors. Defaults to [HEAD] :param exclude: Iterable of SHAs of commits to exclude along with their ancestors, overriding includes. :param order: ORDER_* constant specifying the order of results. Anything other than ORDER_DATE may result in O(n) memory usage. :param reverse: If True, reverse the order of output, requiring O(n) memory. :param max_entries: The maximum number of entries to yield, or None for no limit. :param paths: Iterable of file or subtree paths to show entries for. :param rename_detector: diff.RenameDetector object for detecting renames. :param follow: If True, follow path across renames/copies. Forces a default rename_detector. :param since: Timestamp to list commits after. :param until: Timestamp to list commits before. :param queue_cls: A class to use for a queue of commits, supporting the iterator protocol. The constructor takes a single argument, the Walker. :return: A `Walker` object """fromdulwich.walkimportWalkerifincludeisNone:include=[self.head()]returnWalker(self.object_store,include,*args,**kwargs)defrevision_history(self,head):"""Returns a list of the commits reachable from head. :param head: The SHA of the head to list revision history for. :return: A list of commit objects reachable from head, starting with head itself, in descending commit time order. :raise MissingCommitError: if any missing commits are referenced, including if the head parameter isn't the SHA of a commit. """warnings.warn("Repo.revision_history() is deprecated.""Use dulwich.walker.Walker(repo) instead.",category=DeprecationWarning,stacklevel=2)return[e.commitforeinself.get_walker(include=[head])]def__getitem__(self,name):"""Retrieve a Git object by SHA1 or ref. :param name: A Git object SHA1 or a ref name :return: A `ShaFile` object, such as a Commit or Blob :raise KeyError: when the specified ref or object does not exist """iflen(name)in(20,40):try:returnself.object_store[name]exceptKeyError:passtry:returnself.object_store[self.refs[name]]exceptRefFormatError:raiseKeyError(name)def__contains__(self,name):"""Check if a specific Git object or ref is present. :param name: Git object SHA1 or ref name """iflen(name)in(20,40):returnnameinself.object_storeornameinself.refselse:returnnameinself.refsdef__setitem__(self,name,value):"""Set a ref. :param name: ref name :param value: Ref value - either a ShaFile object, or a hex sha """ifname.startswith("refs/")orname=="HEAD":ifisinstance(value,ShaFile):self.refs[name]=value.idelifisinstance(value,str):self.refs[name]=valueelse:raiseTypeError(value)else:raiseValueError(name)def__delitem__(self,name):"""Remove a ref. :param name: Name of the ref to remove """ifname.startswith("refs/")orname=="HEAD":delself.refs[name]else:raiseValueError(name)def_get_user_identity(self):"""Determine the identity to use for new commits. """config=self.get_config_stack()return"%s <%s>"%(config.get(("user",),"name"),config.get(("user",),"email"))defdo_commit(self,message=None,committer=None,author=None,commit_timestamp=None,commit_timezone=None,author_timestamp=None,author_timezone=None,tree=None,encoding=None,ref='HEAD',merge_heads=None):"""Create a new commit. :param message: Commit message :param committer: Committer fullname :param author: Author fullname (defaults to committer) :param commit_timestamp: Commit timestamp (defaults to now) :param commit_timezone: Commit timestamp timezone (defaults to GMT) :param author_timestamp: Author timestamp (defaults to commit timestamp) :param author_timezone: Author timestamp timezone (defaults to commit timestamp timezone) :param tree: SHA1 of the tree root to use (if not specified the current index will be committed). :param encoding: Encoding :param ref: Optional ref to commit to (defaults to current branch) :param merge_heads: Merge heads (defaults to .git/MERGE_HEADS) :return: New commit SHA1 """importtimec=Commit()iftreeisNone:index=self.open_index()c.tree=index.commit(self.object_store)else:iflen(tree)!=40:raiseValueError("tree must be a 40-byte hex sha string")c.tree=treeifmerge_headsisNone:# FIXME: Read merge heads from .git/MERGE_HEADSmerge_heads=[]ifcommitterisNone:committer=self._get_user_identity()c.committer=committerifcommit_timestampisNone:commit_timestamp=time.time()c.commit_time=int(commit_timestamp)ifcommit_timezoneisNone:# FIXME: Use current user timezone rather than UTCcommit_timezone=0c.commit_timezone=commit_timezoneifauthorisNone:author=committerc.author=authorifauthor_timestampisNone:author_timestamp=commit_timestampc.author_time=int(author_timestamp)ifauthor_timezoneisNone:author_timezone=commit_timezonec.author_timezone=author_timezoneifencodingisnotNone:c.encoding=encodingifmessageisNone:# FIXME: Try to read commit message from .git/MERGE_MSGraiseValueError("No commit message specified")c.message=messagetry:old_head=self.refs[ref]c.parents=[old_head]+merge_headsself.object_store.add_object(c)ok=self.refs.set_if_equals(ref,old_head,c.id)exceptKeyError:c.parents=merge_headsself.object_store.add_object(c)ok=self.refs.add_if_new(ref,c.id)ifnotok:# Fail if the atomic compare-and-swap failed, leaving the commit and# all its objects as garbage.raiseCommitError("%s changed during commit"%(ref,))returnc.idclassRepo(BaseRepo):"""A git repository backed by local disk. To open an existing repository, call the contructor with the path of the repository. To create a new repository, use the Repo.init class method. """def__init__(self,root):ifos.path.isdir(os.path.join(root,".git",OBJECTDIR)):self.bare=Falseself._controldir=os.path.join(root,".git")elif(os.path.isdir(os.path.join(root,OBJECTDIR))andos.path.isdir(os.path.join(root,REFSDIR))):self.bare=Trueself._controldir=rootelif(os.path.isfile(os.path.join(root,".git"))):importrewithopen(os.path.join(root,".git"),'r')asf:_,path=re.match('(gitdir: )(.+$)',f.read()).groups()self.bare=Falseself._controldir=os.path.join(root,path)else:raiseNotGitRepository("No git repository was found at %(path)s"%dict(path=root))self.path=rootobject_store=DiskObjectStore(os.path.join(self.controldir(),OBJECTDIR))refs=DiskRefsContainer(self.controldir())BaseRepo.__init__(self,object_store,refs)defcontroldir(self):"""Return the path of the control directory."""returnself._controldirdef_put_named_file(self,path,contents):"""Write a file to the control dir with the given name and contents. :param path: The path to the file, relative to the control dir. :param contents: A string to write to the file. """path=path.lstrip(os.path.sep)f=GitFile(os.path.join(self.controldir(),path),'wb')try:f.write(contents)finally:f.close()defget_named_file(self,path):"""Get a file from the control dir with a specific name. Although the filename should be interpreted as a filename relative to the control dir in a disk-based Repo, the object returned need not be pointing to a file in that location. :param path: The path to the file, relative to the control dir. :return: An open file object, or None if the file does not exist. """# TODO(dborowitz): sanitize filenames, since this is used directly by# the dumb web serving code.path=path.lstrip(os.path.sep)try:returnopen(os.path.join(self.controldir(),path),'rb')except(IOError,OSError),e:ife.errno==errno.ENOENT:returnNoneraisedefindex_path(self):"""Return path to the index file."""returnos.path.join(self.controldir(),INDEX_FILENAME)defopen_index(self):"""Open the index for this repository. :raise NoIndexPresent: If no index is present :return: The matching `Index` """fromdulwich.indeximportIndexifnotself.has_index():raiseNoIndexPresent()returnIndex(self.index_path())defhas_index(self):"""Check if an index is present."""# Bare repos must never have index files; non-bare repos may have a# missing index file, which is treated as empty.returnnotself.baredefstage(self,paths):"""Stage a set of paths. :param paths: List of paths, relative to the repository path """ifisinstance(paths,basestring):paths=[paths]fromdulwich.indeximportindex_entry_from_statindex=self.open_index()forpathinpaths:full_path=os.path.join(self.path,path)try:st=os.stat(full_path)exceptOSError:# File no longer existstry:delindex[path]exceptKeyError:pass# already removedelse:blob=Blob()f=open(full_path,'rb')try:blob.data=f.read()finally:f.close()self.object_store.add_object(blob)index[path]=index_entry_from_stat(st,blob.id,0)index.write()defclone(self,target_path,mkdir=True,bare=False,origin="origin"):"""Clone this repository. :param target_path: Target path :param mkdir: Create the target directory :param bare: Whether to create a bare repository :param origin: Base name for refs in target repository cloned from this repository :return: Created repository as `Repo` """ifnotbare:target=self.init(target_path,mkdir=mkdir)else:target=self.init_bare(target_path)self.fetch(target)target.refs.import_refs('refs/remotes/'+origin,self.refs.as_dict('refs/heads'))target.refs.import_refs('refs/tags',self.refs.as_dict('refs/tags'))try:target.refs.add_if_new('refs/heads/master',self.refs['refs/heads/master'])exceptKeyError:pass# Update target headhead,head_sha=self.refs._follow('HEAD')ifheadisnotNoneandhead_shaisnotNone: target.refs.set_symbolic_ref('HEAD', head)
target['HEAD'] = head_sha
- if not bare:
- # Checkout HEAD to target dir
- from dulwich.index import build_index_from_tree
- build_index_from_tree(target.path, target.index_path(),
- target.object_store, target['HEAD'].tree)
+ if not bare:
+ # Checkout HEAD to target dir
+from dulwich.index import build_index_from_tree
+ build_index_from_tree(target.path, target.index_path(),
+target.object_store, target['HEAD'].tree)
return target
def__repr__(self):return"<Repo at %r>"%self.path@classmethoddef_init_maybe_bare(cls,path,bare):fordinBASE_DIRECTORIES:os.mkdir(os.path.join(path,*d))DiskObjectStore.init(os.path.join(path,OBJECTDIR))ret=cls(path)ret.refs.set_symbolic_ref("HEAD","refs/heads/master")ret._init_files(bare)returnret@classmethoddefinit(cls,path,mkdir=False):"""Create a new repository. :param path: Path in which to create the repository :param mkdir: Whether to create the directory :return: `Repo` instance """ifmkdir:os.mkdir(path)controldir=os.path.join(path,".git")os.mkdir(controldir)cls._init_maybe_bare(controldir,False)returncls(path)@classmethoddefinit_bare(cls,path):"""Create a new bare repository. ``path`` should already exist and be an emty directory. :param path: Path to create bare repository in :return: a `Repo` instance """returncls._init_maybe_bare(path,True)create=init_bareclassMemoryRepo(BaseRepo):"""Repo that stores refs, objects, and named files in memory. MemoryRepos are always bare: they have no working tree and no index, since those have a stronger dependency on the filesystem. """def__init__(self):BaseRepo.__init__(self,MemoryObjectStore(),DictRefsContainer({}))self._named_files={}self.bare=Truedef_put_named_file(self,path,contents):"""Write a file to the control dir with the given name and contents. :param path: The path to the file, relative to the control dir. :param contents: A string to write to the file. """self._named_files[path]=contentsdefget_named_file(self,path):"""Get a file from the control dir with a specific name. Although the filename should be interpreted as a filename relative to the control dir in a disk-baked Repo, the object returned need not be pointing to a file in that location. :param path: The path to the file, relative to the control dir. :return: An open file object, or None if the file does not exist. """contents=self._named_files.get(path,None)ifcontentsisNone:returnNonereturnStringIO(contents)defopen_index(self):"""Fail to open index for this repo, since it is bare. :raise NoIndexPresent: Raised when no index is present """raiseNoIndexPresent()@classmethoddefinit_bare(cls,objects,refs):"""Create a new bare repository in memory. :param objects: Objects for the new repository, as iterable :param refs: Refs as dictionary, mapping names to object SHA1s """ret=cls()forobjinobjects:ret.object_store.add_object(obj)forrefname,shainrefs.iteritems():ret.refs[refname]=sharet._init_files(bare=True)returnret
Attach a Trello Card
Add a tag
Your session has expired
You are no longer logged in. Please log in and try your request again.
Filter RSS Feed
This RSS feed URL allows you to see the contents of your current filter using any feed reader.
This link includes a special authentication token. If you share the URL with anyone else, they can see this RSS feed's activity. You can disable these tokens when needed.
Your current filter is unsaved; changing it won't affect this RSS feed.