Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/refs.py: 31%
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# refs.py -- For dealing with git refs
2# Copyright (C) 2008-2013 Jelmer Vernooij <jelmer@jelmer.uk>
3#
4# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
5# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
6# General Public License as published by the Free Software Foundation; version 2.0
7# or (at your option) any later version. You can redistribute it and/or
8# modify it under the terms of either of these two licenses.
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16# You should have received a copy of the licenses; if not, see
17# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
18# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
19# License, Version 2.0.
20#
23"""Ref handling."""
25__all__ = [
26 "HEADREF",
27 "LOCAL_BRANCH_PREFIX",
28 "LOCAL_NOTES_PREFIX",
29 "LOCAL_REMOTE_PREFIX",
30 "LOCAL_REPLACE_PREFIX",
31 "LOCAL_TAG_PREFIX",
32 "SYMREF",
33 "DictRefsContainer",
34 "DiskRefsContainer",
35 "NamespacedRefsContainer",
36 "Ref",
37 "RefsContainer",
38 "SymrefLoop",
39 "check_ref_format",
40 "extract_branch_name",
41 "extract_tag_name",
42 "filter_ref_prefix",
43 "is_local_branch",
44 "is_per_worktree_ref",
45 "local_branch_name",
46 "local_replace_name",
47 "local_tag_name",
48 "parse_remote_ref",
49 "parse_symref_value",
50 "read_info_refs",
51 "read_packed_refs",
52 "read_packed_refs_with_peeled",
53 "set_ref_from_raw",
54 "shorten_ref_name",
55 "write_packed_refs",
56]
58import os
59import sys
60import types
61import warnings
62from collections.abc import Callable, Iterable, Iterator, Mapping
63from contextlib import suppress
64from typing import (
65 IO,
66 TYPE_CHECKING,
67 Any,
68 BinaryIO,
69 NewType,
70 overload,
71)
73if sys.version_info >= (3, 11):
74 from typing import Self
75else:
76 from typing_extensions import Self
78if TYPE_CHECKING:
79 from .file import _GitFile
81from .errors import PackedRefsException, RefFormatError
82from .file import GitFile, ensure_dir_exists
83from .objects import ZERO_SHA, ObjectID, git_line, valid_hexsha
85Ref = NewType("Ref", bytes)
87HEADREF = Ref(b"HEAD")
88SYMREF = b"ref: "
89LOCAL_BRANCH_PREFIX = b"refs/heads/"
90LOCAL_TAG_PREFIX = b"refs/tags/"
91LOCAL_REMOTE_PREFIX = b"refs/remotes/"
92LOCAL_NOTES_PREFIX = b"refs/notes/"
93LOCAL_REPLACE_PREFIX = b"refs/replace/"
94BAD_REF_CHARS: set[int] = set(b"\177 ~^:?*[")
97class SymrefLoop(Exception):
98 """There is a loop between one or more symrefs."""
100 def __init__(self, ref: bytes, depth: int) -> None:
101 """Initialize SymrefLoop exception."""
102 self.ref = ref
103 self.depth = depth
106def parse_symref_value(contents: bytes) -> bytes:
107 """Parse a symref value.
109 Args:
110 contents: Contents to parse
111 Returns: Destination
112 """
113 if contents.startswith(SYMREF):
114 return contents[len(SYMREF) :].rstrip(b"\r\n")
115 raise ValueError(contents)
118def check_ref_format(refname: Ref) -> bool:
119 """Check if a refname is correctly formatted.
121 Implements all the same rules as git-check-ref-format[1].
123 [1]
124 http://www.kernel.org/pub/software/scm/git/docs/git-check-ref-format.html
126 Args:
127 refname: The refname to check
128 Returns: True if refname is valid, False otherwise
129 """
130 # These could be combined into one big expression, but are listed
131 # separately to parallel [1].
132 if refname == b"@":
133 return False
134 if b"/" not in refname: # type: ignore[comparison-overlap]
135 return False
136 if b".." in refname: # type: ignore[comparison-overlap]
137 return False
138 for i, c in enumerate(refname):
139 if ord(refname[i : i + 1]) < 0o40 or c in BAD_REF_CHARS:
140 return False
141 if refname[-1] in b"/.":
142 return False
143 if b"@{" in refname: # type: ignore[comparison-overlap]
144 return False
145 if b"\\" in refname: # type: ignore[comparison-overlap]
146 return False
147 for component in refname.split(b"/"):
148 if not component:
149 return False
150 if component.startswith(b"."):
151 return False
152 if component.endswith(b".lock"):
153 return False
154 return True
157def _collapse_slashes(refname: bytes) -> bytes:
158 """Collapse runs of consecutive slashes in a ref name into a single slash."""
159 return b"/".join(component for component in refname.split(b"/") if component)
162def parse_remote_ref(ref: bytes) -> tuple[bytes, bytes]:
163 """Parse a remote ref into remote name and branch name.
165 Args:
166 ref: Remote ref like b"refs/remotes/origin/main"
168 Returns:
169 Tuple of (remote_name, branch_name)
171 Raises:
172 ValueError: If ref is not a valid remote ref
173 """
174 if not ref.startswith(LOCAL_REMOTE_PREFIX):
175 raise ValueError(f"Not a remote ref: {ref!r}")
177 # Remove the prefix
178 remainder = ref[len(LOCAL_REMOTE_PREFIX) :]
180 # Split into remote name and branch name
181 parts = remainder.split(b"/", 1)
182 if len(parts) != 2:
183 raise ValueError(f"Invalid remote ref format: {ref!r}")
185 remote_name, branch_name = parts
186 return (remote_name, branch_name)
189def set_ref_from_raw(refs: "RefsContainer", name: Ref, raw_ref: bytes) -> None:
190 """Set a reference from a raw ref value.
192 This handles both symbolic refs (starting with 'ref: ') and direct ObjectID refs.
194 Args:
195 refs: The RefsContainer to set the ref in
196 name: The ref name to set
197 raw_ref: The raw ref value (either a symbolic ref or an ObjectID)
198 """
199 if raw_ref.startswith(SYMREF):
200 # It's a symbolic ref
201 target = Ref(raw_ref[len(SYMREF) :])
202 refs.set_symbolic_ref(name, target)
203 else:
204 # It's a direct ObjectID
205 refs[name] = ObjectID(raw_ref)
208class RefsContainer:
209 """A container for refs."""
211 def __init__(
212 self,
213 logger: Callable[
214 [bytes, bytes, bytes, bytes | None, int | None, int | None, bytes], None
215 ]
216 | None = None,
217 ) -> None:
218 """Initialize RefsContainer with optional logger function."""
219 self._logger = logger
221 def _log(
222 self,
223 ref: bytes,
224 old_sha: bytes | None,
225 new_sha: bytes | None,
226 committer: bytes | None = None,
227 timestamp: int | None = None,
228 timezone: int | None = None,
229 message: bytes | None = None,
230 ) -> None:
231 if self._logger is None:
232 return
233 if message is None:
234 return
235 # Use ZERO_SHA for None values, matching git behavior
236 if old_sha is None:
237 old_sha = ZERO_SHA
238 if new_sha is None:
239 new_sha = ZERO_SHA
240 self._logger(ref, old_sha, new_sha, committer, timestamp, timezone, message)
242 def set_symbolic_ref(
243 self,
244 name: Ref,
245 other: Ref,
246 committer: bytes | None = None,
247 timestamp: int | None = None,
248 timezone: int | None = None,
249 message: bytes | None = None,
250 ) -> None:
251 """Make a ref point at another ref.
253 Args:
254 name: Name of the ref to set
255 other: Name of the ref to point at
256 committer: Optional committer name/email
257 timestamp: Optional timestamp
258 timezone: Optional timezone
259 message: Optional message
260 """
261 raise NotImplementedError(self.set_symbolic_ref)
263 def get_packed_refs(self) -> dict[Ref, ObjectID]:
264 """Get contents of the packed-refs file.
266 Returns: Dictionary mapping ref names to SHA1s
268 Note: Will return an empty dictionary when no packed-refs file is
269 present.
270 """
271 raise NotImplementedError(self.get_packed_refs)
273 def add_packed_refs(self, new_refs: Mapping[Ref, ObjectID | None]) -> None:
274 """Add the given refs as packed refs.
276 Args:
277 new_refs: A mapping of ref names to targets; if a target is None that
278 means remove the ref
279 """
280 raise NotImplementedError(self.add_packed_refs)
282 def get_peeled(self, name: Ref) -> ObjectID | None:
283 """Return the cached peeled value of a ref, if available.
285 Args:
286 name: Name of the ref to peel
287 Returns: The peeled value of the ref. If the ref is known not point to
288 a tag, this will be the SHA the ref refers to. If the ref may point
289 to a tag, but no cached information is available, None is returned.
290 """
291 return None
293 def import_refs(
294 self,
295 base: Ref,
296 other: Mapping[Ref, ObjectID | None],
297 committer: bytes | None = None,
298 timestamp: bytes | None = None,
299 timezone: bytes | None = None,
300 message: bytes | None = None,
301 prune: bool = False,
302 ) -> None:
303 """Import refs from another repository.
305 Args:
306 base: Base ref to import into (e.g., b'refs/remotes/origin')
307 other: Dictionary of refs to import
308 committer: Optional committer for reflog
309 timestamp: Optional timestamp for reflog
310 timezone: Optional timezone for reflog
311 message: Optional message for reflog
312 prune: If True, remove refs not in other
313 """
314 # Strip a trailing slash so joining ``base`` with a ref name below
315 # does not produce a malformed ref with an empty path component
316 # (e.g. b'refs/tags/' + b'/' + b'v1.0' -> b'refs/tags//v1.0').
317 base = Ref(base.rstrip(b"/"))
318 if prune:
319 to_delete = set(self.subkeys(base))
320 else:
321 to_delete = set()
322 for name, value in other.items():
323 if value is None:
324 to_delete.add(name)
325 else:
326 self.set_if_equals(
327 Ref(b"/".join((base, name))), None, value, message=message
328 )
329 if to_delete:
330 try:
331 to_delete.remove(name)
332 except KeyError:
333 pass
334 for ref in to_delete:
335 self.remove_if_equals(Ref(b"/".join((base, ref))), None, message=message)
337 def allkeys(self) -> set[Ref]:
338 """All refs present in this container."""
339 raise NotImplementedError(self.allkeys)
341 def __iter__(self) -> Iterator[Ref]:
342 """Iterate over all reference keys."""
343 return iter(self.allkeys())
345 def keys(self, base: Ref | None = None) -> set[Ref]:
346 """Refs present in this container.
348 Args:
349 base: An optional base to return refs under.
350 Returns: An unsorted set of valid refs in this container, including
351 packed refs.
352 """
353 if base is not None:
354 return self.subkeys(base)
355 else:
356 return self.allkeys()
358 def subkeys(self, base: Ref) -> set[Ref]:
359 """Refs present in this container under a base.
361 Args:
362 base: The base to return refs under.
363 Returns: A set of valid refs in this container under the base; the base
364 prefix is stripped from the ref names returned.
365 """
366 keys: set[Ref] = set()
367 base_len = len(base) + 1
368 for refname in self.allkeys():
369 if refname.startswith(base):
370 keys.add(Ref(refname[base_len:]))
371 return keys
373 def as_dict(self, base: Ref | None = None) -> dict[Ref, ObjectID]:
374 """Return the contents of this container as a dictionary."""
375 ret: dict[Ref, ObjectID] = {}
376 keys = self.keys(base)
377 base_bytes: bytes
378 if base is None:
379 base_bytes = b""
380 else:
381 base_bytes = base.rstrip(b"/")
382 for key in keys:
383 try:
384 ret[key] = self[Ref((base_bytes + b"/" + key).strip(b"/"))]
385 except (SymrefLoop, KeyError):
386 continue # Unable to resolve
388 return ret
390 def _check_refname(self, name: Ref) -> None:
391 """Ensure a refname is valid and lives in refs or is HEAD.
393 HEAD is not a valid refname according to git-check-ref-format, but this
394 class needs to be able to touch HEAD. Also, check_ref_format expects
395 refnames without the leading 'refs/', but this class requires that
396 so it cannot touch anything outside the refs dir (or HEAD).
398 Args:
399 name: The name of the reference.
401 Raises:
402 KeyError: if a refname is not HEAD or is otherwise not valid.
403 """
404 if name in (HEADREF, Ref(b"refs/stash")):
405 return
406 if not name.startswith(b"refs/"):
407 raise RefFormatError(name)
408 rest = Ref(name[5:])
409 if check_ref_format(rest):
410 return
411 # As of Dulwich 1.2.3 check_ref_format rejects empty path components
412 # (e.g. b'refs/tags//v1.0'). Such names were silently accepted before,
413 # and some callers (e.g. older Poetry releases) still construct them.
414 # Warn rather than raise for now if collapsing repeated slashes would
415 # make the name valid, so the only defect is empty components.
416 if (
417 b"//" in name # type: ignore[comparison-overlap]
418 and check_ref_format(Ref(_collapse_slashes(rest)))
419 ):
420 warnings.warn(
421 f"Ref name {name!r} contains empty path components; "
422 "this will be rejected in a future version of Dulwich.",
423 DeprecationWarning,
424 stacklevel=3,
425 )
426 return
427 raise RefFormatError(name)
429 def _check_ref_value(self, ref: ObjectID) -> None:
430 """Ensure a ref value is a valid object id or a symref.
432 Args:
433 ref: The value a ref is being set to.
435 Raises:
436 ValueError: if the value is neither a valid sha nor a symref.
437 """
438 if not (valid_hexsha(ref) or ref.startswith(SYMREF)):
439 raise ValueError(f"{ref!r} must be a valid sha or a symref")
441 def read_ref(self, refname: Ref) -> bytes | None:
442 """Read a reference without following any references.
444 Args:
445 refname: The name of the reference
446 Returns: The contents of the ref file, or None if it does
447 not exist.
448 """
449 contents = self.read_loose_ref(refname)
450 if not contents:
451 contents = self.get_packed_refs().get(refname, None)
452 return contents
454 def read_loose_ref(self, name: Ref) -> bytes | None:
455 """Read a loose reference and return its contents.
457 Args:
458 name: the refname to read
459 Returns: The contents of the ref file, or None if it does
460 not exist.
461 """
462 raise NotImplementedError(self.read_loose_ref)
464 def follow(self, name: Ref) -> tuple[list[Ref], ObjectID | None]:
465 """Follow a reference name.
467 Returns: a tuple of (refnames, sha), wheres refnames are the names of
468 references in the chain
469 """
470 contents: bytes | None = SYMREF + name
471 depth = 0
472 refnames: list[Ref] = []
473 while contents and contents.startswith(SYMREF):
474 refname = Ref(contents[len(SYMREF) :])
475 refnames.append(refname)
476 contents = self.read_ref(refname)
477 if not contents:
478 break
479 depth += 1
480 if depth > 5:
481 raise SymrefLoop(name, depth)
482 return refnames, ObjectID(contents) if contents else None
484 def __contains__(self, refname: Ref) -> bool:
485 """Check if a reference exists."""
486 if self.read_ref(refname):
487 return True
488 return False
490 def __getitem__(self, name: Ref) -> ObjectID:
491 """Get the SHA1 for a reference name.
493 This method follows all symbolic references.
494 """
495 _, sha = self.follow(name)
496 if sha is None:
497 raise KeyError(name)
498 return sha
500 def set_if_equals(
501 self,
502 name: Ref,
503 old_ref: ObjectID | None,
504 new_ref: ObjectID,
505 committer: bytes | None = None,
506 timestamp: int | None = None,
507 timezone: int | None = None,
508 message: bytes | None = None,
509 ) -> bool:
510 """Set a refname to new_ref only if it currently equals old_ref.
512 This method follows all symbolic references if applicable for the
513 subclass, and can be used to perform an atomic compare-and-swap
514 operation.
516 Args:
517 name: The refname to set.
518 old_ref: The old sha the refname must refer to, or None to set
519 unconditionally.
520 new_ref: The new sha the refname will refer to.
521 committer: Optional committer name/email
522 timestamp: Optional timestamp
523 timezone: Optional timezone
524 message: Message for reflog
525 Returns: True if the set was successful, False otherwise.
526 """
527 raise NotImplementedError(self.set_if_equals)
529 def add_if_new(
530 self,
531 name: Ref,
532 ref: ObjectID,
533 committer: bytes | None = None,
534 timestamp: int | None = None,
535 timezone: int | None = None,
536 message: bytes | None = None,
537 ) -> bool:
538 """Add a new reference only if it does not already exist.
540 Args:
541 name: Ref name
542 ref: Ref value
543 committer: Optional committer name/email
544 timestamp: Optional timestamp
545 timezone: Optional timezone
546 message: Optional message for reflog
547 """
548 raise NotImplementedError(self.add_if_new)
550 def __setitem__(self, name: Ref, ref: ObjectID) -> None:
551 """Set a reference name to point to the given SHA1.
553 This method follows all symbolic references if applicable for the
554 subclass.
556 Note: This method unconditionally overwrites the contents of a
557 reference. To update atomically only if the reference has not
558 changed, use set_if_equals().
560 Args:
561 name: The refname to set.
562 ref: The new sha the refname will refer to.
563 """
564 self._check_ref_value(ref)
565 self.set_if_equals(name, None, ref)
567 def remove_if_equals(
568 self,
569 name: Ref,
570 old_ref: ObjectID | None,
571 committer: bytes | None = None,
572 timestamp: int | None = None,
573 timezone: int | None = None,
574 message: bytes | None = None,
575 ) -> bool:
576 """Remove a refname only if it currently equals old_ref.
578 This method does not follow symbolic references, even if applicable for
579 the subclass. It can be used to perform an atomic compare-and-delete
580 operation.
582 Args:
583 name: The refname to delete.
584 old_ref: The old sha the refname must refer to, or None to
585 delete unconditionally.
586 committer: Optional committer name/email
587 timestamp: Optional timestamp
588 timezone: Optional timezone
589 message: Message for reflog
590 Returns: True if the delete was successful, False otherwise.
591 """
592 raise NotImplementedError(self.remove_if_equals)
594 def __delitem__(self, name: Ref) -> None:
595 """Remove a refname.
597 This method does not follow symbolic references, even if applicable for
598 the subclass.
600 Note: This method unconditionally deletes the contents of a reference.
601 To delete atomically only if the reference has not changed, use
602 remove_if_equals().
604 Args:
605 name: The refname to delete.
606 """
607 self.remove_if_equals(name, None)
609 def get_symrefs(self) -> dict[Ref, Ref]:
610 """Get a dict with all symrefs in this container.
612 Returns: Dictionary mapping source ref to target ref
613 """
614 ret: dict[Ref, Ref] = {}
615 for src in self.allkeys():
616 try:
617 ref_value = self.read_ref(src)
618 assert ref_value is not None
619 dst = parse_symref_value(ref_value)
620 except ValueError:
621 pass
622 else:
623 ret[src] = Ref(dst)
624 return ret
626 def pack_refs(self, all: bool = False) -> None:
627 """Pack loose refs into packed-refs file.
629 Args:
630 all: If True, pack all refs. If False, only pack tags.
631 """
632 raise NotImplementedError(self.pack_refs)
635class DictRefsContainer(RefsContainer):
636 """RefsContainer backed by a simple dict.
638 This container does not support symbolic or packed references and is not
639 threadsafe.
640 """
642 def __init__(
643 self,
644 refs: dict[Ref, bytes],
645 logger: Callable[
646 [
647 bytes,
648 bytes | None,
649 bytes | None,
650 bytes | None,
651 int | None,
652 int | None,
653 bytes | None,
654 ],
655 None,
656 ]
657 | None = None,
658 ) -> None:
659 """Initialize DictRefsContainer with refs dictionary and optional logger."""
660 super().__init__(logger=logger)
661 self._refs = refs
662 self._peeled: dict[Ref, ObjectID] = {}
663 self._watchers: set[Any] = set()
665 def allkeys(self) -> set[Ref]:
666 """Return all reference keys."""
667 return set(self._refs.keys())
669 def read_loose_ref(self, name: Ref) -> bytes | None:
670 """Read a loose reference."""
671 return self._refs.get(name, None)
673 def get_packed_refs(self) -> dict[Ref, ObjectID]:
674 """Get packed references."""
675 return {}
677 def _notify(self, ref: bytes, newsha: bytes | None) -> None:
678 for watcher in self._watchers:
679 watcher._notify((ref, newsha))
681 def set_symbolic_ref(
682 self,
683 name: Ref,
684 other: Ref,
685 committer: bytes | None = None,
686 timestamp: int | None = None,
687 timezone: int | None = None,
688 message: bytes | None = None,
689 ) -> None:
690 """Make a ref point at another ref.
692 Args:
693 name: Name of the ref to set
694 other: Name of the ref to point at
695 committer: Optional committer name for reflog
696 timestamp: Optional timestamp for reflog
697 timezone: Optional timezone for reflog
698 message: Optional message for reflog
699 """
700 old = self.follow(name)[-1]
701 new = SYMREF + other
702 self._refs[name] = new
703 self._notify(name, new)
704 self._log(
705 name,
706 old,
707 new,
708 committer=committer,
709 timestamp=timestamp,
710 timezone=timezone,
711 message=message,
712 )
714 def set_if_equals(
715 self,
716 name: Ref,
717 old_ref: ObjectID | None,
718 new_ref: ObjectID,
719 committer: bytes | None = None,
720 timestamp: int | None = None,
721 timezone: int | None = None,
722 message: bytes | None = None,
723 ) -> bool:
724 """Set a refname to new_ref only if it currently equals old_ref.
726 This method follows all symbolic references, and can be used to perform
727 an atomic compare-and-swap operation.
729 Args:
730 name: The refname to set.
731 old_ref: The old sha the refname must refer to, or None to set
732 unconditionally.
733 new_ref: The new sha the refname will refer to.
734 committer: Optional committer name for reflog
735 timestamp: Optional timestamp for reflog
736 timezone: Optional timezone for reflog
737 message: Optional message for reflog
739 Returns:
740 True if the set was successful, False otherwise.
741 """
742 self._check_ref_value(new_ref)
743 if old_ref is not None and self._refs.get(name, ZERO_SHA) != old_ref:
744 return False
745 # Only update the specific ref requested, not the whole chain
746 self._check_refname(name)
747 old = self._refs.get(name)
748 self._refs[name] = new_ref
749 self._notify(name, new_ref)
750 self._log(
751 name,
752 old,
753 new_ref,
754 committer=committer,
755 timestamp=timestamp,
756 timezone=timezone,
757 message=message,
758 )
759 return True
761 def add_if_new(
762 self,
763 name: Ref,
764 ref: ObjectID,
765 committer: bytes | None = None,
766 timestamp: int | None = None,
767 timezone: int | None = None,
768 message: bytes | None = None,
769 ) -> bool:
770 """Add a new reference only if it does not already exist.
772 Args:
773 name: Ref name
774 ref: Ref value
775 committer: Optional committer name for reflog
776 timestamp: Optional timestamp for reflog
777 timezone: Optional timezone for reflog
778 message: Optional message for reflog
780 Returns:
781 True if the add was successful, False otherwise.
782 """
783 self._check_ref_value(ref)
784 if name in self._refs:
785 return False
786 self._refs[name] = ref
787 self._notify(name, ref)
788 self._log(
789 name,
790 None,
791 ref,
792 committer=committer,
793 timestamp=timestamp,
794 timezone=timezone,
795 message=message,
796 )
797 return True
799 def remove_if_equals(
800 self,
801 name: Ref,
802 old_ref: ObjectID | None,
803 committer: bytes | None = None,
804 timestamp: int | None = None,
805 timezone: int | None = None,
806 message: bytes | None = None,
807 ) -> bool:
808 """Remove a refname only if it currently equals old_ref.
810 This method does not follow symbolic references. It can be used to
811 perform an atomic compare-and-delete operation.
813 Args:
814 name: The refname to delete.
815 old_ref: The old sha the refname must refer to, or None to
816 delete unconditionally.
817 committer: Optional committer name for reflog
818 timestamp: Optional timestamp for reflog
819 timezone: Optional timezone for reflog
820 message: Optional message for reflog
822 Returns:
823 True if the delete was successful, False otherwise.
824 """
825 if old_ref is not None and self._refs.get(name, ZERO_SHA) != old_ref:
826 return False
827 try:
828 old = self._refs.pop(name)
829 except KeyError:
830 pass
831 else:
832 self._notify(name, None)
833 self._log(
834 name,
835 old,
836 None,
837 committer=committer,
838 timestamp=timestamp,
839 timezone=timezone,
840 message=message,
841 )
842 return True
844 def get_peeled(self, name: Ref) -> ObjectID | None:
845 """Get peeled version of a reference."""
846 return self._peeled.get(name)
848 def _update(self, refs: Mapping[Ref, ObjectID]) -> None:
849 """Update multiple refs; intended only for testing."""
850 # TODO(dborowitz): replace this with a public function that uses
851 # set_if_equal.
852 for ref, sha in refs.items():
853 self.set_if_equals(ref, None, sha)
855 def _update_peeled(self, peeled: Mapping[Ref, ObjectID]) -> None:
856 """Update cached peeled refs; intended only for testing."""
857 self._peeled.update(peeled)
860class DiskRefsContainer(RefsContainer):
861 """Refs container that reads refs from disk."""
863 def __init__(
864 self,
865 path: str | bytes | os.PathLike[str],
866 worktree_path: str | bytes | os.PathLike[str] | None = None,
867 logger: Callable[
868 [bytes, bytes, bytes, bytes | None, int | None, int | None, bytes], None
869 ]
870 | None = None,
871 ) -> None:
872 """Initialize DiskRefsContainer."""
873 super().__init__(logger=logger)
874 # Convert path-like objects to strings, then to bytes for Git compatibility
875 self.path = os.fsencode(os.fspath(path))
876 if worktree_path is None:
877 self.worktree_path = self.path
878 else:
879 self.worktree_path = os.fsencode(os.fspath(worktree_path))
880 self._packed_refs: dict[Ref, ObjectID] | None = None
881 self._peeled_refs: dict[Ref, ObjectID] | None = None
883 def __repr__(self) -> str:
884 """Return string representation of DiskRefsContainer."""
885 return f"{self.__class__.__name__}({self.path!r})"
887 def _iter_dir(
888 self,
889 path: bytes,
890 base: bytes,
891 dir_filter: Callable[[bytes], bool] | None = None,
892 ) -> Iterator[Ref]:
893 refspath = os.path.join(path, base.rstrip(b"/"))
894 prefix_len = len(os.path.join(path, b""))
896 for root, dirs, files in os.walk(refspath):
897 directory = root[prefix_len:]
898 if os.path.sep != "/":
899 directory = directory.replace(os.fsencode(os.path.sep), b"/")
900 if dir_filter is not None:
901 dirs[:] = [
902 d for d in dirs if dir_filter(b"/".join([directory, d, b""]))
903 ]
905 for filename in files:
906 refname = b"/".join([directory, filename])
907 if check_ref_format(Ref(refname)):
908 yield Ref(refname)
910 def _iter_loose_refs(self, base: bytes = b"refs/") -> Iterator[Ref]:
911 base = base.rstrip(b"/") + b"/"
912 search_paths: list[tuple[bytes, Callable[[bytes], bool] | None]] = []
913 if base != b"refs/":
914 path = self.worktree_path if is_per_worktree_ref(base) else self.path
915 search_paths.append((path, None))
916 elif self.worktree_path == self.path:
917 # Iterate through all the refs from the main worktree
918 search_paths.append((self.path, None))
919 else:
920 # Iterate through all the shared refs from the commondir, excluding per-worktree refs
921 search_paths.append((self.path, lambda r: not is_per_worktree_ref(r)))
922 # Iterate through all the per-worktree refs from the worktree's gitdir
923 search_paths.append((self.worktree_path, is_per_worktree_ref))
925 for path, dir_filter in search_paths:
926 yield from self._iter_dir(path, base, dir_filter=dir_filter)
928 def subkeys(self, base: Ref) -> set[Ref]:
929 """Return subkeys under a given base reference path."""
930 subkeys: set[Ref] = set()
932 for key in self._iter_loose_refs(base):
933 if key.startswith(base):
934 subkeys.add(Ref(key[len(base) :].strip(b"/")))
936 for key in self.get_packed_refs():
937 if key.startswith(base):
938 subkeys.add(Ref(key[len(base) :].strip(b"/")))
939 return subkeys
941 def allkeys(self) -> set[Ref]:
942 """Return all reference keys."""
943 allkeys: set[Ref] = set()
944 if os.path.exists(self.refpath(HEADREF)):
945 allkeys.add(Ref(HEADREF))
947 allkeys.update(self._iter_loose_refs())
948 allkeys.update(self.get_packed_refs())
949 return allkeys
951 def refpath(self, name: bytes) -> bytes:
952 """Return the disk path of a ref."""
953 path = name
954 if os.path.sep != "/":
955 path = path.replace(b"/", os.fsencode(os.path.sep))
957 root_dir = self.worktree_path if is_per_worktree_ref(name) else self.path
958 return os.path.join(root_dir, path)
960 def get_packed_refs(self) -> dict[Ref, ObjectID]:
961 """Get contents of the packed-refs file.
963 Returns: Dictionary mapping ref names to SHA1s
965 Note: Will return an empty dictionary when no packed-refs file is
966 present.
967 """
968 # TODO: invalidate the cache on repacking
969 if self._packed_refs is None:
970 # set both to empty because we want _peeled_refs to be
971 # None if and only if _packed_refs is also None.
972 self._packed_refs = {}
973 self._peeled_refs = {}
974 path = os.path.join(self.path, b"packed-refs")
975 try:
976 f = GitFile(path, "rb")
977 except FileNotFoundError:
978 return {}
979 with f:
980 first_line = next(iter(f)).rstrip()
981 if first_line.startswith(b"# pack-refs") and b" peeled" in first_line:
982 for sha, name, peeled in read_packed_refs_with_peeled(f):
983 self._packed_refs[name] = sha
984 if peeled:
985 self._peeled_refs[name] = peeled
986 else:
987 f.seek(0)
988 for sha, name in read_packed_refs(f):
989 self._packed_refs[name] = sha
990 return self._packed_refs
992 def add_packed_refs(self, new_refs: Mapping[Ref, ObjectID | None]) -> None:
993 """Add the given refs as packed refs.
995 Args:
996 new_refs: A mapping of ref names to targets; if a target is None that
997 means remove the ref
998 """
999 if not new_refs:
1000 return
1002 path = os.path.join(self.path, b"packed-refs")
1004 with GitFile(path, "wb") as f:
1005 # reread cached refs from disk, while holding the lock
1006 packed_refs = self.get_packed_refs().copy()
1008 for ref, target in new_refs.items():
1009 # sanity check
1010 if ref == HEADREF:
1011 raise ValueError("cannot pack HEAD")
1013 # remove any loose refs pointing to this one -- please
1014 # note that this bypasses remove_if_equals as we don't
1015 # want to affect packed refs in here
1016 with suppress(OSError):
1017 os.remove(self.refpath(ref))
1019 if target is not None:
1020 packed_refs[ref] = target
1021 else:
1022 packed_refs.pop(ref, None)
1024 write_packed_refs(f, packed_refs, self._peeled_refs)
1026 self._packed_refs = packed_refs
1028 def get_peeled(self, name: Ref) -> ObjectID | None:
1029 """Return the cached peeled value of a ref, if available.
1031 Args:
1032 name: Name of the ref to peel
1033 Returns: The peeled value of the ref. If the ref is known not point to
1034 a tag, this will be the SHA the ref refers to. If the ref may point
1035 to a tag, but no cached information is available, None is returned.
1036 """
1037 self.get_packed_refs()
1038 if (
1039 self._peeled_refs is None
1040 or self._packed_refs is None
1041 or name not in self._packed_refs
1042 ):
1043 # No cache: no peeled refs were read, or this ref is loose
1044 return None
1045 if name in self._peeled_refs:
1046 return self._peeled_refs[name]
1047 else:
1048 # Known not peelable
1049 return self[name]
1051 def read_loose_ref(self, name: Ref) -> bytes | None:
1052 """Read a reference file and return its contents.
1054 If the reference file a symbolic reference, only read the first line of
1055 the file. Otherwise, read the hash (40 bytes for SHA1, 64 bytes for SHA256).
1057 Args:
1058 name: the refname to read, relative to refpath
1059 Returns: The contents of the ref file, or None if the file does not
1060 exist.
1062 Raises:
1063 IOError: if any other error occurs
1064 """
1065 # Validate the name before turning it into a path.
1066 try:
1067 self._check_refname(name)
1068 except RefFormatError:
1069 return None
1070 filename = self.refpath(name)
1071 try:
1072 with GitFile(filename, "rb") as f:
1073 header = f.read(len(SYMREF))
1074 if header == SYMREF:
1075 # Read only the first line
1076 return header + next(iter(f)).rstrip(b"\r\n")
1077 else:
1078 # Read the entire line to get the full hash (handles both SHA1 and SHA256)
1079 f.seek(0)
1080 line = f.readline().rstrip(b"\r\n")
1081 return line
1082 except (OSError, UnicodeError):
1083 # don't assume anything specific about the error; in
1084 # particular, invalid or forbidden paths can raise weird
1085 # errors depending on the specific operating system
1086 return None
1088 def _remove_packed_ref(self, name: Ref) -> None:
1089 if self._packed_refs is None:
1090 return
1091 filename = os.path.join(self.path, b"packed-refs")
1092 # reread cached refs from disk, while holding the lock
1093 f = GitFile(filename, "wb")
1094 try:
1095 self._packed_refs = None
1096 self.get_packed_refs()
1098 if self._packed_refs is None or name not in self._packed_refs:
1099 f.abort()
1100 return
1102 del self._packed_refs[name]
1103 if self._peeled_refs is not None:
1104 with suppress(KeyError):
1105 del self._peeled_refs[name]
1106 write_packed_refs(f, self._packed_refs, self._peeled_refs)
1107 f.close()
1108 except BaseException:
1109 f.abort()
1110 raise
1112 def set_symbolic_ref(
1113 self,
1114 name: Ref,
1115 other: Ref,
1116 committer: bytes | None = None,
1117 timestamp: int | None = None,
1118 timezone: int | None = None,
1119 message: bytes | None = None,
1120 ) -> None:
1121 """Make a ref point at another ref.
1123 Args:
1124 name: Name of the ref to set
1125 other: Name of the ref to point at
1126 committer: Optional committer name
1127 timestamp: Optional timestamp
1128 timezone: Optional timezone
1129 message: Optional message to describe the change
1130 """
1131 self._check_refname(name)
1132 self._check_refname(other)
1133 filename = self.refpath(name)
1134 f = GitFile(filename, "wb")
1135 try:
1136 f.write(SYMREF + other + b"\n")
1137 sha = self.follow(name)[-1]
1138 self._log(
1139 name,
1140 sha,
1141 sha,
1142 committer=committer,
1143 timestamp=timestamp,
1144 timezone=timezone,
1145 message=message,
1146 )
1147 except BaseException:
1148 f.abort()
1149 raise
1150 else:
1151 f.close()
1153 def set_if_equals(
1154 self,
1155 name: Ref,
1156 old_ref: ObjectID | None,
1157 new_ref: ObjectID,
1158 committer: bytes | None = None,
1159 timestamp: int | None = None,
1160 timezone: int | None = None,
1161 message: bytes | None = None,
1162 ) -> bool:
1163 """Set a refname to new_ref only if it currently equals old_ref.
1165 This method follows all symbolic references, and can be used to perform
1166 an atomic compare-and-swap operation.
1168 Args:
1169 name: The refname to set.
1170 old_ref: The old sha the refname must refer to, or None to set
1171 unconditionally.
1172 new_ref: The new sha the refname will refer to.
1173 committer: Optional committer name
1174 timestamp: Optional timestamp
1175 timezone: Optional timezone
1176 message: Set message for reflog
1177 Returns: True if the set was successful, False otherwise.
1178 """
1179 self._check_refname(name)
1180 self._check_ref_value(new_ref)
1181 try:
1182 realnames, _ = self.follow(name)
1183 realname = realnames[-1]
1184 except (KeyError, IndexError, SymrefLoop):
1185 realname = name
1186 filename = self.refpath(realname)
1188 # make sure none of the ancestor folders is in packed refs
1189 probe_ref = Ref(os.path.dirname(realname))
1190 packed_refs = self.get_packed_refs()
1191 while probe_ref:
1192 if packed_refs.get(probe_ref, None) is not None:
1193 raise NotADirectoryError(filename)
1194 probe_ref = Ref(os.path.dirname(probe_ref))
1196 ensure_dir_exists(os.path.dirname(filename))
1197 with GitFile(filename, "wb") as f:
1198 if old_ref is not None:
1199 try:
1200 # read again while holding the lock to handle race conditions
1201 orig_ref = self.read_loose_ref(realname)
1202 if orig_ref is None:
1203 orig_ref = self.get_packed_refs().get(realname, ZERO_SHA)
1204 if orig_ref != old_ref:
1205 f.abort()
1206 return False
1207 except OSError:
1208 f.abort()
1209 raise
1211 # Check if ref already has the desired value while holding the lock
1212 # This avoids fsync when ref is unchanged but still detects lock conflicts
1213 current_ref = self.read_loose_ref(realname)
1214 if current_ref is None:
1215 current_ref = packed_refs.get(realname, None)
1217 if current_ref is not None and current_ref == new_ref:
1218 # Ref already has desired value, abort write to avoid fsync
1219 f.abort()
1220 return True
1222 try:
1223 f.write(new_ref + b"\n")
1224 except OSError:
1225 f.abort()
1226 raise
1227 self._log(
1228 realname,
1229 old_ref,
1230 new_ref,
1231 committer=committer,
1232 timestamp=timestamp,
1233 timezone=timezone,
1234 message=message,
1235 )
1236 return True
1238 def add_if_new(
1239 self,
1240 name: Ref,
1241 ref: ObjectID,
1242 committer: bytes | None = None,
1243 timestamp: int | None = None,
1244 timezone: int | None = None,
1245 message: bytes | None = None,
1246 ) -> bool:
1247 """Add a new reference only if it does not already exist.
1249 This method follows symrefs, and only ensures that the last ref in the
1250 chain does not exist.
1252 Args:
1253 name: The refname to set.
1254 ref: The new sha the refname will refer to.
1255 committer: Optional committer name
1256 timestamp: Optional timestamp
1257 timezone: Optional timezone
1258 message: Optional message for reflog
1259 Returns: True if the add was successful, False otherwise.
1260 """
1261 self._check_ref_value(ref)
1262 try:
1263 realnames, contents = self.follow(name)
1264 if contents is not None:
1265 return False
1266 realname = realnames[-1]
1267 except (KeyError, IndexError):
1268 realname = name
1269 self._check_refname(realname)
1270 filename = self.refpath(realname)
1271 ensure_dir_exists(os.path.dirname(filename))
1272 with GitFile(filename, "wb") as f:
1273 if os.path.exists(filename) or name in self.get_packed_refs():
1274 f.abort()
1275 return False
1276 try:
1277 f.write(ref + b"\n")
1278 except OSError:
1279 f.abort()
1280 raise
1281 else:
1282 self._log(
1283 name,
1284 None,
1285 ref,
1286 committer=committer,
1287 timestamp=timestamp,
1288 timezone=timezone,
1289 message=message,
1290 )
1291 return True
1293 def remove_if_equals(
1294 self,
1295 name: Ref,
1296 old_ref: ObjectID | None,
1297 committer: bytes | None = None,
1298 timestamp: int | None = None,
1299 timezone: int | None = None,
1300 message: bytes | None = None,
1301 ) -> bool:
1302 """Remove a refname only if it currently equals old_ref.
1304 This method does not follow symbolic references. It can be used to
1305 perform an atomic compare-and-delete operation.
1307 Args:
1308 name: The refname to delete.
1309 old_ref: The old sha the refname must refer to, or None to
1310 delete unconditionally.
1311 committer: Optional committer name
1312 timestamp: Optional timestamp
1313 timezone: Optional timezone
1314 message: Optional message
1315 Returns: True if the delete was successful, False otherwise.
1316 """
1317 self._check_refname(name)
1318 filename = self.refpath(name)
1319 ensure_dir_exists(os.path.dirname(filename))
1320 f = GitFile(filename, "wb")
1321 try:
1322 if old_ref is not None:
1323 orig_ref = self.read_loose_ref(name)
1324 if orig_ref is None:
1325 orig_ref = self.get_packed_refs().get(name)
1326 if orig_ref is None:
1327 orig_ref = ZERO_SHA
1328 if orig_ref != old_ref:
1329 return False
1331 # remove the reference file itself
1332 try:
1333 found = os.path.lexists(filename)
1334 except OSError:
1335 # may only be packed, or otherwise unstorable
1336 found = False
1338 if found:
1339 os.remove(filename)
1341 self._remove_packed_ref(name)
1342 self._log(
1343 name,
1344 old_ref,
1345 None,
1346 committer=committer,
1347 timestamp=timestamp,
1348 timezone=timezone,
1349 message=message,
1350 )
1351 finally:
1352 # never write, we just wanted the lock
1353 f.abort()
1355 # outside of the lock, clean-up any parent directory that might now
1356 # be empty. this ensures that re-creating a reference of the same
1357 # name of what was previously a directory works as expected
1358 parent = name
1359 while True:
1360 try:
1361 parent_bytes, _ = parent.rsplit(b"/", 1)
1362 parent = Ref(parent_bytes)
1363 except ValueError:
1364 break
1366 if parent == b"refs":
1367 break
1368 parent_filename = self.refpath(parent)
1369 try:
1370 os.rmdir(parent_filename)
1371 except OSError:
1372 # this can be caused by the parent directory being
1373 # removed by another process, being not empty, etc.
1374 # in any case, this is non fatal because we already
1375 # removed the reference, just ignore it
1376 break
1378 return True
1380 def pack_refs(self, all: bool = False) -> None:
1381 """Pack loose refs into packed-refs file.
1383 Args:
1384 all: If True, pack all refs. If False, only pack tags.
1385 """
1386 refs_to_pack: dict[Ref, ObjectID | None] = {}
1387 for ref in self.allkeys():
1388 if ref == HEADREF:
1389 # Never pack HEAD
1390 continue
1391 if all or ref.startswith(LOCAL_TAG_PREFIX):
1392 try:
1393 sha = self[ref]
1394 if sha:
1395 refs_to_pack[ref] = sha
1396 except KeyError:
1397 # Broken ref, skip it
1398 pass
1400 if refs_to_pack:
1401 self.add_packed_refs(refs_to_pack)
1404def _split_ref_line(line: bytes) -> tuple[ObjectID, Ref]:
1405 """Split a single ref line into a tuple of SHA1 and name."""
1406 fields = line.rstrip(b"\n\r").split(b" ")
1407 if len(fields) != 2:
1408 raise PackedRefsException(f"invalid ref line {line!r}")
1409 sha, name = fields
1410 if not valid_hexsha(sha):
1411 raise PackedRefsException(f"Invalid hex sha {sha!r}")
1412 if not check_ref_format(Ref(name)):
1413 raise PackedRefsException(f"invalid ref name {name!r}")
1414 return (ObjectID(sha), Ref(name))
1417def read_packed_refs(f: IO[bytes]) -> Iterator[tuple[ObjectID, Ref]]:
1418 """Read a packed refs file.
1420 Args:
1421 f: file-like object to read from
1422 Returns: Iterator over tuples with SHA1s and ref names.
1423 """
1424 for line in f:
1425 if line.startswith(b"#"):
1426 # Comment
1427 continue
1428 if line.startswith(b"^"):
1429 raise PackedRefsException("found peeled ref in packed-refs without peeled")
1430 yield _split_ref_line(line)
1433def read_packed_refs_with_peeled(
1434 f: IO[bytes],
1435) -> Iterator[tuple[ObjectID, Ref, ObjectID | None]]:
1436 """Read a packed refs file including peeled refs.
1438 Assumes the "# pack-refs with: peeled" line was already read. Yields tuples
1439 with ref names, SHA1s, and peeled SHA1s (or None).
1441 Args:
1442 f: file-like object to read from, seek'ed to the second line
1443 """
1444 last = None
1445 for line in f:
1446 if line.startswith(b"#"):
1447 continue
1448 line = line.rstrip(b"\r\n")
1449 if line.startswith(b"^"):
1450 if not last:
1451 raise PackedRefsException("unexpected peeled ref line")
1452 if not valid_hexsha(line[1:]):
1453 raise PackedRefsException(f"Invalid hex sha {line[1:]!r}")
1454 sha, name = _split_ref_line(last)
1455 last = None
1456 yield (sha, name, ObjectID(line[1:]))
1457 else:
1458 if last:
1459 sha, name = _split_ref_line(last)
1460 yield (sha, name, None)
1461 last = line
1462 if last:
1463 sha, name = _split_ref_line(last)
1464 yield (sha, name, None)
1467def write_packed_refs(
1468 f: IO[bytes],
1469 packed_refs: Mapping[Ref, ObjectID],
1470 peeled_refs: Mapping[Ref, ObjectID] | None = None,
1471) -> None:
1472 """Write a packed refs file.
1474 Args:
1475 f: empty file-like object to write to
1476 packed_refs: dict of refname to sha of packed refs to write
1477 peeled_refs: dict of refname to peeled value of sha
1478 """
1479 if peeled_refs is None:
1480 peeled_refs = {}
1481 else:
1482 f.write(b"# pack-refs with: peeled\n")
1483 for refname in sorted(packed_refs.keys()):
1484 f.write(git_line(packed_refs[refname], refname))
1485 if refname in peeled_refs:
1486 f.write(b"^" + peeled_refs[refname] + b"\n")
1489def read_info_refs(f: BinaryIO) -> dict[Ref, ObjectID]:
1490 """Read info/refs file.
1492 Args:
1493 f: File-like object to read from
1495 Returns:
1496 Dictionary mapping ref names to SHA1s
1497 """
1498 ret: dict[Ref, ObjectID] = {}
1499 for line_no, line in enumerate(f.readlines(), 1):
1500 stripped = line.rstrip(b"\r\n")
1501 parts = stripped.split(b"\t", 1)
1502 if len(parts) != 2:
1503 raise ValueError(
1504 f"Invalid info/refs format at line {line_no}: "
1505 f"expected '<sha>\\t<refname>', got {stripped[:100]!r}"
1506 )
1507 (sha, name) = parts
1508 ret[Ref(name)] = ObjectID(sha)
1509 return ret
1512def is_local_branch(x: bytes) -> bool:
1513 """Check if a ref name is a local branch."""
1514 return x.startswith(LOCAL_BRANCH_PREFIX)
1517def _strip_leading_slash(name: bytes, kind: str) -> bytes:
1518 """Strip a leading slash from a short ref name, warning if one is present.
1520 A leading slash here means the caller stripped a ref prefix incorrectly
1521 (e.g. used ``ref[len(b"refs/tags"):]`` instead of ``len(b"refs/tags/")``).
1522 Joining such a name with a prefix would produce a malformed ref with an
1523 empty path component. Warn rather than raise for now; this will become an
1524 error in a future release.
1525 """
1526 if not name.startswith(b"/"):
1527 return name
1528 warnings.warn(
1529 f"{kind} name must not start with a slash: {name!r}; "
1530 "this will be rejected in a future version of Dulwich.",
1531 DeprecationWarning,
1532 stacklevel=3,
1533 )
1534 return name.lstrip(b"/")
1537def local_branch_name(name: bytes) -> Ref:
1538 """Build a full branch ref from a short name.
1540 Args:
1541 name: Short branch name (e.g., b"master") or full ref
1543 Returns:
1544 Full branch ref name (e.g., b"refs/heads/master")
1546 Examples:
1547 >>> local_branch_name(b"master")
1548 b'refs/heads/master'
1549 >>> local_branch_name(b"refs/heads/master")
1550 b'refs/heads/master'
1551 """
1552 if name.startswith(LOCAL_BRANCH_PREFIX):
1553 return Ref(name)
1554 return Ref(LOCAL_BRANCH_PREFIX + _strip_leading_slash(name, "Branch"))
1557def local_tag_name(name: bytes) -> Ref:
1558 """Build a full tag ref from a short name.
1560 Args:
1561 name: Short tag name (e.g., b"v1.0") or full ref
1563 Returns:
1564 Full tag ref name (e.g., b"refs/tags/v1.0")
1566 Examples:
1567 >>> local_tag_name(b"v1.0")
1568 b'refs/tags/v1.0'
1569 >>> local_tag_name(b"refs/tags/v1.0")
1570 b'refs/tags/v1.0'
1571 """
1572 if name.startswith(LOCAL_TAG_PREFIX):
1573 return Ref(name)
1574 return Ref(LOCAL_TAG_PREFIX + _strip_leading_slash(name, "Tag"))
1577def local_replace_name(name: bytes) -> Ref:
1578 """Build a full replace ref from a short name.
1580 Args:
1581 name: Short replace name (object SHA) or full ref
1583 Returns:
1584 Full replace ref name (e.g., b"refs/replace/<sha>")
1586 Examples:
1587 >>> local_replace_name(b"abc123")
1588 b'refs/replace/abc123'
1589 >>> local_replace_name(b"refs/replace/abc123")
1590 b'refs/replace/abc123'
1591 """
1592 if name.startswith(LOCAL_REPLACE_PREFIX):
1593 return Ref(name)
1594 return Ref(LOCAL_REPLACE_PREFIX + _strip_leading_slash(name, "Replace"))
1597def extract_branch_name(ref: bytes) -> bytes:
1598 """Extract branch name from a full branch ref.
1600 Args:
1601 ref: Full branch ref (e.g., b"refs/heads/master")
1603 Returns:
1604 Short branch name (e.g., b"master")
1606 Raises:
1607 ValueError: If ref is not a local branch
1609 Examples:
1610 >>> extract_branch_name(b"refs/heads/master")
1611 b'master'
1612 >>> extract_branch_name(b"refs/heads/feature/foo")
1613 b'feature/foo'
1614 """
1615 if not ref.startswith(LOCAL_BRANCH_PREFIX):
1616 raise ValueError(f"Not a local branch ref: {ref!r}")
1617 return ref[len(LOCAL_BRANCH_PREFIX) :]
1620def extract_tag_name(ref: bytes) -> bytes:
1621 """Extract tag name from a full tag ref.
1623 Args:
1624 ref: Full tag ref (e.g., b"refs/tags/v1.0")
1626 Returns:
1627 Short tag name (e.g., b"v1.0")
1629 Raises:
1630 ValueError: If ref is not a local tag
1632 Examples:
1633 >>> extract_tag_name(b"refs/tags/v1.0")
1634 b'v1.0'
1635 """
1636 if not ref.startswith(LOCAL_TAG_PREFIX):
1637 raise ValueError(f"Not a local tag ref: {ref!r}")
1638 return ref[len(LOCAL_TAG_PREFIX) :]
1641def shorten_ref_name(ref: bytes) -> bytes:
1642 """Convert a full ref name to its short form.
1644 Args:
1645 ref: Full ref name (e.g., b"refs/heads/master")
1647 Returns:
1648 Short ref name (e.g., b"master")
1650 Examples:
1651 >>> shorten_ref_name(b"refs/heads/master")
1652 b'master'
1653 >>> shorten_ref_name(b"refs/remotes/origin/main")
1654 b'origin/main'
1655 >>> shorten_ref_name(b"refs/tags/v1.0")
1656 b'v1.0'
1657 >>> shorten_ref_name(b"HEAD")
1658 b'HEAD'
1659 """
1660 if ref.startswith(LOCAL_BRANCH_PREFIX):
1661 return ref[len(LOCAL_BRANCH_PREFIX) :]
1662 elif ref.startswith(LOCAL_REMOTE_PREFIX):
1663 return ref[len(LOCAL_REMOTE_PREFIX) :]
1664 elif ref.startswith(LOCAL_TAG_PREFIX):
1665 return ref[len(LOCAL_TAG_PREFIX) :]
1666 return ref
1669def _set_origin_head(
1670 refs: RefsContainer, origin: bytes, origin_head: bytes | None
1671) -> None:
1672 # set refs/remotes/origin/HEAD
1673 origin_base = b"refs/remotes/" + origin + b"/"
1674 if origin_head and origin_head.startswith(LOCAL_BRANCH_PREFIX):
1675 origin_ref = Ref(origin_base + HEADREF)
1676 target_ref = Ref(origin_base + extract_branch_name(origin_head))
1677 if target_ref in refs:
1678 refs.set_symbolic_ref(origin_ref, target_ref)
1681def _set_default_branch(
1682 refs: RefsContainer,
1683 origin: bytes,
1684 origin_head: bytes | None,
1685 branch: bytes | None,
1686 ref_message: bytes | None,
1687) -> bytes:
1688 """Set the default branch."""
1689 origin_base = b"refs/remotes/" + origin + b"/"
1690 if branch:
1691 origin_ref = Ref(origin_base + branch)
1692 if origin_ref in refs:
1693 local_ref = Ref(local_branch_name(branch))
1694 refs.add_if_new(local_ref, refs[origin_ref], ref_message)
1695 head_ref = local_ref
1696 elif Ref(local_tag_name(branch)) in refs:
1697 head_ref = Ref(local_tag_name(branch))
1698 else:
1699 raise ValueError(f"{os.fsencode(branch)!r} is not a valid branch or tag")
1700 elif origin_head:
1701 head_ref = Ref(origin_head)
1702 if origin_head.startswith(LOCAL_BRANCH_PREFIX):
1703 origin_ref = Ref(origin_base + extract_branch_name(origin_head))
1704 else:
1705 origin_ref = Ref(origin_head)
1706 try:
1707 refs.add_if_new(head_ref, refs[origin_ref], ref_message)
1708 except KeyError:
1709 pass
1710 else:
1711 raise ValueError("neither origin_head nor branch are provided")
1712 return head_ref
1715def _set_head(
1716 refs: RefsContainer, head_ref: bytes, ref_message: bytes | None
1717) -> ObjectID | None:
1718 if head_ref.startswith(LOCAL_TAG_PREFIX):
1719 # detach HEAD at specified tag
1720 head = refs[Ref(head_ref)]
1721 del refs[HEADREF]
1722 refs.set_if_equals(HEADREF, None, head, message=ref_message)
1723 else:
1724 # set HEAD to specific branch
1725 try:
1726 head = refs[Ref(head_ref)]
1727 refs.set_symbolic_ref(HEADREF, Ref(head_ref))
1728 refs.set_if_equals(HEADREF, None, head, message=ref_message)
1729 except KeyError:
1730 head = None
1731 return head
1734def _import_remote_refs(
1735 refs_container: RefsContainer,
1736 remote_name: str,
1737 refs: Mapping[Ref, ObjectID | None],
1738 message: bytes | None = None,
1739 prune: bool = False,
1740 prune_tags: bool = False,
1741) -> None:
1742 from .protocol import PEELED_TAG_SUFFIX, strip_peeled_refs
1744 stripped_refs = strip_peeled_refs(refs)
1745 branches: dict[Ref, ObjectID | None] = {
1746 Ref(extract_branch_name(n)): v
1747 for (n, v) in stripped_refs.items()
1748 if n.startswith(LOCAL_BRANCH_PREFIX)
1749 }
1750 refs_container.import_refs(
1751 Ref(b"refs/remotes/" + remote_name.encode()),
1752 branches,
1753 message=message,
1754 prune=prune,
1755 )
1756 tags: dict[Ref, ObjectID | None] = {
1757 Ref(extract_tag_name(n)): v
1758 for (n, v) in stripped_refs.items()
1759 if n.startswith(LOCAL_TAG_PREFIX) and not n.endswith(PEELED_TAG_SUFFIX)
1760 }
1761 refs_container.import_refs(
1762 Ref(b"refs/tags"), tags, message=message, prune=prune_tags
1763 )
1766class locked_ref:
1767 """Lock a ref while making modifications.
1769 Works as a context manager.
1770 """
1772 def __init__(self, refs_container: DiskRefsContainer, refname: Ref) -> None:
1773 """Initialize a locked ref.
1775 Args:
1776 refs_container: The DiskRefsContainer to lock the ref in
1777 refname: The ref name to lock
1778 """
1779 self._refs_container = refs_container
1780 self._refname = refname
1781 self._file: _GitFile | None = None
1782 self._realname: Ref | None = None
1783 self._deleted = False
1785 def __enter__(self) -> Self:
1786 """Enter the context manager and acquire the lock.
1788 Returns:
1789 This locked_ref instance
1791 Raises:
1792 OSError: If the lock cannot be acquired
1793 """
1794 self._refs_container._check_refname(self._refname)
1795 try:
1796 realnames, _ = self._refs_container.follow(self._refname)
1797 self._realname = realnames[-1]
1798 except (KeyError, IndexError, SymrefLoop):
1799 self._realname = self._refname
1801 filename = self._refs_container.refpath(self._realname)
1802 ensure_dir_exists(os.path.dirname(filename))
1803 f = GitFile(filename, "wb")
1804 self._file = f
1805 return self
1807 def __exit__(
1808 self,
1809 exc_type: type | None,
1810 exc_value: BaseException | None,
1811 traceback: types.TracebackType | None,
1812 ) -> None:
1813 """Exit the context manager and release the lock.
1815 Args:
1816 exc_type: Type of exception if one occurred
1817 exc_value: Exception instance if one occurred
1818 traceback: Traceback if an exception occurred
1819 """
1820 if self._file:
1821 if exc_type is not None or self._deleted:
1822 self._file.abort()
1823 else:
1824 self._file.close()
1826 def get(self) -> bytes | None:
1827 """Get the current value of the ref."""
1828 if not self._file:
1829 raise RuntimeError("locked_ref not in context")
1831 assert self._realname is not None
1832 current_ref = self._refs_container.read_loose_ref(self._realname)
1833 if current_ref is None:
1834 current_ref = self._refs_container.get_packed_refs().get(
1835 self._realname, None
1836 )
1837 return current_ref
1839 def ensure_equals(self, expected_value: bytes | None) -> bool:
1840 """Ensure the ref currently equals the expected value.
1842 Args:
1843 expected_value: The expected current value of the ref
1844 Returns:
1845 True if the ref equals the expected value, False otherwise
1846 """
1847 current_value = self.get()
1848 return current_value == expected_value
1850 def set(self, new_ref: bytes) -> None:
1851 """Set the ref to a new value.
1853 Args:
1854 new_ref: The new SHA1 or symbolic ref value
1855 """
1856 if not self._file:
1857 raise RuntimeError("locked_ref not in context")
1859 if not (valid_hexsha(new_ref) or new_ref.startswith(SYMREF)):
1860 raise ValueError(f"{new_ref!r} must be a valid sha or a symref")
1862 self._file.seek(0)
1863 self._file.truncate()
1864 self._file.write(new_ref + b"\n")
1865 self._deleted = False
1867 def set_symbolic_ref(self, target: Ref) -> None:
1868 """Make this ref point at another ref.
1870 Args:
1871 target: Name of the ref to point at
1872 """
1873 if not self._file:
1874 raise RuntimeError("locked_ref not in context")
1876 self._refs_container._check_refname(target)
1877 self._file.seek(0)
1878 self._file.truncate()
1879 self._file.write(SYMREF + target + b"\n")
1880 self._deleted = False
1882 def delete(self) -> None:
1883 """Delete the ref file while holding the lock."""
1884 if not self._file:
1885 raise RuntimeError("locked_ref not in context")
1887 # Delete the actual ref file while holding the lock
1888 if self._realname:
1889 filename = self._refs_container.refpath(self._realname)
1890 try:
1891 if os.path.lexists(filename):
1892 os.remove(filename)
1893 except FileNotFoundError:
1894 pass
1895 self._refs_container._remove_packed_ref(self._realname)
1897 self._deleted = True
1900class NamespacedRefsContainer(RefsContainer):
1901 """Wrapper that adds namespace prefix to all ref operations.
1903 This implements Git's GIT_NAMESPACE feature, which stores refs under
1904 refs/namespaces/<namespace>/ and filters operations to only show refs
1905 within that namespace.
1907 Example:
1908 With namespace "foo", a ref "refs/heads/master" is stored as
1909 "refs/namespaces/foo/refs/heads/master" in the underlying container.
1910 """
1912 def __init__(self, refs: RefsContainer, namespace: bytes) -> None:
1913 """Initialize NamespacedRefsContainer.
1915 Args:
1916 refs: The underlying refs container to wrap
1917 namespace: The namespace prefix (e.g., b"foo" or b"foo/bar")
1918 """
1919 super().__init__(logger=refs._logger)
1920 self._refs = refs
1921 # Build namespace prefix: refs/namespaces/<namespace>/
1922 # Support nested namespaces: foo/bar -> refs/namespaces/foo/refs/namespaces/bar/
1923 namespace_parts = namespace.split(b"/")
1924 self._namespace_prefix = b""
1925 for part in namespace_parts:
1926 self._namespace_prefix += b"refs/namespaces/" + part + b"/"
1928 def _apply_namespace(self, name: bytes) -> bytes:
1929 """Apply namespace prefix to a ref name."""
1930 # HEAD and other special refs are not namespaced
1931 if name == HEADREF or not name.startswith(b"refs/"):
1932 return name
1933 return self._namespace_prefix + name
1935 def _strip_namespace(self, name: bytes) -> bytes | None:
1936 """Remove namespace prefix from a ref name.
1938 Returns None if the ref is not in our namespace.
1939 """
1940 # HEAD and other special refs are not namespaced
1941 if name == HEADREF or not name.startswith(b"refs/"):
1942 return name
1943 if name.startswith(self._namespace_prefix):
1944 return name[len(self._namespace_prefix) :]
1945 return None
1947 def allkeys(self) -> set[Ref]:
1948 """Return all reference keys in this namespace."""
1949 keys: set[Ref] = set()
1950 for key in self._refs.allkeys():
1951 stripped = self._strip_namespace(key)
1952 if stripped is not None:
1953 keys.add(Ref(stripped))
1954 return keys
1956 def read_loose_ref(self, name: Ref) -> bytes | None:
1957 """Read a loose reference."""
1958 return self._refs.read_loose_ref(Ref(self._apply_namespace(name)))
1960 def get_packed_refs(self) -> dict[Ref, ObjectID]:
1961 """Get packed refs within this namespace."""
1962 packed: dict[Ref, ObjectID] = {}
1963 for name, value in self._refs.get_packed_refs().items():
1964 stripped = self._strip_namespace(name)
1965 if stripped is not None:
1966 packed[Ref(stripped)] = value
1967 return packed
1969 def add_packed_refs(self, new_refs: Mapping[Ref, ObjectID | None]) -> None:
1970 """Add packed refs with namespace prefix."""
1971 namespaced_refs: dict[Ref, ObjectID | None] = {
1972 Ref(self._apply_namespace(name)): value for name, value in new_refs.items()
1973 }
1974 self._refs.add_packed_refs(namespaced_refs)
1976 def get_peeled(self, name: Ref) -> ObjectID | None:
1977 """Return the cached peeled value of a ref."""
1978 return self._refs.get_peeled(Ref(self._apply_namespace(name)))
1980 def set_symbolic_ref(
1981 self,
1982 name: Ref,
1983 other: Ref,
1984 committer: bytes | None = None,
1985 timestamp: int | None = None,
1986 timezone: int | None = None,
1987 message: bytes | None = None,
1988 ) -> None:
1989 """Make a ref point at another ref."""
1990 self._refs.set_symbolic_ref(
1991 Ref(self._apply_namespace(name)),
1992 Ref(self._apply_namespace(other)),
1993 committer=committer,
1994 timestamp=timestamp,
1995 timezone=timezone,
1996 message=message,
1997 )
1999 def set_if_equals(
2000 self,
2001 name: Ref,
2002 old_ref: ObjectID | None,
2003 new_ref: ObjectID,
2004 committer: bytes | None = None,
2005 timestamp: int | None = None,
2006 timezone: int | None = None,
2007 message: bytes | None = None,
2008 ) -> bool:
2009 """Set a refname to new_ref only if it currently equals old_ref."""
2010 return self._refs.set_if_equals(
2011 Ref(self._apply_namespace(name)),
2012 old_ref,
2013 new_ref,
2014 committer=committer,
2015 timestamp=timestamp,
2016 timezone=timezone,
2017 message=message,
2018 )
2020 def add_if_new(
2021 self,
2022 name: Ref,
2023 ref: ObjectID,
2024 committer: bytes | None = None,
2025 timestamp: int | None = None,
2026 timezone: int | None = None,
2027 message: bytes | None = None,
2028 ) -> bool:
2029 """Add a new reference only if it does not already exist."""
2030 return self._refs.add_if_new(
2031 Ref(self._apply_namespace(name)),
2032 ref,
2033 committer=committer,
2034 timestamp=timestamp,
2035 timezone=timezone,
2036 message=message,
2037 )
2039 def remove_if_equals(
2040 self,
2041 name: Ref,
2042 old_ref: ObjectID | None,
2043 committer: bytes | None = None,
2044 timestamp: int | None = None,
2045 timezone: int | None = None,
2046 message: bytes | None = None,
2047 ) -> bool:
2048 """Remove a refname only if it currently equals old_ref."""
2049 return self._refs.remove_if_equals(
2050 Ref(self._apply_namespace(name)),
2051 old_ref,
2052 committer=committer,
2053 timestamp=timestamp,
2054 timezone=timezone,
2055 message=message,
2056 )
2058 def pack_refs(self, all: bool = False) -> None:
2059 """Pack loose refs into packed-refs file.
2061 Note: This packs all refs in the underlying container, not just
2062 those in the namespace.
2063 """
2064 self._refs.pack_refs(all=all)
2067@overload
2068def filter_ref_prefix(
2069 refs: dict[Ref, ObjectID], prefixes: Iterable[bytes]
2070) -> dict[Ref, ObjectID]: ...
2073@overload
2074def filter_ref_prefix(
2075 refs: dict[Ref, ObjectID | None], prefixes: Iterable[bytes]
2076) -> dict[Ref, ObjectID | None]: ...
2079def filter_ref_prefix(
2080 refs: dict[Ref, ObjectID] | dict[Ref, ObjectID | None],
2081 prefixes: Iterable[bytes],
2082) -> dict[Ref, ObjectID] | dict[Ref, ObjectID | None]:
2083 """Filter refs to only include those with a given prefix.
2085 Args:
2086 refs: A dictionary of refs.
2087 prefixes: The prefixes to filter by.
2088 """
2089 return {k: v for k, v in refs.items() if any(k.startswith(p) for p in prefixes)}
2092def is_per_worktree_ref(ref: bytes) -> bool:
2093 """Returns whether a reference is stored per worktree or not.
2095 Per-worktree references are:
2096 - all pseudorefs, e.g. HEAD
2097 - all references stored inside "refs/bisect/", "refs/worktree/" and "refs/rewritten/"
2099 All refs starting with "refs/" are shared, except for the ones listed above.
2101 See https://git-scm.com/docs/git-worktree#_refs.
2102 """
2103 return not ref.startswith(b"refs/") or ref.startswith(
2104 (b"refs/bisect/", b"refs/worktree/", b"refs/rewritten/")
2105 )