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 .config import ConfigFile
80 from .file import _GitFile
82from .errors import PackedRefsException, RefFormatError
83from .file import GitFile, ensure_dir_exists
84from .objects import ZERO_SHA, ObjectID, git_line, valid_hexsha
86Ref = NewType("Ref", bytes)
88HEADREF = Ref(b"HEAD")
89SYMREF = b"ref: "
90LOCAL_BRANCH_PREFIX = b"refs/heads/"
91LOCAL_TAG_PREFIX = b"refs/tags/"
92LOCAL_REMOTE_PREFIX = b"refs/remotes/"
93LOCAL_NOTES_PREFIX = b"refs/notes/"
94LOCAL_REPLACE_PREFIX = b"refs/replace/"
95BAD_REF_CHARS: set[int] = set(b"\177 ~^:?*[")
98class SymrefLoop(Exception):
99 """There is a loop between one or more symrefs."""
101 def __init__(self, ref: bytes, depth: int) -> None:
102 """Initialize SymrefLoop exception."""
103 self.ref = ref
104 self.depth = depth
107def parse_symref_value(contents: bytes) -> bytes:
108 """Parse a symref value.
110 Args:
111 contents: Contents to parse
112 Returns: Destination
113 """
114 if contents.startswith(SYMREF):
115 return contents[len(SYMREF) :].rstrip(b"\r\n")
116 raise ValueError(contents)
119def check_ref_format(refname: Ref) -> bool:
120 """Check if a refname is correctly formatted.
122 Implements all the same rules as git-check-ref-format[1].
124 [1]
125 http://www.kernel.org/pub/software/scm/git/docs/git-check-ref-format.html
127 Args:
128 refname: The refname to check
129 Returns: True if refname is valid, False otherwise
130 """
131 # These could be combined into one big expression, but are listed
132 # separately to parallel [1].
133 if refname == b"@":
134 return False
135 if b"/" not in refname: # type: ignore[comparison-overlap]
136 return False
137 if b".." in refname: # type: ignore[comparison-overlap]
138 return False
139 for i, c in enumerate(refname):
140 if ord(refname[i : i + 1]) < 0o40 or c in BAD_REF_CHARS:
141 return False
142 if refname[-1] in b"/.":
143 return False
144 if b"@{" in refname: # type: ignore[comparison-overlap]
145 return False
146 if b"\\" in refname: # type: ignore[comparison-overlap]
147 return False
148 for component in refname.split(b"/"):
149 if not component:
150 return False
151 if component.startswith(b"."):
152 return False
153 if component.endswith(b".lock"):
154 return False
155 return True
158def _collapse_slashes(refname: bytes) -> bytes:
159 """Collapse runs of consecutive slashes in a ref name into a single slash."""
160 return b"/".join(component for component in refname.split(b"/") if component)
163def parse_remote_ref(ref: bytes) -> tuple[bytes, bytes]:
164 """Parse a remote ref into remote name and branch name.
166 Args:
167 ref: Remote ref like b"refs/remotes/origin/main"
169 Returns:
170 Tuple of (remote_name, branch_name)
172 Raises:
173 ValueError: If ref is not a valid remote ref
174 """
175 if not ref.startswith(LOCAL_REMOTE_PREFIX):
176 raise ValueError(f"Not a remote ref: {ref!r}")
178 # Remove the prefix
179 remainder = ref[len(LOCAL_REMOTE_PREFIX) :]
181 # Split into remote name and branch name
182 parts = remainder.split(b"/", 1)
183 if len(parts) != 2:
184 raise ValueError(f"Invalid remote ref format: {ref!r}")
186 remote_name, branch_name = parts
187 return (remote_name, branch_name)
190def set_ref_from_raw(refs: "RefsContainer", name: Ref, raw_ref: bytes) -> None:
191 """Set a reference from a raw ref value.
193 This handles both symbolic refs (starting with 'ref: ') and direct ObjectID refs.
195 Args:
196 refs: The RefsContainer to set the ref in
197 name: The ref name to set
198 raw_ref: The raw ref value (either a symbolic ref or an ObjectID)
199 """
200 if raw_ref.startswith(SYMREF):
201 # It's a symbolic ref
202 target = Ref(raw_ref[len(SYMREF) :])
203 refs.set_symbolic_ref(name, target)
204 else:
205 # It's a direct ObjectID
206 refs[name] = ObjectID(raw_ref)
209class RefsContainer:
210 """A container for refs."""
212 def __init__(
213 self,
214 logger: Callable[
215 [bytes, bytes, bytes, bytes | None, int | None, int | None, bytes], None
216 ]
217 | None = None,
218 ) -> None:
219 """Initialize RefsContainer with optional logger function."""
220 self._logger = logger
222 def _log(
223 self,
224 ref: bytes,
225 old_sha: bytes | None,
226 new_sha: bytes | None,
227 committer: bytes | None = None,
228 timestamp: int | None = None,
229 timezone: int | None = None,
230 message: bytes | None = None,
231 ) -> None:
232 if self._logger is None:
233 return
234 if message is None:
235 return
236 # Use ZERO_SHA for None values, matching git behavior
237 if old_sha is None:
238 old_sha = ZERO_SHA
239 if new_sha is None:
240 new_sha = ZERO_SHA
241 self._logger(ref, old_sha, new_sha, committer, timestamp, timezone, message)
243 def set_symbolic_ref(
244 self,
245 name: Ref,
246 other: Ref,
247 committer: bytes | None = None,
248 timestamp: int | None = None,
249 timezone: int | None = None,
250 message: bytes | None = None,
251 ) -> None:
252 """Make a ref point at another ref.
254 Args:
255 name: Name of the ref to set
256 other: Name of the ref to point at
257 committer: Optional committer name/email
258 timestamp: Optional timestamp
259 timezone: Optional timezone
260 message: Optional message
261 """
262 raise NotImplementedError(self.set_symbolic_ref)
264 def get_packed_refs(self) -> dict[Ref, ObjectID]:
265 """Get contents of the packed-refs file.
267 Returns: Dictionary mapping ref names to SHA1s
269 Note: Will return an empty dictionary when no packed-refs file is
270 present.
271 """
272 raise NotImplementedError(self.get_packed_refs)
274 def add_packed_refs(self, new_refs: Mapping[Ref, ObjectID | None]) -> None:
275 """Add the given refs as packed refs.
277 Args:
278 new_refs: A mapping of ref names to targets; if a target is None that
279 means remove the ref
280 """
281 raise NotImplementedError(self.add_packed_refs)
283 def get_peeled(self, name: Ref) -> ObjectID | None:
284 """Return the cached peeled value of a ref, if available.
286 Args:
287 name: Name of the ref to peel
288 Returns: The peeled value of the ref. If the ref is known not point to
289 a tag, this will be the SHA the ref refers to. If the ref may point
290 to a tag, but no cached information is available, None is returned.
291 """
292 return None
294 def import_refs(
295 self,
296 base: Ref,
297 other: Mapping[Ref, ObjectID | None],
298 committer: bytes | None = None,
299 timestamp: bytes | None = None,
300 timezone: bytes | None = None,
301 message: bytes | None = None,
302 prune: bool = False,
303 ) -> None:
304 """Import refs from another repository.
306 Args:
307 base: Base ref to import into (e.g., b'refs/remotes/origin')
308 other: Dictionary of refs to import
309 committer: Optional committer for reflog
310 timestamp: Optional timestamp for reflog
311 timezone: Optional timezone for reflog
312 message: Optional message for reflog
313 prune: If True, remove refs not in other
314 """
315 # Strip a trailing slash so joining ``base`` with a ref name below
316 # does not produce a malformed ref with an empty path component
317 # (e.g. b'refs/tags/' + b'/' + b'v1.0' -> b'refs/tags//v1.0').
318 base = Ref(base.rstrip(b"/"))
319 if prune:
320 to_delete = set(self.subkeys(base))
321 else:
322 to_delete = set()
323 for name, value in other.items():
324 if value is None:
325 to_delete.add(name)
326 else:
327 self.set_if_equals(
328 Ref(b"/".join((base, name))), None, value, message=message
329 )
330 if to_delete:
331 try:
332 to_delete.remove(name)
333 except KeyError:
334 pass
335 for ref in to_delete:
336 self.remove_if_equals(Ref(b"/".join((base, ref))), None, message=message)
338 def allkeys(self) -> set[Ref]:
339 """All refs present in this container."""
340 raise NotImplementedError(self.allkeys)
342 def __iter__(self) -> Iterator[Ref]:
343 """Iterate over all reference keys."""
344 return iter(self.allkeys())
346 def keys(self, base: Ref | None = None) -> set[Ref]:
347 """Refs present in this container.
349 Args:
350 base: An optional base to return refs under.
351 Returns: An unsorted set of valid refs in this container, including
352 packed refs.
353 """
354 if base is not None:
355 return self.subkeys(base)
356 else:
357 return self.allkeys()
359 def subkeys(self, base: Ref) -> set[Ref]:
360 """Refs present in this container under a base.
362 Args:
363 base: The base to return refs under.
364 Returns: A set of valid refs in this container under the base; the base
365 prefix is stripped from the ref names returned.
366 """
367 keys: set[Ref] = set()
368 base_len = len(base) + 1
369 for refname in self.allkeys():
370 if refname.startswith(base):
371 keys.add(Ref(refname[base_len:]))
372 return keys
374 def as_dict(self, base: Ref | None = None) -> dict[Ref, ObjectID]:
375 """Return the contents of this container as a dictionary."""
376 ret: dict[Ref, ObjectID] = {}
377 keys = self.keys(base)
378 base_bytes: bytes
379 if base is None:
380 base_bytes = b""
381 else:
382 base_bytes = base.rstrip(b"/")
383 for key in keys:
384 try:
385 ret[key] = self[Ref((base_bytes + b"/" + key).strip(b"/"))]
386 except (SymrefLoop, KeyError):
387 continue # Unable to resolve
389 return ret
391 def _check_refname(self, name: Ref) -> None:
392 """Ensure a refname is valid and lives in refs or is HEAD.
394 HEAD is not a valid refname according to git-check-ref-format, but this
395 class needs to be able to touch HEAD. Also, check_ref_format expects
396 refnames without the leading 'refs/', but this class requires that
397 so it cannot touch anything outside the refs dir (or HEAD).
399 Args:
400 name: The name of the reference.
402 Raises:
403 KeyError: if a refname is not HEAD or is otherwise not valid.
404 """
405 if name in (HEADREF, Ref(b"refs/stash")):
406 return
407 if not name.startswith(b"refs/"):
408 raise RefFormatError(name)
409 rest = Ref(name[5:])
410 if check_ref_format(rest):
411 return
412 # As of Dulwich 1.2.3 check_ref_format rejects empty path components
413 # (e.g. b'refs/tags//v1.0'). Such names were silently accepted before,
414 # and some callers (e.g. older Poetry releases) still construct them.
415 # Warn rather than raise for now if collapsing repeated slashes would
416 # make the name valid, so the only defect is empty components.
417 if (
418 b"//" in name # type: ignore[comparison-overlap]
419 and check_ref_format(Ref(_collapse_slashes(rest)))
420 ):
421 warnings.warn(
422 f"Ref name {name!r} contains empty path components; "
423 "this will be rejected in a future version of Dulwich.",
424 DeprecationWarning,
425 stacklevel=3,
426 )
427 return
428 raise RefFormatError(name)
430 def _check_ref_value(self, ref: ObjectID) -> None:
431 """Ensure a ref value is a valid object id or a symref.
433 Args:
434 ref: The value a ref is being set to.
436 Raises:
437 ValueError: if the value is neither a valid sha nor a symref.
438 """
439 if not (valid_hexsha(ref) or ref.startswith(SYMREF)):
440 raise ValueError(f"{ref!r} must be a valid sha or a symref")
442 def read_ref(self, refname: Ref) -> bytes | None:
443 """Read a reference without following any references.
445 Args:
446 refname: The name of the reference
447 Returns: The contents of the ref file, or None if it does
448 not exist.
449 """
450 contents = self.read_loose_ref(refname)
451 if not contents:
452 contents = self.get_packed_refs().get(refname, None)
453 return contents
455 def read_loose_ref(self, name: Ref) -> bytes | None:
456 """Read a loose reference and return its contents.
458 Args:
459 name: the refname to read
460 Returns: The contents of the ref file, or None if it does
461 not exist.
462 """
463 raise NotImplementedError(self.read_loose_ref)
465 def follow(self, name: Ref) -> tuple[list[Ref], ObjectID | None]:
466 """Follow a reference name.
468 Returns: a tuple of (refnames, sha), wheres refnames are the names of
469 references in the chain
470 """
471 contents: bytes | None = SYMREF + name
472 depth = 0
473 refnames: list[Ref] = []
474 while contents and contents.startswith(SYMREF):
475 refname = Ref(contents[len(SYMREF) :])
476 refnames.append(refname)
477 contents = self.read_ref(refname)
478 if not contents:
479 break
480 depth += 1
481 if depth > 5:
482 raise SymrefLoop(name, depth)
483 return refnames, ObjectID(contents) if contents else None
485 def __contains__(self, refname: Ref) -> bool:
486 """Check if a reference exists."""
487 if self.read_ref(refname):
488 return True
489 return False
491 def __getitem__(self, name: Ref) -> ObjectID:
492 """Get the SHA1 for a reference name.
494 This method follows all symbolic references.
495 """
496 _, sha = self.follow(name)
497 if sha is None:
498 raise KeyError(name)
499 return sha
501 def set_if_equals(
502 self,
503 name: Ref,
504 old_ref: ObjectID | None,
505 new_ref: ObjectID,
506 committer: bytes | None = None,
507 timestamp: int | None = None,
508 timezone: int | None = None,
509 message: bytes | None = None,
510 ) -> bool:
511 """Set a refname to new_ref only if it currently equals old_ref.
513 This method follows all symbolic references if applicable for the
514 subclass, and can be used to perform an atomic compare-and-swap
515 operation.
517 Args:
518 name: The refname to set.
519 old_ref: The old sha the refname must refer to, or None to set
520 unconditionally.
521 new_ref: The new sha the refname will refer to.
522 committer: Optional committer name/email
523 timestamp: Optional timestamp
524 timezone: Optional timezone
525 message: Message for reflog
526 Returns: True if the set was successful, False otherwise.
527 """
528 raise NotImplementedError(self.set_if_equals)
530 def add_if_new(
531 self,
532 name: Ref,
533 ref: ObjectID,
534 committer: bytes | None = None,
535 timestamp: int | None = None,
536 timezone: int | None = None,
537 message: bytes | None = None,
538 ) -> bool:
539 """Add a new reference only if it does not already exist.
541 Args:
542 name: Ref name
543 ref: Ref value
544 committer: Optional committer name/email
545 timestamp: Optional timestamp
546 timezone: Optional timezone
547 message: Optional message for reflog
548 """
549 raise NotImplementedError(self.add_if_new)
551 def __setitem__(self, name: Ref, ref: ObjectID) -> None:
552 """Set a reference name to point to the given SHA1.
554 This method follows all symbolic references if applicable for the
555 subclass.
557 Note: This method unconditionally overwrites the contents of a
558 reference. To update atomically only if the reference has not
559 changed, use set_if_equals().
561 Args:
562 name: The refname to set.
563 ref: The new sha the refname will refer to.
564 """
565 self._check_ref_value(ref)
566 self.set_if_equals(name, None, ref)
568 def remove_if_equals(
569 self,
570 name: Ref,
571 old_ref: ObjectID | None,
572 committer: bytes | None = None,
573 timestamp: int | None = None,
574 timezone: int | None = None,
575 message: bytes | None = None,
576 ) -> bool:
577 """Remove a refname only if it currently equals old_ref.
579 This method does not follow symbolic references, even if applicable for
580 the subclass. It can be used to perform an atomic compare-and-delete
581 operation.
583 Args:
584 name: The refname to delete.
585 old_ref: The old sha the refname must refer to, or None to
586 delete unconditionally.
587 committer: Optional committer name/email
588 timestamp: Optional timestamp
589 timezone: Optional timezone
590 message: Message for reflog
591 Returns: True if the delete was successful, False otherwise.
592 """
593 raise NotImplementedError(self.remove_if_equals)
595 def __delitem__(self, name: Ref) -> None:
596 """Remove a refname.
598 This method does not follow symbolic references, even if applicable for
599 the subclass.
601 Note: This method unconditionally deletes the contents of a reference.
602 To delete atomically only if the reference has not changed, use
603 remove_if_equals().
605 Args:
606 name: The refname to delete.
607 """
608 self.remove_if_equals(name, None)
610 def get_symrefs(self) -> dict[Ref, Ref]:
611 """Get a dict with all symrefs in this container.
613 Returns: Dictionary mapping source ref to target ref
614 """
615 ret: dict[Ref, Ref] = {}
616 for src in self.allkeys():
617 try:
618 ref_value = self.read_ref(src)
619 assert ref_value is not None
620 dst = parse_symref_value(ref_value)
621 except ValueError:
622 pass
623 else:
624 ret[src] = Ref(dst)
625 return ret
627 def pack_refs(self, all: bool = False) -> None:
628 """Pack loose refs into packed-refs file.
630 Args:
631 all: If True, pack all refs. If False, only pack tags.
632 """
633 raise NotImplementedError(self.pack_refs)
636class DictRefsContainer(RefsContainer):
637 """RefsContainer backed by a simple dict.
639 This container does not support symbolic or packed references and is not
640 threadsafe.
641 """
643 def __init__(
644 self,
645 refs: dict[Ref, bytes],
646 logger: Callable[
647 [
648 bytes,
649 bytes | None,
650 bytes | None,
651 bytes | None,
652 int | None,
653 int | None,
654 bytes | None,
655 ],
656 None,
657 ]
658 | None = None,
659 ) -> None:
660 """Initialize DictRefsContainer with refs dictionary and optional logger."""
661 super().__init__(logger=logger)
662 self._refs = refs
663 self._peeled: dict[Ref, ObjectID] = {}
664 self._watchers: set[Any] = set()
666 def allkeys(self) -> set[Ref]:
667 """Return all reference keys."""
668 return set(self._refs.keys())
670 def read_loose_ref(self, name: Ref) -> bytes | None:
671 """Read a loose reference."""
672 return self._refs.get(name, None)
674 def get_packed_refs(self) -> dict[Ref, ObjectID]:
675 """Get packed references."""
676 return {}
678 def _notify(self, ref: bytes, newsha: bytes | None) -> None:
679 for watcher in self._watchers:
680 watcher._notify((ref, newsha))
682 def set_symbolic_ref(
683 self,
684 name: Ref,
685 other: Ref,
686 committer: bytes | None = None,
687 timestamp: int | None = None,
688 timezone: int | None = None,
689 message: bytes | None = None,
690 ) -> None:
691 """Make a ref point at another ref.
693 Args:
694 name: Name of the ref to set
695 other: Name of the ref to point at
696 committer: Optional committer name for reflog
697 timestamp: Optional timestamp for reflog
698 timezone: Optional timezone for reflog
699 message: Optional message for reflog
700 """
701 old = self.follow(name)[-1]
702 new = SYMREF + other
703 self._refs[name] = new
704 self._notify(name, new)
705 self._log(
706 name,
707 old,
708 new,
709 committer=committer,
710 timestamp=timestamp,
711 timezone=timezone,
712 message=message,
713 )
715 def set_if_equals(
716 self,
717 name: Ref,
718 old_ref: ObjectID | None,
719 new_ref: ObjectID,
720 committer: bytes | None = None,
721 timestamp: int | None = None,
722 timezone: int | None = None,
723 message: bytes | None = None,
724 ) -> bool:
725 """Set a refname to new_ref only if it currently equals old_ref.
727 This method follows all symbolic references, and can be used to perform
728 an atomic compare-and-swap operation.
730 Args:
731 name: The refname to set.
732 old_ref: The old sha the refname must refer to, or None to set
733 unconditionally.
734 new_ref: The new sha the refname will refer to.
735 committer: Optional committer name for reflog
736 timestamp: Optional timestamp for reflog
737 timezone: Optional timezone for reflog
738 message: Optional message for reflog
740 Returns:
741 True if the set was successful, False otherwise.
742 """
743 self._check_ref_value(new_ref)
744 if old_ref is not None and self._refs.get(name, ZERO_SHA) != old_ref:
745 return False
746 # Only update the specific ref requested, not the whole chain
747 self._check_refname(name)
748 old = self._refs.get(name)
749 self._refs[name] = new_ref
750 self._notify(name, new_ref)
751 self._log(
752 name,
753 old,
754 new_ref,
755 committer=committer,
756 timestamp=timestamp,
757 timezone=timezone,
758 message=message,
759 )
760 return True
762 def add_if_new(
763 self,
764 name: Ref,
765 ref: ObjectID,
766 committer: bytes | None = None,
767 timestamp: int | None = None,
768 timezone: int | None = None,
769 message: bytes | None = None,
770 ) -> bool:
771 """Add a new reference only if it does not already exist.
773 Args:
774 name: Ref name
775 ref: Ref value
776 committer: Optional committer name for reflog
777 timestamp: Optional timestamp for reflog
778 timezone: Optional timezone for reflog
779 message: Optional message for reflog
781 Returns:
782 True if the add was successful, False otherwise.
783 """
784 self._check_ref_value(ref)
785 if name in self._refs:
786 return False
787 self._refs[name] = ref
788 self._notify(name, ref)
789 self._log(
790 name,
791 None,
792 ref,
793 committer=committer,
794 timestamp=timestamp,
795 timezone=timezone,
796 message=message,
797 )
798 return True
800 def remove_if_equals(
801 self,
802 name: Ref,
803 old_ref: ObjectID | None,
804 committer: bytes | None = None,
805 timestamp: int | None = None,
806 timezone: int | None = None,
807 message: bytes | None = None,
808 ) -> bool:
809 """Remove a refname only if it currently equals old_ref.
811 This method does not follow symbolic references. It can be used to
812 perform an atomic compare-and-delete operation.
814 Args:
815 name: The refname to delete.
816 old_ref: The old sha the refname must refer to, or None to
817 delete unconditionally.
818 committer: Optional committer name for reflog
819 timestamp: Optional timestamp for reflog
820 timezone: Optional timezone for reflog
821 message: Optional message for reflog
823 Returns:
824 True if the delete was successful, False otherwise.
825 """
826 if old_ref is not None and self._refs.get(name, ZERO_SHA) != old_ref:
827 return False
828 try:
829 old = self._refs.pop(name)
830 except KeyError:
831 pass
832 else:
833 self._notify(name, None)
834 self._log(
835 name,
836 old,
837 None,
838 committer=committer,
839 timestamp=timestamp,
840 timezone=timezone,
841 message=message,
842 )
843 return True
845 def get_peeled(self, name: Ref) -> ObjectID | None:
846 """Get peeled version of a reference."""
847 return self._peeled.get(name)
849 def _update(self, refs: Mapping[Ref, ObjectID]) -> None:
850 """Update multiple refs; intended only for testing."""
851 # TODO(dborowitz): replace this with a public function that uses
852 # set_if_equal.
853 for ref, sha in refs.items():
854 self.set_if_equals(ref, None, sha)
856 def _update_peeled(self, peeled: Mapping[Ref, ObjectID]) -> None:
857 """Update cached peeled refs; intended only for testing."""
858 self._peeled.update(peeled)
861#: Identity of a particular packed-refs file, used to detect that the file a
862#: cache was populated from has since been replaced.
863_PackedRefsKey = tuple[int, int, int, int, int]
866def _packed_refs_key(st: os.stat_result) -> _PackedRefsKey:
867 """Build a key identifying the packed-refs file described by ``st``.
869 The fields mirror those git compares in ``match_stat_data()`` when
870 deciding whether a file it has cached has been replaced. Size alone is a
871 weak signal here, since repacking refs often produces a file of identical
872 length, so the inode and timestamps do most of the work.
874 Args:
875 st: Stat result for a packed-refs file.
876 Returns: An opaque, comparable key.
877 """
878 return (st.st_ino, st.st_dev, st.st_size, st.st_mtime_ns, st.st_ctime_ns)
881class DiskRefsContainer(RefsContainer):
882 """Refs container that reads refs from disk."""
884 def __init__(
885 self,
886 path: str | bytes | os.PathLike[str],
887 worktree_path: str | bytes | os.PathLike[str] | None = None,
888 logger: Callable[
889 [bytes, bytes, bytes, bytes | None, int | None, int | None, bytes], None
890 ]
891 | None = None,
892 ) -> None:
893 """Initialize DiskRefsContainer."""
894 super().__init__(logger=logger)
895 # Convert path-like objects to strings, then to bytes for Git compatibility
896 self.path = os.fsencode(os.fspath(path))
897 if worktree_path is None:
898 self.worktree_path = self.path
899 else:
900 self.worktree_path = os.fsencode(os.fspath(worktree_path))
901 self._packed_refs: dict[Ref, ObjectID] | None = None
902 self._peeled_refs: dict[Ref, ObjectID] | None = None
903 self._packed_refs_key: _PackedRefsKey | None = None
905 def __repr__(self) -> str:
906 """Return string representation of DiskRefsContainer."""
907 return f"{self.__class__.__name__}({self.path!r})"
909 def _iter_dir(
910 self,
911 path: bytes,
912 base: bytes,
913 dir_filter: Callable[[bytes], bool] | None = None,
914 ) -> Iterator[Ref]:
915 refspath = os.path.join(path, base.rstrip(b"/"))
916 prefix_len = len(os.path.join(path, b""))
918 for root, dirs, files in os.walk(refspath):
919 directory = root[prefix_len:]
920 if os.path.sep != "/":
921 directory = directory.replace(os.fsencode(os.path.sep), b"/")
922 if dir_filter is not None:
923 dirs[:] = [
924 d for d in dirs if dir_filter(b"/".join([directory, d, b""]))
925 ]
927 for filename in files:
928 refname = b"/".join([directory, filename])
929 if check_ref_format(Ref(refname)):
930 yield Ref(refname)
932 def _iter_loose_refs(self, base: bytes = b"refs/") -> Iterator[Ref]:
933 base = base.rstrip(b"/") + b"/"
934 search_paths: list[tuple[bytes, Callable[[bytes], bool] | None]] = []
935 if base != b"refs/":
936 path = self.worktree_path if is_per_worktree_ref(base) else self.path
937 search_paths.append((path, None))
938 elif self.worktree_path == self.path:
939 # Iterate through all the refs from the main worktree
940 search_paths.append((self.path, None))
941 else:
942 # Iterate through all the shared refs from the commondir, excluding per-worktree refs
943 search_paths.append((self.path, lambda r: not is_per_worktree_ref(r)))
944 # Iterate through all the per-worktree refs from the worktree's gitdir
945 search_paths.append((self.worktree_path, is_per_worktree_ref))
947 for path, dir_filter in search_paths:
948 yield from self._iter_dir(path, base, dir_filter=dir_filter)
950 def subkeys(self, base: Ref) -> set[Ref]:
951 """Return subkeys under a given base reference path."""
952 subkeys: set[Ref] = set()
954 for key in self._iter_loose_refs(base):
955 if key.startswith(base):
956 subkeys.add(Ref(key[len(base) :].strip(b"/")))
958 for key in self.get_packed_refs():
959 if key.startswith(base):
960 subkeys.add(Ref(key[len(base) :].strip(b"/")))
961 return subkeys
963 def allkeys(self) -> set[Ref]:
964 """Return all reference keys."""
965 allkeys: set[Ref] = set()
966 if os.path.exists(self.refpath(HEADREF)):
967 allkeys.add(Ref(HEADREF))
969 allkeys.update(self._iter_loose_refs())
970 allkeys.update(self.get_packed_refs())
971 return allkeys
973 def refpath(self, name: bytes) -> bytes:
974 """Return the disk path of a ref."""
975 path = name
976 if os.path.sep != "/":
977 path = path.replace(b"/", os.fsencode(os.path.sep))
979 root_dir = self.worktree_path if is_per_worktree_ref(name) else self.path
980 return os.path.join(root_dir, path)
982 def _current_packed_refs_key(self) -> _PackedRefsKey | None:
983 """Identify the packed-refs file currently on disk.
985 Returns: An opaque key for the current packed-refs file, or None if
986 no packed-refs file is present.
987 """
988 try:
989 st = os.stat(os.path.join(self.path, b"packed-refs"))
990 except FileNotFoundError:
991 return None
992 return _packed_refs_key(st)
994 def _invalidate_packed_refs_cache(self) -> None:
995 """Discard cached packed and peeled refs."""
996 self._packed_refs = None
997 self._peeled_refs = None
998 self._packed_refs_key = None
1000 def get_packed_refs(self) -> dict[Ref, ObjectID]:
1001 """Get contents of the packed-refs file.
1003 Returns: Dictionary mapping ref names to SHA1s
1005 Note: Will return an empty dictionary when no packed-refs file is
1006 present.
1007 """
1008 if (
1009 self._packed_refs is not None
1010 and self._packed_refs_key != self._current_packed_refs_key()
1011 ):
1012 # The packed-refs file was replaced underneath us, most likely by
1013 # another process packing or unpacking refs. Drop the stale cache
1014 # so that it is reloaded below.
1015 self._invalidate_packed_refs_cache()
1017 if self._packed_refs is None:
1018 # set both to empty because we want _peeled_refs to be
1019 # None if and only if _packed_refs is also None.
1020 self._packed_refs = {}
1021 self._peeled_refs = {}
1022 self._packed_refs_key = None
1023 path = os.path.join(self.path, b"packed-refs")
1024 try:
1025 f = GitFile(path, "rb")
1026 except FileNotFoundError:
1027 return {}
1028 with f:
1029 first_line = next(iter(f)).rstrip()
1030 if first_line.startswith(b"# pack-refs") and b" peeled" in first_line:
1031 for sha, name, peeled in read_packed_refs_with_peeled(f):
1032 self._packed_refs[name] = sha
1033 if peeled:
1034 self._peeled_refs[name] = peeled
1035 else:
1036 f.seek(0)
1037 for sha, name in read_packed_refs(f):
1038 self._packed_refs[name] = sha
1039 # Record which file the cache was populated from, so that a
1040 # later replacement of it can be detected. Stat the open file
1041 # rather than the path to avoid picking up a newer file that
1042 # was renamed into place while we were reading.
1043 self._packed_refs_key = _packed_refs_key(os.fstat(f.fileno()))
1044 return self._packed_refs
1046 def add_packed_refs(self, new_refs: Mapping[Ref, ObjectID | None]) -> None:
1047 """Add the given refs as packed refs.
1049 Args:
1050 new_refs: A mapping of ref names to targets; if a target is None that
1051 means remove the ref
1052 """
1053 if not new_refs:
1054 return
1056 path = os.path.join(self.path, b"packed-refs")
1058 try:
1059 with GitFile(path, "wb") as f:
1060 # reread cached refs from disk, while holding the lock
1061 packed_refs = self.get_packed_refs().copy()
1063 for ref, target in new_refs.items():
1064 # sanity check
1065 if ref == HEADREF:
1066 raise ValueError("cannot pack HEAD")
1068 # remove any loose refs pointing to this one -- please
1069 # note that this bypasses remove_if_equals as we don't
1070 # want to affect packed refs in here
1071 with suppress(OSError):
1072 os.remove(self.refpath(ref))
1074 if target is not None:
1075 packed_refs[ref] = target
1076 else:
1077 packed_refs.pop(ref, None)
1079 write_packed_refs(f, packed_refs, self._peeled_refs)
1080 finally:
1081 # Do not stat the path and associate that identity with the data
1082 # just written: another writer can replace packed-refs after the
1083 # lock is released but before the stat. Reload on the next access
1084 # instead.
1085 self._invalidate_packed_refs_cache()
1087 def get_peeled(self, name: Ref) -> ObjectID | None:
1088 """Return the cached peeled value of a ref, if available.
1090 Args:
1091 name: Name of the ref to peel
1092 Returns: The peeled value of the ref. If the ref is known not point to
1093 a tag, this will be the SHA the ref refers to. If the ref may point
1094 to a tag, but no cached information is available, None is returned.
1095 """
1096 self.get_packed_refs()
1097 if (
1098 self._peeled_refs is None
1099 or self._packed_refs is None
1100 or name not in self._packed_refs
1101 ):
1102 # No cache: no peeled refs were read, or this ref is loose
1103 return None
1104 if name in self._peeled_refs:
1105 return self._peeled_refs[name]
1106 else:
1107 # Known not peelable
1108 return self[name]
1110 def read_loose_ref(self, name: Ref) -> bytes | None:
1111 """Read a reference file and return its contents.
1113 If the reference file a symbolic reference, only read the first line of
1114 the file. Otherwise, read the hash (40 bytes for SHA1, 64 bytes for SHA256).
1116 Args:
1117 name: the refname to read, relative to refpath
1118 Returns: The contents of the ref file, or None if the file does not
1119 exist.
1121 Raises:
1122 IOError: if any other error occurs
1123 """
1124 # Validate the name before turning it into a path.
1125 try:
1126 self._check_refname(name)
1127 except RefFormatError:
1128 return None
1129 filename = self.refpath(name)
1130 try:
1131 with GitFile(filename, "rb") as f:
1132 header = f.read(len(SYMREF))
1133 if header == SYMREF:
1134 # Read only the first line
1135 return header + next(iter(f)).rstrip(b"\r\n")
1136 else:
1137 # Read the entire line to get the full hash (handles both SHA1 and SHA256)
1138 f.seek(0)
1139 line = f.readline().rstrip(b"\r\n")
1140 return line
1141 except (OSError, UnicodeError):
1142 # don't assume anything specific about the error; in
1143 # particular, invalid or forbidden paths can raise weird
1144 # errors depending on the specific operating system
1145 return None
1147 def _remove_packed_ref(self, name: Ref) -> None:
1148 if name not in self.get_packed_refs():
1149 return
1151 filename = os.path.join(self.path, b"packed-refs")
1152 # reread cached refs from disk, while holding the lock
1153 f = GitFile(filename, "wb")
1154 try:
1155 self._invalidate_packed_refs_cache()
1156 packed_refs = self.get_packed_refs().copy()
1157 peeled_refs = (
1158 self._peeled_refs.copy() if self._peeled_refs is not None else None
1159 )
1161 if name not in packed_refs:
1162 f.abort()
1163 return
1165 del packed_refs[name]
1166 if peeled_refs is not None:
1167 peeled_refs.pop(name, None)
1168 write_packed_refs(f, packed_refs, peeled_refs)
1169 f.close()
1170 finally:
1171 if not f.closed:
1172 f.abort()
1173 self._invalidate_packed_refs_cache()
1175 def set_symbolic_ref(
1176 self,
1177 name: Ref,
1178 other: Ref,
1179 committer: bytes | None = None,
1180 timestamp: int | None = None,
1181 timezone: int | None = None,
1182 message: bytes | None = None,
1183 ) -> None:
1184 """Make a ref point at another ref.
1186 Args:
1187 name: Name of the ref to set
1188 other: Name of the ref to point at
1189 committer: Optional committer name
1190 timestamp: Optional timestamp
1191 timezone: Optional timezone
1192 message: Optional message to describe the change
1193 """
1194 self._check_refname(name)
1195 self._check_refname(other)
1196 filename = self.refpath(name)
1197 f = GitFile(filename, "wb")
1198 try:
1199 f.write(SYMREF + other + b"\n")
1200 sha = self.follow(name)[-1]
1201 self._log(
1202 name,
1203 sha,
1204 sha,
1205 committer=committer,
1206 timestamp=timestamp,
1207 timezone=timezone,
1208 message=message,
1209 )
1210 except BaseException:
1211 f.abort()
1212 raise
1213 else:
1214 f.close()
1216 def set_if_equals(
1217 self,
1218 name: Ref,
1219 old_ref: ObjectID | None,
1220 new_ref: ObjectID,
1221 committer: bytes | None = None,
1222 timestamp: int | None = None,
1223 timezone: int | None = None,
1224 message: bytes | None = None,
1225 ) -> bool:
1226 """Set a refname to new_ref only if it currently equals old_ref.
1228 This method follows all symbolic references, and can be used to perform
1229 an atomic compare-and-swap operation.
1231 Args:
1232 name: The refname to set.
1233 old_ref: The old sha the refname must refer to, or None to set
1234 unconditionally.
1235 new_ref: The new sha the refname will refer to.
1236 committer: Optional committer name
1237 timestamp: Optional timestamp
1238 timezone: Optional timezone
1239 message: Set message for reflog
1240 Returns: True if the set was successful, False otherwise.
1241 """
1242 self._check_refname(name)
1243 self._check_ref_value(new_ref)
1244 try:
1245 realnames, _ = self.follow(name)
1246 realname = realnames[-1]
1247 except (KeyError, IndexError, SymrefLoop):
1248 realname = name
1249 filename = self.refpath(realname)
1251 # make sure none of the ancestor folders is in packed refs
1252 probe_ref = Ref(os.path.dirname(realname))
1253 packed_refs = self.get_packed_refs()
1254 while probe_ref:
1255 if packed_refs.get(probe_ref, None) is not None:
1256 raise NotADirectoryError(filename)
1257 probe_ref = Ref(os.path.dirname(probe_ref))
1259 ensure_dir_exists(os.path.dirname(filename))
1260 with GitFile(filename, "wb") as f:
1261 if old_ref is not None:
1262 try:
1263 # read again while holding the lock to handle race conditions
1264 orig_ref = self.read_loose_ref(realname)
1265 if orig_ref is None:
1266 orig_ref = self.get_packed_refs().get(realname, ZERO_SHA)
1267 if orig_ref != old_ref:
1268 f.abort()
1269 return False
1270 except OSError:
1271 f.abort()
1272 raise
1274 # Check if ref already has the desired value while holding the lock
1275 # This avoids fsync when ref is unchanged but still detects lock conflicts
1276 current_ref = self.read_loose_ref(realname)
1277 if current_ref is None:
1278 current_ref = packed_refs.get(realname, None)
1280 if current_ref is not None and current_ref == new_ref:
1281 # Ref already has desired value, abort write to avoid fsync
1282 f.abort()
1283 return True
1285 try:
1286 f.write(new_ref + b"\n")
1287 except OSError:
1288 f.abort()
1289 raise
1290 self._log(
1291 realname,
1292 old_ref,
1293 new_ref,
1294 committer=committer,
1295 timestamp=timestamp,
1296 timezone=timezone,
1297 message=message,
1298 )
1299 return True
1301 def add_if_new(
1302 self,
1303 name: Ref,
1304 ref: ObjectID,
1305 committer: bytes | None = None,
1306 timestamp: int | None = None,
1307 timezone: int | None = None,
1308 message: bytes | None = None,
1309 ) -> bool:
1310 """Add a new reference only if it does not already exist.
1312 This method follows symrefs, and only ensures that the last ref in the
1313 chain does not exist.
1315 Args:
1316 name: The refname to set.
1317 ref: The new sha the refname will refer to.
1318 committer: Optional committer name
1319 timestamp: Optional timestamp
1320 timezone: Optional timezone
1321 message: Optional message for reflog
1322 Returns: True if the add was successful, False otherwise.
1323 """
1324 self._check_ref_value(ref)
1325 try:
1326 realnames, contents = self.follow(name)
1327 if contents is not None:
1328 return False
1329 realname = realnames[-1]
1330 except (KeyError, IndexError):
1331 realname = name
1332 self._check_refname(realname)
1333 filename = self.refpath(realname)
1334 ensure_dir_exists(os.path.dirname(filename))
1335 with GitFile(filename, "wb") as f:
1336 if os.path.exists(filename) or name in self.get_packed_refs():
1337 f.abort()
1338 return False
1339 try:
1340 f.write(ref + b"\n")
1341 except OSError:
1342 f.abort()
1343 raise
1344 else:
1345 self._log(
1346 name,
1347 None,
1348 ref,
1349 committer=committer,
1350 timestamp=timestamp,
1351 timezone=timezone,
1352 message=message,
1353 )
1354 return True
1356 def remove_if_equals(
1357 self,
1358 name: Ref,
1359 old_ref: ObjectID | None,
1360 committer: bytes | None = None,
1361 timestamp: int | None = None,
1362 timezone: int | None = None,
1363 message: bytes | None = None,
1364 ) -> bool:
1365 """Remove a refname only if it currently equals old_ref.
1367 This method does not follow symbolic references. It can be used to
1368 perform an atomic compare-and-delete operation.
1370 Args:
1371 name: The refname to delete.
1372 old_ref: The old sha the refname must refer to, or None to
1373 delete unconditionally.
1374 committer: Optional committer name
1375 timestamp: Optional timestamp
1376 timezone: Optional timezone
1377 message: Optional message
1378 Returns: True if the delete was successful, False otherwise.
1379 """
1380 self._check_refname(name)
1381 filename = self.refpath(name)
1382 ensure_dir_exists(os.path.dirname(filename))
1383 f = GitFile(filename, "wb")
1384 try:
1385 if old_ref is not None:
1386 orig_ref = self.read_loose_ref(name)
1387 if orig_ref is None:
1388 orig_ref = self.get_packed_refs().get(name)
1389 if orig_ref is None:
1390 orig_ref = ZERO_SHA
1391 if orig_ref != old_ref:
1392 return False
1394 # remove the reference file itself
1395 try:
1396 found = os.path.lexists(filename)
1397 except OSError:
1398 # may only be packed, or otherwise unstorable
1399 found = False
1401 if found:
1402 os.remove(filename)
1404 self._remove_packed_ref(name)
1405 self._log(
1406 name,
1407 old_ref,
1408 None,
1409 committer=committer,
1410 timestamp=timestamp,
1411 timezone=timezone,
1412 message=message,
1413 )
1414 finally:
1415 # never write, we just wanted the lock
1416 f.abort()
1418 # outside of the lock, clean-up any parent directory that might now
1419 # be empty. this ensures that re-creating a reference of the same
1420 # name of what was previously a directory works as expected
1421 parent = name
1422 while True:
1423 try:
1424 parent_bytes, _ = parent.rsplit(b"/", 1)
1425 parent = Ref(parent_bytes)
1426 except ValueError:
1427 break
1429 if parent == b"refs":
1430 break
1431 parent_filename = self.refpath(parent)
1432 try:
1433 os.rmdir(parent_filename)
1434 except OSError:
1435 # this can be caused by the parent directory being
1436 # removed by another process, being not empty, etc.
1437 # in any case, this is non fatal because we already
1438 # removed the reference, just ignore it
1439 break
1441 return True
1443 def pack_refs(self, all: bool = False) -> None:
1444 """Pack loose refs into packed-refs file.
1446 Args:
1447 all: If True, pack all refs. If False, only pack tags.
1448 """
1449 refs_to_pack: dict[Ref, ObjectID | None] = {}
1450 for ref in self.allkeys():
1451 if ref == HEADREF:
1452 # Never pack HEAD
1453 continue
1454 if all or ref.startswith(LOCAL_TAG_PREFIX):
1455 try:
1456 sha = self[ref]
1457 if sha:
1458 refs_to_pack[ref] = sha
1459 except KeyError:
1460 # Broken ref, skip it
1461 pass
1463 if refs_to_pack:
1464 self.add_packed_refs(refs_to_pack)
1467def _split_ref_line(line: bytes) -> tuple[ObjectID, Ref]:
1468 """Split a single ref line into a tuple of SHA1 and name."""
1469 fields = line.rstrip(b"\n\r").split(b" ")
1470 if len(fields) != 2:
1471 raise PackedRefsException(f"invalid ref line {line!r}")
1472 sha, name = fields
1473 if not valid_hexsha(sha):
1474 raise PackedRefsException(f"Invalid hex sha {sha!r}")
1475 if not check_ref_format(Ref(name)):
1476 raise PackedRefsException(f"invalid ref name {name!r}")
1477 return (ObjectID(sha), Ref(name))
1480def read_packed_refs(f: IO[bytes]) -> Iterator[tuple[ObjectID, Ref]]:
1481 """Read a packed refs file.
1483 Args:
1484 f: file-like object to read from
1485 Returns: Iterator over tuples with SHA1s and ref names.
1486 """
1487 for line in f:
1488 if line.startswith(b"#"):
1489 # Comment
1490 continue
1491 if line.startswith(b"^"):
1492 raise PackedRefsException("found peeled ref in packed-refs without peeled")
1493 yield _split_ref_line(line)
1496def read_packed_refs_with_peeled(
1497 f: IO[bytes],
1498) -> Iterator[tuple[ObjectID, Ref, ObjectID | None]]:
1499 """Read a packed refs file including peeled refs.
1501 Assumes the "# pack-refs with: peeled" line was already read. Yields tuples
1502 with ref names, SHA1s, and peeled SHA1s (or None).
1504 Args:
1505 f: file-like object to read from, seek'ed to the second line
1506 """
1507 last = None
1508 for line in f:
1509 if line.startswith(b"#"):
1510 continue
1511 line = line.rstrip(b"\r\n")
1512 if line.startswith(b"^"):
1513 if not last:
1514 raise PackedRefsException("unexpected peeled ref line")
1515 if not valid_hexsha(line[1:]):
1516 raise PackedRefsException(f"Invalid hex sha {line[1:]!r}")
1517 sha, name = _split_ref_line(last)
1518 last = None
1519 yield (sha, name, ObjectID(line[1:]))
1520 else:
1521 if last:
1522 sha, name = _split_ref_line(last)
1523 yield (sha, name, None)
1524 last = line
1525 if last:
1526 sha, name = _split_ref_line(last)
1527 yield (sha, name, None)
1530def write_packed_refs(
1531 f: IO[bytes],
1532 packed_refs: Mapping[Ref, ObjectID],
1533 peeled_refs: Mapping[Ref, ObjectID] | None = None,
1534) -> None:
1535 """Write a packed refs file.
1537 Args:
1538 f: empty file-like object to write to
1539 packed_refs: dict of refname to sha of packed refs to write
1540 peeled_refs: dict of refname to peeled value of sha
1541 """
1542 if peeled_refs is None:
1543 peeled_refs = {}
1544 else:
1545 f.write(b"# pack-refs with: peeled\n")
1546 for refname in sorted(packed_refs.keys()):
1547 f.write(git_line(packed_refs[refname], refname))
1548 if refname in peeled_refs:
1549 f.write(b"^" + peeled_refs[refname] + b"\n")
1552def read_info_refs(f: BinaryIO) -> dict[Ref, ObjectID]:
1553 """Read info/refs file.
1555 Args:
1556 f: File-like object to read from
1558 Returns:
1559 Dictionary mapping ref names to SHA1s
1560 """
1561 ret: dict[Ref, ObjectID] = {}
1562 for line_no, line in enumerate(f.readlines(), 1):
1563 stripped = line.rstrip(b"\r\n")
1564 parts = stripped.split(b"\t", 1)
1565 if len(parts) != 2:
1566 raise ValueError(
1567 f"Invalid info/refs format at line {line_no}: "
1568 f"expected '<sha>\\t<refname>', got {stripped[:100]!r}"
1569 )
1570 (sha, name) = parts
1571 ret[Ref(name)] = ObjectID(sha)
1572 return ret
1575def is_local_branch(x: bytes) -> bool:
1576 """Check if a ref name is a local branch."""
1577 return x.startswith(LOCAL_BRANCH_PREFIX)
1580def _strip_leading_slash(name: bytes, kind: str) -> bytes:
1581 """Strip a leading slash from a short ref name, warning if one is present.
1583 A leading slash here means the caller stripped a ref prefix incorrectly
1584 (e.g. used ``ref[len(b"refs/tags"):]`` instead of ``len(b"refs/tags/")``).
1585 Joining such a name with a prefix would produce a malformed ref with an
1586 empty path component. Warn rather than raise for now; this will become an
1587 error in a future release.
1588 """
1589 if not name.startswith(b"/"):
1590 return name
1591 warnings.warn(
1592 f"{kind} name must not start with a slash: {name!r}; "
1593 "this will be rejected in a future version of Dulwich.",
1594 DeprecationWarning,
1595 stacklevel=3,
1596 )
1597 return name.lstrip(b"/")
1600def local_branch_name(name: bytes) -> Ref:
1601 """Build a full branch ref from a short name.
1603 Args:
1604 name: Short branch name (e.g., b"master") or full ref
1606 Returns:
1607 Full branch ref name (e.g., b"refs/heads/master")
1609 Examples:
1610 >>> local_branch_name(b"master")
1611 b'refs/heads/master'
1612 >>> local_branch_name(b"refs/heads/master")
1613 b'refs/heads/master'
1614 """
1615 if name.startswith(LOCAL_BRANCH_PREFIX):
1616 return Ref(name)
1617 return Ref(LOCAL_BRANCH_PREFIX + _strip_leading_slash(name, "Branch"))
1620def local_tag_name(name: bytes) -> Ref:
1621 """Build a full tag ref from a short name.
1623 Args:
1624 name: Short tag name (e.g., b"v1.0") or full ref
1626 Returns:
1627 Full tag ref name (e.g., b"refs/tags/v1.0")
1629 Examples:
1630 >>> local_tag_name(b"v1.0")
1631 b'refs/tags/v1.0'
1632 >>> local_tag_name(b"refs/tags/v1.0")
1633 b'refs/tags/v1.0'
1634 """
1635 if name.startswith(LOCAL_TAG_PREFIX):
1636 return Ref(name)
1637 return Ref(LOCAL_TAG_PREFIX + _strip_leading_slash(name, "Tag"))
1640def local_replace_name(name: bytes) -> Ref:
1641 """Build a full replace ref from a short name.
1643 Args:
1644 name: Short replace name (object SHA) or full ref
1646 Returns:
1647 Full replace ref name (e.g., b"refs/replace/<sha>")
1649 Examples:
1650 >>> local_replace_name(b"abc123")
1651 b'refs/replace/abc123'
1652 >>> local_replace_name(b"refs/replace/abc123")
1653 b'refs/replace/abc123'
1654 """
1655 if name.startswith(LOCAL_REPLACE_PREFIX):
1656 return Ref(name)
1657 return Ref(LOCAL_REPLACE_PREFIX + _strip_leading_slash(name, "Replace"))
1660def extract_branch_name(ref: bytes) -> bytes:
1661 """Extract branch name from a full branch ref.
1663 Args:
1664 ref: Full branch ref (e.g., b"refs/heads/master")
1666 Returns:
1667 Short branch name (e.g., b"master")
1669 Raises:
1670 ValueError: If ref is not a local branch
1672 Examples:
1673 >>> extract_branch_name(b"refs/heads/master")
1674 b'master'
1675 >>> extract_branch_name(b"refs/heads/feature/foo")
1676 b'feature/foo'
1677 """
1678 if not ref.startswith(LOCAL_BRANCH_PREFIX):
1679 raise ValueError(f"Not a local branch ref: {ref!r}")
1680 return ref[len(LOCAL_BRANCH_PREFIX) :]
1683def extract_tag_name(ref: bytes) -> bytes:
1684 """Extract tag name from a full tag ref.
1686 Args:
1687 ref: Full tag ref (e.g., b"refs/tags/v1.0")
1689 Returns:
1690 Short tag name (e.g., b"v1.0")
1692 Raises:
1693 ValueError: If ref is not a local tag
1695 Examples:
1696 >>> extract_tag_name(b"refs/tags/v1.0")
1697 b'v1.0'
1698 """
1699 if not ref.startswith(LOCAL_TAG_PREFIX):
1700 raise ValueError(f"Not a local tag ref: {ref!r}")
1701 return ref[len(LOCAL_TAG_PREFIX) :]
1704def shorten_ref_name(ref: bytes) -> bytes:
1705 """Convert a full ref name to its short form.
1707 Args:
1708 ref: Full ref name (e.g., b"refs/heads/master")
1710 Returns:
1711 Short ref name (e.g., b"master")
1713 Examples:
1714 >>> shorten_ref_name(b"refs/heads/master")
1715 b'master'
1716 >>> shorten_ref_name(b"refs/remotes/origin/main")
1717 b'origin/main'
1718 >>> shorten_ref_name(b"refs/tags/v1.0")
1719 b'v1.0'
1720 >>> shorten_ref_name(b"HEAD")
1721 b'HEAD'
1722 """
1723 if ref.startswith(LOCAL_BRANCH_PREFIX):
1724 return ref[len(LOCAL_BRANCH_PREFIX) :]
1725 elif ref.startswith(LOCAL_REMOTE_PREFIX):
1726 return ref[len(LOCAL_REMOTE_PREFIX) :]
1727 elif ref.startswith(LOCAL_TAG_PREFIX):
1728 return ref[len(LOCAL_TAG_PREFIX) :]
1729 return ref
1732def _set_origin_head(
1733 refs: RefsContainer, origin: bytes, origin_head: bytes | None
1734) -> None:
1735 # set refs/remotes/origin/HEAD
1736 origin_base = b"refs/remotes/" + origin + b"/"
1737 if origin_head and origin_head.startswith(LOCAL_BRANCH_PREFIX):
1738 origin_ref = Ref(origin_base + HEADREF)
1739 target_ref = Ref(origin_base + extract_branch_name(origin_head))
1740 if target_ref in refs:
1741 refs.set_symbolic_ref(origin_ref, target_ref)
1744def _set_default_branch(
1745 refs: RefsContainer,
1746 origin: bytes,
1747 origin_head: bytes | None,
1748 branch: bytes | None,
1749 ref_message: bytes | None,
1750) -> bytes:
1751 """Set the default branch."""
1752 origin_base = b"refs/remotes/" + origin + b"/"
1753 if branch:
1754 origin_ref = Ref(origin_base + branch)
1755 if origin_ref in refs:
1756 local_ref = Ref(local_branch_name(branch))
1757 refs.add_if_new(local_ref, refs[origin_ref], ref_message)
1758 head_ref = local_ref
1759 elif Ref(local_tag_name(branch)) in refs:
1760 head_ref = Ref(local_tag_name(branch))
1761 else:
1762 raise ValueError(f"{os.fsencode(branch)!r} is not a valid branch or tag")
1763 elif origin_head:
1764 head_ref = Ref(origin_head)
1765 if origin_head.startswith(LOCAL_BRANCH_PREFIX):
1766 origin_ref = Ref(origin_base + extract_branch_name(origin_head))
1767 else:
1768 origin_ref = Ref(origin_head)
1769 try:
1770 refs.add_if_new(head_ref, refs[origin_ref], ref_message)
1771 except KeyError:
1772 pass
1773 else:
1774 raise ValueError("neither origin_head nor branch are provided")
1775 return head_ref
1778def _set_head(
1779 refs: RefsContainer, head_ref: bytes, ref_message: bytes | None
1780) -> ObjectID | None:
1781 if head_ref.startswith(LOCAL_TAG_PREFIX):
1782 # detach HEAD at specified tag
1783 head = refs[Ref(head_ref)]
1784 del refs[HEADREF]
1785 refs.set_if_equals(HEADREF, None, head, message=ref_message)
1786 else:
1787 # set HEAD to specific branch
1788 try:
1789 head = refs[Ref(head_ref)]
1790 refs.set_symbolic_ref(HEADREF, Ref(head_ref))
1791 refs.set_if_equals(HEADREF, None, head, message=ref_message)
1792 except KeyError:
1793 head = None
1794 return head
1797def _set_branch_tracking(config: "ConfigFile", head_ref: bytes, remote: bytes) -> None:
1798 """Point the branch a clone checked out at its counterpart on the remote.
1800 This is what lets a subsequent "git pull" with no arguments know what to
1801 merge. Like git clone, nothing is written when HEAD ended up detached at a
1802 tag rather than on a branch.
1804 Args:
1805 config: Config of the freshly cloned repository
1806 head_ref: Local ref HEAD was pointed at, e.g. refs/heads/master
1807 remote: Name of the remote the clone came from
1808 """
1809 if not head_ref.startswith(LOCAL_BRANCH_PREFIX):
1810 return
1811 branch = extract_branch_name(Ref(head_ref))
1812 config.set((b"branch", branch), b"remote", remote)
1813 config.set((b"branch", branch), b"merge", head_ref)
1814 config.write_to_path()
1817def _import_remote_refs(
1818 refs_container: RefsContainer,
1819 remote_name: str,
1820 refs: Mapping[Ref, ObjectID | None],
1821 message: bytes | None = None,
1822 prune: bool = False,
1823 prune_tags: bool = False,
1824) -> None:
1825 from .protocol import PEELED_TAG_SUFFIX, strip_peeled_refs
1827 stripped_refs = strip_peeled_refs(refs)
1828 branches: dict[Ref, ObjectID | None] = {
1829 Ref(extract_branch_name(n)): v
1830 for (n, v) in stripped_refs.items()
1831 if n.startswith(LOCAL_BRANCH_PREFIX)
1832 }
1833 refs_container.import_refs(
1834 Ref(b"refs/remotes/" + remote_name.encode()),
1835 branches,
1836 message=message,
1837 prune=prune,
1838 )
1839 tags: dict[Ref, ObjectID | None] = {
1840 Ref(extract_tag_name(n)): v
1841 for (n, v) in stripped_refs.items()
1842 if n.startswith(LOCAL_TAG_PREFIX) and not n.endswith(PEELED_TAG_SUFFIX)
1843 }
1844 refs_container.import_refs(
1845 Ref(b"refs/tags"), tags, message=message, prune=prune_tags
1846 )
1849class locked_ref:
1850 """Lock a ref while making modifications.
1852 Works as a context manager.
1853 """
1855 def __init__(self, refs_container: DiskRefsContainer, refname: Ref) -> None:
1856 """Initialize a locked ref.
1858 Args:
1859 refs_container: The DiskRefsContainer to lock the ref in
1860 refname: The ref name to lock
1861 """
1862 self._refs_container = refs_container
1863 self._refname = refname
1864 self._file: _GitFile | None = None
1865 self._realname: Ref | None = None
1866 self._deleted = False
1868 def __enter__(self) -> Self:
1869 """Enter the context manager and acquire the lock.
1871 Returns:
1872 This locked_ref instance
1874 Raises:
1875 OSError: If the lock cannot be acquired
1876 """
1877 self._refs_container._check_refname(self._refname)
1878 try:
1879 realnames, _ = self._refs_container.follow(self._refname)
1880 self._realname = realnames[-1]
1881 except (KeyError, IndexError, SymrefLoop):
1882 self._realname = self._refname
1884 filename = self._refs_container.refpath(self._realname)
1885 ensure_dir_exists(os.path.dirname(filename))
1886 f = GitFile(filename, "wb")
1887 self._file = f
1888 return self
1890 def __exit__(
1891 self,
1892 exc_type: type | None,
1893 exc_value: BaseException | None,
1894 traceback: types.TracebackType | None,
1895 ) -> None:
1896 """Exit the context manager and release the lock.
1898 Args:
1899 exc_type: Type of exception if one occurred
1900 exc_value: Exception instance if one occurred
1901 traceback: Traceback if an exception occurred
1902 """
1903 if self._file:
1904 if exc_type is not None or self._deleted:
1905 self._file.abort()
1906 else:
1907 self._file.close()
1909 def get(self) -> bytes | None:
1910 """Get the current value of the ref."""
1911 if not self._file:
1912 raise RuntimeError("locked_ref not in context")
1914 assert self._realname is not None
1915 current_ref = self._refs_container.read_loose_ref(self._realname)
1916 if current_ref is None:
1917 current_ref = self._refs_container.get_packed_refs().get(
1918 self._realname, None
1919 )
1920 return current_ref
1922 def ensure_equals(self, expected_value: bytes | None) -> bool:
1923 """Ensure the ref currently equals the expected value.
1925 Args:
1926 expected_value: The expected current value of the ref
1927 Returns:
1928 True if the ref equals the expected value, False otherwise
1929 """
1930 current_value = self.get()
1931 return current_value == expected_value
1933 def set(self, new_ref: bytes) -> None:
1934 """Set the ref to a new value.
1936 Args:
1937 new_ref: The new SHA1 or symbolic ref value
1938 """
1939 if not self._file:
1940 raise RuntimeError("locked_ref not in context")
1942 if not (valid_hexsha(new_ref) or new_ref.startswith(SYMREF)):
1943 raise ValueError(f"{new_ref!r} must be a valid sha or a symref")
1945 self._file.seek(0)
1946 self._file.truncate()
1947 self._file.write(new_ref + b"\n")
1948 self._deleted = False
1950 def set_symbolic_ref(self, target: Ref) -> None:
1951 """Make this ref point at another ref.
1953 Args:
1954 target: Name of the ref to point at
1955 """
1956 if not self._file:
1957 raise RuntimeError("locked_ref not in context")
1959 self._refs_container._check_refname(target)
1960 self._file.seek(0)
1961 self._file.truncate()
1962 self._file.write(SYMREF + target + b"\n")
1963 self._deleted = False
1965 def delete(self) -> None:
1966 """Delete the ref file while holding the lock."""
1967 if not self._file:
1968 raise RuntimeError("locked_ref not in context")
1970 # Delete the actual ref file while holding the lock
1971 if self._realname:
1972 filename = self._refs_container.refpath(self._realname)
1973 try:
1974 if os.path.lexists(filename):
1975 os.remove(filename)
1976 except FileNotFoundError:
1977 pass
1978 self._refs_container._remove_packed_ref(self._realname)
1980 self._deleted = True
1983class NamespacedRefsContainer(RefsContainer):
1984 """Wrapper that adds namespace prefix to all ref operations.
1986 This implements Git's GIT_NAMESPACE feature, which stores refs under
1987 refs/namespaces/<namespace>/ and filters operations to only show refs
1988 within that namespace.
1990 Example:
1991 With namespace "foo", a ref "refs/heads/master" is stored as
1992 "refs/namespaces/foo/refs/heads/master" in the underlying container.
1993 """
1995 def __init__(self, refs: RefsContainer, namespace: bytes) -> None:
1996 """Initialize NamespacedRefsContainer.
1998 Args:
1999 refs: The underlying refs container to wrap
2000 namespace: The namespace prefix (e.g., b"foo" or b"foo/bar")
2001 """
2002 super().__init__(logger=refs._logger)
2003 self._refs = refs
2004 # Build namespace prefix: refs/namespaces/<namespace>/
2005 # Support nested namespaces: foo/bar -> refs/namespaces/foo/refs/namespaces/bar/
2006 namespace_parts = namespace.split(b"/")
2007 self._namespace_prefix = b""
2008 for part in namespace_parts:
2009 self._namespace_prefix += b"refs/namespaces/" + part + b"/"
2011 def _apply_namespace(self, name: bytes) -> bytes:
2012 """Apply namespace prefix to a ref name."""
2013 # HEAD and other special refs are not namespaced
2014 if name == HEADREF or not name.startswith(b"refs/"):
2015 return name
2016 return self._namespace_prefix + name
2018 def _strip_namespace(self, name: bytes) -> bytes | None:
2019 """Remove namespace prefix from a ref name.
2021 Returns None if the ref is not in our namespace.
2022 """
2023 # HEAD and other special refs are not namespaced
2024 if name == HEADREF or not name.startswith(b"refs/"):
2025 return name
2026 if name.startswith(self._namespace_prefix):
2027 return name[len(self._namespace_prefix) :]
2028 return None
2030 def allkeys(self) -> set[Ref]:
2031 """Return all reference keys in this namespace."""
2032 keys: set[Ref] = set()
2033 for key in self._refs.allkeys():
2034 stripped = self._strip_namespace(key)
2035 if stripped is not None:
2036 keys.add(Ref(stripped))
2037 return keys
2039 def read_loose_ref(self, name: Ref) -> bytes | None:
2040 """Read a loose reference."""
2041 return self._refs.read_loose_ref(Ref(self._apply_namespace(name)))
2043 def get_packed_refs(self) -> dict[Ref, ObjectID]:
2044 """Get packed refs within this namespace."""
2045 packed: dict[Ref, ObjectID] = {}
2046 for name, value in self._refs.get_packed_refs().items():
2047 stripped = self._strip_namespace(name)
2048 if stripped is not None:
2049 packed[Ref(stripped)] = value
2050 return packed
2052 def add_packed_refs(self, new_refs: Mapping[Ref, ObjectID | None]) -> None:
2053 """Add packed refs with namespace prefix."""
2054 namespaced_refs: dict[Ref, ObjectID | None] = {
2055 Ref(self._apply_namespace(name)): value for name, value in new_refs.items()
2056 }
2057 self._refs.add_packed_refs(namespaced_refs)
2059 def get_peeled(self, name: Ref) -> ObjectID | None:
2060 """Return the cached peeled value of a ref."""
2061 return self._refs.get_peeled(Ref(self._apply_namespace(name)))
2063 def set_symbolic_ref(
2064 self,
2065 name: Ref,
2066 other: Ref,
2067 committer: bytes | None = None,
2068 timestamp: int | None = None,
2069 timezone: int | None = None,
2070 message: bytes | None = None,
2071 ) -> None:
2072 """Make a ref point at another ref."""
2073 self._refs.set_symbolic_ref(
2074 Ref(self._apply_namespace(name)),
2075 Ref(self._apply_namespace(other)),
2076 committer=committer,
2077 timestamp=timestamp,
2078 timezone=timezone,
2079 message=message,
2080 )
2082 def set_if_equals(
2083 self,
2084 name: Ref,
2085 old_ref: ObjectID | None,
2086 new_ref: ObjectID,
2087 committer: bytes | None = None,
2088 timestamp: int | None = None,
2089 timezone: int | None = None,
2090 message: bytes | None = None,
2091 ) -> bool:
2092 """Set a refname to new_ref only if it currently equals old_ref."""
2093 return self._refs.set_if_equals(
2094 Ref(self._apply_namespace(name)),
2095 old_ref,
2096 new_ref,
2097 committer=committer,
2098 timestamp=timestamp,
2099 timezone=timezone,
2100 message=message,
2101 )
2103 def add_if_new(
2104 self,
2105 name: Ref,
2106 ref: ObjectID,
2107 committer: bytes | None = None,
2108 timestamp: int | None = None,
2109 timezone: int | None = None,
2110 message: bytes | None = None,
2111 ) -> bool:
2112 """Add a new reference only if it does not already exist."""
2113 return self._refs.add_if_new(
2114 Ref(self._apply_namespace(name)),
2115 ref,
2116 committer=committer,
2117 timestamp=timestamp,
2118 timezone=timezone,
2119 message=message,
2120 )
2122 def remove_if_equals(
2123 self,
2124 name: Ref,
2125 old_ref: ObjectID | None,
2126 committer: bytes | None = None,
2127 timestamp: int | None = None,
2128 timezone: int | None = None,
2129 message: bytes | None = None,
2130 ) -> bool:
2131 """Remove a refname only if it currently equals old_ref."""
2132 return self._refs.remove_if_equals(
2133 Ref(self._apply_namespace(name)),
2134 old_ref,
2135 committer=committer,
2136 timestamp=timestamp,
2137 timezone=timezone,
2138 message=message,
2139 )
2141 def pack_refs(self, all: bool = False) -> None:
2142 """Pack loose refs into packed-refs file.
2144 Note: This packs all refs in the underlying container, not just
2145 those in the namespace.
2146 """
2147 self._refs.pack_refs(all=all)
2150@overload
2151def filter_ref_prefix(
2152 refs: dict[Ref, ObjectID], prefixes: Iterable[bytes]
2153) -> dict[Ref, ObjectID]: ...
2156@overload
2157def filter_ref_prefix(
2158 refs: dict[Ref, ObjectID | None], prefixes: Iterable[bytes]
2159) -> dict[Ref, ObjectID | None]: ...
2162def filter_ref_prefix(
2163 refs: dict[Ref, ObjectID] | dict[Ref, ObjectID | None],
2164 prefixes: Iterable[bytes],
2165) -> dict[Ref, ObjectID] | dict[Ref, ObjectID | None]:
2166 """Filter refs to only include those with a given prefix.
2168 Args:
2169 refs: A dictionary of refs.
2170 prefixes: The prefixes to filter by.
2171 """
2172 return {k: v for k, v in refs.items() if any(k.startswith(p) for p in prefixes)}
2175def is_per_worktree_ref(ref: bytes) -> bool:
2176 """Returns whether a reference is stored per worktree or not.
2178 Per-worktree references are:
2179 - all pseudorefs, e.g. HEAD
2180 - all references stored inside "refs/bisect/", "refs/worktree/" and "refs/rewritten/"
2182 All refs starting with "refs/" are shared, except for the ones listed above.
2184 See https://git-scm.com/docs/git-worktree#_refs.
2185 """
2186 return not ref.startswith(b"refs/") or ref.startswith(
2187 (b"refs/bisect/", b"refs/worktree/", b"refs/rewritten/")
2188 )