Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/refs/symbolic.py: 59%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# This module is part of GitPython and is released under the
2# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/
4__all__ = ["SymbolicReference"]
6import os
7from pathlib import Path
9from gitdb.exc import BadName, BadObject
11from git.compat import defenc
12from git.objects.base import Object
13from git.objects.commit import Commit
14from git.refs.log import RefLog
15from git.util import (
16 LockedFD,
17 assure_directory_exists,
18 hex_to_bin,
19 join_path,
20 join_path_native,
21 to_native_path_linux,
22)
24# typing ------------------------------------------------------------------
26from typing import (
27 Any,
28 Iterator,
29 List,
30 TYPE_CHECKING,
31 Tuple,
32 Type,
33 TypeVar,
34 Union,
35 cast,
36)
38from git.types import AnyGitObject, PathLike
40if TYPE_CHECKING:
41 from git.config import GitConfigParser
42 from git.objects.commit import Actor
43 from git.refs.log import RefLogEntry
44 from git.refs.reference import Reference
45 from git.repo import Repo
48T_References = TypeVar("T_References", bound="SymbolicReference")
50# ------------------------------------------------------------------------------
53def _git_dir(repo: "Repo", path: Union[PathLike, None]) -> PathLike:
54 """Find the git dir that is appropriate for the path."""
55 name = f"{path}"
56 if name in ["HEAD", "ORIG_HEAD", "FETCH_HEAD", "index", "logs"]:
57 return repo.git_dir
58 return repo.common_dir
61class SymbolicReference:
62 """A reference that can point to another reference or be detached.
64 An attached :class:`~git.refs.head.HEAD` usually points to a
65 :class:`~git.refs.head.Head`, which itself specifies a commit. A detached
66 :class:`~git.refs.head.HEAD` points directly to a commit instead.
68 Use :attr:`commit` to access the commit in either case, and :attr:`reference`
69 to access the target reference when attached.
70 """
72 __slots__ = ("repo", "path")
74 _resolve_ref_on_create = False
75 _points_to_commits_only = True
76 _common_path_default = ""
77 _remote_common_path_default = "refs/remotes"
78 _id_attribute_ = "name"
79 # Match Git's SYMREF_MAXDEPTH, counting the terminal reference as well.
80 _max_symref_depth = 5
82 def __init__(self, repo: "Repo", path: PathLike, check_path: bool = False) -> None:
83 self.repo = repo
84 self.path: PathLike = path
86 def __str__(self) -> str:
87 return os.fspath(self.path)
89 def __repr__(self) -> str:
90 return '<git.%s "%s">' % (self.__class__.__name__, self.path)
92 def __eq__(self, other: object) -> bool:
93 if hasattr(other, "path"):
94 other = cast(SymbolicReference, other)
95 return self.path == other.path
96 return False
98 def __ne__(self, other: object) -> bool:
99 return not (self == other)
101 def __hash__(self) -> int:
102 return hash(self.path)
104 @property
105 def name(self) -> str:
106 """
107 :return:
108 In case of symbolic references, the shortest assumable name is the path
109 itself.
110 """
111 return os.fspath(self.path)
113 @property
114 def abspath(self) -> PathLike:
115 return join_path_native(_git_dir(self.repo, self.path), self.path)
117 @staticmethod
118 def _get_validated_path(base: PathLike, path: PathLike) -> str:
119 path = os.fspath(path)
120 base_path = os.path.realpath(os.fspath(base))
121 abs_path = os.path.realpath(os.path.join(base_path, path))
122 try:
123 common_path = os.path.commonpath([base_path, abs_path])
124 except ValueError as e:
125 raise ValueError("Reference path %r escapes the repository" % path) from e
126 if common_path != base_path:
127 raise ValueError("Reference path %r escapes the repository" % path)
128 return abs_path
130 @classmethod
131 def _get_validated_ref_path(cls, repo: "Repo", path: PathLike) -> str:
132 """Return the absolute filesystem path for a ref after validating it."""
133 cls._check_ref_name_valid(path)
134 ref_path = os.fspath(path)
135 return cls._get_validated_path(_git_dir(repo, ref_path), ref_path)
137 @classmethod
138 def _get_validated_reflog_path(cls, repo: "Repo", path: PathLike) -> str:
139 """Return the absolute filesystem path for a reflog after validating it."""
140 cls._check_ref_name_valid(path)
141 return cls._get_validated_path(os.path.join(repo.git_dir, "logs"), path)
143 @classmethod
144 def _get_packed_refs_path(cls, repo: "Repo") -> str:
145 return os.path.join(repo.common_dir, "packed-refs")
147 @classmethod
148 def _iter_packed_refs(cls, repo: "Repo") -> Iterator[Tuple[str, str]]:
149 """Return an iterator yielding pairs of sha1/path pairs (as strings) for the
150 corresponding refs.
152 :note:
153 The packed refs file will be kept open as long as we iterate.
154 """
155 try:
156 with open(cls._get_packed_refs_path(repo), "rt", encoding="UTF-8") as fp:
157 for line in fp:
158 line = line.strip()
159 if not line:
160 continue
161 if line.startswith("#"):
162 # "# pack-refs with: peeled fully-peeled sorted"
163 # the git source code shows "peeled",
164 # "fully-peeled" and "sorted" as the keywords
165 # that can go on this line, as per comments in git file
166 # refs/packed-backend.c
167 # I looked at master on 2017-10-11,
168 # commit 111ef79afe, after tag v2.15.0-rc1
169 # from repo https://github.com/git/git.git
170 if line.startswith("# pack-refs with:") and "peeled" not in line:
171 raise TypeError("PackingType of packed-Refs not understood: %r" % line)
172 # END abort if we do not understand the packing scheme
173 continue
174 # END parse comment
176 # Skip dereferenced tag object entries - previous line was actual
177 # tag reference for it.
178 if line[0] == "^":
179 continue
181 yield cast(Tuple[str, str], tuple(line.split(" ", 1)))
182 # END for each line
183 except OSError:
184 return None
185 # END no packed-refs file handling
187 @classmethod
188 def dereference_recursive(cls, repo: "Repo", ref_path: Union[PathLike, None]) -> str:
189 """
190 :return:
191 hexsha stored in the reference at the given `ref_path`, recursively
192 dereferencing all intermediate references as required
194 :param repo:
195 The repository containing the reference at `ref_path`.
197 :raise ValueError:
198 If the reference is missing, invalid, or exceeds Git's limit of five
199 references in a symbolic reference chain (including the terminal ref).
200 """
202 for _ in range(cls._max_symref_depth):
203 hexsha, ref_path = cls._get_ref_info(repo, ref_path)
204 if hexsha is not None:
205 return hexsha
206 # END recursive dereferencing
207 raise ValueError("Too many levels of symbolic references at %r" % ref_path)
209 @staticmethod
210 def _check_ref_name_valid(ref_path: PathLike) -> None:
211 """Check a ref name for validity.
213 This is based on the rules described in :manpage:`git-check-ref-format(1)`.
214 """
215 previous: Union[str, None] = None
216 one_before_previous: Union[str, None] = None
217 for c in os.fspath(ref_path):
218 if c in " ~^:?*[\\":
219 raise ValueError(
220 f"Invalid reference '{ref_path}': references cannot contain spaces, tildes (~), carets (^),"
221 f" colons (:), question marks (?), asterisks (*), open brackets ([) or backslashes (\\)"
222 )
223 elif c == ".":
224 if previous is None or previous == "/":
225 raise ValueError(
226 f"Invalid reference '{ref_path}': references cannot start with a period (.) or contain '/.'"
227 )
228 elif previous == ".":
229 raise ValueError(f"Invalid reference '{ref_path}': references cannot contain '..'")
230 elif c == "/":
231 if previous == "/":
232 raise ValueError(f"Invalid reference '{ref_path}': references cannot contain '//'")
233 elif previous is None:
234 raise ValueError(
235 f"Invalid reference '{ref_path}': references cannot start with forward slashes '/'"
236 )
237 elif c == "{" and previous == "@":
238 raise ValueError(f"Invalid reference '{ref_path}': references cannot contain '@{{'")
239 elif ord(c) < 32 or ord(c) == 127:
240 raise ValueError(f"Invalid reference '{ref_path}': references cannot contain ASCII control characters")
242 one_before_previous = previous
243 previous = c
245 if previous == ".":
246 raise ValueError(f"Invalid reference '{ref_path}': references cannot end with a period (.)")
247 elif previous == "/":
248 raise ValueError(f"Invalid reference '{ref_path}': references cannot end with a forward slash (/)")
249 elif previous == "@" and one_before_previous is None:
250 raise ValueError(f"Invalid reference '{ref_path}': references cannot be '@'")
251 elif any(component.endswith(".lock") for component in Path(ref_path).parts):
252 raise ValueError(
253 f"Invalid reference '{ref_path}': references cannot have slash-separated components that end with"
254 " '.lock'"
255 )
257 @classmethod
258 def _get_ref_info_helper(
259 cls, repo: "Repo", ref_path: Union[PathLike, None]
260 ) -> Union[Tuple[str, None], Tuple[None, str]]:
261 """
262 :return:
263 *(str(sha), str(target_ref_path))*, where:
265 * *sha* is of the file at rela_path points to if available, or ``None``.
266 * *target_ref_path* is the reference we point to, or ``None``.
267 """
268 if ref_path is None:
269 raise ValueError("Reference at %r does not exist" % ref_path)
270 ref_file = cls._get_validated_ref_path(repo, ref_path)
272 tokens: Union[None, List[str], Tuple[str, str]] = None
273 try:
274 with open(ref_file, "rt", encoding="UTF-8") as fp:
275 value = fp.read().rstrip()
276 # Don't only split on spaces, but on whitespace, which allows to parse lines like:
277 # 60b64ef992065e2600bfef6187a97f92398a9144 branch 'master' of git-server:/path/to/repo
278 tokens = value.split()
279 assert len(tokens) != 0
280 except OSError:
281 # Probably we are just packed. Find our entry in the packed refs file.
282 # NOTE: We are not a symbolic ref if we are in a packed file, as these
283 # are excluded explicitly.
284 for sha, path in cls._iter_packed_refs(repo):
285 if path != ref_path:
286 continue
287 # sha will be used.
288 tokens = sha, path
289 break
290 # END for each packed ref
291 # END handle packed refs
292 if tokens is None:
293 raise ValueError("Reference at %r does not exist" % ref_path)
295 # Is it a reference?
296 if tokens[0] == "ref:":
297 return (None, tokens[1])
299 # It's a commit.
300 if repo.re_hexsha_only.match(tokens[0]):
301 return (tokens[0], None)
303 raise ValueError("Failed to parse reference information from %r" % ref_path)
305 @classmethod
306 def _get_ref_info(cls, repo: "Repo", ref_path: Union[PathLike, None]) -> Union[Tuple[str, None], Tuple[None, str]]:
307 """
308 :return:
309 *(str(sha), str(target_ref_path))*, where:
311 * *sha* is of the file at rela_path points to if available, or ``None``.
312 * *target_ref_path* is the reference we point to, or ``None``.
313 """
314 return cls._get_ref_info_helper(repo, ref_path)
316 def _get_object(self) -> AnyGitObject:
317 """
318 :return:
319 The object our ref currently refers to. Refs can be cached, they will always
320 point to the actual object as it gets re-created on each query.
321 """
322 # We have to be dynamic here as we may be a tag which can point to anything.
323 # Our path will be resolved to the hexsha which will be used accordingly.
324 return Object.new_from_sha(self.repo, hex_to_bin(self.dereference_recursive(self.repo, self.path)))
326 def _get_commit(self) -> "Commit":
327 """
328 :return:
329 :class:`~git.objects.commit.Commit` object we point to. This works for
330 detached and non-detached :class:`SymbolicReference` instances. The symbolic
331 reference will be dereferenced recursively.
332 """
333 obj = self._get_object()
334 if obj.type == "tag":
335 obj = obj.object
336 # END dereference tag
338 if obj.type != Commit.type:
339 raise TypeError("Symbolic Reference pointed to object %r, commit was required" % obj)
340 # END handle type
341 return obj
343 def set_commit(
344 self,
345 commit: Union[Commit, "SymbolicReference", str],
346 logmsg: Union[str, None] = None,
347 ) -> "SymbolicReference":
348 """Like :meth:`set_object`, but restricts the type of object to be a
349 :class:`~git.objects.commit.Commit`.
351 :raise ValueError:
352 If `commit` is not a :class:`~git.objects.commit.Commit` object, nor does it
353 point to a commit.
355 :return:
356 self
357 """
358 # Check the type - assume the best if it is a base-string.
359 invalid_type = False
360 if isinstance(commit, Object):
361 invalid_type = commit.type != Commit.type
362 elif isinstance(commit, SymbolicReference):
363 invalid_type = commit.object.type != Commit.type
364 else:
365 try:
366 invalid_type = self.repo.rev_parse(commit).type != Commit.type
367 except (BadObject, BadName) as e:
368 raise ValueError("Invalid object: %s" % commit) from e
369 # END handle exception
370 # END verify type
372 if invalid_type:
373 raise ValueError("Need commit, got %r" % commit)
374 # END handle raise
376 # We leave strings to the rev-parse method below.
377 self.set_object(commit, logmsg)
379 return self
381 def set_object(
382 self,
383 object: Union[AnyGitObject, "SymbolicReference", str],
384 logmsg: Union[str, None] = None,
385 ) -> "SymbolicReference":
386 """Set the object we point to, possibly dereference our symbolic reference
387 first. If the reference does not exist, it will be created.
389 :param object:
390 A refspec, a :class:`SymbolicReference` or an
391 :class:`~git.objects.base.Object` instance.
393 * :class:`SymbolicReference` instances will be dereferenced beforehand to
394 obtain the git object they point to.
395 * :class:`~git.objects.base.Object` instances must represent git objects
396 (:class:`~git.types.AnyGitObject`).
398 :param logmsg:
399 If not ``None``, the message will be used in the reflog entry to be written.
400 Otherwise the reflog is not altered.
402 :note:
403 Plain :class:`SymbolicReference` instances may not actually point to objects
404 by convention.
406 :return:
407 self
409 :raise ValueError:
410 If the symbolic reference chain exceeds Git's limit of five references.
411 """
412 # Validate before recursing through the public method, preserving subclass
413 # dispatch and Reference's HEAD reflog updates.
414 ref_path: Union[PathLike, None] = self.path
415 for _ in range(self._max_symref_depth):
416 try:
417 hexsha, ref_path = self._get_ref_info(self.repo, ref_path)
418 except ValueError:
419 break # Allow creating an unborn terminal reference.
420 if hexsha is not None:
421 break
422 else:
423 raise ValueError("Too many levels of symbolic references at %r" % ref_path)
425 if isinstance(object, SymbolicReference):
426 object = object.object # @ReservedAssignment
427 # END resolve references
429 is_detached = True
430 try:
431 is_detached = self.is_detached
432 except ValueError:
433 pass
434 # END handle non-existing ones
436 if is_detached:
437 return self.set_reference(object, logmsg)
439 # set the commit on our reference
440 return self._get_reference().set_object(object, logmsg)
442 @property
443 def commit(self) -> "Commit":
444 """The commit this reference resolves to, whether detached or symbolic.
446 For example, ``repo.head.commit.hexsha`` returns the current commit ID
447 both on a branch and with a detached HEAD. HEAD must resolve to an
448 existing commit; an unborn branch in an empty repository has none.
450 Assigning updates the commit without changing whether this reference
451 is detached.
452 """
453 return self._get_commit()
455 @commit.setter
456 def commit(self, commit: Union[Commit, "SymbolicReference", str]) -> "SymbolicReference":
457 return self.set_commit(commit)
459 @property
460 def object(self) -> AnyGitObject:
461 """Return the object our ref currently refers to"""
462 return self._get_object()
464 @object.setter
465 def object(self, object: Union[AnyGitObject, "SymbolicReference", str]) -> "SymbolicReference":
466 return self.set_object(object)
468 def _get_reference(self) -> "Reference":
469 """
470 :return:
471 :class:`~git.refs.reference.Reference` object we point to
473 :raise TypeError:
474 If this symbolic reference is detached, hence it doesn't point to a
475 reference, but to a commit.
476 """
477 sha, target_ref_path = self._get_ref_info(self.repo, self.path)
478 if target_ref_path is None:
479 raise TypeError(
480 "%s is a detached symbolic reference as it points to %r. "
481 "Use .commit or .object to access the target directly." % (self, sha)
482 )
483 return cast("Reference", self.from_path(self.repo, target_ref_path))
485 def set_reference(
486 self,
487 ref: Union[AnyGitObject, "SymbolicReference", str],
488 logmsg: Union[str, None] = None,
489 ) -> "SymbolicReference":
490 """Set ourselves to the given `ref`.
492 It will stay a symbol if the `ref` is a :class:`~git.refs.reference.Reference`.
494 Otherwise a git object, specified as a :class:`~git.objects.base.Object`
495 instance or refspec, is assumed. If it is valid, this reference will be set to
496 it, which effectively detaches the reference if it was a purely symbolic one.
498 :param ref:
499 A :class:`SymbolicReference` instance, an :class:`~git.objects.base.Object`
500 instance (specifically an :class:`~git.types.AnyGitObject`), or a refspec
501 string. Only if the ref is a :class:`SymbolicReference` instance, we will
502 point to it. Everything else is dereferenced to obtain the actual object.
504 :param logmsg:
505 If set to a string, the message will be used in the reflog.
506 Otherwise, a reflog entry is not written for the changed reference.
507 The previous commit of the entry will be the commit we point to now.
509 See also: :meth:`log_append`
511 :return:
512 self
514 :note:
515 This symbolic reference will not be dereferenced. For that, see
516 :meth:`set_object`.
517 """
518 write_value = None
519 obj = None
520 if isinstance(ref, SymbolicReference):
521 write_value = "ref: %s" % ref.path
522 elif isinstance(ref, Object):
523 obj = ref
524 write_value = ref.hexsha
525 elif isinstance(ref, str):
526 try:
527 obj = self.repo.rev_parse(ref + "^{}") # Optionally dereference tags.
528 write_value = obj.hexsha
529 except (BadObject, BadName) as e:
530 raise ValueError("Could not extract object from %s" % ref) from e
531 # END end try string
532 else:
533 raise ValueError("Unrecognized Value: %r" % ref)
534 # END try commit attribute
536 # typecheck
537 if obj is not None and self._points_to_commits_only and obj.type != Commit.type:
538 raise TypeError("Require commit, got %r" % obj)
539 # END verify type
541 oldbinsha: bytes = b""
542 if logmsg is not None:
543 try:
544 oldbinsha = self.commit.binsha
545 except ValueError:
546 oldbinsha = Commit.NULL_BIN_SHA
547 # END handle non-existing
548 # END retrieve old hexsha
550 fpath = self._get_validated_ref_path(self.repo, self.path)
551 assure_directory_exists(fpath, is_file=True)
553 lfd = LockedFD(fpath)
554 fd = lfd.open(write=True, stream=True)
555 try:
556 fd.write(write_value.encode("utf-8") + b"\n")
557 lfd.commit()
558 except BaseException:
559 lfd.rollback()
560 raise
561 # Adjust the reflog
562 if logmsg is not None:
563 self.log_append(oldbinsha, logmsg)
565 return self
567 # Aliased reference
568 @property
569 def reference(self) -> "Reference":
570 """The reference we point to, available only when not detached.
572 Check :attr:`is_detached` before reading this property if a target
573 reference is required. To access the target commit or object in either
574 state, use :attr:`commit` or :attr:`object` instead.
576 Assigning a reference keeps this reference symbolic. Assigning a git
577 object or revision string detaches it; reading this property then raises.
579 :raise TypeError:
580 If this reference is detached when reading the property.
581 """
582 return self._get_reference()
584 @reference.setter
585 def reference(self, ref: Union[AnyGitObject, "SymbolicReference", str]) -> "SymbolicReference":
586 return self.set_reference(ref)
588 ref = reference
590 def is_valid(self) -> bool:
591 """
592 :return:
593 ``True`` if the reference is valid, hence it can be read and points to a
594 valid object or reference.
595 """
596 try:
597 self.object # noqa: B018
598 except (OSError, ValueError):
599 return False
600 else:
601 return True
603 @property
604 def is_detached(self) -> bool:
605 """
606 :return:
607 ``True`` if we are a detached reference, hence we point to a specific commit
608 instead to another reference.
609 """
610 # Inspect only this ref. Constructing its target would recursively inspect
611 # symbolic references with nonstandard names through from_path().
612 return self._get_ref_info(self.repo, self.path)[1] is None
614 def log(self) -> "RefLog":
615 """
616 :return:
617 :class:`~git.refs.log.RefLog` for this reference.
618 Its last entry reflects the latest change applied to this reference.
620 :note:
621 As the log is parsed every time, its recommended to cache it for use instead
622 of calling this method repeatedly. It should be considered read-only.
623 """
624 return RefLog.from_file(RefLog.path(self))
626 def log_append(
627 self,
628 oldbinsha: bytes,
629 message: Union[str, None],
630 newbinsha: Union[bytes, None] = None,
631 ) -> "RefLogEntry":
632 """Append a logentry to the logfile of this ref.
634 :param oldbinsha:
635 Binary sha this ref used to point to.
637 :param message:
638 A message describing the change.
640 :param newbinsha:
641 The sha the ref points to now. If None, our current commit sha will be used.
643 :return:
644 The added :class:`~git.refs.log.RefLogEntry` instance.
645 """
646 # NOTE: We use the committer of the currently active commit - this should be
647 # correct to allow overriding the committer on a per-commit level.
648 # See https://github.com/gitpython-developers/GitPython/pull/146.
649 try:
650 committer_or_reader: Union["Actor", "GitConfigParser"] = self.commit.committer
651 except ValueError:
652 committer_or_reader = self.repo.config_reader()
653 # END handle newly cloned repositories
654 if newbinsha is None:
655 newbinsha = self.commit.binsha
657 if message is None:
658 message = ""
660 return RefLog.append_entry(committer_or_reader, RefLog.path(self), oldbinsha, newbinsha, message)
662 def log_entry(self, index: int) -> "RefLogEntry":
663 """
664 :return:
665 :class:`~git.refs.log.RefLogEntry` at the given index
667 :param index:
668 Python list compatible positive or negative index.
670 :note:
671 This method must read part of the reflog during execution, hence it should
672 be used sparingly, or only if you need just one index. In that case, it will
673 be faster than the :meth:`log` method.
674 """
675 return RefLog.entry_at(RefLog.path(self), index)
677 @classmethod
678 def to_full_path(cls, path: Union[PathLike, "SymbolicReference"]) -> PathLike:
679 """
680 :return:
681 String with a full repository-relative path which can be used to initialize
682 a :class:`~git.refs.reference.Reference` instance, for instance by using
683 :meth:`Reference.from_path <git.refs.reference.Reference.from_path>`.
684 """
685 if isinstance(path, SymbolicReference):
686 path = path.path
687 full_ref_path = path
688 if not cls._common_path_default:
689 return full_ref_path
690 if not os.fspath(path).startswith(cls._common_path_default + "/"):
691 full_ref_path = "%s/%s" % (cls._common_path_default, path)
692 return full_ref_path
694 @classmethod
695 def delete(cls, repo: "Repo", path: PathLike) -> None:
696 """Delete the reference at the given path.
698 :param repo:
699 Repository to delete the reference from.
701 :param path:
702 Short or full path pointing to the reference, e.g. ``refs/myreference`` or
703 just ``myreference``, hence ``refs/`` is implied.
704 Alternatively the symbolic reference to be deleted.
705 """
706 full_ref_path = cls.to_full_path(path)
707 abs_path = cls._get_validated_ref_path(repo, full_ref_path)
708 if os.path.exists(abs_path):
709 os.remove(abs_path)
710 else:
711 # Check packed refs.
712 pack_file_path = cls._get_packed_refs_path(repo)
713 try:
714 with open(pack_file_path, "rb") as reader:
715 new_lines = []
716 made_change = False
717 dropped_last_line = False
718 for line_bytes in reader:
719 line = line_bytes.decode(defenc)
720 _, _, line_ref = line.partition(" ")
721 line_ref = line_ref.strip()
722 # Keep line if it is a comment or if the ref to delete is not in
723 # the line.
724 # If we deleted the last line and this one is a tag-reference
725 # object, we drop it as well.
726 if (line.startswith("#") or full_ref_path != line_ref) and (
727 not dropped_last_line or dropped_last_line and not line.startswith("^")
728 ):
729 new_lines.append(line)
730 dropped_last_line = False
731 continue
732 # END skip comments and lines without our path
734 # Drop this line.
735 made_change = True
736 dropped_last_line = True
738 # Write the new lines.
739 if made_change:
740 # Binary writing is required, otherwise Windows will open the file
741 # in text mode and change LF to CRLF!
742 with open(pack_file_path, "wb") as fd:
743 fd.writelines(line.encode(defenc) for line in new_lines)
745 except OSError:
746 pass # It didn't exist at all.
748 # Delete the reflog.
749 reflog_path = RefLog.path(cls(repo, full_ref_path))
750 if os.path.isfile(reflog_path):
751 os.remove(reflog_path)
752 # END remove reflog
754 @classmethod
755 def _create(
756 cls: Type[T_References],
757 repo: "Repo",
758 path: PathLike,
759 resolve: bool,
760 reference: Union["SymbolicReference", str],
761 force: bool,
762 logmsg: Union[str, None] = None,
763 ) -> T_References:
764 """Internal method used to create a new symbolic reference.
766 If `resolve` is ``False``, the reference will be taken as is, creating a proper
767 symbolic reference. Otherwise it will be resolved to the corresponding object
768 and a detached symbolic reference will be created instead.
769 """
770 full_ref_path = cls.to_full_path(path)
771 abs_ref_path = cls._get_validated_ref_path(repo, full_ref_path)
773 # Figure out target data.
774 target = reference
775 if resolve:
776 target = repo.rev_parse(str(reference))
778 if not force and os.path.isfile(abs_ref_path):
779 target_data = str(target)
780 if isinstance(target, SymbolicReference):
781 target_data = os.fspath(target.path)
782 if not resolve:
783 target_data = "ref: " + target_data
784 with open(abs_ref_path, "rb") as fd:
785 existing_data = fd.read().decode(defenc).strip()
786 if existing_data != target_data:
787 raise OSError(
788 "Reference at %r does already exist, pointing to %r, requested was %r"
789 % (full_ref_path, existing_data, target_data)
790 )
791 # END no force handling
793 ref = cls(repo, full_ref_path)
794 ref.set_reference(target, logmsg)
795 return ref
797 @classmethod
798 def create(
799 cls: Type[T_References],
800 repo: "Repo",
801 path: PathLike,
802 reference: Union["SymbolicReference", str] = "HEAD",
803 logmsg: Union[str, None] = None,
804 force: bool = False,
805 **kwargs: Any,
806 ) -> T_References:
807 """Create a new symbolic reference: a reference pointing to another reference.
809 :param repo:
810 Repository to create the reference in.
812 :param path:
813 Full path at which the new symbolic reference is supposed to be created at,
814 e.g. ``NEW_HEAD`` or ``symrefs/my_new_symref``.
816 :param reference:
817 The reference which the new symbolic reference should point to.
818 If it is a commit-ish, the symbolic ref will be detached.
820 :param force:
821 If ``True``, force creation even if a symbolic reference with that name
822 already exists. Raise :exc:`OSError` otherwise.
824 :param logmsg:
825 If not ``None``, the message to append to the reflog.
826 If ``None``, no reflog entry is written.
828 :return:
829 Newly created symbolic reference
831 :raise OSError:
832 If a (Symbolic)Reference with the same name but different contents already
833 exists.
835 :note:
836 This does not alter the current HEAD, index or working tree.
837 """
838 return cls._create(repo, path, cls._resolve_ref_on_create, reference, force, logmsg)
840 def rename(self, new_path: PathLike, force: bool = False) -> "SymbolicReference":
841 """Rename self to a new path.
843 :param new_path:
844 Either a simple name or a full path, e.g. ``new_name`` or
845 ``features/new_name``.
846 The prefix ``refs/`` is implied for references and will be set as needed.
847 In case this is a symbolic ref, there is no implied prefix.
849 :param force:
850 If ``True``, the rename will succeed even if a head with the target name
851 already exists. It will be overwritten in that case.
853 :return:
854 self
856 :raise OSError:
857 If a file at path but with different contents already exists.
858 """
859 new_path = self.to_full_path(new_path)
860 if self.path == new_path:
861 return self
863 new_abs_path = self._get_validated_ref_path(self.repo, new_path)
864 cur_abs_path = self._get_validated_ref_path(self.repo, self.path)
865 if os.path.isfile(new_abs_path):
866 if not force:
867 # If they point to the same file, it's not an error.
868 with open(new_abs_path, "rb") as fd1:
869 f1 = fd1.read().strip()
870 with open(cur_abs_path, "rb") as fd2:
871 f2 = fd2.read().strip()
872 if f1 != f2:
873 raise OSError("File at path %r already exists" % new_abs_path)
874 # else: We could remove ourselves and use the other one, but...
875 # ...for clarity, we just continue as usual.
876 # END not force handling
877 os.remove(new_abs_path)
878 # END handle existing target file
880 dname = os.path.dirname(new_abs_path)
881 if not os.path.isdir(dname):
882 os.makedirs(dname)
883 # END create directory
885 os.rename(cur_abs_path, new_abs_path)
886 self.path = new_path
888 return self
890 @classmethod
891 def _iter_items(
892 cls: Type[T_References], repo: "Repo", common_path: Union[PathLike, None] = None
893 ) -> Iterator[T_References]:
894 if common_path is None:
895 common_path = cls._common_path_default
896 rela_paths = set()
898 # Walk loose refs.
899 # Currently we do not follow links.
900 for root, dirs, files in os.walk(join_path_native(repo.common_dir, common_path)):
901 if "refs" not in root.split(os.sep): # Skip non-refs subfolders.
902 refs_id = [d for d in dirs if d == "refs"]
903 if refs_id:
904 dirs[0:] = ["refs"]
905 # END prune non-refs folders
907 for f in files:
908 if f == "packed-refs":
909 continue
910 abs_path = to_native_path_linux(join_path(root, f))
911 rela_paths.add(abs_path.replace(to_native_path_linux(repo.common_dir) + "/", ""))
912 # END for each file in root directory
913 # END for each directory to walk
915 # Read packed refs.
916 for _sha, rela_path in cls._iter_packed_refs(repo):
917 if rela_path.startswith(os.fspath(common_path)):
918 rela_paths.add(rela_path)
919 # END relative path matches common path
920 # END packed refs reading
922 # Yield paths in sorted order.
923 for path in sorted(rela_paths):
924 try:
925 yield cls.from_path(repo, path)
926 except ValueError:
927 continue
928 # END for each sorted relative refpath
930 @classmethod
931 def iter_items(
932 cls: Type[T_References],
933 repo: "Repo",
934 common_path: Union[PathLike, None] = None,
935 *args: Any,
936 **kwargs: Any,
937 ) -> Iterator[T_References]:
938 """Find all refs in the repository.
940 :param repo:
941 The :class:`~git.repo.base.Repo`.
943 :param common_path:
944 Optional keyword argument to the path which is to be shared by all returned
945 Ref objects.
946 Defaults to class specific portion if ``None``, ensuring that only refs
947 suitable for the actual class are returned.
949 :return:
950 A list of :class:`SymbolicReference`, each guaranteed to be a symbolic ref
951 which is not detached and pointing to a valid ref.
953 The list is lexicographically sorted. The returned objects are instances of
954 concrete subclasses, such as :class:`~git.refs.head.Head` or
955 :class:`~git.refs.tag.TagReference`.
956 """
957 return (r for r in cls._iter_items(repo, common_path) if r.__class__ is SymbolicReference or not r.is_detached)
959 @classmethod
960 def from_path(cls: Type[T_References], repo: "Repo", path: PathLike) -> T_References:
961 """Make a symbolic reference from a path.
963 :param path:
964 Full ``.git``-directory-relative path name to the Reference to instantiate.
966 :note:
967 Use :meth:`to_full_path` if you only have a partial path of a known
968 Reference type.
970 :return:
971 Instance of type :class:`~git.refs.reference.Reference`,
972 :class:`~git.refs.head.Head`, or :class:`~git.refs.tag.Tag`, depending on
973 the given path.
974 """
975 if not path:
976 raise ValueError("Cannot create Reference from %r" % path)
978 # Names like HEAD are inserted after the refs module is imported - we have an
979 # import dependency cycle and don't want to import these names in-function.
980 from . import HEAD, Head, RemoteReference, TagReference, Reference
982 for ref_type in (
983 HEAD,
984 Head,
985 RemoteReference,
986 TagReference,
987 Reference,
988 SymbolicReference,
989 ):
990 try:
991 instance = cast(T_References, ref_type(repo, path))
992 if instance.__class__ is SymbolicReference and instance.is_detached:
993 raise ValueError("SymbolicRef was detached, we drop it")
994 else:
995 return instance
997 except ValueError:
998 pass
999 # END exception handling
1000 # END for each type to try
1001 raise ValueError("Could not find reference type suitable to handle path %r" % path)
1003 def is_remote(self) -> bool:
1004 """:return: True if this symbolic reference points to a remote branch"""
1005 return os.fspath(self.path).startswith(self._remote_common_path_default + "/")