Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/refs.py: 32%
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 "BAD_REF_CHARS",
27 "HEADREF",
28 "LOCAL_BRANCH_PREFIX",
29 "LOCAL_NOTES_PREFIX",
30 "LOCAL_REMOTE_PREFIX",
31 "LOCAL_REPLACE_PREFIX",
32 "LOCAL_TAG_PREFIX",
33 "SYMREF",
34 "DictRefsContainer",
35 "DiskRefsContainer",
36 "NamespacedRefsContainer",
37 "Ref",
38 "RefsContainer",
39 "SymrefLoop",
40 "check_ref_format",
41 "extract_branch_name",
42 "extract_tag_name",
43 "filter_ref_prefix",
44 "is_local_branch",
45 "is_per_worktree_ref",
46 "local_branch_name",
47 "local_replace_name",
48 "local_tag_name",
49 "locked_ref",
50 "parse_remote_ref",
51 "parse_symref_value",
52 "read_info_refs",
53 "read_packed_refs",
54 "read_packed_refs_with_peeled",
55 "refname_is_safe",
56 "set_ref_from_raw",
57 "shorten_ref_name",
58 "write_packed_refs",
59]
61import os
62import sys
63import types
64import warnings
65from collections.abc import Callable, Iterable, Iterator, Mapping
66from contextlib import suppress
67from typing import (
68 IO,
69 TYPE_CHECKING,
70 Any,
71 BinaryIO,
72 NewType,
73 overload,
74)
76if sys.version_info >= (3, 11):
77 from typing import Self
78else:
79 from typing_extensions import Self
81if TYPE_CHECKING:
82 from .config import ConfigFile
83 from .file import _GitFile
85from .errors import PackedRefsException, RefFormatError
86from .file import GitFile, ensure_dir_exists
87from .objects import ZERO_SHA, ObjectID, git_line, valid_hexsha
89Ref = NewType("Ref", bytes)
91HEADREF = Ref(b"HEAD")
92SYMREF = b"ref: "
93LOCAL_BRANCH_PREFIX = b"refs/heads/"
94LOCAL_TAG_PREFIX = b"refs/tags/"
95LOCAL_REMOTE_PREFIX = b"refs/remotes/"
96LOCAL_NOTES_PREFIX = b"refs/notes/"
97LOCAL_REPLACE_PREFIX = b"refs/replace/"
98BAD_REF_CHARS: set[int] = set(b"\177 ~^:?*[")
101class SymrefLoop(Exception):
102 """There is a loop between one or more symrefs."""
104 def __init__(self, ref: bytes, depth: int) -> None:
105 """Initialize SymrefLoop exception."""
106 self.ref = ref
107 self.depth = depth
110def parse_symref_value(contents: bytes) -> bytes:
111 """Parse a symref value.
113 Args:
114 contents: Contents to parse
115 Returns: Destination
116 """
117 if contents.startswith(SYMREF):
118 return contents[len(SYMREF) :].rstrip(b"\r\n")
119 raise ValueError(contents)
122def check_ref_format(refname: Ref) -> bool:
123 """Check if a refname is correctly formatted.
125 Implements all the same rules as git-check-ref-format[1].
127 [1]
128 http://www.kernel.org/pub/software/scm/git/docs/git-check-ref-format.html
130 Args:
131 refname: The refname to check
132 Returns: True if refname is valid, False otherwise
133 """
134 # These could be combined into one big expression, but are listed
135 # separately to parallel [1].
136 if refname == b"@":
137 return False
138 if b"/" not in refname: # type: ignore[comparison-overlap]
139 return False
140 if b".." in refname: # type: ignore[comparison-overlap]
141 return False
142 for i, c in enumerate(refname):
143 if ord(refname[i : i + 1]) < 0o40 or c in BAD_REF_CHARS:
144 return False
145 if refname[-1] in b"/.":
146 return False
147 if b"@{" in refname: # type: ignore[comparison-overlap]
148 return False
149 if b"\\" in refname: # type: ignore[comparison-overlap]
150 return False
151 for component in refname.split(b"/"):
152 if not component:
153 return False
154 if component.startswith(b"."):
155 return False
156 if component.endswith(b".lock"):
157 return False
158 return True
161def _collapse_slashes(refname: bytes) -> bytes:
162 """Collapse runs of consecutive slashes in a ref name into a single slash."""
163 return b"/".join(component for component in refname.split(b"/") if component)
166# git's isupper() is ASCII-only, so non-ASCII (e.g. UTF-8) bytes never match.
167_PSEUDOREF_CHARS = frozenset(b"ABCDEFGHIJKLMNOPQRSTUVWXYZ_")
170def refname_is_safe(refname: Ref) -> bool:
171 """Check if a refname is safe to use as a path within the ref store.
173 This is the equivalent of git's ``refname_is_safe()``: a weaker check than
174 :func:`check_ref_format`, meant for refnames that have already been
175 resolved (such as the target of a symref read from disk) and only
176 establishes that turning the name into a path cannot escape the ref store.
178 A name under ``refs/`` must already be normalized, so that neither
179 ``refs/../x`` nor ``refs/heads/../../x`` is accepted. Any other name must
180 be a pseudoref, i.e. consist only of uppercase letters and underscores
181 (e.g. ``HEAD`` or ``MERGE_HEAD``).
183 Args:
184 refname: The refname to check
185 Returns: True if the refname is safe, False otherwise
186 """
187 if refname.startswith(b"refs/"):
188 rest = refname[len(b"refs/") :]
189 # rest must not be empty or end with "/"
190 if not rest or rest.endswith(b"/"):
191 return False
192 # refpath() maps "/" onto os.sep, so on Windows a backslash would act
193 # as a second separator; check_ref_format rejects it as well.
194 if b"\\" in rest:
195 return False
196 # Unlike git, repeated slashes are tolerated here rather than treated
197 # as a denormalized name, because _check_refname still only deprecates
198 # them; only "." and ".." components make a name unsafe.
199 return not any(c in (b".", b"..") for c in rest.split(b"/"))
200 return bool(refname) and set(refname) <= _PSEUDOREF_CHARS
203def parse_remote_ref(ref: bytes) -> tuple[bytes, bytes]:
204 """Parse a remote ref into remote name and branch name.
206 Args:
207 ref: Remote ref like b"refs/remotes/origin/main"
209 Returns:
210 Tuple of (remote_name, branch_name)
212 Raises:
213 ValueError: If ref is not a valid remote ref
214 """
215 if not ref.startswith(LOCAL_REMOTE_PREFIX):
216 raise ValueError(f"Not a remote ref: {ref!r}")
218 # Remove the prefix
219 remainder = ref[len(LOCAL_REMOTE_PREFIX) :]
221 # Split into remote name and branch name
222 parts = remainder.split(b"/", 1)
223 if len(parts) != 2:
224 raise ValueError(f"Invalid remote ref format: {ref!r}")
226 remote_name, branch_name = parts
227 return (remote_name, branch_name)
230def set_ref_from_raw(refs: "RefsContainer", name: Ref, raw_ref: bytes) -> None:
231 """Set a reference from a raw ref value.
233 This handles both symbolic refs (starting with 'ref: ') and direct ObjectID refs.
235 Args:
236 refs: The RefsContainer to set the ref in
237 name: The ref name to set
238 raw_ref: The raw ref value (either a symbolic ref or an ObjectID)
239 """
240 if raw_ref.startswith(SYMREF):
241 # It's a symbolic ref
242 target = Ref(raw_ref[len(SYMREF) :])
243 refs.set_symbolic_ref(name, target)
244 else:
245 # It's a direct ObjectID
246 refs[name] = ObjectID(raw_ref)
249class RefsContainer:
250 """A container for refs."""
252 def __init__(
253 self,
254 logger: Callable[
255 [bytes, bytes, bytes, bytes | None, int | None, int | None, bytes], None
256 ]
257 | None = None,
258 ) -> None:
259 """Initialize RefsContainer with optional logger function."""
260 self._logger = logger
262 def _log(
263 self,
264 ref: bytes,
265 old_sha: bytes | None,
266 new_sha: bytes | None,
267 committer: bytes | None = None,
268 timestamp: int | None = None,
269 timezone: int | None = None,
270 message: bytes | None = None,
271 ) -> None:
272 if self._logger is None:
273 return
274 if message is None:
275 return
276 # Use ZERO_SHA for None values, matching git behavior
277 if old_sha is None:
278 old_sha = ZERO_SHA
279 if new_sha is None:
280 new_sha = ZERO_SHA
281 self._logger(ref, old_sha, new_sha, committer, timestamp, timezone, message)
283 def set_symbolic_ref(
284 self,
285 name: Ref,
286 other: Ref,
287 committer: bytes | None = None,
288 timestamp: int | None = None,
289 timezone: int | None = None,
290 message: bytes | None = None,
291 ) -> None:
292 """Make a ref point at another ref.
294 Args:
295 name: Name of the ref to set
296 other: Name of the ref to point at
297 committer: Optional committer name/email
298 timestamp: Optional timestamp
299 timezone: Optional timezone
300 message: Optional message
301 """
302 raise NotImplementedError(self.set_symbolic_ref)
304 def get_packed_refs(self) -> dict[Ref, ObjectID]:
305 """Get contents of the packed-refs file.
307 Returns: Dictionary mapping ref names to SHA1s
309 Note: Will return an empty dictionary when no packed-refs file is
310 present.
311 """
312 raise NotImplementedError(self.get_packed_refs)
314 def add_packed_refs(self, new_refs: Mapping[Ref, ObjectID | None]) -> None:
315 """Add the given refs as packed refs.
317 Args:
318 new_refs: A mapping of ref names to targets; if a target is None that
319 means remove the ref
320 """
321 raise NotImplementedError(self.add_packed_refs)
323 def get_peeled(self, name: Ref) -> ObjectID | None:
324 """Return the cached peeled value of a ref, if available.
326 Args:
327 name: Name of the ref to peel
328 Returns: The peeled value of the ref. If the ref is known not point to
329 a tag, this will be the SHA the ref refers to. If the ref may point
330 to a tag, but no cached information is available, None is returned.
331 """
332 return None
334 def import_refs(
335 self,
336 base: Ref,
337 other: Mapping[Ref, ObjectID | None],
338 committer: bytes | None = None,
339 timestamp: bytes | None = None,
340 timezone: bytes | None = None,
341 message: bytes | None = None,
342 prune: bool = False,
343 ) -> None:
344 """Import refs from another repository.
346 Args:
347 base: Base ref to import into (e.g., b'refs/remotes/origin')
348 other: Dictionary of refs to import
349 committer: Optional committer for reflog
350 timestamp: Optional timestamp for reflog
351 timezone: Optional timezone for reflog
352 message: Optional message for reflog
353 prune: If True, remove refs not in other
354 """
355 # Strip a trailing slash so joining ``base`` with a ref name below
356 # does not produce a malformed ref with an empty path component
357 # (e.g. b'refs/tags/' + b'/' + b'v1.0' -> b'refs/tags//v1.0').
358 base = Ref(base.rstrip(b"/"))
359 if prune:
360 to_delete = set(self.subkeys(base))
361 else:
362 to_delete = set()
363 for name, value in other.items():
364 if value is None:
365 to_delete.add(name)
366 else:
367 self.set_if_equals(
368 Ref(b"/".join((base, name))), None, value, message=message
369 )
370 if to_delete:
371 try:
372 to_delete.remove(name)
373 except KeyError:
374 pass
375 for ref in to_delete:
376 self.remove_if_equals(Ref(b"/".join((base, ref))), None, message=message)
378 def allkeys(self) -> set[Ref]:
379 """All refs present in this container."""
380 raise NotImplementedError(self.allkeys)
382 def __iter__(self) -> Iterator[Ref]:
383 """Iterate over all reference keys."""
384 return iter(self.allkeys())
386 def keys(self, base: Ref | None = None) -> set[Ref]:
387 """Refs present in this container.
389 Args:
390 base: An optional base to return refs under.
391 Returns: An unsorted set of valid refs in this container, including
392 packed refs.
393 """
394 if base is not None:
395 return self.subkeys(base)
396 else:
397 return self.allkeys()
399 def subkeys(self, base: Ref) -> set[Ref]:
400 """Refs present in this container under a base.
402 Args:
403 base: The base to return refs under.
404 Returns: A set of valid refs in this container under the base; the base
405 prefix is stripped from the ref names returned.
406 """
407 keys: set[Ref] = set()
408 base_len = len(base) + 1
409 for refname in self.allkeys():
410 if refname.startswith(base):
411 keys.add(Ref(refname[base_len:]))
412 return keys
414 def as_dict(self, base: Ref | None = None) -> dict[Ref, ObjectID]:
415 """Return the contents of this container as a dictionary."""
416 ret: dict[Ref, ObjectID] = {}
417 keys = self.keys(base)
418 base_bytes: bytes
419 if base is None:
420 base_bytes = b""
421 else:
422 base_bytes = base.rstrip(b"/")
423 for key in keys:
424 try:
425 ret[key] = self[Ref((base_bytes + b"/" + key).strip(b"/"))]
426 except (SymrefLoop, KeyError):
427 continue # Unable to resolve
429 return ret
431 def _check_refname(self, name: Ref) -> None:
432 """Ensure a refname is valid and lives in refs or is HEAD.
434 HEAD is not a valid refname according to git-check-ref-format, but this
435 class needs to be able to touch HEAD. Also, check_ref_format expects
436 refnames without the leading 'refs/', but this class requires that
437 so it cannot touch anything outside the refs dir (or HEAD).
439 Args:
440 name: The name of the reference.
442 Raises:
443 KeyError: if a refname is not HEAD or is otherwise not valid.
444 """
445 if name in (HEADREF, Ref(b"refs/stash")):
446 return
447 if not name.startswith(b"refs/"):
448 raise RefFormatError(name)
449 rest = Ref(name[5:])
450 if check_ref_format(rest):
451 return
452 # As of Dulwich 1.2.3 check_ref_format rejects empty path components
453 # (e.g. b'refs/tags//v1.0'). Such names were silently accepted before,
454 # and some callers (e.g. older Poetry releases) still construct them.
455 # Warn rather than raise for now if collapsing repeated slashes would
456 # make the name valid, so the only defect is empty components.
457 if (
458 b"//" in name # type: ignore[comparison-overlap]
459 and check_ref_format(Ref(_collapse_slashes(rest)))
460 ):
461 warnings.warn(
462 f"Ref name {name!r} contains empty path components; "
463 "this will be rejected in a future version of Dulwich.",
464 DeprecationWarning,
465 stacklevel=3,
466 )
467 return
468 raise RefFormatError(name)
470 def _check_ref_value(self, ref: ObjectID) -> None:
471 """Ensure a ref value is a valid object id or a symref.
473 Args:
474 ref: The value a ref is being set to.
476 Raises:
477 ValueError: if the value is neither a valid sha nor a symref.
478 """
479 if not (valid_hexsha(ref) or ref.startswith(SYMREF)):
480 raise ValueError(f"{ref!r} must be a valid sha or a symref")
482 def read_ref(self, refname: Ref) -> bytes | None:
483 """Read a reference without following any references.
485 Args:
486 refname: The name of the reference
487 Returns: The contents of the ref file, or None if it does
488 not exist.
489 """
490 contents = self.read_loose_ref(refname)
491 if not contents:
492 contents = self.get_packed_refs().get(refname, None)
493 return contents
495 def read_loose_ref(self, name: Ref) -> bytes | None:
496 """Read a loose reference and return its contents.
498 Args:
499 name: the refname to read
500 Returns: The contents of the ref file, or None if it does
501 not exist.
502 """
503 raise NotImplementedError(self.read_loose_ref)
505 def follow(self, name: Ref) -> tuple[list[Ref], ObjectID | None]:
506 """Follow a reference name.
508 Returns: a tuple of (refnames, sha), wheres refnames are the names of
509 references in the chain
510 """
511 contents: bytes | None = SYMREF + name
512 depth = 0
513 refnames: list[Ref] = []
514 while contents and contents.startswith(SYMREF):
515 refname = Ref(contents[len(SYMREF) :])
516 refnames.append(refname)
517 contents = self.read_ref(refname)
518 if not contents:
519 break
520 depth += 1
521 if depth > 5:
522 raise SymrefLoop(name, depth)
523 return refnames, ObjectID(contents) if contents else None
525 def __contains__(self, refname: Ref) -> bool:
526 """Check if a reference exists."""
527 if self.read_ref(refname):
528 return True
529 return False
531 def __getitem__(self, name: Ref) -> ObjectID:
532 """Get the SHA1 for a reference name.
534 This method follows all symbolic references.
535 """
536 _, sha = self.follow(name)
537 if sha is None:
538 raise KeyError(name)
539 return sha
541 def set_if_equals(
542 self,
543 name: Ref,
544 old_ref: ObjectID | None,
545 new_ref: ObjectID,
546 committer: bytes | None = None,
547 timestamp: int | None = None,
548 timezone: int | None = None,
549 message: bytes | None = None,
550 ) -> bool:
551 """Set a refname to new_ref only if it currently equals old_ref.
553 This method follows all symbolic references if applicable for the
554 subclass, and can be used to perform an atomic compare-and-swap
555 operation.
557 Args:
558 name: The refname to set.
559 old_ref: The old sha the refname must refer to, or None to set
560 unconditionally.
561 new_ref: The new sha the refname will refer to.
562 committer: Optional committer name/email
563 timestamp: Optional timestamp
564 timezone: Optional timezone
565 message: Message for reflog
566 Returns: True if the set was successful, False otherwise.
567 """
568 raise NotImplementedError(self.set_if_equals)
570 def add_if_new(
571 self,
572 name: Ref,
573 ref: ObjectID,
574 committer: bytes | None = None,
575 timestamp: int | None = None,
576 timezone: int | None = None,
577 message: bytes | None = None,
578 ) -> bool:
579 """Add a new reference only if it does not already exist.
581 Args:
582 name: Ref name
583 ref: Ref value
584 committer: Optional committer name/email
585 timestamp: Optional timestamp
586 timezone: Optional timezone
587 message: Optional message for reflog
588 """
589 raise NotImplementedError(self.add_if_new)
591 def __setitem__(self, name: Ref, ref: ObjectID) -> None:
592 """Set a reference name to point to the given SHA1.
594 This method follows all symbolic references if applicable for the
595 subclass.
597 Note: This method unconditionally overwrites the contents of a
598 reference. To update atomically only if the reference has not
599 changed, use set_if_equals().
601 Args:
602 name: The refname to set.
603 ref: The new sha the refname will refer to.
604 """
605 self._check_ref_value(ref)
606 self.set_if_equals(name, None, ref)
608 def remove_if_equals(
609 self,
610 name: Ref,
611 old_ref: ObjectID | None,
612 committer: bytes | None = None,
613 timestamp: int | None = None,
614 timezone: int | None = None,
615 message: bytes | None = None,
616 ) -> bool:
617 """Remove a refname only if it currently equals old_ref.
619 This method does not follow symbolic references, even if applicable for
620 the subclass. It can be used to perform an atomic compare-and-delete
621 operation.
623 Args:
624 name: The refname to delete.
625 old_ref: The old sha the refname must refer to, or None to
626 delete unconditionally.
627 committer: Optional committer name/email
628 timestamp: Optional timestamp
629 timezone: Optional timezone
630 message: Message for reflog
631 Returns: True if the delete was successful, False otherwise.
632 """
633 raise NotImplementedError(self.remove_if_equals)
635 def __delitem__(self, name: Ref) -> None:
636 """Remove a refname.
638 This method does not follow symbolic references, even if applicable for
639 the subclass.
641 Note: This method unconditionally deletes the contents of a reference.
642 To delete atomically only if the reference has not changed, use
643 remove_if_equals().
645 Args:
646 name: The refname to delete.
647 """
648 self.remove_if_equals(name, None)
650 def get_symrefs(self) -> dict[Ref, Ref]:
651 """Get a dict with all symrefs in this container.
653 Returns: Dictionary mapping source ref to target ref
654 """
655 ret: dict[Ref, Ref] = {}
656 for src in self.allkeys():
657 try:
658 ref_value = self.read_ref(src)
659 assert ref_value is not None
660 dst = parse_symref_value(ref_value)
661 except ValueError:
662 pass
663 else:
664 ret[src] = Ref(dst)
665 return ret
667 def pack_refs(self, all: bool = False) -> None:
668 """Pack loose refs into packed-refs file.
670 Args:
671 all: If True, pack all refs. If False, only pack tags.
672 """
673 raise NotImplementedError(self.pack_refs)
676class DictRefsContainer(RefsContainer):
677 """RefsContainer backed by a simple dict.
679 This container does not support symbolic or packed references and is not
680 threadsafe.
681 """
683 def __init__(
684 self,
685 refs: dict[Ref, bytes],
686 logger: Callable[
687 [
688 bytes,
689 bytes | None,
690 bytes | None,
691 bytes | None,
692 int | None,
693 int | None,
694 bytes | None,
695 ],
696 None,
697 ]
698 | None = None,
699 ) -> None:
700 """Initialize DictRefsContainer with refs dictionary and optional logger."""
701 super().__init__(logger=logger)
702 self._refs = refs
703 self._peeled: dict[Ref, ObjectID] = {}
704 self._watchers: set[Any] = set()
706 def allkeys(self) -> set[Ref]:
707 """Return all reference keys."""
708 return set(self._refs.keys())
710 def read_loose_ref(self, name: Ref) -> bytes | None:
711 """Read a loose reference."""
712 return self._refs.get(name, None)
714 def get_packed_refs(self) -> dict[Ref, ObjectID]:
715 """Get packed references."""
716 return {}
718 def _notify(self, ref: bytes, newsha: bytes | None) -> None:
719 for watcher in self._watchers:
720 watcher._notify((ref, newsha))
722 def set_symbolic_ref(
723 self,
724 name: Ref,
725 other: Ref,
726 committer: bytes | None = None,
727 timestamp: int | None = None,
728 timezone: int | None = None,
729 message: bytes | None = None,
730 ) -> None:
731 """Make a ref point at another ref.
733 Args:
734 name: Name of the ref to set
735 other: Name of the ref to point at
736 committer: Optional committer name for reflog
737 timestamp: Optional timestamp for reflog
738 timezone: Optional timezone for reflog
739 message: Optional message for reflog
740 """
741 old = self.follow(name)[-1]
742 new = SYMREF + other
743 self._refs[name] = new
744 self._notify(name, new)
745 self._log(
746 name,
747 old,
748 new,
749 committer=committer,
750 timestamp=timestamp,
751 timezone=timezone,
752 message=message,
753 )
755 def set_if_equals(
756 self,
757 name: Ref,
758 old_ref: ObjectID | None,
759 new_ref: ObjectID,
760 committer: bytes | None = None,
761 timestamp: int | None = None,
762 timezone: int | None = None,
763 message: bytes | None = None,
764 ) -> bool:
765 """Set a refname to new_ref only if it currently equals old_ref.
767 This method follows all symbolic references, and can be used to perform
768 an atomic compare-and-swap operation.
770 Args:
771 name: The refname to set.
772 old_ref: The old sha the refname must refer to, or None to set
773 unconditionally.
774 new_ref: The new sha the refname will refer to.
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 set was successful, False otherwise.
782 """
783 self._check_ref_value(new_ref)
784 if old_ref is not None and self._refs.get(name, ZERO_SHA) != old_ref:
785 return False
786 # Only update the specific ref requested, not the whole chain
787 self._check_refname(name)
788 old = self._refs.get(name)
789 self._refs[name] = new_ref
790 self._notify(name, new_ref)
791 self._log(
792 name,
793 old,
794 new_ref,
795 committer=committer,
796 timestamp=timestamp,
797 timezone=timezone,
798 message=message,
799 )
800 return True
802 def add_if_new(
803 self,
804 name: Ref,
805 ref: ObjectID,
806 committer: bytes | None = None,
807 timestamp: int | None = None,
808 timezone: int | None = None,
809 message: bytes | None = None,
810 ) -> bool:
811 """Add a new reference only if it does not already exist.
813 Args:
814 name: Ref name
815 ref: Ref value
816 committer: Optional committer name for reflog
817 timestamp: Optional timestamp for reflog
818 timezone: Optional timezone for reflog
819 message: Optional message for reflog
821 Returns:
822 True if the add was successful, False otherwise.
823 """
824 self._check_ref_value(ref)
825 if name in self._refs:
826 return False
827 self._refs[name] = ref
828 self._notify(name, ref)
829 self._log(
830 name,
831 None,
832 ref,
833 committer=committer,
834 timestamp=timestamp,
835 timezone=timezone,
836 message=message,
837 )
838 return True
840 def remove_if_equals(
841 self,
842 name: Ref,
843 old_ref: ObjectID | None,
844 committer: bytes | None = None,
845 timestamp: int | None = None,
846 timezone: int | None = None,
847 message: bytes | None = None,
848 ) -> bool:
849 """Remove a refname only if it currently equals old_ref.
851 This method does not follow symbolic references. It can be used to
852 perform an atomic compare-and-delete operation.
854 Args:
855 name: The refname to delete.
856 old_ref: The old sha the refname must refer to, or None to
857 delete unconditionally.
858 committer: Optional committer name for reflog
859 timestamp: Optional timestamp for reflog
860 timezone: Optional timezone for reflog
861 message: Optional message for reflog
863 Returns:
864 True if the delete was successful, False otherwise.
865 """
866 if old_ref is not None and self._refs.get(name, ZERO_SHA) != old_ref:
867 return False
868 try:
869 old = self._refs.pop(name)
870 except KeyError:
871 pass
872 else:
873 self._notify(name, None)
874 self._log(
875 name,
876 old,
877 None,
878 committer=committer,
879 timestamp=timestamp,
880 timezone=timezone,
881 message=message,
882 )
883 return True
885 def get_peeled(self, name: Ref) -> ObjectID | None:
886 """Get peeled version of a reference."""
887 return self._peeled.get(name)
889 def _update(self, refs: Mapping[Ref, ObjectID]) -> None:
890 """Update multiple refs; intended only for testing."""
891 # TODO(dborowitz): replace this with a public function that uses
892 # set_if_equal.
893 for ref, sha in refs.items():
894 self.set_if_equals(ref, None, sha)
896 def _update_peeled(self, peeled: Mapping[Ref, ObjectID]) -> None:
897 """Update cached peeled refs; intended only for testing."""
898 self._peeled.update(peeled)
901#: Identity of a particular packed-refs file, used to detect that the file a
902#: cache was populated from has since been replaced.
903_PackedRefsKey = tuple[int, int, int, int, int]
906def _packed_refs_key(st: os.stat_result) -> _PackedRefsKey:
907 """Build a key identifying the packed-refs file described by ``st``.
909 The fields mirror those git compares in ``match_stat_data()`` when
910 deciding whether a file it has cached has been replaced. Size alone is a
911 weak signal here, since repacking refs often produces a file of identical
912 length, so the inode and timestamps do most of the work.
914 Args:
915 st: Stat result for a packed-refs file.
916 Returns: An opaque, comparable key.
917 """
918 return (st.st_ino, st.st_dev, st.st_size, st.st_mtime_ns, st.st_ctime_ns)
921class DiskRefsContainer(RefsContainer):
922 """Refs container that reads refs from disk."""
924 def __init__(
925 self,
926 path: str | bytes | os.PathLike[str],
927 worktree_path: str | bytes | os.PathLike[str] | None = None,
928 logger: Callable[
929 [bytes, bytes, bytes, bytes | None, int | None, int | None, bytes], None
930 ]
931 | None = None,
932 ) -> None:
933 """Initialize DiskRefsContainer."""
934 super().__init__(logger=logger)
935 # Convert path-like objects to strings, then to bytes for Git compatibility
936 self.path = os.fsencode(os.fspath(path))
937 if worktree_path is None:
938 self.worktree_path = self.path
939 else:
940 self.worktree_path = os.fsencode(os.fspath(worktree_path))
941 self._packed_refs: dict[Ref, ObjectID] | None = None
942 self._peeled_refs: dict[Ref, ObjectID] | None = None
943 self._packed_refs_key: _PackedRefsKey | None = None
945 def __repr__(self) -> str:
946 """Return string representation of DiskRefsContainer."""
947 return f"{self.__class__.__name__}({self.path!r})"
949 def _iter_dir(
950 self,
951 path: bytes,
952 base: bytes,
953 dir_filter: Callable[[bytes], bool] | None = None,
954 ) -> Iterator[Ref]:
955 refspath = os.path.join(path, base.rstrip(b"/"))
956 prefix_len = len(os.path.join(path, b""))
958 for root, dirs, files in os.walk(refspath):
959 directory = root[prefix_len:]
960 if os.path.sep != "/":
961 directory = directory.replace(os.fsencode(os.path.sep), b"/")
962 if dir_filter is not None:
963 dirs[:] = [
964 d for d in dirs if dir_filter(b"/".join([directory, d, b""]))
965 ]
967 for filename in files:
968 refname = b"/".join([directory, filename])
969 if check_ref_format(Ref(refname)):
970 yield Ref(refname)
972 def _iter_loose_refs(self, base: bytes = b"refs/") -> Iterator[Ref]:
973 base = base.rstrip(b"/") + b"/"
974 search_paths: list[tuple[bytes, Callable[[bytes], bool] | None]] = []
975 if base != b"refs/":
976 path = self.worktree_path if is_per_worktree_ref(base) else self.path
977 search_paths.append((path, None))
978 elif self.worktree_path == self.path:
979 # Iterate through all the refs from the main worktree
980 search_paths.append((self.path, None))
981 else:
982 # Iterate through all the shared refs from the commondir, excluding per-worktree refs
983 search_paths.append((self.path, lambda r: not is_per_worktree_ref(r)))
984 # Iterate through all the per-worktree refs from the worktree's gitdir
985 search_paths.append((self.worktree_path, is_per_worktree_ref))
987 for path, dir_filter in search_paths:
988 yield from self._iter_dir(path, base, dir_filter=dir_filter)
990 def subkeys(self, base: Ref) -> set[Ref]:
991 """Return subkeys under a given base reference path."""
992 subkeys: set[Ref] = set()
994 for key in self._iter_loose_refs(base):
995 if key.startswith(base):
996 subkeys.add(Ref(key[len(base) :].strip(b"/")))
998 for key in self.get_packed_refs():
999 if key.startswith(base):
1000 subkeys.add(Ref(key[len(base) :].strip(b"/")))
1001 return subkeys
1003 def allkeys(self) -> set[Ref]:
1004 """Return all reference keys."""
1005 allkeys: set[Ref] = set()
1006 if os.path.exists(self.refpath(HEADREF)):
1007 allkeys.add(Ref(HEADREF))
1009 allkeys.update(self._iter_loose_refs())
1010 allkeys.update(self.get_packed_refs())
1011 return allkeys
1013 def refpath(self, name: bytes) -> bytes:
1014 """Return the disk path of a ref."""
1015 path = name
1016 if os.path.sep != "/":
1017 path = path.replace(b"/", os.fsencode(os.path.sep))
1019 root_dir = self.worktree_path if is_per_worktree_ref(name) else self.path
1020 return os.path.join(root_dir, path)
1022 def _current_packed_refs_key(self) -> _PackedRefsKey | None:
1023 """Identify the packed-refs file currently on disk.
1025 Returns: An opaque key for the current packed-refs file, or None if
1026 no packed-refs file is present.
1027 """
1028 try:
1029 st = os.stat(os.path.join(self.path, b"packed-refs"))
1030 except FileNotFoundError:
1031 return None
1032 return _packed_refs_key(st)
1034 def _invalidate_packed_refs_cache(self) -> None:
1035 """Discard cached packed and peeled refs."""
1036 self._packed_refs = None
1037 self._peeled_refs = None
1038 self._packed_refs_key = None
1040 def get_packed_refs(self) -> dict[Ref, ObjectID]:
1041 """Get contents of the packed-refs file.
1043 Returns: Dictionary mapping ref names to SHA1s
1045 Note: Will return an empty dictionary when no packed-refs file is
1046 present.
1047 """
1048 if (
1049 self._packed_refs is not None
1050 and self._packed_refs_key != self._current_packed_refs_key()
1051 ):
1052 # The packed-refs file was replaced underneath us, most likely by
1053 # another process packing or unpacking refs. Drop the stale cache
1054 # so that it is reloaded below.
1055 self._invalidate_packed_refs_cache()
1057 if self._packed_refs is None:
1058 # set both to empty because we want _peeled_refs to be
1059 # None if and only if _packed_refs is also None.
1060 self._packed_refs = {}
1061 self._peeled_refs = {}
1062 self._packed_refs_key = None
1063 path = os.path.join(self.path, b"packed-refs")
1064 try:
1065 f = GitFile(path, "rb")
1066 except FileNotFoundError:
1067 return {}
1068 with f:
1069 first_line = next(iter(f)).rstrip()
1070 if first_line.startswith(b"# pack-refs") and b" peeled" in first_line:
1071 for sha, name, peeled in read_packed_refs_with_peeled(f):
1072 self._packed_refs[name] = sha
1073 if peeled:
1074 self._peeled_refs[name] = peeled
1075 else:
1076 f.seek(0)
1077 for sha, name in read_packed_refs(f):
1078 self._packed_refs[name] = sha
1079 # Record which file the cache was populated from, so that a
1080 # later replacement of it can be detected. Stat the open file
1081 # rather than the path to avoid picking up a newer file that
1082 # was renamed into place while we were reading.
1083 self._packed_refs_key = _packed_refs_key(os.fstat(f.fileno()))
1084 return self._packed_refs
1086 def add_packed_refs(self, new_refs: Mapping[Ref, ObjectID | None]) -> None:
1087 """Add the given refs as packed refs.
1089 Args:
1090 new_refs: A mapping of ref names to targets; if a target is None that
1091 means remove the ref
1092 """
1093 if not new_refs:
1094 return
1096 path = os.path.join(self.path, b"packed-refs")
1098 try:
1099 with GitFile(path, "wb") as f:
1100 # reread cached refs from disk, while holding the lock
1101 packed_refs = self.get_packed_refs().copy()
1103 for ref, target in new_refs.items():
1104 # sanity check
1105 if ref == HEADREF:
1106 raise ValueError("cannot pack HEAD")
1108 # remove any loose refs pointing to this one -- please
1109 # note that this bypasses remove_if_equals as we don't
1110 # want to affect packed refs in here
1111 with suppress(OSError):
1112 os.remove(self.refpath(ref))
1114 if target is not None:
1115 packed_refs[ref] = target
1116 else:
1117 packed_refs.pop(ref, None)
1119 write_packed_refs(f, packed_refs, self._peeled_refs)
1120 finally:
1121 # Do not stat the path and associate that identity with the data
1122 # just written: another writer can replace packed-refs after the
1123 # lock is released but before the stat. Reload on the next access
1124 # instead.
1125 self._invalidate_packed_refs_cache()
1127 def get_peeled(self, name: Ref) -> ObjectID | None:
1128 """Return the cached peeled value of a ref, if available.
1130 Args:
1131 name: Name of the ref to peel
1132 Returns: The peeled value of the ref. If the ref is known not point to
1133 a tag, this will be the SHA the ref refers to. If the ref may point
1134 to a tag, but no cached information is available, None is returned.
1135 """
1136 self.get_packed_refs()
1137 if (
1138 self._peeled_refs is None
1139 or self._packed_refs is None
1140 or name not in self._packed_refs
1141 ):
1142 # No cache: no peeled refs were read, or this ref is loose
1143 return None
1144 if name in self._peeled_refs:
1145 return self._peeled_refs[name]
1146 else:
1147 # Known not peelable
1148 return self[name]
1150 def read_loose_ref(self, name: Ref) -> bytes | None:
1151 """Read a reference file and return its contents.
1153 If the reference file a symbolic reference, only read the first line of
1154 the file. Otherwise, read the hash (40 bytes for SHA1, 64 bytes for SHA256).
1156 Args:
1157 name: the refname to read, relative to refpath
1158 Returns: The contents of the ref file, or None if the file does not
1159 exist.
1161 Raises:
1162 IOError: if any other error occurs
1163 """
1164 # Validate the name before turning it into a path.
1165 try:
1166 self._check_refname(name)
1167 except RefFormatError:
1168 return None
1169 return self._read_loose_ref_file(name)
1171 def _read_loose_ref_file(self, name: Ref) -> bytes | None:
1172 """Read a reference file without validating its name.
1174 Callers must have checked the name with :func:`refname_is_safe`.
1175 """
1176 filename = self.refpath(name)
1177 try:
1178 with GitFile(filename, "rb") as f:
1179 header = f.read(len(SYMREF))
1180 if header == SYMREF:
1181 # Read only the first line
1182 return header + next(iter(f)).rstrip(b"\r\n")
1183 else:
1184 # Read the entire line to get the full hash (handles both SHA1 and SHA256)
1185 f.seek(0)
1186 line = f.readline().rstrip(b"\r\n")
1187 return line
1188 except (OSError, UnicodeError):
1189 # don't assume anything specific about the error; in
1190 # particular, invalid or forbidden paths can raise weird
1191 # errors depending on the specific operating system
1192 return None
1194 def _remove_packed_ref(self, name: Ref) -> None:
1195 if name not in self.get_packed_refs():
1196 return
1198 filename = os.path.join(self.path, b"packed-refs")
1199 # reread cached refs from disk, while holding the lock
1200 f = GitFile(filename, "wb")
1201 try:
1202 self._invalidate_packed_refs_cache()
1203 packed_refs = self.get_packed_refs().copy()
1204 peeled_refs = (
1205 self._peeled_refs.copy() if self._peeled_refs is not None else None
1206 )
1208 if name not in packed_refs:
1209 f.abort()
1210 return
1212 del packed_refs[name]
1213 if peeled_refs is not None:
1214 peeled_refs.pop(name, None)
1215 write_packed_refs(f, packed_refs, peeled_refs)
1216 f.close()
1217 finally:
1218 if not f.closed:
1219 f.abort()
1220 self._invalidate_packed_refs_cache()
1222 def set_symbolic_ref(
1223 self,
1224 name: Ref,
1225 other: Ref,
1226 committer: bytes | None = None,
1227 timestamp: int | None = None,
1228 timezone: int | None = None,
1229 message: bytes | None = None,
1230 ) -> None:
1231 """Make a ref point at another ref.
1233 Args:
1234 name: Name of the ref to set
1235 other: Name of the ref to point at
1236 committer: Optional committer name
1237 timestamp: Optional timestamp
1238 timezone: Optional timezone
1239 message: Optional message to describe the change
1240 """
1241 self._check_refname(name)
1242 self._check_refname(other)
1243 filename = self.refpath(name)
1244 f = GitFile(filename, "wb")
1245 try:
1246 f.write(SYMREF + other + b"\n")
1247 sha = self.follow(name)[-1]
1248 self._log(
1249 name,
1250 sha,
1251 sha,
1252 committer=committer,
1253 timestamp=timestamp,
1254 timezone=timezone,
1255 message=message,
1256 )
1257 except BaseException:
1258 f.abort()
1259 raise
1260 else:
1261 f.close()
1263 def set_if_equals(
1264 self,
1265 name: Ref,
1266 old_ref: ObjectID | None,
1267 new_ref: ObjectID,
1268 committer: bytes | None = None,
1269 timestamp: int | None = None,
1270 timezone: int | None = None,
1271 message: bytes | None = None,
1272 ) -> bool:
1273 """Set a refname to new_ref only if it currently equals old_ref.
1275 This method follows all symbolic references, and can be used to perform
1276 an atomic compare-and-swap operation.
1278 Args:
1279 name: The refname to set.
1280 old_ref: The old sha the refname must refer to, or None to set
1281 unconditionally.
1282 new_ref: The new sha the refname will refer to.
1283 committer: Optional committer name
1284 timestamp: Optional timestamp
1285 timezone: Optional timezone
1286 message: Set message for reflog
1287 Returns: True if the set was successful, False otherwise.
1288 """
1289 self._check_refname(name)
1290 self._check_ref_value(new_ref)
1291 try:
1292 realnames, _ = self.follow(name)
1293 realname = realnames[-1]
1294 except (KeyError, IndexError, SymrefLoop):
1295 realname = name
1296 # The resolved symref target may escape the ref store.
1297 if not refname_is_safe(realname):
1298 raise RefFormatError(realname)
1299 filename = self.refpath(realname)
1301 # make sure none of the ancestor folders is in packed refs
1302 probe_ref = Ref(os.path.dirname(realname))
1303 packed_refs = self.get_packed_refs()
1304 while probe_ref:
1305 if packed_refs.get(probe_ref, None) is not None:
1306 raise NotADirectoryError(filename)
1307 probe_ref = Ref(os.path.dirname(probe_ref))
1309 ensure_dir_exists(os.path.dirname(filename))
1310 with GitFile(filename, "wb") as f:
1311 if old_ref is not None:
1312 try:
1313 # read again while holding the lock to handle race conditions
1314 orig_ref = self._read_loose_ref_file(realname)
1315 if orig_ref is None:
1316 orig_ref = self.get_packed_refs().get(realname, ZERO_SHA)
1317 if orig_ref != old_ref:
1318 f.abort()
1319 return False
1320 except OSError:
1321 f.abort()
1322 raise
1324 # Check if ref already has the desired value while holding the lock
1325 # This avoids fsync when ref is unchanged but still detects lock conflicts
1326 current_ref = self._read_loose_ref_file(realname)
1327 if current_ref is None:
1328 current_ref = packed_refs.get(realname, None)
1330 if current_ref is not None and current_ref == new_ref:
1331 # Ref already has desired value, abort write to avoid fsync
1332 f.abort()
1333 return True
1335 try:
1336 f.write(new_ref + b"\n")
1337 except OSError:
1338 f.abort()
1339 raise
1340 self._log(
1341 realname,
1342 old_ref,
1343 new_ref,
1344 committer=committer,
1345 timestamp=timestamp,
1346 timezone=timezone,
1347 message=message,
1348 )
1349 return True
1351 def add_if_new(
1352 self,
1353 name: Ref,
1354 ref: ObjectID,
1355 committer: bytes | None = None,
1356 timestamp: int | None = None,
1357 timezone: int | None = None,
1358 message: bytes | None = None,
1359 ) -> bool:
1360 """Add a new reference only if it does not already exist.
1362 This method follows symrefs, and only ensures that the last ref in the
1363 chain does not exist.
1365 Args:
1366 name: The refname to set.
1367 ref: The new sha the refname will refer to.
1368 committer: Optional committer name
1369 timestamp: Optional timestamp
1370 timezone: Optional timezone
1371 message: Optional message for reflog
1372 Returns: True if the add was successful, False otherwise.
1373 """
1374 self._check_refname(name)
1375 self._check_ref_value(ref)
1376 try:
1377 realnames, contents = self.follow(name)
1378 if contents is not None:
1379 return False
1380 realname = realnames[-1]
1381 except (KeyError, IndexError):
1382 realname = name
1383 if not refname_is_safe(realname):
1384 raise RefFormatError(realname)
1385 filename = self.refpath(realname)
1386 ensure_dir_exists(os.path.dirname(filename))
1387 with GitFile(filename, "wb") as f:
1388 if os.path.exists(filename) or name in self.get_packed_refs():
1389 f.abort()
1390 return False
1391 try:
1392 f.write(ref + b"\n")
1393 except OSError:
1394 f.abort()
1395 raise
1396 else:
1397 self._log(
1398 name,
1399 None,
1400 ref,
1401 committer=committer,
1402 timestamp=timestamp,
1403 timezone=timezone,
1404 message=message,
1405 )
1406 return True
1408 def remove_if_equals(
1409 self,
1410 name: Ref,
1411 old_ref: ObjectID | None,
1412 committer: bytes | None = None,
1413 timestamp: int | None = None,
1414 timezone: int | None = None,
1415 message: bytes | None = None,
1416 ) -> bool:
1417 """Remove a refname only if it currently equals old_ref.
1419 This method does not follow symbolic references. It can be used to
1420 perform an atomic compare-and-delete operation.
1422 Args:
1423 name: The refname to delete.
1424 old_ref: The old sha the refname must refer to, or None to
1425 delete unconditionally.
1426 committer: Optional committer name
1427 timestamp: Optional timestamp
1428 timezone: Optional timezone
1429 message: Optional message
1430 Returns: True if the delete was successful, False otherwise.
1431 """
1432 self._check_refname(name)
1433 filename = self.refpath(name)
1434 ensure_dir_exists(os.path.dirname(filename))
1435 f = GitFile(filename, "wb")
1436 try:
1437 if old_ref is not None:
1438 orig_ref = self.read_loose_ref(name)
1439 if orig_ref is None:
1440 orig_ref = self.get_packed_refs().get(name)
1441 if orig_ref is None:
1442 orig_ref = ZERO_SHA
1443 if orig_ref != old_ref:
1444 return False
1446 # remove the reference file itself
1447 try:
1448 found = os.path.lexists(filename)
1449 except OSError:
1450 # may only be packed, or otherwise unstorable
1451 found = False
1453 if found:
1454 os.remove(filename)
1456 self._remove_packed_ref(name)
1457 self._log(
1458 name,
1459 old_ref,
1460 None,
1461 committer=committer,
1462 timestamp=timestamp,
1463 timezone=timezone,
1464 message=message,
1465 )
1466 finally:
1467 # never write, we just wanted the lock
1468 f.abort()
1470 # outside of the lock, clean-up any parent directory that might now
1471 # be empty. this ensures that re-creating a reference of the same
1472 # name of what was previously a directory works as expected
1473 parent = name
1474 while True:
1475 try:
1476 parent_bytes, _ = parent.rsplit(b"/", 1)
1477 parent = Ref(parent_bytes)
1478 except ValueError:
1479 break
1481 if parent == b"refs":
1482 break
1483 parent_filename = self.refpath(parent)
1484 try:
1485 os.rmdir(parent_filename)
1486 except OSError:
1487 # this can be caused by the parent directory being
1488 # removed by another process, being not empty, etc.
1489 # in any case, this is non fatal because we already
1490 # removed the reference, just ignore it
1491 break
1493 return True
1495 def pack_refs(self, all: bool = False) -> None:
1496 """Pack loose refs into packed-refs file.
1498 Args:
1499 all: If True, pack all refs. If False, only pack tags.
1500 """
1501 refs_to_pack: dict[Ref, ObjectID | None] = {}
1502 for ref in self.allkeys():
1503 if ref == HEADREF:
1504 # Never pack HEAD
1505 continue
1506 if all or ref.startswith(LOCAL_TAG_PREFIX):
1507 try:
1508 sha = self[ref]
1509 if sha:
1510 refs_to_pack[ref] = sha
1511 except KeyError:
1512 # Broken ref, skip it
1513 pass
1515 if refs_to_pack:
1516 self.add_packed_refs(refs_to_pack)
1519def _split_ref_line(line: bytes) -> tuple[ObjectID, Ref]:
1520 """Split a single ref line into a tuple of SHA1 and name."""
1521 fields = line.rstrip(b"\n\r").split(b" ")
1522 if len(fields) != 2:
1523 raise PackedRefsException(f"invalid ref line {line!r}")
1524 sha, name = fields
1525 if not valid_hexsha(sha):
1526 raise PackedRefsException(f"Invalid hex sha {sha!r}")
1527 if not check_ref_format(Ref(name)):
1528 raise PackedRefsException(f"invalid ref name {name!r}")
1529 return (ObjectID(sha), Ref(name))
1532def read_packed_refs(f: IO[bytes]) -> Iterator[tuple[ObjectID, Ref]]:
1533 """Read a packed refs file.
1535 Args:
1536 f: file-like object to read from
1537 Returns: Iterator over tuples with SHA1s and ref names.
1538 """
1539 for line in f:
1540 if line.startswith(b"#"):
1541 # Comment
1542 continue
1543 if line.startswith(b"^"):
1544 raise PackedRefsException("found peeled ref in packed-refs without peeled")
1545 yield _split_ref_line(line)
1548def read_packed_refs_with_peeled(
1549 f: IO[bytes],
1550) -> Iterator[tuple[ObjectID, Ref, ObjectID | None]]:
1551 """Read a packed refs file including peeled refs.
1553 Assumes the "# pack-refs with: peeled" line was already read. Yields tuples
1554 with ref names, SHA1s, and peeled SHA1s (or None).
1556 Args:
1557 f: file-like object to read from, seek'ed to the second line
1558 """
1559 last = None
1560 for line in f:
1561 if line.startswith(b"#"):
1562 continue
1563 line = line.rstrip(b"\r\n")
1564 if line.startswith(b"^"):
1565 if not last:
1566 raise PackedRefsException("unexpected peeled ref line")
1567 if not valid_hexsha(line[1:]):
1568 raise PackedRefsException(f"Invalid hex sha {line[1:]!r}")
1569 sha, name = _split_ref_line(last)
1570 last = None
1571 yield (sha, name, ObjectID(line[1:]))
1572 else:
1573 if last:
1574 sha, name = _split_ref_line(last)
1575 yield (sha, name, None)
1576 last = line
1577 if last:
1578 sha, name = _split_ref_line(last)
1579 yield (sha, name, None)
1582def write_packed_refs(
1583 f: IO[bytes],
1584 packed_refs: Mapping[Ref, ObjectID],
1585 peeled_refs: Mapping[Ref, ObjectID] | None = None,
1586) -> None:
1587 """Write a packed refs file.
1589 Args:
1590 f: empty file-like object to write to
1591 packed_refs: dict of refname to sha of packed refs to write
1592 peeled_refs: dict of refname to peeled value of sha
1593 """
1594 if peeled_refs is None:
1595 peeled_refs = {}
1596 else:
1597 f.write(b"# pack-refs with: peeled\n")
1598 for refname in sorted(packed_refs.keys()):
1599 f.write(git_line(packed_refs[refname], refname))
1600 if refname in peeled_refs:
1601 f.write(b"^" + peeled_refs[refname] + b"\n")
1604def read_info_refs(f: BinaryIO) -> dict[Ref, ObjectID]:
1605 """Read info/refs file.
1607 Args:
1608 f: File-like object to read from
1610 Returns:
1611 Dictionary mapping ref names to SHA1s
1612 """
1613 ret: dict[Ref, ObjectID] = {}
1614 for line_no, line in enumerate(f.readlines(), 1):
1615 stripped = line.rstrip(b"\r\n")
1616 parts = stripped.split(b"\t", 1)
1617 if len(parts) != 2:
1618 raise ValueError(
1619 f"Invalid info/refs format at line {line_no}: "
1620 f"expected '<sha>\\t<refname>', got {stripped[:100]!r}"
1621 )
1622 (sha, name) = parts
1623 ret[Ref(name)] = ObjectID(sha)
1624 return ret
1627def is_local_branch(x: bytes) -> bool:
1628 """Check if a ref name is a local branch."""
1629 return x.startswith(LOCAL_BRANCH_PREFIX)
1632def _strip_leading_slash(name: bytes, kind: str) -> bytes:
1633 """Strip a leading slash from a short ref name, warning if one is present.
1635 A leading slash here means the caller stripped a ref prefix incorrectly
1636 (e.g. used ``ref[len(b"refs/tags"):]`` instead of ``len(b"refs/tags/")``).
1637 Joining such a name with a prefix would produce a malformed ref with an
1638 empty path component. Warn rather than raise for now; this will become an
1639 error in a future release.
1640 """
1641 if not name.startswith(b"/"):
1642 return name
1643 warnings.warn(
1644 f"{kind} name must not start with a slash: {name!r}; "
1645 "this will be rejected in a future version of Dulwich.",
1646 DeprecationWarning,
1647 stacklevel=3,
1648 )
1649 return name.lstrip(b"/")
1652def local_branch_name(name: bytes) -> Ref:
1653 """Build a full branch ref from a short name.
1655 Args:
1656 name: Short branch name (e.g., b"master") or full ref
1658 Returns:
1659 Full branch ref name (e.g., b"refs/heads/master")
1661 Examples:
1662 >>> local_branch_name(b"master")
1663 b'refs/heads/master'
1664 >>> local_branch_name(b"refs/heads/master")
1665 b'refs/heads/master'
1666 """
1667 if name.startswith(LOCAL_BRANCH_PREFIX):
1668 return Ref(name)
1669 return Ref(LOCAL_BRANCH_PREFIX + _strip_leading_slash(name, "Branch"))
1672def local_tag_name(name: bytes) -> Ref:
1673 """Build a full tag ref from a short name.
1675 Args:
1676 name: Short tag name (e.g., b"v1.0") or full ref
1678 Returns:
1679 Full tag ref name (e.g., b"refs/tags/v1.0")
1681 Examples:
1682 >>> local_tag_name(b"v1.0")
1683 b'refs/tags/v1.0'
1684 >>> local_tag_name(b"refs/tags/v1.0")
1685 b'refs/tags/v1.0'
1686 """
1687 if name.startswith(LOCAL_TAG_PREFIX):
1688 return Ref(name)
1689 return Ref(LOCAL_TAG_PREFIX + _strip_leading_slash(name, "Tag"))
1692def local_replace_name(name: bytes) -> Ref:
1693 """Build a full replace ref from a short name.
1695 Args:
1696 name: Short replace name (object SHA) or full ref
1698 Returns:
1699 Full replace ref name (e.g., b"refs/replace/<sha>")
1701 Examples:
1702 >>> local_replace_name(b"abc123")
1703 b'refs/replace/abc123'
1704 >>> local_replace_name(b"refs/replace/abc123")
1705 b'refs/replace/abc123'
1706 """
1707 if name.startswith(LOCAL_REPLACE_PREFIX):
1708 return Ref(name)
1709 return Ref(LOCAL_REPLACE_PREFIX + _strip_leading_slash(name, "Replace"))
1712def extract_branch_name(ref: bytes) -> bytes:
1713 """Extract branch name from a full branch ref.
1715 Args:
1716 ref: Full branch ref (e.g., b"refs/heads/master")
1718 Returns:
1719 Short branch name (e.g., b"master")
1721 Raises:
1722 ValueError: If ref is not a local branch
1724 Examples:
1725 >>> extract_branch_name(b"refs/heads/master")
1726 b'master'
1727 >>> extract_branch_name(b"refs/heads/feature/foo")
1728 b'feature/foo'
1729 """
1730 if not ref.startswith(LOCAL_BRANCH_PREFIX):
1731 raise ValueError(f"Not a local branch ref: {ref!r}")
1732 return ref[len(LOCAL_BRANCH_PREFIX) :]
1735def extract_tag_name(ref: bytes) -> bytes:
1736 """Extract tag name from a full tag ref.
1738 Args:
1739 ref: Full tag ref (e.g., b"refs/tags/v1.0")
1741 Returns:
1742 Short tag name (e.g., b"v1.0")
1744 Raises:
1745 ValueError: If ref is not a local tag
1747 Examples:
1748 >>> extract_tag_name(b"refs/tags/v1.0")
1749 b'v1.0'
1750 """
1751 if not ref.startswith(LOCAL_TAG_PREFIX):
1752 raise ValueError(f"Not a local tag ref: {ref!r}")
1753 return ref[len(LOCAL_TAG_PREFIX) :]
1756def shorten_ref_name(ref: bytes) -> bytes:
1757 """Convert a full ref name to its short form.
1759 Args:
1760 ref: Full ref name (e.g., b"refs/heads/master")
1762 Returns:
1763 Short ref name (e.g., b"master")
1765 Examples:
1766 >>> shorten_ref_name(b"refs/heads/master")
1767 b'master'
1768 >>> shorten_ref_name(b"refs/remotes/origin/main")
1769 b'origin/main'
1770 >>> shorten_ref_name(b"refs/tags/v1.0")
1771 b'v1.0'
1772 >>> shorten_ref_name(b"HEAD")
1773 b'HEAD'
1774 """
1775 if ref.startswith(LOCAL_BRANCH_PREFIX):
1776 return ref[len(LOCAL_BRANCH_PREFIX) :]
1777 elif ref.startswith(LOCAL_REMOTE_PREFIX):
1778 return ref[len(LOCAL_REMOTE_PREFIX) :]
1779 elif ref.startswith(LOCAL_TAG_PREFIX):
1780 return ref[len(LOCAL_TAG_PREFIX) :]
1781 return ref
1784def _set_origin_head(
1785 refs: RefsContainer, origin: bytes, origin_head: bytes | None
1786) -> None:
1787 # set refs/remotes/origin/HEAD
1788 origin_base = b"refs/remotes/" + origin + b"/"
1789 if origin_head and origin_head.startswith(LOCAL_BRANCH_PREFIX):
1790 origin_ref = Ref(origin_base + HEADREF)
1791 target_ref = Ref(origin_base + extract_branch_name(origin_head))
1792 if target_ref in refs:
1793 refs.set_symbolic_ref(origin_ref, target_ref)
1796def _set_default_branch(
1797 refs: RefsContainer,
1798 origin: bytes,
1799 origin_head: bytes | None,
1800 branch: bytes | None,
1801 ref_message: bytes | None,
1802) -> bytes:
1803 """Set the default branch."""
1804 origin_base = b"refs/remotes/" + origin + b"/"
1805 if branch:
1806 origin_ref = Ref(origin_base + branch)
1807 if origin_ref in refs:
1808 local_ref = Ref(local_branch_name(branch))
1809 refs.add_if_new(local_ref, refs[origin_ref], ref_message)
1810 head_ref = local_ref
1811 elif Ref(local_tag_name(branch)) in refs:
1812 head_ref = Ref(local_tag_name(branch))
1813 else:
1814 raise ValueError(f"{os.fsencode(branch)!r} is not a valid branch or tag")
1815 elif origin_head:
1816 head_ref = Ref(origin_head)
1817 if origin_head.startswith(LOCAL_BRANCH_PREFIX):
1818 origin_ref = Ref(origin_base + extract_branch_name(origin_head))
1819 else:
1820 origin_ref = Ref(origin_head)
1821 try:
1822 refs.add_if_new(head_ref, refs[origin_ref], ref_message)
1823 except KeyError:
1824 pass
1825 else:
1826 raise ValueError("neither origin_head nor branch are provided")
1827 return head_ref
1830def _set_head(
1831 refs: RefsContainer, head_ref: bytes, ref_message: bytes | None
1832) -> ObjectID | None:
1833 if head_ref.startswith(LOCAL_TAG_PREFIX):
1834 # detach HEAD at specified tag
1835 head = refs[Ref(head_ref)]
1836 del refs[HEADREF]
1837 refs.set_if_equals(HEADREF, None, head, message=ref_message)
1838 else:
1839 # set HEAD to specific branch
1840 try:
1841 head = refs[Ref(head_ref)]
1842 refs.set_symbolic_ref(HEADREF, Ref(head_ref))
1843 refs.set_if_equals(HEADREF, None, head, message=ref_message)
1844 except KeyError:
1845 head = None
1846 return head
1849def _set_branch_tracking(config: "ConfigFile", head_ref: bytes, remote: bytes) -> None:
1850 """Point the branch a clone checked out at its counterpart on the remote.
1852 This is what lets a subsequent "git pull" with no arguments know what to
1853 merge. Like git clone, nothing is written when HEAD ended up detached at a
1854 tag rather than on a branch.
1856 Args:
1857 config: Config of the freshly cloned repository
1858 head_ref: Local ref HEAD was pointed at, e.g. refs/heads/master
1859 remote: Name of the remote the clone came from
1860 """
1861 if not head_ref.startswith(LOCAL_BRANCH_PREFIX):
1862 return
1863 branch = extract_branch_name(Ref(head_ref))
1864 config.set((b"branch", branch), b"remote", remote)
1865 config.set((b"branch", branch), b"merge", head_ref)
1866 config.write_to_path()
1869def _import_remote_refs(
1870 refs_container: RefsContainer,
1871 remote_name: str,
1872 refs: Mapping[Ref, ObjectID | None],
1873 message: bytes | None = None,
1874 prune: bool = False,
1875 prune_tags: bool = False,
1876) -> None:
1877 from .protocol import PEELED_TAG_SUFFIX, strip_peeled_refs
1879 stripped_refs = strip_peeled_refs(refs)
1880 branches: dict[Ref, ObjectID | None] = {
1881 Ref(extract_branch_name(n)): v
1882 for (n, v) in stripped_refs.items()
1883 if n.startswith(LOCAL_BRANCH_PREFIX)
1884 }
1885 refs_container.import_refs(
1886 Ref(b"refs/remotes/" + remote_name.encode()),
1887 branches,
1888 message=message,
1889 prune=prune,
1890 )
1891 tags: dict[Ref, ObjectID | None] = {
1892 Ref(extract_tag_name(n)): v
1893 for (n, v) in stripped_refs.items()
1894 if n.startswith(LOCAL_TAG_PREFIX) and not n.endswith(PEELED_TAG_SUFFIX)
1895 }
1896 refs_container.import_refs(
1897 Ref(b"refs/tags"), tags, message=message, prune=prune_tags
1898 )
1901class locked_ref:
1902 """Lock a ref while making modifications.
1904 Works as a context manager.
1905 """
1907 def __init__(self, refs_container: DiskRefsContainer, refname: Ref) -> None:
1908 """Initialize a locked ref.
1910 Args:
1911 refs_container: The DiskRefsContainer to lock the ref in
1912 refname: The ref name to lock
1913 """
1914 self._refs_container = refs_container
1915 self._refname = refname
1916 self._file: _GitFile | None = None
1917 self._realname: Ref | None = None
1918 self._deleted = False
1920 def __enter__(self) -> Self:
1921 """Enter the context manager and acquire the lock.
1923 Returns:
1924 This locked_ref instance
1926 Raises:
1927 OSError: If the lock cannot be acquired
1928 """
1929 self._refs_container._check_refname(self._refname)
1930 try:
1931 realnames, _ = self._refs_container.follow(self._refname)
1932 self._realname = realnames[-1]
1933 except (KeyError, IndexError, SymrefLoop):
1934 self._realname = self._refname
1936 # The resolved symref target may escape the ref store.
1937 if not refname_is_safe(self._realname):
1938 raise RefFormatError(self._realname)
1939 filename = self._refs_container.refpath(self._realname)
1940 ensure_dir_exists(os.path.dirname(filename))
1941 f = GitFile(filename, "wb")
1942 self._file = f
1943 return self
1945 def __exit__(
1946 self,
1947 exc_type: type | None,
1948 exc_value: BaseException | None,
1949 traceback: types.TracebackType | None,
1950 ) -> None:
1951 """Exit the context manager and release the lock.
1953 Args:
1954 exc_type: Type of exception if one occurred
1955 exc_value: Exception instance if one occurred
1956 traceback: Traceback if an exception occurred
1957 """
1958 if self._file:
1959 if exc_type is not None or self._deleted:
1960 self._file.abort()
1961 else:
1962 self._file.close()
1964 def get(self) -> bytes | None:
1965 """Get the current value of the ref."""
1966 if not self._file:
1967 raise RuntimeError("locked_ref not in context")
1969 assert self._realname is not None
1970 current_ref = self._refs_container._read_loose_ref_file(self._realname)
1971 if current_ref is None:
1972 current_ref = self._refs_container.get_packed_refs().get(
1973 self._realname, None
1974 )
1975 return current_ref
1977 def ensure_equals(self, expected_value: bytes | None) -> bool:
1978 """Ensure the ref currently equals the expected value.
1980 Args:
1981 expected_value: The expected current value of the ref
1982 Returns:
1983 True if the ref equals the expected value, False otherwise
1984 """
1985 current_value = self.get()
1986 return current_value == expected_value
1988 def set(self, new_ref: bytes) -> None:
1989 """Set the ref to a new value.
1991 Args:
1992 new_ref: The new SHA1 or symbolic ref value
1993 """
1994 if not self._file:
1995 raise RuntimeError("locked_ref not in context")
1997 if not (valid_hexsha(new_ref) or new_ref.startswith(SYMREF)):
1998 raise ValueError(f"{new_ref!r} must be a valid sha or a symref")
2000 self._file.seek(0)
2001 self._file.truncate()
2002 self._file.write(new_ref + b"\n")
2003 self._deleted = False
2005 def set_symbolic_ref(self, target: Ref) -> None:
2006 """Make this ref point at another ref.
2008 Args:
2009 target: Name of the ref to point at
2010 """
2011 if not self._file:
2012 raise RuntimeError("locked_ref not in context")
2014 self._refs_container._check_refname(target)
2015 self._file.seek(0)
2016 self._file.truncate()
2017 self._file.write(SYMREF + target + b"\n")
2018 self._deleted = False
2020 def delete(self) -> None:
2021 """Delete the ref file while holding the lock."""
2022 if not self._file:
2023 raise RuntimeError("locked_ref not in context")
2025 # Delete the actual ref file while holding the lock
2026 if self._realname:
2027 filename = self._refs_container.refpath(self._realname)
2028 try:
2029 if os.path.lexists(filename):
2030 os.remove(filename)
2031 except FileNotFoundError:
2032 pass
2033 self._refs_container._remove_packed_ref(self._realname)
2035 self._deleted = True
2038class NamespacedRefsContainer(RefsContainer):
2039 """Wrapper that adds namespace prefix to all ref operations.
2041 This implements Git's GIT_NAMESPACE feature, which stores refs under
2042 refs/namespaces/<namespace>/ and filters operations to only show refs
2043 within that namespace.
2045 Example:
2046 With namespace "foo", a ref "refs/heads/master" is stored as
2047 "refs/namespaces/foo/refs/heads/master" in the underlying container.
2048 """
2050 def __init__(self, refs: RefsContainer, namespace: bytes) -> None:
2051 """Initialize NamespacedRefsContainer.
2053 Args:
2054 refs: The underlying refs container to wrap
2055 namespace: The namespace prefix (e.g., b"foo" or b"foo/bar")
2056 """
2057 super().__init__(logger=refs._logger)
2058 self._refs = refs
2059 # Build namespace prefix: refs/namespaces/<namespace>/
2060 # Support nested namespaces: foo/bar -> refs/namespaces/foo/refs/namespaces/bar/
2061 namespace_parts = namespace.split(b"/")
2062 self._namespace_prefix = b""
2063 for part in namespace_parts:
2064 self._namespace_prefix += b"refs/namespaces/" + part + b"/"
2066 def _apply_namespace(self, name: bytes) -> bytes:
2067 """Apply namespace prefix to a ref name."""
2068 # HEAD and other special refs are not namespaced
2069 if name == HEADREF or not name.startswith(b"refs/"):
2070 return name
2071 return self._namespace_prefix + name
2073 def _strip_namespace(self, name: bytes) -> bytes | None:
2074 """Remove namespace prefix from a ref name.
2076 Returns None if the ref is not in our namespace.
2077 """
2078 # HEAD and other special refs are not namespaced
2079 if name == HEADREF or not name.startswith(b"refs/"):
2080 return name
2081 if name.startswith(self._namespace_prefix):
2082 return name[len(self._namespace_prefix) :]
2083 return None
2085 def allkeys(self) -> set[Ref]:
2086 """Return all reference keys in this namespace."""
2087 keys: set[Ref] = set()
2088 for key in self._refs.allkeys():
2089 stripped = self._strip_namespace(key)
2090 if stripped is not None:
2091 keys.add(Ref(stripped))
2092 return keys
2094 def read_loose_ref(self, name: Ref) -> bytes | None:
2095 """Read a loose reference."""
2096 return self._refs.read_loose_ref(Ref(self._apply_namespace(name)))
2098 def get_packed_refs(self) -> dict[Ref, ObjectID]:
2099 """Get packed refs within this namespace."""
2100 packed: dict[Ref, ObjectID] = {}
2101 for name, value in self._refs.get_packed_refs().items():
2102 stripped = self._strip_namespace(name)
2103 if stripped is not None:
2104 packed[Ref(stripped)] = value
2105 return packed
2107 def add_packed_refs(self, new_refs: Mapping[Ref, ObjectID | None]) -> None:
2108 """Add packed refs with namespace prefix."""
2109 namespaced_refs: dict[Ref, ObjectID | None] = {
2110 Ref(self._apply_namespace(name)): value for name, value in new_refs.items()
2111 }
2112 self._refs.add_packed_refs(namespaced_refs)
2114 def get_peeled(self, name: Ref) -> ObjectID | None:
2115 """Return the cached peeled value of a ref."""
2116 return self._refs.get_peeled(Ref(self._apply_namespace(name)))
2118 def set_symbolic_ref(
2119 self,
2120 name: Ref,
2121 other: Ref,
2122 committer: bytes | None = None,
2123 timestamp: int | None = None,
2124 timezone: int | None = None,
2125 message: bytes | None = None,
2126 ) -> None:
2127 """Make a ref point at another ref."""
2128 self._refs.set_symbolic_ref(
2129 Ref(self._apply_namespace(name)),
2130 Ref(self._apply_namespace(other)),
2131 committer=committer,
2132 timestamp=timestamp,
2133 timezone=timezone,
2134 message=message,
2135 )
2137 def set_if_equals(
2138 self,
2139 name: Ref,
2140 old_ref: ObjectID | None,
2141 new_ref: ObjectID,
2142 committer: bytes | None = None,
2143 timestamp: int | None = None,
2144 timezone: int | None = None,
2145 message: bytes | None = None,
2146 ) -> bool:
2147 """Set a refname to new_ref only if it currently equals old_ref."""
2148 return self._refs.set_if_equals(
2149 Ref(self._apply_namespace(name)),
2150 old_ref,
2151 new_ref,
2152 committer=committer,
2153 timestamp=timestamp,
2154 timezone=timezone,
2155 message=message,
2156 )
2158 def add_if_new(
2159 self,
2160 name: Ref,
2161 ref: ObjectID,
2162 committer: bytes | None = None,
2163 timestamp: int | None = None,
2164 timezone: int | None = None,
2165 message: bytes | None = None,
2166 ) -> bool:
2167 """Add a new reference only if it does not already exist."""
2168 return self._refs.add_if_new(
2169 Ref(self._apply_namespace(name)),
2170 ref,
2171 committer=committer,
2172 timestamp=timestamp,
2173 timezone=timezone,
2174 message=message,
2175 )
2177 def remove_if_equals(
2178 self,
2179 name: Ref,
2180 old_ref: ObjectID | None,
2181 committer: bytes | None = None,
2182 timestamp: int | None = None,
2183 timezone: int | None = None,
2184 message: bytes | None = None,
2185 ) -> bool:
2186 """Remove a refname only if it currently equals old_ref."""
2187 return self._refs.remove_if_equals(
2188 Ref(self._apply_namespace(name)),
2189 old_ref,
2190 committer=committer,
2191 timestamp=timestamp,
2192 timezone=timezone,
2193 message=message,
2194 )
2196 def pack_refs(self, all: bool = False) -> None:
2197 """Pack loose refs into packed-refs file.
2199 Note: This packs all refs in the underlying container, not just
2200 those in the namespace.
2201 """
2202 self._refs.pack_refs(all=all)
2205@overload
2206def filter_ref_prefix(
2207 refs: dict[Ref, ObjectID], prefixes: Iterable[bytes]
2208) -> dict[Ref, ObjectID]: ...
2211@overload
2212def filter_ref_prefix(
2213 refs: dict[Ref, ObjectID | None], prefixes: Iterable[bytes]
2214) -> dict[Ref, ObjectID | None]: ...
2217def filter_ref_prefix(
2218 refs: dict[Ref, ObjectID] | dict[Ref, ObjectID | None],
2219 prefixes: Iterable[bytes],
2220) -> dict[Ref, ObjectID] | dict[Ref, ObjectID | None]:
2221 """Filter refs to only include those with a given prefix.
2223 Args:
2224 refs: A dictionary of refs.
2225 prefixes: The prefixes to filter by.
2226 """
2227 return {k: v for k, v in refs.items() if any(k.startswith(p) for p in prefixes)}
2230def is_per_worktree_ref(ref: bytes) -> bool:
2231 """Returns whether a reference is stored per worktree or not.
2233 Per-worktree references are:
2234 - all pseudorefs, e.g. HEAD
2235 - all references stored inside "refs/bisect/", "refs/worktree/" and "refs/rewritten/"
2237 All refs starting with "refs/" are shared, except for the ones listed above.
2239 See https://git-scm.com/docs/git-worktree#_refs.
2240 """
2241 return not ref.startswith(b"refs/") or ref.startswith(
2242 (b"refs/bisect/", b"refs/worktree/", b"refs/rewritten/")
2243 )