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 """Special case of a reference that is symbolic.
64 This does not point to a specific commit, but to another
65 :class:`~git.refs.head.Head`, which itself specifies a commit.
67 A typical example for a symbolic reference is :class:`~git.refs.head.HEAD`.
68 """
70 __slots__ = ("repo", "path")
72 _resolve_ref_on_create = False
73 _points_to_commits_only = True
74 _common_path_default = ""
75 _remote_common_path_default = "refs/remotes"
76 _id_attribute_ = "name"
78 def __init__(self, repo: "Repo", path: PathLike, check_path: bool = False) -> None:
79 self.repo = repo
80 self.path: PathLike = path
82 def __str__(self) -> str:
83 return os.fspath(self.path)
85 def __repr__(self) -> str:
86 return '<git.%s "%s">' % (self.__class__.__name__, self.path)
88 def __eq__(self, other: object) -> bool:
89 if hasattr(other, "path"):
90 other = cast(SymbolicReference, other)
91 return self.path == other.path
92 return False
94 def __ne__(self, other: object) -> bool:
95 return not (self == other)
97 def __hash__(self) -> int:
98 return hash(self.path)
100 @property
101 def name(self) -> str:
102 """
103 :return:
104 In case of symbolic references, the shortest assumable name is the path
105 itself.
106 """
107 return os.fspath(self.path)
109 @property
110 def abspath(self) -> PathLike:
111 return join_path_native(_git_dir(self.repo, self.path), self.path)
113 @staticmethod
114 def _get_validated_path(base: PathLike, path: PathLike) -> str:
115 path = os.fspath(path)
116 base_path = os.path.realpath(os.fspath(base))
117 abs_path = os.path.realpath(os.path.join(base_path, path))
118 try:
119 common_path = os.path.commonpath([base_path, abs_path])
120 except ValueError as e:
121 raise ValueError("Reference path %r escapes the repository" % path) from e
122 if os.path.normcase(common_path) != os.path.normcase(base_path):
123 raise ValueError("Reference path %r escapes the repository" % path)
124 return abs_path
126 @classmethod
127 def _get_validated_ref_path(cls, repo: "Repo", path: PathLike) -> str:
128 """Return the absolute filesystem path for a ref after validating it."""
129 cls._check_ref_name_valid(path)
130 ref_path = os.fspath(path)
131 return cls._get_validated_path(_git_dir(repo, ref_path), ref_path)
133 @classmethod
134 def _get_validated_reflog_path(cls, repo: "Repo", path: PathLike) -> str:
135 """Return the absolute filesystem path for a reflog after validating it."""
136 cls._check_ref_name_valid(path)
137 return cls._get_validated_path(os.path.join(repo.git_dir, "logs"), path)
139 @classmethod
140 def _get_packed_refs_path(cls, repo: "Repo") -> str:
141 return os.path.join(repo.common_dir, "packed-refs")
143 @classmethod
144 def _iter_packed_refs(cls, repo: "Repo") -> Iterator[Tuple[str, str]]:
145 """Return an iterator yielding pairs of sha1/path pairs (as strings) for the
146 corresponding refs.
148 :note:
149 The packed refs file will be kept open as long as we iterate.
150 """
151 try:
152 with open(cls._get_packed_refs_path(repo), "rt", encoding="UTF-8") as fp:
153 for line in fp:
154 line = line.strip()
155 if not line:
156 continue
157 if line.startswith("#"):
158 # "# pack-refs with: peeled fully-peeled sorted"
159 # the git source code shows "peeled",
160 # "fully-peeled" and "sorted" as the keywords
161 # that can go on this line, as per comments in git file
162 # refs/packed-backend.c
163 # I looked at master on 2017-10-11,
164 # commit 111ef79afe, after tag v2.15.0-rc1
165 # from repo https://github.com/git/git.git
166 if line.startswith("# pack-refs with:") and "peeled" not in line:
167 raise TypeError("PackingType of packed-Refs not understood: %r" % line)
168 # END abort if we do not understand the packing scheme
169 continue
170 # END parse comment
172 # Skip dereferenced tag object entries - previous line was actual
173 # tag reference for it.
174 if line[0] == "^":
175 continue
177 yield cast(Tuple[str, str], tuple(line.split(" ", 1)))
178 # END for each line
179 except OSError:
180 return None
181 # END no packed-refs file handling
183 @classmethod
184 def dereference_recursive(cls, repo: "Repo", ref_path: Union[PathLike, None]) -> str:
185 """
186 :return:
187 hexsha stored in the reference at the given `ref_path`, recursively
188 dereferencing all intermediate references as required
190 :param repo:
191 The repository containing the reference at `ref_path`.
192 """
194 while True:
195 hexsha, ref_path = cls._get_ref_info(repo, ref_path)
196 if hexsha is not None:
197 return hexsha
198 # END recursive dereferencing
200 @staticmethod
201 def _check_ref_name_valid(ref_path: PathLike) -> None:
202 """Check a ref name for validity.
204 This is based on the rules described in :manpage:`git-check-ref-format(1)`.
205 """
206 previous: Union[str, None] = None
207 one_before_previous: Union[str, None] = None
208 for c in os.fspath(ref_path):
209 if c in " ~^:?*[\\":
210 raise ValueError(
211 f"Invalid reference '{ref_path}': references cannot contain spaces, tildes (~), carets (^),"
212 f" colons (:), question marks (?), asterisks (*), open brackets ([) or backslashes (\\)"
213 )
214 elif c == ".":
215 if previous is None or previous == "/":
216 raise ValueError(
217 f"Invalid reference '{ref_path}': references cannot start with a period (.) or contain '/.'"
218 )
219 elif previous == ".":
220 raise ValueError(f"Invalid reference '{ref_path}': references cannot contain '..'")
221 elif c == "/":
222 if previous == "/":
223 raise ValueError(f"Invalid reference '{ref_path}': references cannot contain '//'")
224 elif previous is None:
225 raise ValueError(
226 f"Invalid reference '{ref_path}': references cannot start with forward slashes '/'"
227 )
228 elif c == "{" and previous == "@":
229 raise ValueError(f"Invalid reference '{ref_path}': references cannot contain '@{{'")
230 elif ord(c) < 32 or ord(c) == 127:
231 raise ValueError(f"Invalid reference '{ref_path}': references cannot contain ASCII control characters")
233 one_before_previous = previous
234 previous = c
236 if previous == ".":
237 raise ValueError(f"Invalid reference '{ref_path}': references cannot end with a period (.)")
238 elif previous == "/":
239 raise ValueError(f"Invalid reference '{ref_path}': references cannot end with a forward slash (/)")
240 elif previous == "@" and one_before_previous is None:
241 raise ValueError(f"Invalid reference '{ref_path}': references cannot be '@'")
242 elif any(component.endswith(".lock") for component in Path(ref_path).parts):
243 raise ValueError(
244 f"Invalid reference '{ref_path}': references cannot have slash-separated components that end with"
245 " '.lock'"
246 )
248 @classmethod
249 def _get_ref_info_helper(
250 cls, repo: "Repo", ref_path: Union[PathLike, None]
251 ) -> Union[Tuple[str, None], Tuple[None, str]]:
252 """
253 :return:
254 *(str(sha), str(target_ref_path))*, where:
256 * *sha* is of the file at rela_path points to if available, or ``None``.
257 * *target_ref_path* is the reference we point to, or ``None``.
258 """
259 if ref_path:
260 cls._check_ref_name_valid(ref_path)
262 tokens: Union[None, List[str], Tuple[str, str]] = None
263 repodir = _git_dir(repo, ref_path)
264 try:
265 with open(os.path.join(repodir, ref_path), "rt", encoding="UTF-8") as fp: # type: ignore[arg-type]
266 value = fp.read().rstrip()
267 # Don't only split on spaces, but on whitespace, which allows to parse lines like:
268 # 60b64ef992065e2600bfef6187a97f92398a9144 branch 'master' of git-server:/path/to/repo
269 tokens = value.split()
270 assert len(tokens) != 0
271 except OSError:
272 # Probably we are just packed. Find our entry in the packed refs file.
273 # NOTE: We are not a symbolic ref if we are in a packed file, as these
274 # are excluded explicitly.
275 for sha, path in cls._iter_packed_refs(repo):
276 if path != ref_path:
277 continue
278 # sha will be used.
279 tokens = sha, path
280 break
281 # END for each packed ref
282 # END handle packed refs
283 if tokens is None:
284 raise ValueError("Reference at %r does not exist" % ref_path)
286 # Is it a reference?
287 if tokens[0] == "ref:":
288 return (None, tokens[1])
290 # It's a commit.
291 if repo.re_hexsha_only.match(tokens[0]):
292 return (tokens[0], None)
294 raise ValueError("Failed to parse reference information from %r" % ref_path)
296 @classmethod
297 def _get_ref_info(cls, repo: "Repo", ref_path: Union[PathLike, None]) -> Union[Tuple[str, None], Tuple[None, str]]:
298 """
299 :return:
300 *(str(sha), str(target_ref_path))*, where:
302 * *sha* is of the file at rela_path points to if available, or ``None``.
303 * *target_ref_path* is the reference we point to, or ``None``.
304 """
305 return cls._get_ref_info_helper(repo, ref_path)
307 def _get_object(self) -> AnyGitObject:
308 """
309 :return:
310 The object our ref currently refers to. Refs can be cached, they will always
311 point to the actual object as it gets re-created on each query.
312 """
313 # We have to be dynamic here as we may be a tag which can point to anything.
314 # Our path will be resolved to the hexsha which will be used accordingly.
315 return Object.new_from_sha(self.repo, hex_to_bin(self.dereference_recursive(self.repo, self.path)))
317 def _get_commit(self) -> "Commit":
318 """
319 :return:
320 :class:`~git.objects.commit.Commit` object we point to. This works for
321 detached and non-detached :class:`SymbolicReference` instances. The symbolic
322 reference will be dereferenced recursively.
323 """
324 obj = self._get_object()
325 if obj.type == "tag":
326 obj = obj.object
327 # END dereference tag
329 if obj.type != Commit.type:
330 raise TypeError("Symbolic Reference pointed to object %r, commit was required" % obj)
331 # END handle type
332 return obj
334 def set_commit(
335 self,
336 commit: Union[Commit, "SymbolicReference", str],
337 logmsg: Union[str, None] = None,
338 ) -> "SymbolicReference":
339 """Like :meth:`set_object`, but restricts the type of object to be a
340 :class:`~git.objects.commit.Commit`.
342 :raise ValueError:
343 If `commit` is not a :class:`~git.objects.commit.Commit` object, nor does it
344 point to a commit.
346 :return:
347 self
348 """
349 # Check the type - assume the best if it is a base-string.
350 invalid_type = False
351 if isinstance(commit, Object):
352 invalid_type = commit.type != Commit.type
353 elif isinstance(commit, SymbolicReference):
354 invalid_type = commit.object.type != Commit.type
355 else:
356 try:
357 invalid_type = self.repo.rev_parse(commit).type != Commit.type
358 except (BadObject, BadName) as e:
359 raise ValueError("Invalid object: %s" % commit) from e
360 # END handle exception
361 # END verify type
363 if invalid_type:
364 raise ValueError("Need commit, got %r" % commit)
365 # END handle raise
367 # We leave strings to the rev-parse method below.
368 self.set_object(commit, logmsg)
370 return self
372 def set_object(
373 self,
374 object: Union[AnyGitObject, "SymbolicReference", str],
375 logmsg: Union[str, None] = None,
376 ) -> "SymbolicReference":
377 """Set the object we point to, possibly dereference our symbolic reference
378 first. If the reference does not exist, it will be created.
380 :param object:
381 A refspec, a :class:`SymbolicReference` or an
382 :class:`~git.objects.base.Object` instance.
384 * :class:`SymbolicReference` instances will be dereferenced beforehand to
385 obtain the git object they point to.
386 * :class:`~git.objects.base.Object` instances must represent git objects
387 (:class:`~git.types.AnyGitObject`).
389 :param logmsg:
390 If not ``None``, the message will be used in the reflog entry to be written.
391 Otherwise the reflog is not altered.
393 :note:
394 Plain :class:`SymbolicReference` instances may not actually point to objects
395 by convention.
397 :return:
398 self
399 """
400 if isinstance(object, SymbolicReference):
401 object = object.object # @ReservedAssignment
402 # END resolve references
404 is_detached = True
405 try:
406 is_detached = self.is_detached
407 except ValueError:
408 pass
409 # END handle non-existing ones
411 if is_detached:
412 return self.set_reference(object, logmsg)
414 # set the commit on our reference
415 return self._get_reference().set_object(object, logmsg)
417 @property
418 def commit(self) -> "Commit":
419 """Query or set commits directly"""
420 return self._get_commit()
422 @commit.setter
423 def commit(self, commit: Union[Commit, "SymbolicReference", str]) -> "SymbolicReference":
424 return self.set_commit(commit)
426 @property
427 def object(self) -> AnyGitObject:
428 """Return the object our ref currently refers to"""
429 return self._get_object()
431 @object.setter
432 def object(self, object: Union[AnyGitObject, "SymbolicReference", str]) -> "SymbolicReference":
433 return self.set_object(object)
435 def _get_reference(self) -> "Reference":
436 """
437 :return:
438 :class:`~git.refs.reference.Reference` object we point to
440 :raise TypeError:
441 If this symbolic reference is detached, hence it doesn't point to a
442 reference, but to a commit.
443 """
444 sha, target_ref_path = self._get_ref_info(self.repo, self.path)
445 if target_ref_path is None:
446 raise TypeError("%s is a detached symbolic reference as it points to %r" % (self, sha))
447 return cast("Reference", self.from_path(self.repo, target_ref_path))
449 def set_reference(
450 self,
451 ref: Union[AnyGitObject, "SymbolicReference", str],
452 logmsg: Union[str, None] = None,
453 ) -> "SymbolicReference":
454 """Set ourselves to the given `ref`.
456 It will stay a symbol if the `ref` is a :class:`~git.refs.reference.Reference`.
458 Otherwise a git object, specified as a :class:`~git.objects.base.Object`
459 instance or refspec, is assumed. If it is valid, this reference will be set to
460 it, which effectively detaches the reference if it was a purely symbolic one.
462 :param ref:
463 A :class:`SymbolicReference` instance, an :class:`~git.objects.base.Object`
464 instance (specifically an :class:`~git.types.AnyGitObject`), or a refspec
465 string. Only if the ref is a :class:`SymbolicReference` instance, we will
466 point to it. Everything else is dereferenced to obtain the actual object.
468 :param logmsg:
469 If set to a string, the message will be used in the reflog.
470 Otherwise, a reflog entry is not written for the changed reference.
471 The previous commit of the entry will be the commit we point to now.
473 See also: :meth:`log_append`
475 :return:
476 self
478 :note:
479 This symbolic reference will not be dereferenced. For that, see
480 :meth:`set_object`.
481 """
482 write_value = None
483 obj = None
484 if isinstance(ref, SymbolicReference):
485 write_value = "ref: %s" % ref.path
486 elif isinstance(ref, Object):
487 obj = ref
488 write_value = ref.hexsha
489 elif isinstance(ref, str):
490 try:
491 obj = self.repo.rev_parse(ref + "^{}") # Optionally dereference tags.
492 write_value = obj.hexsha
493 except (BadObject, BadName) as e:
494 raise ValueError("Could not extract object from %s" % ref) from e
495 # END end try string
496 else:
497 raise ValueError("Unrecognized Value: %r" % ref)
498 # END try commit attribute
500 # typecheck
501 if obj is not None and self._points_to_commits_only and obj.type != Commit.type:
502 raise TypeError("Require commit, got %r" % obj)
503 # END verify type
505 oldbinsha: bytes = b""
506 if logmsg is not None:
507 try:
508 oldbinsha = self.commit.binsha
509 except ValueError:
510 oldbinsha = Commit.NULL_BIN_SHA
511 # END handle non-existing
512 # END retrieve old hexsha
514 fpath = self._get_validated_ref_path(self.repo, self.path)
515 assure_directory_exists(fpath, is_file=True)
517 lfd = LockedFD(fpath)
518 fd = lfd.open(write=True, stream=True)
519 try:
520 fd.write(write_value.encode("utf-8") + b"\n")
521 lfd.commit()
522 except BaseException:
523 lfd.rollback()
524 raise
525 # Adjust the reflog
526 if logmsg is not None:
527 self.log_append(oldbinsha, logmsg)
529 return self
531 # Aliased reference
532 @property
533 def reference(self) -> "Reference":
534 return self._get_reference()
536 @reference.setter
537 def reference(self, ref: Union[AnyGitObject, "SymbolicReference", str]) -> "SymbolicReference":
538 return self.set_reference(ref)
540 ref = reference
542 def is_valid(self) -> bool:
543 """
544 :return:
545 ``True`` if the reference is valid, hence it can be read and points to a
546 valid object or reference.
547 """
548 try:
549 self.object # noqa: B018
550 except (OSError, ValueError):
551 return False
552 else:
553 return True
555 @property
556 def is_detached(self) -> bool:
557 """
558 :return:
559 ``True`` if we are a detached reference, hence we point to a specific commit
560 instead to another reference.
561 """
562 try:
563 self.ref # noqa: B018
564 return False
565 except TypeError:
566 return True
568 def log(self) -> "RefLog":
569 """
570 :return:
571 :class:`~git.refs.log.RefLog` for this reference.
572 Its last entry reflects the latest change applied to this reference.
574 :note:
575 As the log is parsed every time, its recommended to cache it for use instead
576 of calling this method repeatedly. It should be considered read-only.
577 """
578 return RefLog.from_file(RefLog.path(self))
580 def log_append(
581 self,
582 oldbinsha: bytes,
583 message: Union[str, None],
584 newbinsha: Union[bytes, None] = None,
585 ) -> "RefLogEntry":
586 """Append a logentry to the logfile of this ref.
588 :param oldbinsha:
589 Binary sha this ref used to point to.
591 :param message:
592 A message describing the change.
594 :param newbinsha:
595 The sha the ref points to now. If None, our current commit sha will be used.
597 :return:
598 The added :class:`~git.refs.log.RefLogEntry` instance.
599 """
600 # NOTE: We use the committer of the currently active commit - this should be
601 # correct to allow overriding the committer on a per-commit level.
602 # See https://github.com/gitpython-developers/GitPython/pull/146.
603 try:
604 committer_or_reader: Union["Actor", "GitConfigParser"] = self.commit.committer
605 except ValueError:
606 committer_or_reader = self.repo.config_reader()
607 # END handle newly cloned repositories
608 if newbinsha is None:
609 newbinsha = self.commit.binsha
611 if message is None:
612 message = ""
614 return RefLog.append_entry(committer_or_reader, RefLog.path(self), oldbinsha, newbinsha, message)
616 def log_entry(self, index: int) -> "RefLogEntry":
617 """
618 :return:
619 :class:`~git.refs.log.RefLogEntry` at the given index
621 :param index:
622 Python list compatible positive or negative index.
624 :note:
625 This method must read part of the reflog during execution, hence it should
626 be used sparingly, or only if you need just one index. In that case, it will
627 be faster than the :meth:`log` method.
628 """
629 return RefLog.entry_at(RefLog.path(self), index)
631 @classmethod
632 def to_full_path(cls, path: Union[PathLike, "SymbolicReference"]) -> PathLike:
633 """
634 :return:
635 String with a full repository-relative path which can be used to initialize
636 a :class:`~git.refs.reference.Reference` instance, for instance by using
637 :meth:`Reference.from_path <git.refs.reference.Reference.from_path>`.
638 """
639 if isinstance(path, SymbolicReference):
640 path = path.path
641 full_ref_path = path
642 if not cls._common_path_default:
643 return full_ref_path
644 if not os.fspath(path).startswith(cls._common_path_default + "/"):
645 full_ref_path = "%s/%s" % (cls._common_path_default, path)
646 return full_ref_path
648 @classmethod
649 def delete(cls, repo: "Repo", path: PathLike) -> None:
650 """Delete the reference at the given path.
652 :param repo:
653 Repository to delete the reference from.
655 :param path:
656 Short or full path pointing to the reference, e.g. ``refs/myreference`` or
657 just ``myreference``, hence ``refs/`` is implied.
658 Alternatively the symbolic reference to be deleted.
659 """
660 full_ref_path = cls.to_full_path(path)
661 abs_path = cls._get_validated_ref_path(repo, full_ref_path)
662 if os.path.exists(abs_path):
663 os.remove(abs_path)
664 else:
665 # Check packed refs.
666 pack_file_path = cls._get_packed_refs_path(repo)
667 try:
668 with open(pack_file_path, "rb") as reader:
669 new_lines = []
670 made_change = False
671 dropped_last_line = False
672 for line_bytes in reader:
673 line = line_bytes.decode(defenc)
674 _, _, line_ref = line.partition(" ")
675 line_ref = line_ref.strip()
676 # Keep line if it is a comment or if the ref to delete is not in
677 # the line.
678 # If we deleted the last line and this one is a tag-reference
679 # object, we drop it as well.
680 if (line.startswith("#") or full_ref_path != line_ref) and (
681 not dropped_last_line or dropped_last_line and not line.startswith("^")
682 ):
683 new_lines.append(line)
684 dropped_last_line = False
685 continue
686 # END skip comments and lines without our path
688 # Drop this line.
689 made_change = True
690 dropped_last_line = True
692 # Write the new lines.
693 if made_change:
694 # Binary writing is required, otherwise Windows will open the file
695 # in text mode and change LF to CRLF!
696 with open(pack_file_path, "wb") as fd:
697 fd.writelines(line.encode(defenc) for line in new_lines)
699 except OSError:
700 pass # It didn't exist at all.
702 # Delete the reflog.
703 reflog_path = RefLog.path(cls(repo, full_ref_path))
704 if os.path.isfile(reflog_path):
705 os.remove(reflog_path)
706 # END remove reflog
708 @classmethod
709 def _create(
710 cls: Type[T_References],
711 repo: "Repo",
712 path: PathLike,
713 resolve: bool,
714 reference: Union["SymbolicReference", str],
715 force: bool,
716 logmsg: Union[str, None] = None,
717 ) -> T_References:
718 """Internal method used to create a new symbolic reference.
720 If `resolve` is ``False``, the reference will be taken as is, creating a proper
721 symbolic reference. Otherwise it will be resolved to the corresponding object
722 and a detached symbolic reference will be created instead.
723 """
724 full_ref_path = cls.to_full_path(path)
725 abs_ref_path = cls._get_validated_ref_path(repo, full_ref_path)
727 # Figure out target data.
728 target = reference
729 if resolve:
730 target = repo.rev_parse(str(reference))
732 if not force and os.path.isfile(abs_ref_path):
733 target_data = str(target)
734 if isinstance(target, SymbolicReference):
735 target_data = os.fspath(target.path)
736 if not resolve:
737 target_data = "ref: " + target_data
738 with open(abs_ref_path, "rb") as fd:
739 existing_data = fd.read().decode(defenc).strip()
740 if existing_data != target_data:
741 raise OSError(
742 "Reference at %r does already exist, pointing to %r, requested was %r"
743 % (full_ref_path, existing_data, target_data)
744 )
745 # END no force handling
747 ref = cls(repo, full_ref_path)
748 ref.set_reference(target, logmsg)
749 return ref
751 @classmethod
752 def create(
753 cls: Type[T_References],
754 repo: "Repo",
755 path: PathLike,
756 reference: Union["SymbolicReference", str] = "HEAD",
757 logmsg: Union[str, None] = None,
758 force: bool = False,
759 **kwargs: Any,
760 ) -> T_References:
761 """Create a new symbolic reference: a reference pointing to another reference.
763 :param repo:
764 Repository to create the reference in.
766 :param path:
767 Full path at which the new symbolic reference is supposed to be created at,
768 e.g. ``NEW_HEAD`` or ``symrefs/my_new_symref``.
770 :param reference:
771 The reference which the new symbolic reference should point to.
772 If it is a commit-ish, the symbolic ref will be detached.
774 :param force:
775 If ``True``, force creation even if a symbolic reference with that name
776 already exists. Raise :exc:`OSError` otherwise.
778 :param logmsg:
779 If not ``None``, the message to append to the reflog.
780 If ``None``, no reflog entry is written.
782 :return:
783 Newly created symbolic reference
785 :raise OSError:
786 If a (Symbolic)Reference with the same name but different contents already
787 exists.
789 :note:
790 This does not alter the current HEAD, index or working tree.
791 """
792 return cls._create(repo, path, cls._resolve_ref_on_create, reference, force, logmsg)
794 def rename(self, new_path: PathLike, force: bool = False) -> "SymbolicReference":
795 """Rename self to a new path.
797 :param new_path:
798 Either a simple name or a full path, e.g. ``new_name`` or
799 ``features/new_name``.
800 The prefix ``refs/`` is implied for references and will be set as needed.
801 In case this is a symbolic ref, there is no implied prefix.
803 :param force:
804 If ``True``, the rename will succeed even if a head with the target name
805 already exists. It will be overwritten in that case.
807 :return:
808 self
810 :raise OSError:
811 If a file at path but with different contents already exists.
812 """
813 new_path = self.to_full_path(new_path)
814 if self.path == new_path:
815 return self
817 new_abs_path = self._get_validated_ref_path(self.repo, new_path)
818 cur_abs_path = self._get_validated_ref_path(self.repo, self.path)
819 if os.path.isfile(new_abs_path):
820 if not force:
821 # If they point to the same file, it's not an error.
822 with open(new_abs_path, "rb") as fd1:
823 f1 = fd1.read().strip()
824 with open(cur_abs_path, "rb") as fd2:
825 f2 = fd2.read().strip()
826 if f1 != f2:
827 raise OSError("File at path %r already exists" % new_abs_path)
828 # else: We could remove ourselves and use the other one, but...
829 # ...for clarity, we just continue as usual.
830 # END not force handling
831 os.remove(new_abs_path)
832 # END handle existing target file
834 dname = os.path.dirname(new_abs_path)
835 if not os.path.isdir(dname):
836 os.makedirs(dname)
837 # END create directory
839 os.rename(cur_abs_path, new_abs_path)
840 self.path = new_path
842 return self
844 @classmethod
845 def _iter_items(
846 cls: Type[T_References], repo: "Repo", common_path: Union[PathLike, None] = None
847 ) -> Iterator[T_References]:
848 if common_path is None:
849 common_path = cls._common_path_default
850 rela_paths = set()
852 # Walk loose refs.
853 # Currently we do not follow links.
854 for root, dirs, files in os.walk(join_path_native(repo.common_dir, common_path)):
855 if "refs" not in root.split(os.sep): # Skip non-refs subfolders.
856 refs_id = [d for d in dirs if d == "refs"]
857 if refs_id:
858 dirs[0:] = ["refs"]
859 # END prune non-refs folders
861 for f in files:
862 if f == "packed-refs":
863 continue
864 abs_path = to_native_path_linux(join_path(root, f))
865 rela_paths.add(abs_path.replace(to_native_path_linux(repo.common_dir) + "/", ""))
866 # END for each file in root directory
867 # END for each directory to walk
869 # Read packed refs.
870 for _sha, rela_path in cls._iter_packed_refs(repo):
871 if rela_path.startswith(os.fspath(common_path)):
872 rela_paths.add(rela_path)
873 # END relative path matches common path
874 # END packed refs reading
876 # Yield paths in sorted order.
877 for path in sorted(rela_paths):
878 try:
879 yield cls.from_path(repo, path)
880 except ValueError:
881 continue
882 # END for each sorted relative refpath
884 @classmethod
885 def iter_items(
886 cls: Type[T_References],
887 repo: "Repo",
888 common_path: Union[PathLike, None] = None,
889 *args: Any,
890 **kwargs: Any,
891 ) -> Iterator[T_References]:
892 """Find all refs in the repository.
894 :param repo:
895 The :class:`~git.repo.base.Repo`.
897 :param common_path:
898 Optional keyword argument to the path which is to be shared by all returned
899 Ref objects.
900 Defaults to class specific portion if ``None``, ensuring that only refs
901 suitable for the actual class are returned.
903 :return:
904 A list of :class:`SymbolicReference`, each guaranteed to be a symbolic ref
905 which is not detached and pointing to a valid ref.
907 The list is lexicographically sorted. The returned objects are instances of
908 concrete subclasses, such as :class:`~git.refs.head.Head` or
909 :class:`~git.refs.tag.TagReference`.
910 """
911 return (r for r in cls._iter_items(repo, common_path) if r.__class__ is SymbolicReference or not r.is_detached)
913 @classmethod
914 def from_path(cls: Type[T_References], repo: "Repo", path: PathLike) -> T_References:
915 """Make a symbolic reference from a path.
917 :param path:
918 Full ``.git``-directory-relative path name to the Reference to instantiate.
920 :note:
921 Use :meth:`to_full_path` if you only have a partial path of a known
922 Reference type.
924 :return:
925 Instance of type :class:`~git.refs.reference.Reference`,
926 :class:`~git.refs.head.Head`, or :class:`~git.refs.tag.Tag`, depending on
927 the given path.
928 """
929 if not path:
930 raise ValueError("Cannot create Reference from %r" % path)
932 # Names like HEAD are inserted after the refs module is imported - we have an
933 # import dependency cycle and don't want to import these names in-function.
934 from . import HEAD, Head, RemoteReference, TagReference, Reference
936 for ref_type in (
937 HEAD,
938 Head,
939 RemoteReference,
940 TagReference,
941 Reference,
942 SymbolicReference,
943 ):
944 try:
945 instance = cast(T_References, ref_type(repo, path))
946 if instance.__class__ is SymbolicReference and instance.is_detached:
947 raise ValueError("SymbolicRef was detached, we drop it")
948 else:
949 return instance
951 except ValueError:
952 pass
953 # END exception handling
954 # END for each type to try
955 raise ValueError("Could not find reference type suitable to handle path %r" % path)
957 def is_remote(self) -> bool:
958 """:return: True if this symbolic reference points to a remote branch"""
959 return os.fspath(self.path).startswith(self._remote_common_path_default + "/")