Uh oh!
There was an error while loading.Please reload this page.
- Notifications
You must be signed in to change notification settings - Fork966
Expand file tree
/
Copy pathsymbolic.py
More file actions
678 lines (574 loc) · 26.3 KB
/
symbolic.py
File metadata and controls
678 lines (574 loc) · 26.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
importos
fromgit.compatimportdefenc
fromgit.objectsimportObject,Commit
fromgit.utilimport (
join_path,
join_path_native,
to_native_path_linux,
assure_directory_exists,
hex_to_bin,
LockedFD
)
fromgitdb.excimport (
BadObject,
BadName
)
importos.pathasosp
from .logimportRefLog
__all__= ["SymbolicReference"]
def_git_dir(repo,path):
""" Find the git dir that's appropriate for the path"""
name="%s"% (path,)
ifnamein ['HEAD','ORIG_HEAD','FETCH_HEAD','index','logs']:
returnrepo.git_dir
returnrepo.common_dir
classSymbolicReference(object):
"""Represents a special case of a reference such that this reference is symbolic.
It does not point to a specific commit, but to another Head, which itself
specifies a commit.
A typical example for a symbolic reference is HEAD."""
__slots__= ("repo","path")
_resolve_ref_on_create=False
_points_to_commits_only=True
_common_path_default=""
_remote_common_path_default="refs/remotes"
_id_attribute_="name"
def__init__(self,repo,path):
self.repo=repo
self.path=path
def__str__(self):
returnself.path
def__repr__(self):
return'<git.%s "%s">'% (self.__class__.__name__,self.path)
def__eq__(self,other):
ifhasattr(other,'path'):
returnself.path==other.path
returnFalse
def__ne__(self,other):
returnnot (self==other)
def__hash__(self):
returnhash(self.path)
@property
defname(self):
"""
:return:
In case of symbolic references, the shortest assumable name
is the path itself."""
returnself.path
@property
defabspath(self):
returnjoin_path_native(_git_dir(self.repo,self.path),self.path)
@classmethod
def_get_packed_refs_path(cls,repo):
returnosp.join(repo.common_dir,'packed-refs')
@classmethod
def_iter_packed_refs(cls,repo):
"""Returns an iterator yielding pairs of sha1/path pairs (as bytes) for the corresponding refs.
:note: The packed refs file will be kept open as long as we iterate"""
try:
withopen(cls._get_packed_refs_path(repo),'rt')asfp:
forlineinfp:
line=line.strip()
ifnotline:
continue
ifline.startswith('#'):
# "# pack-refs with: peeled fully-peeled sorted"
# the git source code shows "peeled",
# "fully-peeled" and "sorted" as the keywords
# that can go on this line, as per comments in git file
# refs/packed-backend.c
# I looked at master on 2017-10-11,
# commit 111ef79afe, after tag v2.15.0-rc1
# from repo https://github.com/git/git.git
ifline.startswith('# pack-refs with:')and'peeled'notinline:
raiseTypeError("PackingType of packed-Refs not understood: %r"%line)
# END abort if we do not understand the packing scheme
continue
# END parse comment
# skip dereferenced tag object entries - previous line was actual
# tag reference for it
ifline[0]=='^':
continue
yieldtuple(line.split(' ',1))
# END for each line
except (OSError,IOError):
return
# END no packed-refs file handling
# NOTE: Had try-finally block around here to close the fp,
# but some python version wouldn't allow yields within that.
# I believe files are closing themselves on destruction, so it is
# alright.
@classmethod
defdereference_recursive(cls,repo,ref_path):
"""
:return: hexsha stored in the reference at the given ref_path, recursively dereferencing all
intermediate references as required
:param repo: the repository containing the reference at ref_path"""
whileTrue:
hexsha,ref_path=cls._get_ref_info(repo,ref_path)
ifhexshaisnotNone:
returnhexsha
# END recursive dereferencing
@classmethod
def_get_ref_info_helper(cls,repo,ref_path):
"""Return: (str(sha), str(target_ref_path)) if available, the sha the file at
rela_path points to, or None. target_ref_path is the reference we
point to, or None"""
tokens=None
repodir=_git_dir(repo,ref_path)
try:
withopen(osp.join(repodir,ref_path),'rt',encoding='UTF-8')asfp:
value=fp.read().rstrip()
# Don't only split on spaces, but on whitespace, which allows to parse lines like
# 60b64ef992065e2600bfef6187a97f92398a9144 branch 'master' of git-server:/path/to/repo
tokens=value.split()
assert(len(tokens)!=0)
except (OSError,IOError):
# Probably we are just packed, find our entry in the packed refs file
# NOTE: We are not a symbolic ref if we are in a packed file, as these
# are excluded explicitly
forsha,pathincls._iter_packed_refs(repo):
ifpath!=ref_path:
continue
# sha will be used
tokens=sha,path
break
# END for each packed ref
# END handle packed refs
iftokensisNone:
raiseValueError("Reference at %r does not exist"%ref_path)
# is it a reference ?
iftokens[0]=='ref:':
return (None,tokens[1])
# its a commit
ifrepo.re_hexsha_only.match(tokens[0]):
return (tokens[0],None)
raiseValueError("Failed to parse reference information from %r"%ref_path)
@classmethod
def_get_ref_info(cls,repo,ref_path):
"""Return: (str(sha), str(target_ref_path)) if available, the sha the file at
rela_path points to, or None. target_ref_path is the reference we
point to, or None"""
returncls._get_ref_info_helper(repo,ref_path)
def_get_object(self):
"""
:return:
The object our ref currently refers to. Refs can be cached, they will
always point to the actual object as it gets re-created on each query"""
# have to be dynamic here as we may be a tag which can point to anything
# Our path will be resolved to the hexsha which will be used accordingly
returnObject.new_from_sha(self.repo,hex_to_bin(self.dereference_recursive(self.repo,self.path)))
def_get_commit(self):
"""
:return:
Commit object we point to, works for detached and non-detached
SymbolicReferences. The symbolic reference will be dereferenced recursively."""
obj=self._get_object()
ifobj.type=='tag':
obj=obj.object
# END dereference tag
ifobj.type!=Commit.type:
raiseTypeError("Symbolic Reference pointed to object %r, commit was required"%obj)
# END handle type
returnobj
defset_commit(self,commit,logmsg=None):
"""As set_object, but restricts the type of object to be a Commit
:raise ValueError: If commit is not a Commit object or doesn't point to
a commit
:return: self"""
# check the type - assume the best if it is a base-string
invalid_type=False
ifisinstance(commit,Object):
invalid_type=commit.type!=Commit.type
elifisinstance(commit,SymbolicReference):
invalid_type=commit.object.type!=Commit.type
else:
try:
invalid_type=self.repo.rev_parse(commit).type!=Commit.type
except (BadObject,BadName)ase:
raiseValueError("Invalid object: %s"%commit)frome
# END handle exception
# END verify type
ifinvalid_type:
raiseValueError("Need commit, got %r"%commit)
# END handle raise
# we leave strings to the rev-parse method below
self.set_object(commit,logmsg)
returnself
defset_object(self,object,logmsg=None):# @ReservedAssignment
"""Set the object we point to, possibly dereference our symbolic reference first.
If the reference does not exist, it will be created
:param object: a refspec, a SymbolicReference or an Object instance. SymbolicReferences
will be dereferenced beforehand to obtain the object they point to
:param logmsg: If not None, the message will be used in the reflog entry to be
written. Otherwise the reflog is not altered
:note: plain SymbolicReferences may not actually point to objects by convention
:return: self"""
ifisinstance(object,SymbolicReference):
object=object.object# @ReservedAssignment
# END resolve references
is_detached=True
try:
is_detached=self.is_detached
exceptValueError:
pass
# END handle non-existing ones
ifis_detached:
returnself.set_reference(object,logmsg)
# set the commit on our reference
returnself._get_reference().set_object(object,logmsg)
commit=property(_get_commit,set_commit,doc="Query or set commits directly")
object=property(_get_object,set_object,doc="Return the object our ref currently refers to")
def_get_reference(self):
""":return: Reference Object we point to
:raise TypeError: If this symbolic reference is detached, hence it doesn't point
to a reference, but to a commit"""
sha,target_ref_path=self._get_ref_info(self.repo,self.path)
iftarget_ref_pathisNone:
raiseTypeError("%s is a detached symbolic reference as it points to %r"% (self,sha))
returnself.from_path(self.repo,target_ref_path)
defset_reference(self,ref,logmsg=None):
"""Set ourselves to the given ref. It will stay a symbol if the ref is a Reference.
Otherwise an Object, given as Object instance or refspec, is assumed and if valid,
will be set which effectively detaches the refererence if it was a purely
symbolic one.
:param ref: SymbolicReference instance, Object instance or refspec string
Only if the ref is a SymbolicRef instance, we will point to it. Everything
else is dereferenced to obtain the actual object.
:param logmsg: If set to a string, the message will be used in the reflog.
Otherwise, a reflog entry is not written for the changed reference.
The previous commit of the entry will be the commit we point to now.
See also: log_append()
:return: self
:note: This symbolic reference will not be dereferenced. For that, see
``set_object(...)``"""
write_value=None
obj=None
ifisinstance(ref,SymbolicReference):
write_value="ref: %s"%ref.path
elifisinstance(ref,Object):
obj=ref
write_value=ref.hexsha
elifisinstance(ref,str):
try:
obj=self.repo.rev_parse(ref+"^{}")# optionally deref tags
write_value=obj.hexsha
except (BadObject,BadName)ase:
raiseValueError("Could not extract object from %s"%ref)frome
# END end try string
else:
raiseValueError("Unrecognized Value: %r"%ref)
# END try commit attribute
# typecheck
ifobjisnotNoneandself._points_to_commits_onlyandobj.type!=Commit.type:
raiseTypeError("Require commit, got %r"%obj)
# END verify type
oldbinsha=None
iflogmsgisnotNone:
try:
oldbinsha=self.commit.binsha
exceptValueError:
oldbinsha=Commit.NULL_BIN_SHA
# END handle non-existing
# END retrieve old hexsha
fpath=self.abspath
assure_directory_exists(fpath,is_file=True)
lfd=LockedFD(fpath)
fd=lfd.open(write=True,stream=True)
ok=True
try:
fd.write(write_value.encode('ascii')+b'\n')
lfd.commit()
ok=True
finally:
ifnotok:
lfd.rollback()
# Adjust the reflog
iflogmsgisnotNone:
self.log_append(oldbinsha,logmsg)
returnself
# aliased reference
reference=property(_get_reference,set_reference,doc="Returns the Reference we point to")
ref=reference
defis_valid(self):
"""
:return:
True if the reference is valid, hence it can be read and points to
a valid object or reference."""
try:
self.object
except (OSError,ValueError):
returnFalse
else:
returnTrue
@property
defis_detached(self):
"""
:return:
True if we are a detached reference, hence we point to a specific commit
instead to another reference"""
try:
self.ref
returnFalse
exceptTypeError:
returnTrue
deflog(self):
"""
:return: RefLog for this reference. Its last entry reflects the latest change
applied to this reference
.. note:: As the log is parsed every time, its recommended to cache it for use
instead of calling this method repeatedly. It should be considered read-only."""
returnRefLog.from_file(RefLog.path(self))
deflog_append(self,oldbinsha,message,newbinsha=None):
"""Append a logentry to the logfile of this ref
:param oldbinsha: binary sha this ref used to point to
:param message: A message describing the change
:param newbinsha: The sha the ref points to now. If None, our current commit sha
will be used
:return: added RefLogEntry instance"""
# NOTE: we use the committer of the currently active commit - this should be
# correct to allow overriding the committer on a per-commit level.
# See https://github.com/gitpython-developers/GitPython/pull/146
try:
committer_or_reader=self.commit.committer
exceptValueError:
committer_or_reader=self.repo.config_reader()
# end handle newly cloned repositories
returnRefLog.append_entry(committer_or_reader,RefLog.path(self),oldbinsha,
(newbinshaisNoneandself.commit.binsha)ornewbinsha,
message)
deflog_entry(self,index):
""":return: RefLogEntry at the given index
:param index: python list compatible positive or negative index
.. note:: This method must read part of the reflog during execution, hence
it should be used sparringly, or only if you need just one index.
In that case, it will be faster than the ``log()`` method"""
returnRefLog.entry_at(RefLog.path(self),index)
@classmethod
defto_full_path(cls,path):
"""
:return: string with a full repository-relative path which can be used to initialize
a Reference instance, for instance by using ``Reference.from_path``"""
ifisinstance(path,SymbolicReference):
path=path.path
full_ref_path=path
ifnotcls._common_path_default:
returnfull_ref_path
ifnotpath.startswith(cls._common_path_default+"/"):
full_ref_path='%s/%s'% (cls._common_path_default,path)
returnfull_ref_path
@classmethod
defdelete(cls,repo,path):
"""Delete the reference at the given path
:param repo:
Repository to delete the reference from
:param path:
Short or full path pointing to the reference, i.e. refs/myreference
or just "myreference", hence 'refs/' is implied.
Alternatively the symbolic reference to be deleted"""
full_ref_path=cls.to_full_path(path)
abs_path=osp.join(repo.common_dir,full_ref_path)
ifosp.exists(abs_path):
os.remove(abs_path)
else:
# check packed refs
pack_file_path=cls._get_packed_refs_path(repo)
try:
withopen(pack_file_path,'rb')asreader:
new_lines= []
made_change=False
dropped_last_line=False
forlineinreader:
line=line.decode(defenc)
_,_,line_ref=line.partition(' ')
line_ref=line_ref.strip()
# keep line if it is a comment or if the ref to delete is not
# in the line
# If we deleted the last line and this one is a tag-reference object,
# we drop it as well
if (line.startswith('#')orfull_ref_path!=line_ref)and \
(notdropped_last_lineordropped_last_lineandnotline.startswith('^')):
new_lines.append(line)
dropped_last_line=False
continue
# END skip comments and lines without our path
# drop this line
made_change=True
dropped_last_line=True
# write the new lines
ifmade_change:
# write-binary is required, otherwise windows will
# open the file in text mode and change LF to CRLF !
withopen(pack_file_path,'wb')asfd:
fd.writelines(line.encode(defenc)forlineinnew_lines)
except (OSError,IOError):
pass# it didn't exist at all
# delete the reflog
reflog_path=RefLog.path(cls(repo,full_ref_path))
ifosp.isfile(reflog_path):
os.remove(reflog_path)
# END remove reflog
@classmethod
def_create(cls,repo,path,resolve,reference,force,logmsg=None):
"""internal method used to create a new symbolic reference.
If resolve is False, the reference will be taken as is, creating
a proper symbolic reference. Otherwise it will be resolved to the
corresponding object and a detached symbolic reference will be created
instead"""
git_dir=_git_dir(repo,path)
full_ref_path=cls.to_full_path(path)
abs_ref_path=osp.join(git_dir,full_ref_path)
# figure out target data
target=reference
ifresolve:
target=repo.rev_parse(str(reference))
ifnotforceandosp.isfile(abs_ref_path):
target_data=str(target)
ifisinstance(target,SymbolicReference):
target_data=target.path
ifnotresolve:
target_data="ref: "+target_data
withopen(abs_ref_path,'rb')asfd:
existing_data=fd.read().decode(defenc).strip()
ifexisting_data!=target_data:
raiseOSError("Reference at %r does already exist, pointing to %r, requested was %r"%
(full_ref_path,existing_data,target_data))
# END no force handling
ref=cls(repo,full_ref_path)
ref.set_reference(target,logmsg)
returnref
@classmethod
defcreate(cls,repo,path,reference='HEAD',force=False,logmsg=None):
"""Create a new symbolic reference, hence a reference pointing to another reference.
:param repo:
Repository to create the reference in
:param path:
full path at which the new symbolic reference is supposed to be
created at, i.e. "NEW_HEAD" or "symrefs/my_new_symref"
:param reference:
The reference to which the new symbolic reference should point to.
If it is a commit'ish, the symbolic ref will be detached.
:param force:
if True, force creation even if a symbolic reference with that name already exists.
Raise OSError otherwise
:param logmsg:
If not None, the message to append to the reflog. Otherwise no reflog
entry is written.
:return: Newly created symbolic Reference
:raise OSError:
If a (Symbolic)Reference with the same name but different contents
already exists.
:note: This does not alter the current HEAD, index or Working Tree"""
returncls._create(repo,path,cls._resolve_ref_on_create,reference,force,logmsg)
defrename(self,new_path,force=False):
"""Rename self to a new path
:param new_path:
Either a simple name or a full path, i.e. new_name or features/new_name.
The prefix refs/ is implied for references and will be set as needed.
In case this is a symbolic ref, there is no implied prefix
:param force:
If True, the rename will succeed even if a head with the target name
already exists. It will be overwritten in that case
:return: self
:raise OSError: In case a file at path but a different contents already exists """
new_path=self.to_full_path(new_path)
ifself.path==new_path:
returnself
new_abs_path=osp.join(_git_dir(self.repo,new_path),new_path)
cur_abs_path=osp.join(_git_dir(self.repo,self.path),self.path)
ifosp.isfile(new_abs_path):
ifnotforce:
# if they point to the same file, its not an error
withopen(new_abs_path,'rb')asfd1:
f1=fd1.read().strip()
withopen(cur_abs_path,'rb')asfd2:
f2=fd2.read().strip()
iff1!=f2:
raiseOSError("File at path %r already exists"%new_abs_path)
# else: we could remove ourselves and use the otherone, but
# but clarity we just continue as usual
# END not force handling
os.remove(new_abs_path)
# END handle existing target file
dname=osp.dirname(new_abs_path)
ifnotosp.isdir(dname):
os.makedirs(dname)
# END create directory
os.rename(cur_abs_path,new_abs_path)
self.path=new_path
returnself
@classmethod
def_iter_items(cls,repo,common_path=None):
ifcommon_pathisNone:
common_path=cls._common_path_default
rela_paths=set()
# walk loose refs
# Currently we do not follow links
forroot,dirs,filesinos.walk(join_path_native(repo.common_dir,common_path)):
if'refs'notinroot.split(os.sep):# skip non-refs subfolders
refs_id= [dfordindirsifd=='refs']
ifrefs_id:
dirs[0:]= ['refs']
# END prune non-refs folders
forfinfiles:
iff=='packed-refs':
continue
abs_path=to_native_path_linux(join_path(root,f))
rela_paths.add(abs_path.replace(to_native_path_linux(repo.common_dir)+'/',""))
# END for each file in root directory
# END for each directory to walk
# read packed refs
for_sha,rela_pathincls._iter_packed_refs(repo):
ifrela_path.startswith(common_path):
rela_paths.add(rela_path)
# END relative path matches common path
# END packed refs reading
# return paths in sorted order
forpathinsorted(rela_paths):
try:
yieldcls.from_path(repo,path)
exceptValueError:
continue
# END for each sorted relative refpath
@classmethod
defiter_items(cls,repo,common_path=None):
"""Find all refs in the repository
:param repo: is the Repo
:param common_path:
Optional keyword argument to the path which is to be shared by all
returned Ref objects.
Defaults to class specific portion if None assuring that only
refs suitable for the actual class are returned.
:return:
git.SymbolicReference[], each of them is guaranteed to be a symbolic
ref which is not detached and pointing to a valid ref
List is lexicographically sorted
The returned objects represent actual subclasses, such as Head or TagReference"""
return (rforrincls._iter_items(repo,common_path)ifr.__class__==SymbolicReferenceornotr.is_detached)
@classmethod
deffrom_path(cls,repo,path):
"""
:param path: full .git-directory-relative path name to the Reference to instantiate
:note: use to_full_path() if you only have a partial path of a known Reference Type
:return:
Instance of type Reference, Head, or Tag
depending on the given path"""
ifnotpath:
raiseValueError("Cannot create Reference from %r"%path)
# Names like HEAD are inserted after the refs module is imported - we have an import dependency
# cycle and don't want to import these names in-function
from .importHEAD,Head,RemoteReference,TagReference,Reference
forref_typein (HEAD,Head,RemoteReference,TagReference,Reference,SymbolicReference):
try:
instance=ref_type(repo,path)
ifinstance.__class__==SymbolicReferenceandinstance.is_detached:
raiseValueError("SymbolRef was detached, we drop it")
returninstance
exceptValueError:
pass
# END exception handling
# END for each type to try
raiseValueError("Could not find reference type suitable to handle path %r"%path)
defis_remote(self):
""":return: True if this symbolic reference points to a remote branch"""
returnself.path.startswith(self._remote_common_path_default+"/")