Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/repo.py: 38%
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# repo.py -- For dealing with git repositories.
2# Copyright (C) 2007 James Westby <jw+debian@jameswestby.net>
3# Copyright (C) 2008-2013 Jelmer Vernooij <jelmer@jelmer.uk>
4#
5# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
6# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
7# General Public License as published by the Free Software Foundation; version 2.0
8# or (at your option) any later version. You can redistribute it and/or
9# modify it under the terms of either of these two licenses.
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS,
13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14# See the License for the specific language governing permissions and
15# limitations under the License.
16#
17# You should have received a copy of the licenses; if not, see
18# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
19# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
20# License, Version 2.0.
21#
24"""Repository access.
26This module contains the base class for git repositories
27(BaseRepo) and an implementation which uses a repository on
28local disk (Repo).
30"""
32__all__ = [
33 "BASE_DIRECTORIES",
34 "COMMONDIR",
35 "CONTROLDIR",
36 "DEFAULT_BRANCH",
37 "DEFAULT_OFS_DELTA",
38 "GITDIR",
39 "INDEX_FILENAME",
40 "OBJECTDIR",
41 "REFSDIR",
42 "REFSDIR_HEADS",
43 "REFSDIR_TAGS",
44 "WORKTREES",
45 "BaseRepo",
46 "DefaultIdentityNotFound",
47 "InvalidUserIdentity",
48 "MemoryRepo",
49 "ParentsProvider",
50 "Repo",
51 "UnsupportedExtension",
52 "UnsupportedVersion",
53 "check_user_identity",
54 "get_user_identity",
55 "parse_graftpoints",
56 "parse_shared_repository",
57 "read_gitfile",
58 "serialize_graftpoints",
59]
61import logging
62import os
63import stat
64import sys
65import time
66import warnings
67from collections.abc import Callable, Generator, Iterable, Iterator, Mapping, Sequence
68from io import BytesIO
69from types import TracebackType
70from typing import (
71 TYPE_CHECKING,
72 Any,
73 BinaryIO,
74 TypeVar,
75)
77if sys.version_info >= (3, 11):
78 from typing import Self
79else:
80 from typing_extensions import Self
82if TYPE_CHECKING:
83 # There are no circular imports here, but we try to defer imports as long
84 # as possible to reduce start-up time for anything that doesn't need
85 # these imports.
86 from .attrs import GitAttributes
87 from .config import ConditionMatcher, Config, ConfigFile, StackedConfig
88 from .diff_tree import RenameDetector
89 from .filters import FilterBlobNormalizer, FilterContext
90 from .index import Index
91 from .notes import Notes
92 from .object_format import ObjectFormat
93 from .object_store import BaseObjectStore, GraphWalker
94 from .pack import UnpackedObject
95 from .rebase import RebaseStateManager
96 from .walk import Walker
97 from .worktree import WorkTree
99from . import reflog
100from .errors import (
101 NoIndexPresent,
102 NotBlobError,
103 NotCommitError,
104 NotGitRepository,
105 NotTagError,
106 NotTreeError,
107 RefFormatError,
108)
109from .file import GitFile
110from .hooks import (
111 CommitMsgShellHook,
112 Hook,
113 PostCommitShellHook,
114 PostReceiveShellHook,
115 PreCommitShellHook,
116 PreReceiveShellHook,
117 UpdateShellHook,
118)
119from .object_store import (
120 DiskObjectStore,
121 MemoryObjectStore,
122 MissingObjectFinder,
123 ObjectStoreGraphWalker,
124 PackBasedObjectStore,
125 PackCapableObjectStore,
126 find_shallow,
127 peel_sha,
128)
129from .objects import (
130 Blob,
131 Commit,
132 ObjectID,
133 RawObjectID,
134 ShaFile,
135 Tag,
136 Tree,
137 check_hexsha,
138 valid_hexsha,
139)
140from .pack import generate_unpacked_objects
141from .refs import (
142 HEADREF,
143 LOCAL_TAG_PREFIX, # noqa: F401
144 SYMREF, # noqa: F401
145 DictRefsContainer,
146 DiskRefsContainer,
147 Ref,
148 RefsContainer,
149 _set_default_branch,
150 _set_head,
151 _set_origin_head,
152 check_ref_format, # noqa: F401
153 extract_branch_name,
154 is_per_worktree_ref,
155 local_branch_name,
156 read_packed_refs, # noqa: F401
157 read_packed_refs_with_peeled, # noqa: F401
158 write_packed_refs, # noqa: F401
159)
161logger = logging.getLogger(__name__)
163CONTROLDIR = ".git"
164OBJECTDIR = "objects"
165DEFAULT_OFS_DELTA = True
167T = TypeVar("T", bound="ShaFile")
168REFSDIR = "refs"
169REFSDIR_TAGS = "tags"
170REFSDIR_HEADS = "heads"
171INDEX_FILENAME = "index"
172COMMONDIR = "commondir"
173GITDIR = "gitdir"
174WORKTREES = "worktrees"
176BASE_DIRECTORIES = [
177 ["branches"],
178 [REFSDIR],
179 [REFSDIR, REFSDIR_TAGS],
180 [REFSDIR, REFSDIR_HEADS],
181 ["hooks"],
182 ["info"],
183]
185DEFAULT_BRANCH = b"master"
188class InvalidUserIdentity(Exception):
189 """User identity is not of the format 'user <email>'."""
191 def __init__(self, identity: str) -> None:
192 """Initialize InvalidUserIdentity exception."""
193 self.identity = identity
196class DefaultIdentityNotFound(Exception):
197 """Default identity could not be determined."""
200# TODO(jelmer): Cache?
201def _get_default_identity(env: Mapping[str, str] | None = None) -> tuple[str, str]:
202 import socket
204 if env is None:
205 env = os.environ
207 for name in ("LOGNAME", "USER", "LNAME", "USERNAME"):
208 username = env.get(name)
209 if username:
210 break
211 else:
212 username = None
214 try:
215 import pwd
216 except ImportError:
217 fullname = None
218 else:
219 try:
220 entry = pwd.getpwuid(os.getuid()) # type: ignore[attr-defined,unused-ignore]
221 except KeyError:
222 fullname = None
223 else:
224 if getattr(entry, "gecos", None):
225 fullname = entry.pw_gecos.split(",")[0]
226 else:
227 fullname = None
228 if username is None:
229 username = entry.pw_name
230 if not fullname:
231 if username is None:
232 raise DefaultIdentityNotFound("no username found")
233 fullname = username
234 email = env.get("EMAIL")
235 if email is None:
236 if username is None:
237 raise DefaultIdentityNotFound("no username found")
238 email = f"{username}@{socket.gethostname()}"
239 return (fullname, email)
242def get_user_identity(config: "Config", kind: str | None = None) -> bytes:
243 """Determine the identity to use for new commits.
245 If kind is set, this first checks
246 GIT_${KIND}_NAME and GIT_${KIND}_EMAIL.
248 If those variables are not set, then it will fall back
249 to reading the user.name and user.email settings from
250 the specified configuration.
252 If that also fails, then it will fall back to using
253 the current users' identity as obtained from the host
254 system (e.g. the gecos field, $EMAIL, $USER@$(hostname -f).
256 Args:
257 config: Configuration stack to read from
258 kind: Optional kind to return identity for,
259 usually either "AUTHOR" or "COMMITTER".
261 Returns:
262 A user identity
263 """
264 user: bytes | None = None
265 email: bytes | None = None
266 if kind:
267 user_uc = os.environ.get("GIT_" + kind + "_NAME")
268 if user_uc is not None:
269 user = user_uc.encode("utf-8")
270 email_uc = os.environ.get("GIT_" + kind + "_EMAIL")
271 if email_uc is not None:
272 email = email_uc.encode("utf-8")
273 if user is None:
274 try:
275 user = config.get(("user",), "name")
276 except KeyError:
277 user = None
278 if email is None:
279 try:
280 email = config.get(("user",), "email")
281 except KeyError:
282 email = None
283 default_user, default_email = _get_default_identity()
284 if user is None:
285 user = default_user.encode("utf-8")
286 if email is None:
287 email = default_email.encode("utf-8")
288 if email.startswith(b"<") and email.endswith(b">"):
289 email = email[1:-1]
290 return user + b" <" + email + b">"
293def check_user_identity(identity: bytes) -> None:
294 """Verify that a user identity is formatted correctly.
296 Args:
297 identity: User identity bytestring
298 Raises:
299 InvalidUserIdentity: Raised when identity is invalid
300 """
301 try:
302 _fst, snd = identity.split(b" <", 1)
303 except ValueError as exc:
304 raise InvalidUserIdentity(identity.decode("utf-8", "replace")) from exc
305 if b">" not in snd:
306 raise InvalidUserIdentity(identity.decode("utf-8", "replace"))
307 if b"\0" in identity or b"\n" in identity:
308 raise InvalidUserIdentity(identity.decode("utf-8", "replace"))
311def parse_graftpoints(
312 graftpoints: Iterable[bytes],
313) -> dict[ObjectID, list[ObjectID]]:
314 """Convert a list of graftpoints into a dict.
316 Args:
317 graftpoints: Iterator of graftpoint lines
319 Each line is formatted as:
320 <commit sha1> <parent sha1> [<parent sha1>]*
322 Resulting dictionary is:
323 <commit sha1>: [<parent sha1>*]
325 https://git.wiki.kernel.org/index.php/GraftPoint
326 """
327 grafts: dict[ObjectID, list[ObjectID]] = {}
328 for line in graftpoints:
329 raw_graft = line.split(None, 1)
331 commit = ObjectID(raw_graft[0])
332 if len(raw_graft) == 2:
333 parents = [ObjectID(p) for p in raw_graft[1].split()]
334 else:
335 parents = []
337 for sha in [commit, *parents]:
338 check_hexsha(sha, "Invalid graftpoint")
340 grafts[commit] = parents
341 return grafts
344def serialize_graftpoints(graftpoints: Mapping[ObjectID, Sequence[ObjectID]]) -> bytes:
345 """Convert a dictionary of grafts into string.
347 The graft dictionary is:
348 <commit sha1>: [<parent sha1>*]
350 Each line is formatted as:
351 <commit sha1> <parent sha1> [<parent sha1>]*
353 https://git.wiki.kernel.org/index.php/GraftPoint
355 """
356 graft_lines = []
357 for commit, parents in graftpoints.items():
358 if parents:
359 graft_lines.append(commit + b" " + b" ".join(parents))
360 else:
361 graft_lines.append(commit)
362 return b"\n".join(graft_lines)
365def _set_filesystem_hidden(path: str) -> None:
366 """Mark path as to be hidden if supported by platform and filesystem.
368 On win32 uses SetFileAttributesW api:
369 <https://docs.microsoft.com/windows/desktop/api/fileapi/nf-fileapi-setfileattributesw>
370 """
371 if sys.platform == "win32":
372 import ctypes
373 from ctypes.wintypes import BOOL, DWORD, LPCWSTR
375 FILE_ATTRIBUTE_HIDDEN = 2
376 SetFileAttributesW = ctypes.WINFUNCTYPE(BOOL, LPCWSTR, DWORD)(
377 ("SetFileAttributesW", ctypes.windll.kernel32)
378 )
380 if isinstance(path, bytes):
381 path = os.fsdecode(path)
382 if not SetFileAttributesW(path, FILE_ATTRIBUTE_HIDDEN):
383 pass # Could raise or log `ctypes.WinError()` here
385 # Could implement other platform specific filesystem hiding here
388def parse_shared_repository(
389 value: str | bytes | bool,
390) -> tuple[int | None, int | None]:
391 """Parse core.sharedRepository configuration value.
393 Args:
394 value: Configuration value (string, bytes, or boolean)
396 Returns:
397 tuple of (file_mask, directory_mask) or (None, None) if not shared
399 The masks are permission bits to apply via chmod.
400 """
401 if isinstance(value, bytes):
402 value = value.decode("utf-8", errors="replace")
404 # Handle boolean values
405 if isinstance(value, bool):
406 if value:
407 # true = group (same as "group")
408 return (0o664, 0o2775)
409 else:
410 # false = umask (use system umask, no adjustment)
411 return (None, None)
413 # Handle string values
414 value_lower = value.lower()
416 if value_lower in ("false", "0", ""):
417 # Use umask (no adjustment)
418 return (None, None)
420 if value_lower in ("true", "1", "group"):
421 # Group writable (with setgid bit)
422 return (0o664, 0o2775)
424 if value_lower in ("all", "world", "everybody", "2"):
425 # World readable/writable (with setgid bit)
426 return (0o666, 0o2777)
428 if value_lower == "umask":
429 # Explicitly use umask
430 return (None, None)
432 # Try to parse as octal
433 if value.startswith("0"):
434 try:
435 mode = int(value, 8)
436 # For directories, add execute bits where read bits are set
437 # and add setgid bit for shared repositories
438 dir_mode = mode | 0o2000 # Add setgid bit
439 if mode & 0o004:
440 dir_mode |= 0o001
441 if mode & 0o040:
442 dir_mode |= 0o010
443 if mode & 0o400:
444 dir_mode |= 0o100
445 return (mode, dir_mode)
446 except ValueError:
447 pass
449 # Default to umask for unrecognized values
450 return (None, None)
453def _enable_relative_worktrees_extension(repo: "Repo") -> None:
454 """Enable the relativeworktrees extension in repository config.
456 This sets core.repositoryformatversion to 1 (if not already) and
457 enables the extensions.relativeworktrees extension.
459 Args:
460 repo: The repository to configure
461 """
462 config = repo.get_config()
464 # Ensure repository format version is at least 1
465 try:
466 version = int(config.get(("core",), "repositoryformatversion"))
467 except KeyError:
468 version = 0
470 if version < 1:
471 config.set(("core",), "repositoryformatversion", "1")
473 # Enable the relativeworktrees extension
474 config.set(("extensions",), "relativeworktrees", True)
475 config.write_to_path()
478class ParentsProvider:
479 """Provider for commit parent information."""
481 def __init__(
482 self,
483 store: "BaseObjectStore",
484 grafts: dict[ObjectID, list[ObjectID]] = {},
485 shallows: Iterable[ObjectID] = [],
486 ) -> None:
487 """Initialize ParentsProvider.
489 Args:
490 store: Object store to use
491 grafts: Graft information
492 shallows: Shallow commit SHAs
493 """
494 self.store = store
495 self.grafts = grafts
496 self.shallows = set(shallows)
498 # Get commit graph once at initialization for performance
499 self.commit_graph = store.get_commit_graph()
501 def get_parents(
502 self, commit_id: ObjectID, commit: Commit | None = None
503 ) -> list[ObjectID]:
504 """Get parents for a commit using the parents provider."""
505 try:
506 return self.grafts[commit_id]
507 except KeyError:
508 pass
509 if commit_id in self.shallows:
510 return []
512 # Try to use commit graph for faster parent lookup
513 if self.commit_graph:
514 parents = self.commit_graph.get_parents(commit_id)
515 if parents is not None:
516 return parents
518 # Fallback to reading the commit object
519 if commit is None:
520 obj = self.store[commit_id]
521 if not isinstance(obj, Commit):
522 raise ValueError(
523 f"Expected Commit object for commit_id {commit_id.decode()}, "
524 f"got {type(obj).__name__}. This usually means a reference "
525 f"points to a {type(obj).__name__} object instead of a Commit."
526 )
527 commit = obj
528 result: list[ObjectID] = commit.parents
529 return result
532class BaseRepo:
533 """Base class for a git repository.
535 This base class is meant to be used for Repository implementations that e.g.
536 work on top of a different transport than a standard filesystem path.
538 Attributes:
539 object_store: Dictionary-like object for accessing
540 the objects
541 refs: Dictionary-like object with the refs in this
542 repository
543 """
545 def __init__(
546 self,
547 object_store: "PackCapableObjectStore",
548 refs: RefsContainer,
549 object_format: "ObjectFormat | None" = None,
550 ) -> None:
551 """Open a repository.
553 This shouldn't be called directly, but rather through one of the
554 base classes, such as MemoryRepo or Repo.
556 Args:
557 object_store: Object store to use
558 refs: Refs container to use
559 object_format: Hash algorithm to use (if None, will use object_store's format)
560 """
561 self.object_store = object_store
562 self.refs = refs
564 self._graftpoints: dict[ObjectID, list[ObjectID]] = {}
565 self.hooks: dict[str, Hook] = {}
566 if object_format is None:
567 self.object_format: ObjectFormat = object_store.object_format
568 else:
569 self.object_format = object_format
571 def _determine_file_mode(self) -> bool:
572 """Probe the file-system to determine whether permissions can be trusted.
574 Returns: True if permissions can be trusted, False otherwise.
575 """
576 raise NotImplementedError(self._determine_file_mode)
578 def _determine_symlinks(self) -> bool:
579 """Probe the filesystem to determine whether symlinks can be created.
581 Returns: True if symlinks can be created, False otherwise.
582 """
583 # For now, just mimic the old behaviour
584 return sys.platform != "win32"
586 def _init_files(
587 self,
588 bare: bool,
589 symlinks: bool | None = None,
590 format: int | None = None,
591 shared_repository: str | bool | None = None,
592 object_format: str | None = None,
593 ) -> None:
594 """Initialize a default set of named files."""
595 from .config import ConfigFile
597 self._put_named_file("description", b"Unnamed repository")
598 f = BytesIO()
599 cf = ConfigFile()
601 # Determine the appropriate format version
602 if object_format == "sha256":
603 # SHA256 requires format version 1
604 if format is None:
605 format = 1
606 elif format != 1:
607 raise ValueError(
608 "SHA256 object format requires repository format version 1"
609 )
610 else:
611 # SHA1 (default) can use format 0 or 1
612 if format is None:
613 format = 0
615 if format not in (0, 1):
616 raise ValueError(f"Unsupported repository format version: {format}")
618 cf.set("core", "repositoryformatversion", str(format))
620 # Set object format extension if using SHA256
621 if object_format == "sha256":
622 cf.set("extensions", "objectformat", "sha256")
624 # Set hash algorithm based on object format
625 from .object_format import get_object_format
627 self.object_format = get_object_format(object_format)
629 if self._determine_file_mode():
630 cf.set("core", "filemode", True)
631 else:
632 cf.set("core", "filemode", False)
634 if symlinks is None and not bare:
635 symlinks = self._determine_symlinks()
637 if symlinks is False:
638 cf.set("core", "symlinks", symlinks)
640 # On macOS, set precomposeunicode to true since HFS+/APFS
641 # returns filenames in NFD (decomposed) Unicode form
642 if sys.platform == "darwin":
643 cf.set("core", "precomposeunicode", True)
645 cf.set("core", "bare", bare)
646 cf.set("core", "logallrefupdates", True)
648 # Set shared repository if specified
649 if shared_repository is not None:
650 if isinstance(shared_repository, bool):
651 cf.set("core", "sharedRepository", shared_repository)
652 else:
653 cf.set("core", "sharedRepository", shared_repository)
655 cf.write_to_file(f)
656 self._put_named_file("config", f.getvalue())
657 self._put_named_file(os.path.join("info", "exclude"), b"")
659 # Allow subclasses to handle config initialization
660 self._init_config(cf)
662 def _init_config(self, config: "ConfigFile") -> None:
663 """Initialize repository configuration.
665 This method can be overridden by subclasses to handle config initialization.
667 Args:
668 config: The ConfigFile object that was just created
669 """
670 # Default implementation does nothing
672 def get_named_file(self, path: str) -> BinaryIO | None:
673 """Get a file from the control dir with a specific name.
675 Although the filename should be interpreted as a filename relative to
676 the control dir in a disk-based Repo, the object returned need not be
677 pointing to a file in that location.
679 Args:
680 path: The path to the file, relative to the control dir.
681 Returns: An open file object, or None if the file does not exist.
682 """
683 raise NotImplementedError(self.get_named_file)
685 def _put_named_file(self, path: str, contents: bytes) -> None:
686 """Write a file to the control dir with the given name and contents.
688 Args:
689 path: The path to the file, relative to the control dir.
690 contents: A string to write to the file.
691 """
692 raise NotImplementedError(self._put_named_file)
694 def _del_named_file(self, path: str) -> None:
695 """Delete a file in the control directory with the given name."""
696 raise NotImplementedError(self._del_named_file)
698 def open_index(self, config: "Config | None" = None) -> "Index":
699 """Open the index for this repository.
701 Args:
702 config: Configuration to consult for index settings. If None,
703 implementations may fall back to ``self.get_config_stack()``.
705 Raises:
706 NoIndexPresent: If no index is present
707 Returns: The matching `Index`
708 """
709 raise NotImplementedError(self.open_index)
711 def _change_object_format(self, object_format_name: str) -> None:
712 """Change the object format of this repository.
714 This can only be done if the object store is empty (no objects written yet).
716 Args:
717 object_format_name: Name of the new object format (e.g., "sha1", "sha256")
719 Raises:
720 AssertionError: If the object store is not empty
721 """
722 # Check if object store has any objects
723 for _ in self.object_store:
724 raise AssertionError(
725 "Cannot change object format: repository already contains objects"
726 )
728 # Update the object format
729 from .object_format import get_object_format
731 new_format = get_object_format(object_format_name)
732 self.object_format = new_format
733 self.object_store.object_format = new_format
735 # Update config file
736 config = self.get_config()
738 if object_format_name == "sha1":
739 # For SHA-1, explicitly remove objectformat extension if present
740 try:
741 config.remove("extensions", "objectformat")
742 except KeyError:
743 pass
744 else:
745 # For non-SHA-1 formats, set repositoryformatversion to 1 and objectformat extension
746 config.set("core", "repositoryformatversion", "1")
747 config.set("extensions", "objectformat", object_format_name)
749 config.write_to_path()
751 def fetch(
752 self,
753 target: "BaseRepo",
754 determine_wants: Callable[[Mapping[Ref, ObjectID], int | None], list[ObjectID]]
755 | None = None,
756 progress: Callable[..., None] | None = None,
757 depth: int | None = None,
758 ) -> dict[Ref, ObjectID]:
759 """Fetch objects into another repository.
761 Args:
762 target: The target repository
763 determine_wants: Optional function to determine what refs to
764 fetch.
765 progress: Optional progress function
766 depth: Optional shallow fetch depth
767 Returns: The local refs
768 """
769 # Fix object format if needed
770 if self.object_format != target.object_format:
771 # Change the target repo's format if it's empty
772 target._change_object_format(self.object_format.name)
774 if determine_wants is None:
775 determine_wants = target.object_store.determine_wants_all
776 count, pack_data = self.fetch_pack_data(
777 determine_wants,
778 target.get_graph_walker(),
779 progress=progress,
780 depth=depth,
781 )
782 target.object_store.add_pack_data(count, pack_data, progress)
783 return self.get_refs()
785 def fetch_pack_data(
786 self,
787 determine_wants: Callable[[Mapping[Ref, ObjectID], int | None], list[ObjectID]],
788 graph_walker: "GraphWalker",
789 progress: Callable[[bytes], None] | None,
790 *,
791 get_tagged: Callable[[], dict[ObjectID, ObjectID]] | None = None,
792 depth: int | None = None,
793 ) -> tuple[int, Iterator["UnpackedObject"]]:
794 """Fetch the pack data required for a set of revisions.
796 Args:
797 determine_wants: Function that takes a dictionary with heads
798 and returns the list of heads to fetch.
799 graph_walker: Object that can iterate over the list of revisions
800 to fetch and has an "ack" method that will be called to acknowledge
801 that a revision is present.
802 progress: Simple progress function that will be called with
803 updated progress strings.
804 get_tagged: Function that returns a dict of pointed-to sha ->
805 tag sha for including tags.
806 depth: Shallow fetch depth
807 Returns: count and iterator over pack data
808 """
809 missing_objects = self.find_missing_objects(
810 determine_wants, graph_walker, progress, get_tagged=get_tagged, depth=depth
811 )
812 if missing_objects is None:
813 return 0, iter([])
814 remote_has = missing_objects.get_remote_has()
815 object_ids = list(missing_objects)
816 return len(object_ids), generate_unpacked_objects(
817 self.object_store, object_ids, progress=progress, other_haves=remote_has
818 )
820 def find_missing_objects(
821 self,
822 determine_wants: Callable[[Mapping[Ref, ObjectID], int | None], list[ObjectID]],
823 graph_walker: "GraphWalker",
824 progress: Callable[[bytes], None] | None,
825 *,
826 get_tagged: Callable[[], dict[ObjectID, ObjectID]] | None = None,
827 depth: int | None = None,
828 ) -> MissingObjectFinder | None:
829 """Fetch the missing objects required for a set of revisions.
831 Args:
832 determine_wants: Function that takes a dictionary with heads
833 and returns the list of heads to fetch.
834 graph_walker: Object that can iterate over the list of revisions
835 to fetch and has an "ack" method that will be called to acknowledge
836 that a revision is present.
837 progress: Simple progress function that will be called with
838 updated progress strings.
839 get_tagged: Function that returns a dict of pointed-to sha ->
840 tag sha for including tags.
841 depth: Shallow fetch depth
842 Returns: iterator over objects, with __len__ implemented
843 """
844 # Filter out refs pointing to missing objects to avoid errors downstream.
845 # This makes Dulwich more robust when dealing with broken refs on disk.
846 # Previously serialize_refs() did this filtering as a side-effect.
847 all_refs = self.get_refs()
848 refs: dict[Ref, ObjectID] = {}
849 for ref, sha in all_refs.items():
850 if sha in self.object_store:
851 refs[ref] = sha
852 else:
853 logger.warning(
854 "ref %s points at non-present sha %s",
855 ref.decode("utf-8", "replace"),
856 sha.decode("ascii"),
857 )
859 wants = determine_wants(refs, depth)
860 if not isinstance(wants, list):
861 raise TypeError("determine_wants() did not return a list")
863 current_shallow = set(getattr(graph_walker, "shallow", set()))
865 unshallow: set[ObjectID] = set()
866 if depth not in (None, 0):
867 assert depth is not None
868 shallow, not_shallow = find_shallow(self.object_store, wants, depth)
869 # Only update if graph_walker has shallow attribute
870 walker_shallow: set[ObjectID] | None = getattr(
871 graph_walker, "shallow", None
872 )
873 if walker_shallow is not None:
874 unshallow = not_shallow & current_shallow
875 # Commits that were a shallow boundary but are now being
876 # deepened past must drop out of the boundary, otherwise their
877 # parents stay unreachable and never get transferred.
878 walker_shallow.update(shallow - not_shallow)
879 walker_shallow.difference_update(unshallow)
880 new_shallow = walker_shallow - current_shallow
881 setattr(graph_walker, "unshallow", unshallow)
882 update_shallow = getattr(graph_walker, "update_shallow", None)
883 if update_shallow is not None:
884 update_shallow(new_shallow, unshallow)
885 else:
886 unshallow = getattr(graph_walker, "unshallow", set())
888 if wants == []:
889 # TODO(dborowitz): find a way to short-circuit that doesn't change
890 # this interface.
892 if getattr(graph_walker, "shallow", set()) or unshallow:
893 # Do not send a pack in shallow short-circuit path
894 return None
896 # Return an actual MissingObjectFinder with empty wants
897 return MissingObjectFinder(
898 self.object_store,
899 haves=[],
900 wants=[],
901 )
903 # If the graph walker is set up with an implementation that can
904 # ACK/NAK to the wire, it will write data to the client through
905 # this call as a side-effect.
906 haves = self.object_store.find_common_revisions(graph_walker)
908 # Deal with shallow requests separately because the haves do
909 # not reflect what objects are missing
910 if getattr(graph_walker, "shallow", set()) or unshallow:
911 # TODO: filter the haves commits from iter_shas. the specific
912 # commits aren't missing.
913 haves = []
915 parents_provider = ParentsProvider(
916 self.object_store,
917 shallows=getattr(graph_walker, "shallow", current_shallow),
918 )
920 def get_parents(commit: Commit) -> list[ObjectID]:
921 """Get parents for a commit using the parents provider.
923 Args:
924 commit: Commit object
926 Returns:
927 List of parent commit SHAs
928 """
929 return parents_provider.get_parents(commit.id, commit)
931 return MissingObjectFinder(
932 self.object_store,
933 haves=haves,
934 wants=wants,
935 shallow=getattr(graph_walker, "shallow", set()),
936 progress=progress,
937 get_tagged=get_tagged,
938 get_parents=get_parents,
939 )
941 def generate_pack_data(
942 self,
943 have: set[ObjectID],
944 want: set[ObjectID],
945 *,
946 shallow: set[ObjectID] | None = None,
947 progress: Callable[[str], None] | None = None,
948 ofs_delta: bool | None = None,
949 ) -> tuple[int, Iterator["UnpackedObject"]]:
950 """Generate pack data objects for a set of wants/haves.
952 Args:
953 have: List of SHA1s of objects that should not be sent
954 want: List of SHA1s of objects that should be sent
955 shallow: Set of shallow commit SHA1s to skip (defaults to repo's shallow commits)
956 ofs_delta: Whether OFS deltas can be included
957 progress: Optional progress reporting method
958 """
959 if shallow is None:
960 shallow = self.get_shallow()
961 return self.object_store.generate_pack_data(
962 have,
963 want,
964 shallow=shallow,
965 progress=progress,
966 ofs_delta=ofs_delta if ofs_delta is not None else DEFAULT_OFS_DELTA,
967 )
969 def get_graph_walker(
970 self, heads: list[ObjectID] | None = None
971 ) -> ObjectStoreGraphWalker:
972 """Retrieve a graph walker.
974 A graph walker is used by a remote repository (or proxy)
975 to find out which objects are present in this repository.
977 Args:
978 heads: Repository heads to use (optional)
979 Returns: A graph walker object
980 """
981 if heads is None:
982 heads = [
983 sha
984 for sha in self.refs.as_dict(Ref(b"refs/heads")).values()
985 if sha in self.object_store
986 ]
987 parents_provider = ParentsProvider(self.object_store)
988 return ObjectStoreGraphWalker(
989 heads,
990 parents_provider.get_parents,
991 shallow=self.get_shallow(),
992 update_shallow=self.update_shallow,
993 )
995 def get_refs(self) -> dict[Ref, ObjectID]:
996 """Get dictionary with all refs.
998 Returns: A ``dict`` mapping ref names to SHA1s
999 """
1000 return self.refs.as_dict()
1002 def head(self) -> ObjectID:
1003 """Return the SHA1 pointed at by HEAD."""
1004 # TODO: move this method to WorkTree
1005 return self.refs[HEADREF]
1007 def _get_object(self, sha: ObjectID | RawObjectID, cls: type[T]) -> T:
1008 assert len(sha) in (
1009 self.object_format.oid_length,
1010 self.object_format.hex_length,
1011 )
1012 ret = self.get_object(sha)
1013 if not isinstance(ret, cls):
1014 if cls is Commit:
1015 raise NotCommitError(ret.id)
1016 elif cls is Blob:
1017 raise NotBlobError(ret.id)
1018 elif cls is Tree:
1019 raise NotTreeError(ret.id)
1020 elif cls is Tag:
1021 raise NotTagError(ret.id)
1022 else:
1023 raise Exception(f"Type invalid: {ret.type_name!r} != {cls.type_name!r}")
1024 return ret
1026 def get_object(self, sha: ObjectID | RawObjectID) -> ShaFile:
1027 """Retrieve the object with the specified SHA.
1029 Args:
1030 sha: SHA to retrieve
1031 Returns: A ShaFile object
1032 Raises:
1033 KeyError: when the object can not be found
1034 """
1035 return self.object_store[sha]
1037 def parents_provider(self) -> ParentsProvider:
1038 """Get a parents provider for this repository.
1040 Returns:
1041 ParentsProvider instance configured with grafts and shallows
1042 """
1043 return ParentsProvider(
1044 self.object_store,
1045 grafts=self._graftpoints,
1046 shallows=self.get_shallow(),
1047 )
1049 def get_parents(
1050 self, sha: ObjectID, commit: Commit | None = None
1051 ) -> list[ObjectID]:
1052 """Retrieve the parents of a specific commit.
1054 If the specific commit is a graftpoint, the graft parents
1055 will be returned instead.
1057 Args:
1058 sha: SHA of the commit for which to retrieve the parents
1059 commit: Optional commit matching the sha
1060 Returns: List of parents
1061 """
1062 return self.parents_provider().get_parents(sha, commit)
1064 def get_config(self) -> "ConfigFile":
1065 """Retrieve the config object.
1067 Returns: `ConfigFile` object for the ``.git/config`` file.
1068 """
1069 raise NotImplementedError(self.get_config)
1071 def get_worktree_config(self) -> "ConfigFile":
1072 """Retrieve the worktree config object."""
1073 raise NotImplementedError(self.get_worktree_config)
1075 def get_description(self) -> bytes | None:
1076 """Retrieve the description for this repository.
1078 Returns: Bytes with the description of the repository
1079 as set by the user.
1080 """
1081 raise NotImplementedError(self.get_description)
1083 def set_description(self, description: bytes) -> None:
1084 """Set the description for this repository.
1086 Args:
1087 description: Text to set as description for this repository.
1088 """
1089 raise NotImplementedError(self.set_description)
1091 def get_rebase_state_manager(self) -> "RebaseStateManager":
1092 """Get the appropriate rebase state manager for this repository.
1094 Returns: RebaseStateManager instance
1095 """
1096 raise NotImplementedError(self.get_rebase_state_manager)
1098 def get_blob_normalizer(
1099 self, config: "Config | None" = None
1100 ) -> "FilterBlobNormalizer":
1101 """Return a BlobNormalizer object for checkin/checkout operations.
1103 Args:
1104 config: Configuration to consult for filter setup. If None,
1105 implementations may fall back to ``self.get_config_stack()``.
1107 Returns: BlobNormalizer instance
1108 """
1109 raise NotImplementedError(self.get_blob_normalizer)
1111 def get_gitattributes(self, tree: bytes | None = None) -> "GitAttributes":
1112 """Read gitattributes for the repository.
1114 Args:
1115 tree: Tree SHA to read .gitattributes from (defaults to HEAD)
1117 Returns:
1118 GitAttributes object that can be used to match paths
1119 """
1120 raise NotImplementedError(self.get_gitattributes)
1122 def get_config_stack(self) -> "StackedConfig":
1123 """Return a config stack for this repository.
1125 This stack accesses the configuration for both this repository
1126 itself (.git/config) and the global configuration, which usually
1127 lives in ~/.gitconfig.
1129 Returns: `Config` instance for this repository
1130 """
1131 from .config import ConfigFile, StackedConfig
1133 local_config = self.get_config()
1134 backends: list[ConfigFile] = [local_config]
1135 if local_config.get_boolean((b"extensions",), b"worktreeconfig", False):
1136 backends.append(self.get_worktree_config())
1138 backends += StackedConfig.default_backends()
1139 return StackedConfig(backends, writable=local_config)
1141 def get_shallow(self) -> set[ObjectID]:
1142 """Get the set of shallow commits.
1144 Returns: Set of shallow commits.
1145 """
1146 f = self.get_named_file("shallow")
1147 if f is None:
1148 return set()
1149 with f:
1150 return {ObjectID(line.strip()) for line in f}
1152 def update_shallow(
1153 self, new_shallow: set[ObjectID] | None, new_unshallow: set[ObjectID] | None
1154 ) -> None:
1155 """Update the list of shallow objects.
1157 Args:
1158 new_shallow: Newly shallow objects
1159 new_unshallow: Newly no longer shallow objects
1160 """
1161 shallow = self.get_shallow()
1162 if new_shallow:
1163 shallow.update(new_shallow)
1164 if new_unshallow:
1165 shallow.difference_update(new_unshallow)
1166 if shallow:
1167 self._put_named_file("shallow", b"".join([sha + b"\n" for sha in shallow]))
1168 else:
1169 self._del_named_file("shallow")
1171 def get_peeled(self, ref: Ref) -> ObjectID:
1172 """Get the peeled value of a ref.
1174 Args:
1175 ref: The refname to peel.
1176 Returns: The fully-peeled SHA1 of a tag object, after peeling all
1177 intermediate tags; if the original ref does not point to a tag,
1178 this will equal the original SHA1.
1179 """
1180 cached = self.refs.get_peeled(ref)
1181 if cached is not None:
1182 return cached
1183 return peel_sha(self.object_store, self.refs[ref])[1].id
1185 @property
1186 def notes(self) -> "Notes":
1187 """Access notes functionality for this repository.
1189 Returns:
1190 Notes object for accessing notes
1191 """
1192 from .notes import Notes
1194 return Notes(self.object_store, self.refs)
1196 def get_walker(
1197 self,
1198 include: Sequence[ObjectID] | None = None,
1199 exclude: Sequence[ObjectID] | None = None,
1200 order: str = "date",
1201 reverse: bool = False,
1202 max_entries: int | None = None,
1203 paths: Sequence[bytes] | None = None,
1204 rename_detector: "RenameDetector | None" = None,
1205 follow: bool = False,
1206 since: int | None = None,
1207 until: int | None = None,
1208 queue_cls: type | None = None,
1209 ) -> "Walker":
1210 """Obtain a walker for this repository.
1212 Args:
1213 include: Iterable of SHAs of commits to include along with their
1214 ancestors. Defaults to [HEAD]
1215 exclude: Iterable of SHAs of commits to exclude along with their
1216 ancestors, overriding includes.
1217 order: ORDER_* constant specifying the order of results.
1218 Anything other than ORDER_DATE may result in O(n) memory usage.
1219 reverse: If True, reverse the order of output, requiring O(n)
1220 memory.
1221 max_entries: The maximum number of entries to yield, or None for
1222 no limit.
1223 paths: Iterable of file or subtree paths to show entries for.
1224 rename_detector: diff.RenameDetector object for detecting
1225 renames.
1226 follow: If True, follow path across renames/copies. Forces a
1227 default rename_detector.
1228 since: Timestamp to list commits after.
1229 until: Timestamp to list commits before.
1230 queue_cls: A class to use for a queue of commits, supporting the
1231 iterator protocol. The constructor takes a single argument, the Walker.
1233 Returns: A `Walker` object
1234 """
1235 from .walk import Walker, _CommitTimeQueue
1237 if include is None:
1238 include = [self.head()]
1240 # Pass all arguments to Walker explicitly to avoid type issues with **kwargs
1241 return Walker(
1242 self.object_store,
1243 include,
1244 exclude=exclude,
1245 order=order,
1246 reverse=reverse,
1247 max_entries=max_entries,
1248 paths=paths,
1249 rename_detector=rename_detector,
1250 follow=follow,
1251 since=since,
1252 until=until,
1253 get_parents=lambda commit: self.get_parents(commit.id, commit),
1254 queue_cls=queue_cls if queue_cls is not None else _CommitTimeQueue,
1255 )
1257 def __getitem__(self, name: ObjectID | Ref | bytes) -> "ShaFile":
1258 """Retrieve a Git object by SHA1 or ref.
1260 Args:
1261 name: A Git object SHA1 or a ref name
1262 Returns: A `ShaFile` object, such as a Commit or Blob
1263 Raises:
1264 KeyError: when the specified ref or object does not exist
1265 """
1266 if not isinstance(name, bytes):
1267 raise TypeError(f"'name' must be bytestring, not {type(name).__name__:.80}")
1268 # If it looks like a ref name, only try refs
1269 if name == b"HEAD" or name.startswith(b"refs/"):
1270 try:
1271 return self.object_store[self.refs[Ref(name)]]
1272 except (RefFormatError, KeyError):
1273 pass
1274 # Otherwise, try as object ID if length matches
1275 if len(name) in (
1276 self.object_store.object_format.oid_length,
1277 self.object_store.object_format.hex_length,
1278 ):
1279 try:
1280 return self.object_store[
1281 ObjectID(name)
1282 if len(name) == self.object_store.object_format.hex_length
1283 else RawObjectID(name)
1284 ]
1285 except (KeyError, ValueError):
1286 pass
1287 # If nothing worked, raise KeyError
1288 raise KeyError(name)
1290 def __contains__(self, name: bytes) -> bool:
1291 """Check if a specific Git object or ref is present.
1293 Args:
1294 name: Git object SHA1/SHA256 or ref name
1295 """
1296 # Check if it's a binary or hex SHA
1297 if len(name) == self.object_format.hex_length and valid_hexsha(name):
1298 return ObjectID(name) in self.object_store or Ref(name) in self.refs
1299 return Ref(name) in self.refs
1301 def __setitem__(self, name: bytes, value: ShaFile | bytes) -> None:
1302 """Set a ref.
1304 Args:
1305 name: ref name
1306 value: Ref value - either a ShaFile object, or a hex sha
1307 """
1308 if name.startswith(b"refs/") or name == HEADREF:
1309 ref_name = Ref(name)
1310 if isinstance(value, ShaFile):
1311 self.refs[ref_name] = value.id
1312 elif isinstance(value, bytes):
1313 self.refs[ref_name] = ObjectID(value)
1314 else:
1315 raise TypeError(value)
1316 else:
1317 raise ValueError(name)
1319 def __delitem__(self, name: bytes) -> None:
1320 """Remove a ref.
1322 Args:
1323 name: Name of the ref to remove
1324 """
1325 if name.startswith(b"refs/") or name == HEADREF:
1326 del self.refs[Ref(name)]
1327 else:
1328 raise ValueError(name)
1330 def _get_user_identity(
1331 self, config: "StackedConfig", kind: str | None = None
1332 ) -> bytes:
1333 """Determine the identity to use for new commits."""
1334 warnings.warn(
1335 "use get_user_identity() rather than Repo._get_user_identity",
1336 DeprecationWarning,
1337 )
1338 return get_user_identity(config)
1340 def _add_graftpoints(
1341 self, updated_graftpoints: dict[ObjectID, list[ObjectID]]
1342 ) -> None:
1343 """Add or modify graftpoints.
1345 Args:
1346 updated_graftpoints: Dict of commit shas to list of parent shas
1347 """
1348 # Simple validation
1349 for commit, parents in updated_graftpoints.items():
1350 for sha in [commit, *parents]:
1351 check_hexsha(sha, "Invalid graftpoint")
1353 self._graftpoints.update(updated_graftpoints)
1355 def _remove_graftpoints(self, to_remove: Sequence[ObjectID] = ()) -> None:
1356 """Remove graftpoints.
1358 Args:
1359 to_remove: List of commit shas
1360 """
1361 for sha in to_remove:
1362 del self._graftpoints[sha]
1364 def _read_heads(self, name: str) -> list[ObjectID]:
1365 f = self.get_named_file(name)
1366 if f is None:
1367 return []
1368 with f:
1369 return [ObjectID(line.strip()) for line in f.readlines() if line.strip()]
1371 def get_worktree(self) -> "WorkTree":
1372 """Get the working tree for this repository.
1374 Returns:
1375 WorkTree instance for performing working tree operations
1377 Raises:
1378 NotImplementedError: If the repository doesn't support working trees
1379 """
1380 raise NotImplementedError(
1381 "Working tree operations not supported by this repository type"
1382 )
1385def read_gitfile(f: BinaryIO) -> str:
1386 """Read a ``.git`` file.
1388 The first line of the file should start with "gitdir: "
1390 Args:
1391 f: File-like object to read from
1392 Returns: A path
1393 """
1394 cs = f.read()
1395 if not cs.startswith(b"gitdir: "):
1396 raise ValueError("Expected file to start with 'gitdir: '")
1397 return cs[len(b"gitdir: ") :].rstrip(b"\r\n").decode("utf-8")
1400class UnsupportedVersion(Exception):
1401 """Unsupported repository version."""
1403 def __init__(self, version: int) -> None:
1404 """Initialize UnsupportedVersion exception.
1406 Args:
1407 version: The unsupported repository version
1408 """
1409 self.version = version
1412class UnsupportedExtension(Exception):
1413 """Unsupported repository extension."""
1415 def __init__(self, extension: str) -> None:
1416 """Initialize UnsupportedExtension exception.
1418 Args:
1419 extension: The unsupported repository extension
1420 """
1421 self.extension = extension
1424class InvalidWorktreeConfiguration(Exception):
1425 """core.worktree is set on a bare repository."""
1428class Repo(BaseRepo):
1429 """A git repository backed by local disk.
1431 To open an existing repository, call the constructor with
1432 the path of the repository.
1434 To create a new repository, use the Repo.init class method.
1436 Note that a repository object may hold on to resources such
1437 as file handles for performance reasons; call .close() to free
1438 up those resources.
1440 Attributes:
1441 path: Path to the working copy (if it exists) or repository control
1442 directory (if the repository is bare). ``core.worktree`` overrides
1443 this, so it is not necessarily the parent of the control directory.
1444 bare: Whether this is a bare repository
1445 """
1447 path: str
1448 bare: bool
1449 object_store: DiskObjectStore
1450 filter_context: "FilterContext | None"
1452 def __init__(
1453 self,
1454 root: str | bytes | os.PathLike[str],
1455 object_store: PackBasedObjectStore | None = None,
1456 bare: bool | None = None,
1457 ) -> None:
1458 """Open a repository on disk.
1460 Args:
1461 root: Path to the repository's root.
1462 object_store: ObjectStore to use; if omitted, we use the
1463 repository's default object store
1464 bare: True if this is a bare repository.
1465 """
1466 root = os.fspath(root)
1467 if isinstance(root, bytes):
1468 root = os.fsdecode(root)
1469 hidden_path = os.path.join(root, CONTROLDIR)
1470 if bare is None:
1471 if os.path.isfile(hidden_path) or os.path.isdir(
1472 os.path.join(hidden_path, OBJECTDIR)
1473 ):
1474 bare = False
1475 elif os.path.isdir(os.path.join(root, OBJECTDIR)) and os.path.isdir(
1476 os.path.join(root, REFSDIR)
1477 ):
1478 bare = True
1479 else:
1480 raise NotGitRepository(
1481 "No git repository was found at {path}".format(**dict(path=root))
1482 )
1484 self.bare = bare
1485 if bare is False:
1486 if os.path.isfile(hidden_path):
1487 with open(hidden_path, "rb") as f:
1488 path = read_gitfile(f)
1489 self._controldir = os.path.join(root, path)
1490 else:
1491 self._controldir = hidden_path
1492 else:
1493 self._controldir = root
1494 commondir = self.get_named_file(COMMONDIR)
1495 if commondir is not None:
1496 with commondir:
1497 self._commondir = os.path.join(
1498 self.controldir(),
1499 os.fsdecode(commondir.read().rstrip(b"\r\n")),
1500 )
1501 else:
1502 self._commondir = self._controldir
1503 self.path = root
1505 # Initialize refs early so they're available for config condition matchers
1506 self.refs = DiskRefsContainer(
1507 self.commondir(), self._controldir, logger=self._write_reflog
1508 )
1510 # Initialize worktrees container
1511 from .worktree import WorkTreeContainer
1513 self.worktrees = WorkTreeContainer(self)
1515 config = self.get_config()
1516 try:
1517 repository_format_version = config.get("core", "repositoryformatversion")
1518 format_version = (
1519 0
1520 if repository_format_version is None
1521 else int(repository_format_version)
1522 )
1523 except KeyError:
1524 format_version = 0
1526 if format_version not in (0, 1):
1527 raise UnsupportedVersion(format_version)
1529 try:
1530 worktree = config.get((b"core",), b"worktree")
1531 except KeyError:
1532 pass
1533 else:
1534 # A repository is bare because core.bare says so, not because it
1535 # was opened at its control directory: a submodule or a
1536 # --separate-git-dir repository is opened that way but still has a
1537 # working tree, named here.
1538 if config.get_boolean((b"core",), b"bare", False):
1539 raise InvalidWorktreeConfiguration(
1540 "core.bare and core.worktree are incompatible"
1541 )
1542 self.bare = False
1543 # Relative paths are resolved against the control directory, not
1544 # against the current directory or the repository root.
1545 self.path = os.path.join(self._controldir, os.fsdecode(worktree))
1547 # Track extensions we encounter
1548 has_reftable_extension = False
1549 for extension, value in config.items((b"extensions",)):
1550 if extension.lower() == b"refstorage":
1551 if value == b"reftable":
1552 has_reftable_extension = True
1553 else:
1554 raise UnsupportedExtension(f"refStorage = {value.decode()}")
1555 elif extension.lower() not in (
1556 b"worktreeconfig",
1557 b"objectformat",
1558 b"relativeworktrees",
1559 ):
1560 raise UnsupportedExtension(extension.decode("utf-8"))
1562 if object_store is None:
1563 # Get shared repository permissions from config
1564 try:
1565 shared_value = config.get(("core",), "sharedRepository")
1566 file_mode, dir_mode = parse_shared_repository(shared_value)
1567 except KeyError:
1568 file_mode, dir_mode = None, None
1570 object_store = DiskObjectStore.from_config(
1571 os.path.join(self.commondir(), OBJECTDIR),
1572 config,
1573 file_mode=file_mode,
1574 dir_mode=dir_mode,
1575 )
1577 # Use reftable if extension is configured
1578 if has_reftable_extension:
1579 from .reftable import ReftableRefsContainer
1581 self.refs = ReftableRefsContainer(self.commondir())
1582 # Update worktrees container after refs change
1583 self.worktrees = WorkTreeContainer(self)
1584 BaseRepo.__init__(self, object_store, self.refs)
1586 # Determine hash algorithm from config if not already set
1587 if self.object_format is None:
1588 from .object_format import DEFAULT_OBJECT_FORMAT, get_object_format
1590 if format_version == 1:
1591 try:
1592 object_format = config.get((b"extensions",), b"objectformat")
1593 self.object_format = get_object_format(
1594 object_format.decode("ascii")
1595 )
1596 except KeyError:
1597 self.object_format = DEFAULT_OBJECT_FORMAT
1598 else:
1599 self.object_format = DEFAULT_OBJECT_FORMAT
1601 self._graftpoints = {}
1602 graft_file = self.get_named_file(
1603 os.path.join("info", "grafts"), basedir=self.commondir()
1604 )
1605 if graft_file:
1606 with graft_file:
1607 self._graftpoints.update(parse_graftpoints(graft_file))
1608 graft_file = self.get_named_file("shallow", basedir=self.commondir())
1609 if graft_file:
1610 with graft_file:
1611 self._graftpoints.update(parse_graftpoints(graft_file))
1613 self.hooks["pre-commit"] = PreCommitShellHook(self.path, self.controldir())
1614 self.hooks["commit-msg"] = CommitMsgShellHook(self.controldir())
1615 self.hooks["post-commit"] = PostCommitShellHook(self.controldir())
1616 self.hooks["pre-receive"] = PreReceiveShellHook(self.controldir())
1617 self.hooks["update"] = UpdateShellHook(self.controldir())
1618 self.hooks["post-receive"] = PostReceiveShellHook(self.controldir())
1620 # Initialize filter context as None, will be created lazily
1621 self.filter_context = None
1623 def get_worktree(self) -> "WorkTree":
1624 """Get the working tree for this repository.
1626 Returns:
1627 WorkTree instance for performing working tree operations
1628 """
1629 from .worktree import WorkTree
1631 return WorkTree(self, self.path)
1633 def _write_reflog(
1634 self,
1635 ref: bytes,
1636 old_sha: bytes,
1637 new_sha: bytes,
1638 committer: bytes | None,
1639 timestamp: int | None,
1640 timezone: int | None,
1641 message: bytes,
1642 ) -> None:
1643 from .reflog import format_reflog_line
1645 path = self._reflog_path(ref)
1647 # Get shared repository permissions
1648 file_mode, dir_mode = self._get_shared_repository_permissions()
1650 # Create directory with appropriate permissions
1651 parent_dir = os.path.dirname(path)
1652 # Create directory tree, setting permissions on each level if needed
1653 parts = []
1654 current = parent_dir
1655 while current and not os.path.exists(current):
1656 parts.append(current)
1657 current = os.path.dirname(current)
1658 parts.reverse()
1659 for part in parts:
1660 os.mkdir(part)
1661 if dir_mode is not None:
1662 os.chmod(part, dir_mode)
1663 if committer is None:
1664 config = self.get_config_stack()
1665 committer = get_user_identity(config)
1666 check_user_identity(committer)
1667 if timestamp is None:
1668 timestamp = int(time.time())
1669 if timezone is None:
1670 timezone = 0 # FIXME
1671 with open(path, "ab") as f:
1672 f.write(
1673 format_reflog_line(
1674 old_sha, new_sha, committer, timestamp, timezone, message
1675 )
1676 + b"\n"
1677 )
1679 # Set file permissions (open() respects umask, so we need chmod to set the actual mode)
1680 # Always chmod to ensure correct permissions even if file already existed
1681 if file_mode is not None:
1682 os.chmod(path, file_mode)
1684 def _reflog_path(self, ref: bytes) -> str:
1685 if ref.startswith((b"main-worktree/", b"worktrees/")):
1686 raise NotImplementedError(f"refs {ref.decode()} are not supported")
1688 base = self.controldir() if is_per_worktree_ref(ref) else self.commondir()
1689 return os.path.join(base, "logs", os.fsdecode(ref))
1691 def read_reflog(self, ref: bytes) -> Generator[reflog.Entry, None, None]:
1692 """Read reflog entries for a reference.
1694 Args:
1695 ref: Reference name (e.g. b'HEAD', b'refs/heads/master')
1697 Yields:
1698 reflog.Entry objects in chronological order (oldest first)
1699 """
1700 from .reflog import read_reflog
1702 path = self._reflog_path(ref)
1703 try:
1704 with open(path, "rb") as f:
1705 yield from read_reflog(f)
1706 except FileNotFoundError:
1707 return
1709 @classmethod
1710 def discover(cls, start: str | bytes | os.PathLike[str] = ".") -> "Repo":
1711 """Iterate parent directories to discover a repository.
1713 Return a Repo object for the first parent directory that looks like a
1714 Git repository.
1716 Args:
1717 start: The directory to start discovery from (defaults to '.')
1718 """
1719 path = os.path.abspath(start)
1720 while True:
1721 try:
1722 return cls(path)
1723 except NotGitRepository:
1724 new_path, _tail = os.path.split(path)
1725 if new_path == path: # Root reached
1726 break
1727 path = new_path
1728 start_str = os.fspath(start)
1729 if isinstance(start_str, bytes):
1730 start_str = start_str.decode("utf-8")
1731 raise NotGitRepository(f"No git repository was found at {start_str}")
1733 def controldir(self) -> str:
1734 """Return the path of the control directory."""
1735 return self._controldir
1737 def commondir(self) -> str:
1738 """Return the path of the common directory.
1740 For a main working tree, it is identical to controldir().
1742 For a linked working tree, it is the control directory of the
1743 main working tree.
1744 """
1745 return self._commondir
1747 def _determine_file_mode(self) -> bool:
1748 """Probe the file-system to determine whether permissions can be trusted.
1750 Returns: True if permissions can be trusted, False otherwise.
1751 """
1752 fname = os.path.join(self.path, ".probe-permissions")
1753 with open(fname, "w") as f:
1754 f.write("")
1756 st1 = os.lstat(fname)
1757 try:
1758 os.chmod(fname, st1.st_mode ^ stat.S_IXUSR)
1759 except PermissionError:
1760 return False
1761 st2 = os.lstat(fname)
1763 os.unlink(fname)
1765 mode_differs = st1.st_mode != st2.st_mode
1766 st2_has_exec = (st2.st_mode & stat.S_IXUSR) != 0
1768 return mode_differs and st2_has_exec
1770 def _determine_symlinks(self) -> bool:
1771 """Probe the filesystem to determine whether symlinks can be created.
1773 Returns: True if symlinks can be created, False otherwise.
1774 """
1775 # TODO(jelmer): Actually probe disk / look at filesystem
1776 return sys.platform != "win32"
1778 def _get_shared_repository_permissions(
1779 self,
1780 ) -> tuple[int | None, int | None]:
1781 """Get shared repository file and directory permissions from config.
1783 Returns:
1784 tuple of (file_mask, directory_mask) or (None, None) if not shared
1785 """
1786 try:
1787 config = self.get_config()
1788 value = config.get(("core",), "sharedRepository")
1789 return parse_shared_repository(value)
1790 except KeyError:
1791 return (None, None)
1793 def _put_named_file(self, path: str, contents: bytes) -> None:
1794 """Write a file to the control dir with the given name and contents.
1796 Args:
1797 path: The path to the file, relative to the control dir.
1798 contents: A string to write to the file.
1799 """
1800 path = path.lstrip(os.path.sep)
1802 # Get shared repository permissions
1803 file_mode, _ = self._get_shared_repository_permissions()
1805 # Create file with appropriate permissions
1806 if file_mode is not None:
1807 with GitFile(
1808 os.path.join(self.controldir(), path), "wb", mask=file_mode
1809 ) as f:
1810 f.write(contents)
1811 else:
1812 with GitFile(os.path.join(self.controldir(), path), "wb") as f:
1813 f.write(contents)
1815 def _del_named_file(self, path: str) -> None:
1816 try:
1817 os.unlink(os.path.join(self.controldir(), path))
1818 except FileNotFoundError:
1819 return
1821 def get_named_file(
1822 self,
1823 path: str | bytes,
1824 basedir: str | None = None,
1825 ) -> BinaryIO | None:
1826 """Get a file from the control dir with a specific name.
1828 Although the filename should be interpreted as a filename relative to
1829 the control dir in a disk-based Repo, the object returned need not be
1830 pointing to a file in that location.
1832 Args:
1833 path: The path to the file, relative to the control dir.
1834 basedir: Optional argument that specifies an alternative to the
1835 control dir.
1836 Returns: An open file object, or None if the file does not exist.
1837 """
1838 # TODO(dborowitz): sanitize filenames, since this is used directly by
1839 # the dumb web serving code.
1840 if basedir is None:
1841 basedir = self.controldir()
1842 if isinstance(path, bytes):
1843 path = path.decode("utf-8")
1844 path = path.lstrip(os.path.sep)
1845 try:
1846 return open(os.path.join(basedir, path), "rb")
1847 except FileNotFoundError:
1848 return None
1850 def index_path(self) -> str:
1851 """Return path to the index file."""
1852 return os.path.join(self.controldir(), INDEX_FILENAME)
1854 def open_index(self, config: "Config | None" = None) -> "Index":
1855 """Open the index for this repository.
1857 Args:
1858 config: Configuration to consult for index settings. If None,
1859 falls back to ``self.get_config_stack()``.
1861 Raises:
1862 NoIndexPresent: If no index is present
1863 Returns: The matching `Index`
1864 """
1865 from .index import Index, make_path_normalizer
1867 if not self.has_index():
1868 raise NoIndexPresent
1870 if config is None:
1871 config = self.get_config_stack()
1872 many_files = config.get_boolean(b"feature", b"manyFiles", False)
1873 skip_hash = False
1874 index_version = None
1876 if many_files:
1877 # When feature.manyFiles is enabled, set index.version=4 and index.skipHash=true
1878 try:
1879 index_version_str = config.get(b"index", b"version")
1880 index_version = int(index_version_str)
1881 except KeyError:
1882 index_version = 4 # Default to version 4 for manyFiles
1883 skip_hash = config.get_boolean(b"index", b"skipHash", True)
1884 else:
1885 # Check for explicit index settings
1886 try:
1887 index_version_str = config.get(b"index", b"version")
1888 index_version = int(index_version_str)
1889 except KeyError:
1890 index_version = None
1891 skip_hash = config.get_boolean(b"index", b"skipHash", False)
1893 # Get shared repository permissions for index file
1894 file_mode, _ = self._get_shared_repository_permissions()
1896 return Index(
1897 self.index_path(),
1898 skip_hash=skip_hash,
1899 version=index_version,
1900 file_mode=file_mode,
1901 path_normalizer=make_path_normalizer(config),
1902 )
1904 def has_index(self) -> bool:
1905 """Check if an index is present."""
1906 # Bare repos must never have index files; non-bare repos may have a
1907 # missing index file, which is treated as empty.
1908 return not self.bare
1910 def clone(
1911 self,
1912 target_path: str | bytes | os.PathLike[str],
1913 *,
1914 mkdir: bool = True,
1915 bare: bool = False,
1916 origin: bytes = b"origin",
1917 checkout: bool | None = None,
1918 branch: bytes | None = None,
1919 progress: Callable[[str], None] | None = None,
1920 depth: int | None = None,
1921 symlinks: bool | None = None,
1922 ) -> "Repo":
1923 """Clone this repository.
1925 Args:
1926 target_path: Target path
1927 mkdir: Create the target directory
1928 bare: Whether to create a bare repository
1929 checkout: Whether or not to check-out HEAD after cloning
1930 origin: Base name for refs in target repository
1931 cloned from this repository
1932 branch: Optional branch or tag to be used as HEAD in the new repository
1933 instead of this repository's HEAD.
1934 progress: Optional progress function
1935 depth: Depth at which to fetch
1936 symlinks: Symlinks setting (default to autodetect)
1937 Returns: Created repository as `Repo`
1938 """
1939 encoded_path = os.fsencode(self.path)
1941 if mkdir:
1942 os.mkdir(target_path)
1944 try:
1945 if not bare:
1946 target = Repo.init(target_path, symlinks=symlinks)
1947 if checkout is None:
1948 checkout = True
1949 else:
1950 if checkout:
1951 raise ValueError("checkout and bare are incompatible")
1952 target = Repo.init_bare(target_path)
1954 try:
1955 target_config = target.get_config()
1956 target_config.set((b"remote", origin), b"url", encoded_path)
1957 target_config.set(
1958 (b"remote", origin),
1959 b"fetch",
1960 b"+refs/heads/*:refs/remotes/" + origin + b"/*",
1961 )
1962 target_config.write_to_path()
1964 ref_message = b"clone: from " + encoded_path
1965 self.fetch(target, depth=depth)
1966 target.refs.import_refs(
1967 Ref(b"refs/remotes/" + origin),
1968 self.refs.as_dict(Ref(b"refs/heads")),
1969 message=ref_message,
1970 )
1971 target.refs.import_refs(
1972 Ref(b"refs/tags"),
1973 self.refs.as_dict(Ref(b"refs/tags")),
1974 message=ref_message,
1975 )
1977 head_chain, origin_sha = self.refs.follow(HEADREF)
1978 origin_head = head_chain[-1] if head_chain else None
1979 head: ObjectID | None = None
1980 if origin_sha and not origin_head:
1981 # set detached HEAD
1982 target.refs[HEADREF] = origin_sha
1983 head = origin_sha
1984 else:
1985 _set_origin_head(target.refs, origin, origin_head)
1986 head_ref = _set_default_branch(
1987 target.refs, origin, origin_head, branch, ref_message
1988 )
1990 # Update target head
1991 if head_ref:
1992 head = _set_head(target.refs, head_ref, ref_message)
1993 else:
1994 head = None
1996 if checkout and head is not None:
1997 target.get_worktree().reset_index(config=target.get_config_stack())
1998 except BaseException:
1999 target.close()
2000 raise
2001 except BaseException:
2002 if mkdir:
2003 import shutil
2005 shutil.rmtree(target_path)
2006 raise
2007 return target
2009 def _get_config_condition_matchers(self) -> dict[str, "ConditionMatcher"]:
2010 """Get condition matchers for includeIf conditions.
2012 Returns a dict of condition prefix to matcher function.
2013 """
2014 from pathlib import Path
2016 from .config import ConditionMatcher, match_glob_pattern
2018 # Add gitdir matchers
2019 def match_gitdir(pattern: str, case_sensitive: bool = True) -> bool:
2020 """Match gitdir against a pattern.
2022 Args:
2023 pattern: Pattern to match against
2024 case_sensitive: Whether to match case-sensitively
2026 Returns:
2027 True if gitdir matches pattern
2028 """
2029 # Handle relative patterns (starting with ./)
2030 if pattern.startswith("./"):
2031 # Can't handle relative patterns without config directory context
2032 return False
2034 # Normalize repository path
2035 try:
2036 repo_path = str(Path(self._controldir).resolve())
2037 except (OSError, ValueError):
2038 return False
2040 # Expand ~ in pattern and normalize
2041 pattern = os.path.expanduser(pattern)
2043 # Normalize pattern following Git's rules
2044 pattern = pattern.replace("\\", "/")
2045 if not pattern.startswith(("~/", "./", "/", "**")):
2046 # Check for Windows absolute path
2047 if len(pattern) >= 2 and pattern[1] == ":":
2048 pass
2049 else:
2050 pattern = "**/" + pattern
2051 if pattern.endswith("/"):
2052 pattern = pattern + "**"
2054 # Use the existing _match_gitdir_pattern function
2055 from .config import _match_gitdir_pattern
2057 pattern_bytes = pattern.encode("utf-8", errors="replace")
2058 repo_path_bytes = repo_path.encode("utf-8", errors="replace")
2060 return _match_gitdir_pattern(
2061 repo_path_bytes, pattern_bytes, ignorecase=not case_sensitive
2062 )
2064 # Add onbranch matcher
2065 def match_onbranch(pattern: str) -> bool:
2066 """Match current branch against a pattern.
2068 Args:
2069 pattern: Pattern to match against
2071 Returns:
2072 True if current branch matches pattern
2073 """
2074 try:
2075 # Get the current branch using refs
2076 ref_chain, _ = self.refs.follow(HEADREF)
2077 head_ref = ref_chain[-1] # Get the final resolved ref
2078 except KeyError:
2079 pass
2080 else:
2081 if head_ref and head_ref.startswith(b"refs/heads/"):
2082 # Extract branch name from ref
2083 branch = extract_branch_name(head_ref).decode(
2084 "utf-8", errors="replace"
2085 )
2086 return match_glob_pattern(branch, pattern)
2087 return False
2089 matchers: dict[str, ConditionMatcher] = {
2090 "onbranch:": match_onbranch,
2091 "gitdir:": lambda pattern: match_gitdir(pattern, True),
2092 "gitdir/i:": lambda pattern: match_gitdir(pattern, False),
2093 }
2095 return matchers
2097 def get_worktree_config(self) -> "ConfigFile":
2098 """Get the worktree-specific config.
2100 Returns:
2101 ConfigFile object for the worktree config
2102 """
2103 from .config import ConfigFile
2105 path = os.path.join(self.commondir(), "config.worktree")
2106 try:
2107 # Pass condition matchers for includeIf evaluation
2108 condition_matchers = self._get_config_condition_matchers()
2109 return ConfigFile.from_path(path, condition_matchers=condition_matchers)
2110 except FileNotFoundError:
2111 cf = ConfigFile()
2112 cf.path = path
2113 return cf
2115 def get_config(self) -> "ConfigFile":
2116 """Retrieve the config object.
2118 Returns: `ConfigFile` object for the ``.git/config`` file.
2119 """
2120 from .config import ConfigFile
2122 path = os.path.join(self._commondir, "config")
2123 try:
2124 # Pass condition matchers for includeIf evaluation
2125 condition_matchers = self._get_config_condition_matchers()
2126 return ConfigFile.from_path(path, condition_matchers=condition_matchers)
2127 except FileNotFoundError:
2128 ret = ConfigFile()
2129 ret.path = path
2130 return ret
2132 def get_rebase_state_manager(self) -> "RebaseStateManager":
2133 """Get the appropriate rebase state manager for this repository.
2135 Returns: DiskRebaseStateManager instance
2136 """
2137 import os
2139 from .rebase import DiskRebaseStateManager
2141 path = os.path.join(self.controldir(), "rebase-merge")
2142 return DiskRebaseStateManager(path)
2144 def get_description(self) -> bytes | None:
2145 """Retrieve the description of this repository.
2147 Returns: Description as bytes or None.
2148 """
2149 path = os.path.join(self._controldir, "description")
2150 try:
2151 with GitFile(path, "rb") as f:
2152 return f.read()
2153 except FileNotFoundError:
2154 return None
2156 def __repr__(self) -> str:
2157 """Return string representation of this repository."""
2158 return f"<Repo at {self.path!r}>"
2160 def set_description(self, description: bytes) -> None:
2161 """Set the description for this repository.
2163 Args:
2164 description: Text to set as description for this repository.
2165 """
2166 self._put_named_file("description", description)
2168 @classmethod
2169 def _init_maybe_bare(
2170 cls,
2171 path: str | bytes | os.PathLike[str],
2172 controldir: str | bytes | os.PathLike[str],
2173 bare: bool,
2174 object_store: PackBasedObjectStore | None = None,
2175 config: "StackedConfig | None" = None,
2176 default_branch: bytes | None = None,
2177 symlinks: bool | None = None,
2178 format: int | None = None,
2179 shared_repository: str | bool | None = None,
2180 object_format: str | None = None,
2181 ) -> "Repo":
2182 path = os.fspath(path)
2183 if isinstance(path, bytes):
2184 path = os.fsdecode(path)
2185 controldir = os.fspath(controldir)
2186 if isinstance(controldir, bytes):
2187 controldir = os.fsdecode(controldir)
2189 # Determine shared repository permissions early
2190 file_mode: int | None = None
2191 dir_mode: int | None = None
2192 if shared_repository is not None:
2193 file_mode, dir_mode = parse_shared_repository(shared_repository)
2195 # Create base directories with appropriate permissions
2196 for d in BASE_DIRECTORIES:
2197 dir_path = os.path.join(controldir, *d)
2198 os.mkdir(dir_path)
2199 if dir_mode is not None:
2200 os.chmod(dir_path, dir_mode)
2202 # Determine hash algorithm
2203 from .object_format import get_object_format
2205 hash_alg = get_object_format(object_format)
2207 if object_store is None:
2208 object_store = DiskObjectStore.init(
2209 os.path.join(controldir, OBJECTDIR),
2210 file_mode=file_mode,
2211 dir_mode=dir_mode,
2212 object_format=hash_alg,
2213 )
2214 ret = cls(path, bare=bare, object_store=object_store)
2215 if default_branch is None:
2216 if config is None:
2217 from .config import StackedConfig
2219 config = StackedConfig.default()
2220 try:
2221 default_branch = config.get("init", "defaultBranch")
2222 except KeyError:
2223 default_branch = DEFAULT_BRANCH
2224 ret.refs.set_symbolic_ref(HEADREF, local_branch_name(default_branch))
2225 ret._init_files(
2226 bare=bare,
2227 symlinks=symlinks,
2228 format=format,
2229 shared_repository=shared_repository,
2230 object_format=object_format,
2231 )
2232 return ret
2234 @classmethod
2235 def init(
2236 cls,
2237 path: str | bytes | os.PathLike[str],
2238 *,
2239 mkdir: bool = False,
2240 config: "StackedConfig | None" = None,
2241 default_branch: bytes | None = None,
2242 symlinks: bool | None = None,
2243 format: int | None = None,
2244 shared_repository: str | bool | None = None,
2245 object_format: str | None = None,
2246 ) -> "Repo":
2247 """Create a new repository.
2249 Args:
2250 path: Path in which to create the repository
2251 mkdir: Whether to create the directory
2252 config: Configuration object
2253 default_branch: Default branch name
2254 symlinks: Whether to support symlinks
2255 format: Repository format version (defaults to 0)
2256 shared_repository: Shared repository setting (group, all, umask, or octal)
2257 object_format: Object format to use ("sha1" or "sha256", defaults to "sha1")
2258 Returns: `Repo` instance
2259 """
2260 path = os.fspath(path)
2261 if isinstance(path, bytes):
2262 path = os.fsdecode(path)
2263 if mkdir:
2264 os.mkdir(path)
2265 controldir = os.path.join(path, CONTROLDIR)
2266 os.mkdir(controldir)
2267 _set_filesystem_hidden(controldir)
2268 return cls._init_maybe_bare(
2269 path,
2270 controldir,
2271 False,
2272 config=config,
2273 default_branch=default_branch,
2274 symlinks=symlinks,
2275 format=format,
2276 shared_repository=shared_repository,
2277 object_format=object_format,
2278 )
2280 @classmethod
2281 def _init_new_working_directory(
2282 cls,
2283 path: str | bytes | os.PathLike[str],
2284 main_repo: "Repo",
2285 identifier: str | None = None,
2286 mkdir: bool = False,
2287 relative_paths: bool = False,
2288 ) -> "Repo":
2289 """Create a new working directory linked to a repository.
2291 Args:
2292 path: Path in which to create the working tree.
2293 main_repo: Main repository to reference
2294 identifier: Worktree identifier
2295 mkdir: Whether to create the directory
2296 relative_paths: Whether to use relative paths for gitdir references
2297 Returns: `Repo` instance
2298 """
2299 path = os.fspath(path)
2300 if isinstance(path, bytes):
2301 path = os.fsdecode(path)
2302 if mkdir:
2303 os.mkdir(path)
2304 if identifier is None:
2305 identifier = os.path.basename(path)
2306 # Ensure we use absolute path for the worktree control directory
2307 main_controldir = os.path.abspath(main_repo.controldir())
2308 main_worktreesdir = os.path.join(main_controldir, WORKTREES)
2309 worktree_controldir = os.path.join(main_worktreesdir, identifier)
2310 gitdirfile_abs = os.path.abspath(os.path.join(path, CONTROLDIR))
2312 # Write gitdir reference in .git file (can be relative)
2313 # Import helper from worktree module to avoid duplication
2314 from .worktree import _compute_gitdir_path
2316 gitdir_ref = _compute_gitdir_path(
2317 main_repo,
2318 worktree_controldir,
2319 os.path.dirname(gitdirfile_abs),
2320 relative_paths,
2321 )
2323 with open(gitdirfile_abs, "wb") as f:
2324 f.write(b"gitdir: " + os.fsencode(gitdir_ref) + b"\n")
2326 # Get shared repository permissions from main repository
2327 _, dir_mode = main_repo._get_shared_repository_permissions()
2329 # Create directories with appropriate permissions
2330 try:
2331 os.mkdir(main_worktreesdir)
2332 if dir_mode is not None:
2333 os.chmod(main_worktreesdir, dir_mode)
2334 except FileExistsError:
2335 pass
2336 try:
2337 os.mkdir(worktree_controldir)
2338 if dir_mode is not None:
2339 os.chmod(worktree_controldir, dir_mode)
2340 except FileExistsError:
2341 pass
2343 # Write gitdir path in control directory (can be relative)
2344 gitdir_path = _compute_gitdir_path(
2345 main_repo, gitdirfile_abs, worktree_controldir, relative_paths
2346 )
2348 with open(os.path.join(worktree_controldir, GITDIR), "wb") as f:
2349 f.write(os.fsencode(gitdir_path) + b"\n")
2350 with open(os.path.join(worktree_controldir, COMMONDIR), "wb") as f:
2351 f.write(b"../..\n")
2352 with open(os.path.join(worktree_controldir, "HEAD"), "wb") as f:
2353 f.write(main_repo.head() + b"\n")
2354 r = cls(os.path.normpath(path))
2355 r.get_worktree().reset_index(config=r.get_config_stack())
2356 return r
2358 @classmethod
2359 def init_bare(
2360 cls,
2361 path: str | bytes | os.PathLike[str],
2362 *,
2363 mkdir: bool = False,
2364 object_store: PackBasedObjectStore | None = None,
2365 config: "StackedConfig | None" = None,
2366 default_branch: bytes | None = None,
2367 format: int | None = None,
2368 shared_repository: str | bool | None = None,
2369 object_format: str | None = None,
2370 ) -> "Repo":
2371 """Create a new bare repository.
2373 ``path`` should already exist and be an empty directory.
2375 Args:
2376 path: Path to create bare repository in
2377 mkdir: Whether to create the directory
2378 object_store: Object store to use
2379 config: Configuration object
2380 default_branch: Default branch name
2381 format: Repository format version (defaults to 0)
2382 shared_repository: Shared repository setting (group, all, umask, or octal)
2383 object_format: Object format to use ("sha1" or "sha256", defaults to "sha1")
2384 Returns: a `Repo` instance
2385 """
2386 path = os.fspath(path)
2387 if isinstance(path, bytes):
2388 path = os.fsdecode(path)
2389 if mkdir:
2390 os.mkdir(path)
2391 return cls._init_maybe_bare(
2392 path,
2393 path,
2394 True,
2395 object_store=object_store,
2396 config=config,
2397 default_branch=default_branch,
2398 format=format,
2399 shared_repository=shared_repository,
2400 object_format=object_format,
2401 )
2403 create = init_bare
2405 def close(self) -> None:
2406 """Close any files opened by this repository."""
2407 self.object_store.close()
2408 # Clean up filter context if it was created
2409 if self.filter_context is not None:
2410 self.filter_context.close()
2411 self.filter_context = None
2413 def __enter__(self) -> Self:
2414 """Enter context manager."""
2415 return self
2417 def __exit__(
2418 self,
2419 exc_type: type[BaseException] | None,
2420 exc_val: BaseException | None,
2421 exc_tb: TracebackType | None,
2422 ) -> None:
2423 """Exit context manager and close repository."""
2424 self.close()
2426 def _read_gitattributes(self) -> dict[bytes, dict[bytes, bytes]]:
2427 """Read .gitattributes file from working tree.
2429 Returns:
2430 Dictionary mapping file patterns to attributes
2431 """
2432 gitattributes = {}
2433 gitattributes_path = os.path.join(self.path, ".gitattributes")
2435 if os.path.exists(gitattributes_path):
2436 with open(gitattributes_path, "rb") as f:
2437 for line in f:
2438 line = line.strip()
2439 if not line or line.startswith(b"#"):
2440 continue
2442 parts = line.split()
2443 if len(parts) < 2:
2444 continue
2446 pattern = parts[0]
2447 attrs = {}
2449 for attr in parts[1:]:
2450 if attr.startswith(b"-"):
2451 # Unset attribute
2452 attrs[attr[1:]] = b"false"
2453 elif b"=" in attr:
2454 # Set to value
2455 key, value = attr.split(b"=", 1)
2456 attrs[key] = value
2457 else:
2458 # Set attribute
2459 attrs[attr] = b"true"
2461 gitattributes[pattern] = attrs
2463 return gitattributes
2465 def get_blob_normalizer(
2466 self, config: "Config | None" = None
2467 ) -> "FilterBlobNormalizer":
2468 """Return a BlobNormalizer object.
2470 Args:
2471 config: Configuration to consult for filter setup. If None,
2472 falls back to ``self.get_config_stack()``.
2473 """
2474 from .filters import FilterBlobNormalizer, FilterContext, FilterRegistry
2476 if config is None:
2477 config = self.get_config_stack()
2478 git_attributes = self.get_gitattributes()
2480 # Lazily create FilterContext if needed
2481 if self.filter_context is None:
2482 filter_registry = FilterRegistry(config, self)
2483 self.filter_context = FilterContext(filter_registry)
2484 else:
2485 # Refresh the context with current config to handle config changes
2486 self.filter_context.refresh_config(config)
2488 return FilterBlobNormalizer(
2489 config, git_attributes, filter_context=self.filter_context
2490 )
2492 def get_gitattributes(self, tree: bytes | None = None) -> "GitAttributes":
2493 """Read gitattributes for the repository.
2495 Args:
2496 tree: Tree SHA to read .gitattributes from (defaults to HEAD)
2498 Returns:
2499 GitAttributes object that can be used to match paths
2500 """
2501 from .attrs import (
2502 GitAttributes,
2503 Pattern,
2504 parse_git_attributes,
2505 )
2507 patterns = []
2509 # Read system gitattributes (TODO: implement this)
2510 # Read global gitattributes (TODO: implement this)
2512 # Read repository .gitattributes from index/tree
2513 if tree is None:
2514 try:
2515 # Try to get from HEAD
2516 head = self[b"HEAD"]
2517 # Peel tags to get to the underlying commit
2518 while isinstance(head, Tag):
2519 _cls, obj = head.object
2520 head = self.get_object(obj)
2521 if not isinstance(head, Commit):
2522 raise ValueError(
2523 f"Expected HEAD to point to a Commit, got {type(head).__name__}. "
2524 f"This usually means HEAD points to a {type(head).__name__} object "
2525 f"instead of a Commit."
2526 )
2527 tree = head.tree
2528 except KeyError:
2529 # No HEAD, no attributes from tree
2530 pass
2532 if tree is not None:
2533 try:
2534 tree_obj = self[tree]
2535 assert isinstance(tree_obj, Tree)
2536 if b".gitattributes" in tree_obj:
2537 _, attrs_sha = tree_obj[b".gitattributes"]
2538 attrs_blob = self[attrs_sha]
2539 if isinstance(attrs_blob, Blob):
2540 attrs_data = BytesIO(attrs_blob.data)
2541 for pattern_bytes, attrs in parse_git_attributes(attrs_data):
2542 pattern = Pattern(pattern_bytes)
2543 patterns.append((pattern, attrs))
2544 except (KeyError, NotTreeError):
2545 pass
2547 # Read .git/info/attributes
2548 info_attrs_path = os.path.join(self.controldir(), "info", "attributes")
2549 if os.path.exists(info_attrs_path):
2550 with open(info_attrs_path, "rb") as f:
2551 for pattern_bytes, attrs in parse_git_attributes(f):
2552 pattern = Pattern(pattern_bytes)
2553 patterns.append((pattern, attrs))
2555 # Read .gitattributes from working directory (if it exists)
2556 working_attrs_path = os.path.join(self.path, ".gitattributes")
2557 if os.path.exists(working_attrs_path):
2558 with open(working_attrs_path, "rb") as f:
2559 for pattern_bytes, attrs in parse_git_attributes(f):
2560 pattern = Pattern(pattern_bytes)
2561 patterns.append((pattern, attrs))
2563 return GitAttributes(patterns)
2566class MemoryRepo(BaseRepo):
2567 """Repo that stores refs, objects, and named files in memory.
2569 MemoryRepos are always bare: they have no working tree and no index, since
2570 those have a stronger dependency on the filesystem.
2571 """
2573 filter_context: "FilterContext | None"
2575 def __init__(self) -> None:
2576 """Create a new repository in memory."""
2577 from .config import ConfigFile
2578 from .object_format import DEFAULT_OBJECT_FORMAT
2580 self._reflog: list[Any] = []
2581 refs_container = DictRefsContainer({}, logger=self._append_reflog)
2582 BaseRepo.__init__(self, MemoryObjectStore(), refs_container)
2583 self._named_files: dict[str, bytes] = {}
2584 self.bare = True
2585 self._config = ConfigFile()
2586 self._description: bytes | None = None
2587 self.filter_context = None
2588 # MemoryRepo defaults to default object format
2589 self.object_format = DEFAULT_OBJECT_FORMAT
2591 def _append_reflog(
2592 self,
2593 ref: bytes,
2594 old_sha: bytes | None,
2595 new_sha: bytes | None,
2596 committer: bytes | None,
2597 timestamp: int | None,
2598 timezone: int | None,
2599 message: bytes | None,
2600 ) -> None:
2601 self._reflog.append(
2602 (ref, old_sha, new_sha, committer, timestamp, timezone, message)
2603 )
2605 def set_description(self, description: bytes) -> None:
2606 """Set the description for this repository.
2608 Args:
2609 description: Text to set as description
2610 """
2611 self._description = description
2613 def get_description(self) -> bytes | None:
2614 """Get the description of this repository.
2616 Returns:
2617 Repository description as bytes
2618 """
2619 return self._description
2621 def _determine_file_mode(self) -> bool:
2622 """Probe the file-system to determine whether permissions can be trusted.
2624 Returns: True if permissions can be trusted, False otherwise.
2625 """
2626 return sys.platform != "win32"
2628 def _determine_symlinks(self) -> bool:
2629 """Probe the file-system to determine whether permissions can be trusted.
2631 Returns: True if permissions can be trusted, False otherwise.
2632 """
2633 return sys.platform != "win32"
2635 def _put_named_file(self, path: str, contents: bytes) -> None:
2636 """Write a file to the control dir with the given name and contents.
2638 Args:
2639 path: The path to the file, relative to the control dir.
2640 contents: A string to write to the file.
2641 """
2642 self._named_files[path] = contents
2644 def _del_named_file(self, path: str) -> None:
2645 try:
2646 del self._named_files[path]
2647 except KeyError:
2648 pass
2650 def get_named_file(
2651 self,
2652 path: str | bytes,
2653 basedir: str | None = None,
2654 ) -> BytesIO | None:
2655 """Get a file from the control dir with a specific name.
2657 Although the filename should be interpreted as a filename relative to
2658 the control dir in a disk-baked Repo, the object returned need not be
2659 pointing to a file in that location.
2661 Args:
2662 path: The path to the file, relative to the control dir.
2663 basedir: Optional base directory for the path
2664 Returns: An open file object, or None if the file does not exist.
2665 """
2666 path_str = path.decode() if isinstance(path, bytes) else path
2667 contents = self._named_files.get(path_str, None)
2668 if contents is None:
2669 return None
2670 return BytesIO(contents)
2672 def open_index(self, config: "Config | None" = None) -> "Index":
2673 """Fail to open index for this repo, since it is bare.
2675 Args:
2676 config: Unused; kept for signature compatibility with ``BaseRepo``.
2678 Raises:
2679 NoIndexPresent: Raised when no index is present
2680 """
2681 raise NoIndexPresent
2683 def _init_config(self, config: "ConfigFile") -> None:
2684 """Initialize repository configuration for MemoryRepo."""
2685 self._config = config
2687 def get_config(self) -> "ConfigFile":
2688 """Retrieve the config object.
2690 Returns: `ConfigFile` object.
2691 """
2692 return self._config
2694 def get_rebase_state_manager(self) -> "RebaseStateManager":
2695 """Get the appropriate rebase state manager for this repository.
2697 Returns: MemoryRebaseStateManager instance
2698 """
2699 from .rebase import MemoryRebaseStateManager
2701 return MemoryRebaseStateManager(self)
2703 def get_blob_normalizer(
2704 self, config: "Config | None" = None
2705 ) -> "FilterBlobNormalizer":
2706 """Return a BlobNormalizer object for checkin/checkout operations.
2708 Args:
2709 config: Configuration to consult for filter setup. If None,
2710 falls back to ``self.get_config_stack()``.
2711 """
2712 from .filters import FilterBlobNormalizer, FilterContext, FilterRegistry
2714 if config is None:
2715 config = self.get_config_stack()
2716 git_attributes = self.get_gitattributes()
2718 if self.filter_context is None:
2719 filter_registry = FilterRegistry(config, self)
2720 self.filter_context = FilterContext(filter_registry)
2721 else:
2722 self.filter_context.refresh_config(config)
2724 return FilterBlobNormalizer(
2725 config, git_attributes, filter_context=self.filter_context
2726 )
2728 def get_gitattributes(self, tree: bytes | None = None) -> "GitAttributes":
2729 """Read gitattributes for the repository."""
2730 from .attrs import GitAttributes
2732 # Memory repos don't have working trees or gitattributes files
2733 # Return empty GitAttributes
2734 return GitAttributes([])
2736 def close(self) -> None:
2737 """Close any resources opened by this repository."""
2738 # Clean up filter context if it was created
2739 if self.filter_context is not None:
2740 self.filter_context.close()
2741 self.filter_context = None
2742 # Close object store to release pack files
2743 self.object_store.close()
2745 def do_commit(
2746 self,
2747 message: bytes | None = None,
2748 committer: bytes | None = None,
2749 author: bytes | None = None,
2750 commit_timestamp: float | None = None,
2751 commit_timezone: int | None = None,
2752 author_timestamp: float | None = None,
2753 author_timezone: int | None = None,
2754 tree: ObjectID | None = None,
2755 encoding: bytes | None = None,
2756 ref: Ref | None = HEADREF,
2757 merge_heads: list[ObjectID] | None = None,
2758 no_verify: bool = False,
2759 sign: bool = False,
2760 config: "Config | None" = None,
2761 ) -> bytes:
2762 """Create a new commit.
2764 This is a simplified implementation for in-memory repositories that
2765 doesn't support worktree operations or hooks.
2767 Args:
2768 message: Commit message
2769 committer: Committer fullname
2770 author: Author fullname
2771 commit_timestamp: Commit timestamp (defaults to now)
2772 commit_timezone: Commit timestamp timezone (defaults to GMT)
2773 author_timestamp: Author timestamp (defaults to commit timestamp)
2774 author_timezone: Author timestamp timezone (defaults to commit timezone)
2775 tree: SHA1 of the tree root to use
2776 encoding: Encoding
2777 ref: Optional ref to commit to (defaults to current branch).
2778 If None, creates a dangling commit without updating any ref.
2779 merge_heads: Merge heads
2780 no_verify: Skip pre-commit and commit-msg hooks (ignored for MemoryRepo)
2781 sign: GPG Sign the commit (ignored for MemoryRepo)
2782 config: Configuration to consult for committer/author identity. If
2783 None, falls back to ``self.get_config_stack()``.
2785 Returns:
2786 New commit SHA1
2787 """
2788 import time
2790 from .objects import Commit
2792 if tree is None:
2793 raise ValueError("tree must be specified for MemoryRepo")
2795 c = Commit()
2796 if len(tree) != self.object_format.hex_length:
2797 raise ValueError(
2798 f"tree must be a {self.object_format.hex_length}-character hex sha string"
2799 )
2800 c.tree = tree
2802 if config is None:
2803 config = self.get_config_stack()
2804 if merge_heads is None:
2805 merge_heads = []
2806 if committer is None:
2807 committer = get_user_identity(config, kind="COMMITTER")
2808 check_user_identity(committer)
2809 c.committer = committer
2810 if commit_timestamp is None:
2811 commit_timestamp = time.time()
2812 c.commit_time = int(commit_timestamp)
2813 if commit_timezone is None:
2814 commit_timezone = 0
2815 c.commit_timezone = commit_timezone
2816 if author is None:
2817 author = get_user_identity(config, kind="AUTHOR")
2818 c.author = author
2819 check_user_identity(author)
2820 if author_timestamp is None:
2821 author_timestamp = commit_timestamp
2822 c.author_time = int(author_timestamp)
2823 if author_timezone is None:
2824 author_timezone = commit_timezone
2825 c.author_timezone = author_timezone
2826 if encoding is None:
2827 try:
2828 encoding = config.get(("i18n",), "commitEncoding")
2829 except KeyError:
2830 pass
2831 if encoding is not None:
2832 c.encoding = encoding
2834 # Handle message (for MemoryRepo, we don't support callable messages)
2835 if callable(message):
2836 message = message(self, c)
2837 if message is None:
2838 raise ValueError("Message callback returned None")
2840 if message is None:
2841 raise ValueError("No commit message specified")
2843 c.message = message
2845 if ref is None:
2846 # Create a dangling commit
2847 c.parents = merge_heads
2848 self.object_store.add_object(c)
2849 else:
2850 try:
2851 old_head = self.refs[ref]
2852 c.parents = [old_head, *merge_heads]
2853 self.object_store.add_object(c)
2854 ok = self.refs.set_if_equals(
2855 ref,
2856 old_head,
2857 c.id,
2858 message=b"commit: " + message,
2859 committer=committer,
2860 timestamp=int(commit_timestamp),
2861 timezone=commit_timezone,
2862 )
2863 except KeyError:
2864 c.parents = merge_heads
2865 self.object_store.add_object(c)
2866 ok = self.refs.add_if_new(
2867 ref,
2868 c.id,
2869 message=b"commit: " + message,
2870 committer=committer,
2871 timestamp=int(commit_timestamp),
2872 timezone=commit_timezone,
2873 )
2874 if not ok:
2875 from .errors import CommitError
2877 raise CommitError(f"{ref!r} changed during commit")
2879 return c.id
2881 @classmethod
2882 def init_bare(
2883 cls,
2884 objects: Iterable[ShaFile],
2885 refs: Mapping[Ref, ObjectID],
2886 format: int | None = None,
2887 object_format: str | None = None,
2888 ) -> "MemoryRepo":
2889 """Create a new bare repository in memory.
2891 Args:
2892 objects: Objects for the new repository,
2893 as iterable
2894 refs: Refs as dictionary, mapping names
2895 to object SHA1s
2896 format: Repository format version (defaults to 0)
2897 object_format: Object format to use ("sha1" or "sha256", defaults to "sha1")
2898 """
2899 ret = cls()
2900 for obj in objects:
2901 ret.object_store.add_object(obj)
2902 for refname, sha in refs.items():
2903 ret.refs.add_if_new(refname, sha)
2904 ret._init_files(bare=True, format=format, object_format=object_format)
2905 return ret