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() -> tuple[str, str]:
202 import socket
204 for name in ("LOGNAME", "USER", "LNAME", "USERNAME"):
205 username = os.environ.get(name)
206 if username:
207 break
208 else:
209 username = None
211 try:
212 import pwd
213 except ImportError:
214 fullname = None
215 else:
216 try:
217 entry = pwd.getpwuid(os.getuid()) # type: ignore[attr-defined,unused-ignore]
218 except KeyError:
219 fullname = None
220 else:
221 if getattr(entry, "gecos", None):
222 fullname = entry.pw_gecos.split(",")[0]
223 else:
224 fullname = None
225 if username is None:
226 username = entry.pw_name
227 if not fullname:
228 if username is None:
229 raise DefaultIdentityNotFound("no username found")
230 fullname = username
231 email = os.environ.get("EMAIL")
232 if email is None:
233 if username is None:
234 raise DefaultIdentityNotFound("no username found")
235 email = f"{username}@{socket.gethostname()}"
236 return (fullname, email)
239def get_user_identity(config: "Config", kind: str | None = None) -> bytes:
240 """Determine the identity to use for new commits.
242 If kind is set, this first checks
243 GIT_${KIND}_NAME and GIT_${KIND}_EMAIL.
245 If those variables are not set, then it will fall back
246 to reading the user.name and user.email settings from
247 the specified configuration.
249 If that also fails, then it will fall back to using
250 the current users' identity as obtained from the host
251 system (e.g. the gecos field, $EMAIL, $USER@$(hostname -f).
253 Args:
254 config: Configuration stack to read from
255 kind: Optional kind to return identity for,
256 usually either "AUTHOR" or "COMMITTER".
258 Returns:
259 A user identity
260 """
261 user: bytes | None = None
262 email: bytes | None = None
263 if kind:
264 user_uc = os.environ.get("GIT_" + kind + "_NAME")
265 if user_uc is not None:
266 user = user_uc.encode("utf-8")
267 email_uc = os.environ.get("GIT_" + kind + "_EMAIL")
268 if email_uc is not None:
269 email = email_uc.encode("utf-8")
270 if user is None:
271 try:
272 user = config.get(("user",), "name")
273 except KeyError:
274 user = None
275 if email is None:
276 try:
277 email = config.get(("user",), "email")
278 except KeyError:
279 email = None
280 default_user, default_email = _get_default_identity()
281 if user is None:
282 user = default_user.encode("utf-8")
283 if email is None:
284 email = default_email.encode("utf-8")
285 if email.startswith(b"<") and email.endswith(b">"):
286 email = email[1:-1]
287 return user + b" <" + email + b">"
290def check_user_identity(identity: bytes) -> None:
291 """Verify that a user identity is formatted correctly.
293 Args:
294 identity: User identity bytestring
295 Raises:
296 InvalidUserIdentity: Raised when identity is invalid
297 """
298 try:
299 _fst, snd = identity.split(b" <", 1)
300 except ValueError as exc:
301 raise InvalidUserIdentity(identity.decode("utf-8", "replace")) from exc
302 if b">" not in snd:
303 raise InvalidUserIdentity(identity.decode("utf-8", "replace"))
304 if b"\0" in identity or b"\n" in identity:
305 raise InvalidUserIdentity(identity.decode("utf-8", "replace"))
308def parse_graftpoints(
309 graftpoints: Iterable[bytes],
310) -> dict[ObjectID, list[ObjectID]]:
311 """Convert a list of graftpoints into a dict.
313 Args:
314 graftpoints: Iterator of graftpoint lines
316 Each line is formatted as:
317 <commit sha1> <parent sha1> [<parent sha1>]*
319 Resulting dictionary is:
320 <commit sha1>: [<parent sha1>*]
322 https://git.wiki.kernel.org/index.php/GraftPoint
323 """
324 grafts: dict[ObjectID, list[ObjectID]] = {}
325 for line in graftpoints:
326 raw_graft = line.split(None, 1)
328 commit = ObjectID(raw_graft[0])
329 if len(raw_graft) == 2:
330 parents = [ObjectID(p) for p in raw_graft[1].split()]
331 else:
332 parents = []
334 for sha in [commit, *parents]:
335 check_hexsha(sha, "Invalid graftpoint")
337 grafts[commit] = parents
338 return grafts
341def serialize_graftpoints(graftpoints: Mapping[ObjectID, Sequence[ObjectID]]) -> bytes:
342 """Convert a dictionary of grafts into string.
344 The graft dictionary is:
345 <commit sha1>: [<parent sha1>*]
347 Each line is formatted as:
348 <commit sha1> <parent sha1> [<parent sha1>]*
350 https://git.wiki.kernel.org/index.php/GraftPoint
352 """
353 graft_lines = []
354 for commit, parents in graftpoints.items():
355 if parents:
356 graft_lines.append(commit + b" " + b" ".join(parents))
357 else:
358 graft_lines.append(commit)
359 return b"\n".join(graft_lines)
362def _set_filesystem_hidden(path: str) -> None:
363 """Mark path as to be hidden if supported by platform and filesystem.
365 On win32 uses SetFileAttributesW api:
366 <https://docs.microsoft.com/windows/desktop/api/fileapi/nf-fileapi-setfileattributesw>
367 """
368 if sys.platform == "win32":
369 import ctypes
370 from ctypes.wintypes import BOOL, DWORD, LPCWSTR
372 FILE_ATTRIBUTE_HIDDEN = 2
373 SetFileAttributesW = ctypes.WINFUNCTYPE(BOOL, LPCWSTR, DWORD)(
374 ("SetFileAttributesW", ctypes.windll.kernel32)
375 )
377 if isinstance(path, bytes):
378 path = os.fsdecode(path)
379 if not SetFileAttributesW(path, FILE_ATTRIBUTE_HIDDEN):
380 pass # Could raise or log `ctypes.WinError()` here
382 # Could implement other platform specific filesystem hiding here
385def parse_shared_repository(
386 value: str | bytes | bool,
387) -> tuple[int | None, int | None]:
388 """Parse core.sharedRepository configuration value.
390 Args:
391 value: Configuration value (string, bytes, or boolean)
393 Returns:
394 tuple of (file_mask, directory_mask) or (None, None) if not shared
396 The masks are permission bits to apply via chmod.
397 """
398 if isinstance(value, bytes):
399 value = value.decode("utf-8", errors="replace")
401 # Handle boolean values
402 if isinstance(value, bool):
403 if value:
404 # true = group (same as "group")
405 return (0o664, 0o2775)
406 else:
407 # false = umask (use system umask, no adjustment)
408 return (None, None)
410 # Handle string values
411 value_lower = value.lower()
413 if value_lower in ("false", "0", ""):
414 # Use umask (no adjustment)
415 return (None, None)
417 if value_lower in ("true", "1", "group"):
418 # Group writable (with setgid bit)
419 return (0o664, 0o2775)
421 if value_lower in ("all", "world", "everybody", "2"):
422 # World readable/writable (with setgid bit)
423 return (0o666, 0o2777)
425 if value_lower == "umask":
426 # Explicitly use umask
427 return (None, None)
429 # Try to parse as octal
430 if value.startswith("0"):
431 try:
432 mode = int(value, 8)
433 # For directories, add execute bits where read bits are set
434 # and add setgid bit for shared repositories
435 dir_mode = mode | 0o2000 # Add setgid bit
436 if mode & 0o004:
437 dir_mode |= 0o001
438 if mode & 0o040:
439 dir_mode |= 0o010
440 if mode & 0o400:
441 dir_mode |= 0o100
442 return (mode, dir_mode)
443 except ValueError:
444 pass
446 # Default to umask for unrecognized values
447 return (None, None)
450def _enable_relative_worktrees_extension(repo: "Repo") -> None:
451 """Enable the relativeworktrees extension in repository config.
453 This sets core.repositoryformatversion to 1 (if not already) and
454 enables the extensions.relativeworktrees extension.
456 Args:
457 repo: The repository to configure
458 """
459 config = repo.get_config()
461 # Ensure repository format version is at least 1
462 try:
463 version = int(config.get(("core",), "repositoryformatversion"))
464 except KeyError:
465 version = 0
467 if version < 1:
468 config.set(("core",), "repositoryformatversion", "1")
470 # Enable the relativeworktrees extension
471 config.set(("extensions",), "relativeworktrees", True)
472 config.write_to_path()
475class ParentsProvider:
476 """Provider for commit parent information."""
478 def __init__(
479 self,
480 store: "BaseObjectStore",
481 grafts: dict[ObjectID, list[ObjectID]] = {},
482 shallows: Iterable[ObjectID] = [],
483 ) -> None:
484 """Initialize ParentsProvider.
486 Args:
487 store: Object store to use
488 grafts: Graft information
489 shallows: Shallow commit SHAs
490 """
491 self.store = store
492 self.grafts = grafts
493 self.shallows = set(shallows)
495 # Get commit graph once at initialization for performance
496 self.commit_graph = store.get_commit_graph()
498 def get_parents(
499 self, commit_id: ObjectID, commit: Commit | None = None
500 ) -> list[ObjectID]:
501 """Get parents for a commit using the parents provider."""
502 try:
503 return self.grafts[commit_id]
504 except KeyError:
505 pass
506 if commit_id in self.shallows:
507 return []
509 # Try to use commit graph for faster parent lookup
510 if self.commit_graph:
511 parents = self.commit_graph.get_parents(commit_id)
512 if parents is not None:
513 return parents
515 # Fallback to reading the commit object
516 if commit is None:
517 obj = self.store[commit_id]
518 if not isinstance(obj, Commit):
519 raise ValueError(
520 f"Expected Commit object for commit_id {commit_id.decode()}, "
521 f"got {type(obj).__name__}. This usually means a reference "
522 f"points to a {type(obj).__name__} object instead of a Commit."
523 )
524 commit = obj
525 result: list[ObjectID] = commit.parents
526 return result
529class BaseRepo:
530 """Base class for a git repository.
532 This base class is meant to be used for Repository implementations that e.g.
533 work on top of a different transport than a standard filesystem path.
535 Attributes:
536 object_store: Dictionary-like object for accessing
537 the objects
538 refs: Dictionary-like object with the refs in this
539 repository
540 """
542 def __init__(
543 self,
544 object_store: "PackCapableObjectStore",
545 refs: RefsContainer,
546 object_format: "ObjectFormat | None" = None,
547 ) -> None:
548 """Open a repository.
550 This shouldn't be called directly, but rather through one of the
551 base classes, such as MemoryRepo or Repo.
553 Args:
554 object_store: Object store to use
555 refs: Refs container to use
556 object_format: Hash algorithm to use (if None, will use object_store's format)
557 """
558 self.object_store = object_store
559 self.refs = refs
561 self._graftpoints: dict[ObjectID, list[ObjectID]] = {}
562 self.hooks: dict[str, Hook] = {}
563 if object_format is None:
564 self.object_format: ObjectFormat = object_store.object_format
565 else:
566 self.object_format = object_format
568 def _determine_file_mode(self) -> bool:
569 """Probe the file-system to determine whether permissions can be trusted.
571 Returns: True if permissions can be trusted, False otherwise.
572 """
573 raise NotImplementedError(self._determine_file_mode)
575 def _determine_symlinks(self) -> bool:
576 """Probe the filesystem to determine whether symlinks can be created.
578 Returns: True if symlinks can be created, False otherwise.
579 """
580 # For now, just mimic the old behaviour
581 return sys.platform != "win32"
583 def _init_files(
584 self,
585 bare: bool,
586 symlinks: bool | None = None,
587 format: int | None = None,
588 shared_repository: str | bool | None = None,
589 object_format: str | None = None,
590 ) -> None:
591 """Initialize a default set of named files."""
592 from .config import ConfigFile
594 self._put_named_file("description", b"Unnamed repository")
595 f = BytesIO()
596 cf = ConfigFile()
598 # Determine the appropriate format version
599 if object_format == "sha256":
600 # SHA256 requires format version 1
601 if format is None:
602 format = 1
603 elif format != 1:
604 raise ValueError(
605 "SHA256 object format requires repository format version 1"
606 )
607 else:
608 # SHA1 (default) can use format 0 or 1
609 if format is None:
610 format = 0
612 if format not in (0, 1):
613 raise ValueError(f"Unsupported repository format version: {format}")
615 cf.set("core", "repositoryformatversion", str(format))
617 # Set object format extension if using SHA256
618 if object_format == "sha256":
619 cf.set("extensions", "objectformat", "sha256")
621 # Set hash algorithm based on object format
622 from .object_format import get_object_format
624 self.object_format = get_object_format(object_format)
626 if self._determine_file_mode():
627 cf.set("core", "filemode", True)
628 else:
629 cf.set("core", "filemode", False)
631 if symlinks is None and not bare:
632 symlinks = self._determine_symlinks()
634 if symlinks is False:
635 cf.set("core", "symlinks", symlinks)
637 # On macOS, set precomposeunicode to true since HFS+/APFS
638 # returns filenames in NFD (decomposed) Unicode form
639 if sys.platform == "darwin":
640 cf.set("core", "precomposeunicode", True)
642 cf.set("core", "bare", bare)
643 cf.set("core", "logallrefupdates", True)
645 # Set shared repository if specified
646 if shared_repository is not None:
647 if isinstance(shared_repository, bool):
648 cf.set("core", "sharedRepository", shared_repository)
649 else:
650 cf.set("core", "sharedRepository", shared_repository)
652 cf.write_to_file(f)
653 self._put_named_file("config", f.getvalue())
654 self._put_named_file(os.path.join("info", "exclude"), b"")
656 # Allow subclasses to handle config initialization
657 self._init_config(cf)
659 def _init_config(self, config: "ConfigFile") -> None:
660 """Initialize repository configuration.
662 This method can be overridden by subclasses to handle config initialization.
664 Args:
665 config: The ConfigFile object that was just created
666 """
667 # Default implementation does nothing
669 def get_named_file(self, path: str) -> BinaryIO | None:
670 """Get a file from the control dir with a specific name.
672 Although the filename should be interpreted as a filename relative to
673 the control dir in a disk-based Repo, the object returned need not be
674 pointing to a file in that location.
676 Args:
677 path: The path to the file, relative to the control dir.
678 Returns: An open file object, or None if the file does not exist.
679 """
680 raise NotImplementedError(self.get_named_file)
682 def _put_named_file(self, path: str, contents: bytes) -> None:
683 """Write a file to the control dir with the given name and contents.
685 Args:
686 path: The path to the file, relative to the control dir.
687 contents: A string to write to the file.
688 """
689 raise NotImplementedError(self._put_named_file)
691 def _del_named_file(self, path: str) -> None:
692 """Delete a file in the control directory with the given name."""
693 raise NotImplementedError(self._del_named_file)
695 def open_index(self, config: "Config | None" = None) -> "Index":
696 """Open the index for this repository.
698 Args:
699 config: Configuration to consult for index settings. If None,
700 implementations may fall back to ``self.get_config_stack()``.
702 Raises:
703 NoIndexPresent: If no index is present
704 Returns: The matching `Index`
705 """
706 raise NotImplementedError(self.open_index)
708 def _change_object_format(self, object_format_name: str) -> None:
709 """Change the object format of this repository.
711 This can only be done if the object store is empty (no objects written yet).
713 Args:
714 object_format_name: Name of the new object format (e.g., "sha1", "sha256")
716 Raises:
717 AssertionError: If the object store is not empty
718 """
719 # Check if object store has any objects
720 for _ in self.object_store:
721 raise AssertionError(
722 "Cannot change object format: repository already contains objects"
723 )
725 # Update the object format
726 from .object_format import get_object_format
728 new_format = get_object_format(object_format_name)
729 self.object_format = new_format
730 self.object_store.object_format = new_format
732 # Update config file
733 config = self.get_config()
735 if object_format_name == "sha1":
736 # For SHA-1, explicitly remove objectformat extension if present
737 try:
738 config.remove("extensions", "objectformat")
739 except KeyError:
740 pass
741 else:
742 # For non-SHA-1 formats, set repositoryformatversion to 1 and objectformat extension
743 config.set("core", "repositoryformatversion", "1")
744 config.set("extensions", "objectformat", object_format_name)
746 config.write_to_path()
748 def fetch(
749 self,
750 target: "BaseRepo",
751 determine_wants: Callable[[Mapping[Ref, ObjectID], int | None], list[ObjectID]]
752 | None = None,
753 progress: Callable[..., None] | None = None,
754 depth: int | None = None,
755 ) -> dict[Ref, ObjectID]:
756 """Fetch objects into another repository.
758 Args:
759 target: The target repository
760 determine_wants: Optional function to determine what refs to
761 fetch.
762 progress: Optional progress function
763 depth: Optional shallow fetch depth
764 Returns: The local refs
765 """
766 # Fix object format if needed
767 if self.object_format != target.object_format:
768 # Change the target repo's format if it's empty
769 target._change_object_format(self.object_format.name)
771 if determine_wants is None:
772 determine_wants = target.object_store.determine_wants_all
773 count, pack_data = self.fetch_pack_data(
774 determine_wants,
775 target.get_graph_walker(),
776 progress=progress,
777 depth=depth,
778 )
779 target.object_store.add_pack_data(count, pack_data, progress)
780 return self.get_refs()
782 def fetch_pack_data(
783 self,
784 determine_wants: Callable[[Mapping[Ref, ObjectID], int | None], list[ObjectID]],
785 graph_walker: "GraphWalker",
786 progress: Callable[[bytes], None] | None,
787 *,
788 get_tagged: Callable[[], dict[ObjectID, ObjectID]] | None = None,
789 depth: int | None = None,
790 ) -> tuple[int, Iterator["UnpackedObject"]]:
791 """Fetch the pack data required for a set of revisions.
793 Args:
794 determine_wants: Function that takes a dictionary with heads
795 and returns the list of heads to fetch.
796 graph_walker: Object that can iterate over the list of revisions
797 to fetch and has an "ack" method that will be called to acknowledge
798 that a revision is present.
799 progress: Simple progress function that will be called with
800 updated progress strings.
801 get_tagged: Function that returns a dict of pointed-to sha ->
802 tag sha for including tags.
803 depth: Shallow fetch depth
804 Returns: count and iterator over pack data
805 """
806 missing_objects = self.find_missing_objects(
807 determine_wants, graph_walker, progress, get_tagged=get_tagged, depth=depth
808 )
809 if missing_objects is None:
810 return 0, iter([])
811 remote_has = missing_objects.get_remote_has()
812 object_ids = list(missing_objects)
813 return len(object_ids), generate_unpacked_objects(
814 self.object_store, object_ids, progress=progress, other_haves=remote_has
815 )
817 def find_missing_objects(
818 self,
819 determine_wants: Callable[[Mapping[Ref, ObjectID], int | None], list[ObjectID]],
820 graph_walker: "GraphWalker",
821 progress: Callable[[bytes], None] | None,
822 *,
823 get_tagged: Callable[[], dict[ObjectID, ObjectID]] | None = None,
824 depth: int | None = None,
825 ) -> MissingObjectFinder | None:
826 """Fetch the missing objects required for a set of revisions.
828 Args:
829 determine_wants: Function that takes a dictionary with heads
830 and returns the list of heads to fetch.
831 graph_walker: Object that can iterate over the list of revisions
832 to fetch and has an "ack" method that will be called to acknowledge
833 that a revision is present.
834 progress: Simple progress function that will be called with
835 updated progress strings.
836 get_tagged: Function that returns a dict of pointed-to sha ->
837 tag sha for including tags.
838 depth: Shallow fetch depth
839 Returns: iterator over objects, with __len__ implemented
840 """
841 # Filter out refs pointing to missing objects to avoid errors downstream.
842 # This makes Dulwich more robust when dealing with broken refs on disk.
843 # Previously serialize_refs() did this filtering as a side-effect.
844 all_refs = self.get_refs()
845 refs: dict[Ref, ObjectID] = {}
846 for ref, sha in all_refs.items():
847 if sha in self.object_store:
848 refs[ref] = sha
849 else:
850 logger.warning(
851 "ref %s points at non-present sha %s",
852 ref.decode("utf-8", "replace"),
853 sha.decode("ascii"),
854 )
856 wants = determine_wants(refs, depth)
857 if not isinstance(wants, list):
858 raise TypeError("determine_wants() did not return a list")
860 current_shallow = set(getattr(graph_walker, "shallow", set()))
862 unshallow: set[ObjectID] = set()
863 if depth not in (None, 0):
864 assert depth is not None
865 shallow, not_shallow = find_shallow(self.object_store, wants, depth)
866 # Only update if graph_walker has shallow attribute
867 walker_shallow: set[ObjectID] | None = getattr(
868 graph_walker, "shallow", None
869 )
870 if walker_shallow is not None:
871 unshallow = not_shallow & current_shallow
872 # Commits that were a shallow boundary but are now being
873 # deepened past must drop out of the boundary, otherwise their
874 # parents stay unreachable and never get transferred.
875 walker_shallow.update(shallow - not_shallow)
876 walker_shallow.difference_update(unshallow)
877 new_shallow = walker_shallow - current_shallow
878 setattr(graph_walker, "unshallow", unshallow)
879 update_shallow = getattr(graph_walker, "update_shallow", None)
880 if update_shallow is not None:
881 update_shallow(new_shallow, unshallow)
882 else:
883 unshallow = getattr(graph_walker, "unshallow", set())
885 if wants == []:
886 # TODO(dborowitz): find a way to short-circuit that doesn't change
887 # this interface.
889 if getattr(graph_walker, "shallow", set()) or unshallow:
890 # Do not send a pack in shallow short-circuit path
891 return None
893 # Return an actual MissingObjectFinder with empty wants
894 return MissingObjectFinder(
895 self.object_store,
896 haves=[],
897 wants=[],
898 )
900 # If the graph walker is set up with an implementation that can
901 # ACK/NAK to the wire, it will write data to the client through
902 # this call as a side-effect.
903 haves = self.object_store.find_common_revisions(graph_walker)
905 # Deal with shallow requests separately because the haves do
906 # not reflect what objects are missing
907 if getattr(graph_walker, "shallow", set()) or unshallow:
908 # TODO: filter the haves commits from iter_shas. the specific
909 # commits aren't missing.
910 haves = []
912 parents_provider = ParentsProvider(
913 self.object_store,
914 shallows=getattr(graph_walker, "shallow", current_shallow),
915 )
917 def get_parents(commit: Commit) -> list[ObjectID]:
918 """Get parents for a commit using the parents provider.
920 Args:
921 commit: Commit object
923 Returns:
924 List of parent commit SHAs
925 """
926 return parents_provider.get_parents(commit.id, commit)
928 return MissingObjectFinder(
929 self.object_store,
930 haves=haves,
931 wants=wants,
932 shallow=getattr(graph_walker, "shallow", set()),
933 progress=progress,
934 get_tagged=get_tagged,
935 get_parents=get_parents,
936 )
938 def generate_pack_data(
939 self,
940 have: set[ObjectID],
941 want: set[ObjectID],
942 *,
943 shallow: set[ObjectID] | None = None,
944 progress: Callable[[str], None] | None = None,
945 ofs_delta: bool | None = None,
946 ) -> tuple[int, Iterator["UnpackedObject"]]:
947 """Generate pack data objects for a set of wants/haves.
949 Args:
950 have: List of SHA1s of objects that should not be sent
951 want: List of SHA1s of objects that should be sent
952 shallow: Set of shallow commit SHA1s to skip (defaults to repo's shallow commits)
953 ofs_delta: Whether OFS deltas can be included
954 progress: Optional progress reporting method
955 """
956 if shallow is None:
957 shallow = self.get_shallow()
958 return self.object_store.generate_pack_data(
959 have,
960 want,
961 shallow=shallow,
962 progress=progress,
963 ofs_delta=ofs_delta if ofs_delta is not None else DEFAULT_OFS_DELTA,
964 )
966 def get_graph_walker(
967 self, heads: list[ObjectID] | None = None
968 ) -> ObjectStoreGraphWalker:
969 """Retrieve a graph walker.
971 A graph walker is used by a remote repository (or proxy)
972 to find out which objects are present in this repository.
974 Args:
975 heads: Repository heads to use (optional)
976 Returns: A graph walker object
977 """
978 if heads is None:
979 heads = [
980 sha
981 for sha in self.refs.as_dict(Ref(b"refs/heads")).values()
982 if sha in self.object_store
983 ]
984 parents_provider = ParentsProvider(self.object_store)
985 return ObjectStoreGraphWalker(
986 heads,
987 parents_provider.get_parents,
988 shallow=self.get_shallow(),
989 update_shallow=self.update_shallow,
990 )
992 def get_refs(self) -> dict[Ref, ObjectID]:
993 """Get dictionary with all refs.
995 Returns: A ``dict`` mapping ref names to SHA1s
996 """
997 return self.refs.as_dict()
999 def head(self) -> ObjectID:
1000 """Return the SHA1 pointed at by HEAD."""
1001 # TODO: move this method to WorkTree
1002 return self.refs[HEADREF]
1004 def _get_object(self, sha: ObjectID | RawObjectID, cls: type[T]) -> T:
1005 assert len(sha) in (
1006 self.object_format.oid_length,
1007 self.object_format.hex_length,
1008 )
1009 ret = self.get_object(sha)
1010 if not isinstance(ret, cls):
1011 if cls is Commit:
1012 raise NotCommitError(ret.id)
1013 elif cls is Blob:
1014 raise NotBlobError(ret.id)
1015 elif cls is Tree:
1016 raise NotTreeError(ret.id)
1017 elif cls is Tag:
1018 raise NotTagError(ret.id)
1019 else:
1020 raise Exception(f"Type invalid: {ret.type_name!r} != {cls.type_name!r}")
1021 return ret
1023 def get_object(self, sha: ObjectID | RawObjectID) -> ShaFile:
1024 """Retrieve the object with the specified SHA.
1026 Args:
1027 sha: SHA to retrieve
1028 Returns: A ShaFile object
1029 Raises:
1030 KeyError: when the object can not be found
1031 """
1032 return self.object_store[sha]
1034 def parents_provider(self) -> ParentsProvider:
1035 """Get a parents provider for this repository.
1037 Returns:
1038 ParentsProvider instance configured with grafts and shallows
1039 """
1040 return ParentsProvider(
1041 self.object_store,
1042 grafts=self._graftpoints,
1043 shallows=self.get_shallow(),
1044 )
1046 def get_parents(
1047 self, sha: ObjectID, commit: Commit | None = None
1048 ) -> list[ObjectID]:
1049 """Retrieve the parents of a specific commit.
1051 If the specific commit is a graftpoint, the graft parents
1052 will be returned instead.
1054 Args:
1055 sha: SHA of the commit for which to retrieve the parents
1056 commit: Optional commit matching the sha
1057 Returns: List of parents
1058 """
1059 return self.parents_provider().get_parents(sha, commit)
1061 def get_config(self) -> "ConfigFile":
1062 """Retrieve the config object.
1064 Returns: `ConfigFile` object for the ``.git/config`` file.
1065 """
1066 raise NotImplementedError(self.get_config)
1068 def get_worktree_config(self) -> "ConfigFile":
1069 """Retrieve the worktree config object."""
1070 raise NotImplementedError(self.get_worktree_config)
1072 def get_description(self) -> bytes | None:
1073 """Retrieve the description for this repository.
1075 Returns: Bytes with the description of the repository
1076 as set by the user.
1077 """
1078 raise NotImplementedError(self.get_description)
1080 def set_description(self, description: bytes) -> None:
1081 """Set the description for this repository.
1083 Args:
1084 description: Text to set as description for this repository.
1085 """
1086 raise NotImplementedError(self.set_description)
1088 def get_rebase_state_manager(self) -> "RebaseStateManager":
1089 """Get the appropriate rebase state manager for this repository.
1091 Returns: RebaseStateManager instance
1092 """
1093 raise NotImplementedError(self.get_rebase_state_manager)
1095 def get_blob_normalizer(
1096 self, config: "Config | None" = None
1097 ) -> "FilterBlobNormalizer":
1098 """Return a BlobNormalizer object for checkin/checkout operations.
1100 Args:
1101 config: Configuration to consult for filter setup. If None,
1102 implementations may fall back to ``self.get_config_stack()``.
1104 Returns: BlobNormalizer instance
1105 """
1106 raise NotImplementedError(self.get_blob_normalizer)
1108 def get_gitattributes(self, tree: bytes | None = None) -> "GitAttributes":
1109 """Read gitattributes for the repository.
1111 Args:
1112 tree: Tree SHA to read .gitattributes from (defaults to HEAD)
1114 Returns:
1115 GitAttributes object that can be used to match paths
1116 """
1117 raise NotImplementedError(self.get_gitattributes)
1119 def get_config_stack(self) -> "StackedConfig":
1120 """Return a config stack for this repository.
1122 This stack accesses the configuration for both this repository
1123 itself (.git/config) and the global configuration, which usually
1124 lives in ~/.gitconfig.
1126 Returns: `Config` instance for this repository
1127 """
1128 from .config import ConfigFile, StackedConfig
1130 local_config = self.get_config()
1131 backends: list[ConfigFile] = [local_config]
1132 if local_config.get_boolean((b"extensions",), b"worktreeconfig", False):
1133 backends.append(self.get_worktree_config())
1135 backends += StackedConfig.default_backends()
1136 return StackedConfig(backends, writable=local_config)
1138 def get_shallow(self) -> set[ObjectID]:
1139 """Get the set of shallow commits.
1141 Returns: Set of shallow commits.
1142 """
1143 f = self.get_named_file("shallow")
1144 if f is None:
1145 return set()
1146 with f:
1147 return {ObjectID(line.strip()) for line in f}
1149 def update_shallow(
1150 self, new_shallow: set[ObjectID] | None, new_unshallow: set[ObjectID] | None
1151 ) -> None:
1152 """Update the list of shallow objects.
1154 Args:
1155 new_shallow: Newly shallow objects
1156 new_unshallow: Newly no longer shallow objects
1157 """
1158 shallow = self.get_shallow()
1159 if new_shallow:
1160 shallow.update(new_shallow)
1161 if new_unshallow:
1162 shallow.difference_update(new_unshallow)
1163 if shallow:
1164 self._put_named_file("shallow", b"".join([sha + b"\n" for sha in shallow]))
1165 else:
1166 self._del_named_file("shallow")
1168 def get_peeled(self, ref: Ref) -> ObjectID:
1169 """Get the peeled value of a ref.
1171 Args:
1172 ref: The refname to peel.
1173 Returns: The fully-peeled SHA1 of a tag object, after peeling all
1174 intermediate tags; if the original ref does not point to a tag,
1175 this will equal the original SHA1.
1176 """
1177 cached = self.refs.get_peeled(ref)
1178 if cached is not None:
1179 return cached
1180 return peel_sha(self.object_store, self.refs[ref])[1].id
1182 @property
1183 def notes(self) -> "Notes":
1184 """Access notes functionality for this repository.
1186 Returns:
1187 Notes object for accessing notes
1188 """
1189 from .notes import Notes
1191 return Notes(self.object_store, self.refs)
1193 def get_walker(
1194 self,
1195 include: Sequence[ObjectID] | None = None,
1196 exclude: Sequence[ObjectID] | None = None,
1197 order: str = "date",
1198 reverse: bool = False,
1199 max_entries: int | None = None,
1200 paths: Sequence[bytes] | None = None,
1201 rename_detector: "RenameDetector | None" = None,
1202 follow: bool = False,
1203 since: int | None = None,
1204 until: int | None = None,
1205 queue_cls: type | None = None,
1206 ) -> "Walker":
1207 """Obtain a walker for this repository.
1209 Args:
1210 include: Iterable of SHAs of commits to include along with their
1211 ancestors. Defaults to [HEAD]
1212 exclude: Iterable of SHAs of commits to exclude along with their
1213 ancestors, overriding includes.
1214 order: ORDER_* constant specifying the order of results.
1215 Anything other than ORDER_DATE may result in O(n) memory usage.
1216 reverse: If True, reverse the order of output, requiring O(n)
1217 memory.
1218 max_entries: The maximum number of entries to yield, or None for
1219 no limit.
1220 paths: Iterable of file or subtree paths to show entries for.
1221 rename_detector: diff.RenameDetector object for detecting
1222 renames.
1223 follow: If True, follow path across renames/copies. Forces a
1224 default rename_detector.
1225 since: Timestamp to list commits after.
1226 until: Timestamp to list commits before.
1227 queue_cls: A class to use for a queue of commits, supporting the
1228 iterator protocol. The constructor takes a single argument, the Walker.
1230 Returns: A `Walker` object
1231 """
1232 from .walk import Walker, _CommitTimeQueue
1234 if include is None:
1235 include = [self.head()]
1237 # Pass all arguments to Walker explicitly to avoid type issues with **kwargs
1238 return Walker(
1239 self.object_store,
1240 include,
1241 exclude=exclude,
1242 order=order,
1243 reverse=reverse,
1244 max_entries=max_entries,
1245 paths=paths,
1246 rename_detector=rename_detector,
1247 follow=follow,
1248 since=since,
1249 until=until,
1250 get_parents=lambda commit: self.get_parents(commit.id, commit),
1251 queue_cls=queue_cls if queue_cls is not None else _CommitTimeQueue,
1252 )
1254 def __getitem__(self, name: ObjectID | Ref | bytes) -> "ShaFile":
1255 """Retrieve a Git object by SHA1 or ref.
1257 Args:
1258 name: A Git object SHA1 or a ref name
1259 Returns: A `ShaFile` object, such as a Commit or Blob
1260 Raises:
1261 KeyError: when the specified ref or object does not exist
1262 """
1263 if not isinstance(name, bytes):
1264 raise TypeError(f"'name' must be bytestring, not {type(name).__name__:.80}")
1265 # If it looks like a ref name, only try refs
1266 if name == b"HEAD" or name.startswith(b"refs/"):
1267 try:
1268 return self.object_store[self.refs[Ref(name)]]
1269 except (RefFormatError, KeyError):
1270 pass
1271 # Otherwise, try as object ID if length matches
1272 if len(name) in (
1273 self.object_store.object_format.oid_length,
1274 self.object_store.object_format.hex_length,
1275 ):
1276 try:
1277 return self.object_store[
1278 ObjectID(name)
1279 if len(name) == self.object_store.object_format.hex_length
1280 else RawObjectID(name)
1281 ]
1282 except (KeyError, ValueError):
1283 pass
1284 # If nothing worked, raise KeyError
1285 raise KeyError(name)
1287 def __contains__(self, name: bytes) -> bool:
1288 """Check if a specific Git object or ref is present.
1290 Args:
1291 name: Git object SHA1/SHA256 or ref name
1292 """
1293 # Check if it's a binary or hex SHA
1294 if len(name) == self.object_format.hex_length and valid_hexsha(name):
1295 return ObjectID(name) in self.object_store or Ref(name) in self.refs
1296 return Ref(name) in self.refs
1298 def __setitem__(self, name: bytes, value: ShaFile | bytes) -> None:
1299 """Set a ref.
1301 Args:
1302 name: ref name
1303 value: Ref value - either a ShaFile object, or a hex sha
1304 """
1305 if name.startswith(b"refs/") or name == HEADREF:
1306 ref_name = Ref(name)
1307 if isinstance(value, ShaFile):
1308 self.refs[ref_name] = value.id
1309 elif isinstance(value, bytes):
1310 self.refs[ref_name] = ObjectID(value)
1311 else:
1312 raise TypeError(value)
1313 else:
1314 raise ValueError(name)
1316 def __delitem__(self, name: bytes) -> None:
1317 """Remove a ref.
1319 Args:
1320 name: Name of the ref to remove
1321 """
1322 if name.startswith(b"refs/") or name == HEADREF:
1323 del self.refs[Ref(name)]
1324 else:
1325 raise ValueError(name)
1327 def _get_user_identity(
1328 self, config: "StackedConfig", kind: str | None = None
1329 ) -> bytes:
1330 """Determine the identity to use for new commits."""
1331 warnings.warn(
1332 "use get_user_identity() rather than Repo._get_user_identity",
1333 DeprecationWarning,
1334 )
1335 return get_user_identity(config)
1337 def _add_graftpoints(
1338 self, updated_graftpoints: dict[ObjectID, list[ObjectID]]
1339 ) -> None:
1340 """Add or modify graftpoints.
1342 Args:
1343 updated_graftpoints: Dict of commit shas to list of parent shas
1344 """
1345 # Simple validation
1346 for commit, parents in updated_graftpoints.items():
1347 for sha in [commit, *parents]:
1348 check_hexsha(sha, "Invalid graftpoint")
1350 self._graftpoints.update(updated_graftpoints)
1352 def _remove_graftpoints(self, to_remove: Sequence[ObjectID] = ()) -> None:
1353 """Remove graftpoints.
1355 Args:
1356 to_remove: List of commit shas
1357 """
1358 for sha in to_remove:
1359 del self._graftpoints[sha]
1361 def _read_heads(self, name: str) -> list[ObjectID]:
1362 f = self.get_named_file(name)
1363 if f is None:
1364 return []
1365 with f:
1366 return [ObjectID(line.strip()) for line in f.readlines() if line.strip()]
1368 def get_worktree(self) -> "WorkTree":
1369 """Get the working tree for this repository.
1371 Returns:
1372 WorkTree instance for performing working tree operations
1374 Raises:
1375 NotImplementedError: If the repository doesn't support working trees
1376 """
1377 raise NotImplementedError(
1378 "Working tree operations not supported by this repository type"
1379 )
1382def read_gitfile(f: BinaryIO) -> str:
1383 """Read a ``.git`` file.
1385 The first line of the file should start with "gitdir: "
1387 Args:
1388 f: File-like object to read from
1389 Returns: A path
1390 """
1391 cs = f.read()
1392 if not cs.startswith(b"gitdir: "):
1393 raise ValueError("Expected file to start with 'gitdir: '")
1394 return cs[len(b"gitdir: ") :].rstrip(b"\r\n").decode("utf-8")
1397class UnsupportedVersion(Exception):
1398 """Unsupported repository version."""
1400 def __init__(self, version: int) -> None:
1401 """Initialize UnsupportedVersion exception.
1403 Args:
1404 version: The unsupported repository version
1405 """
1406 self.version = version
1409class UnsupportedExtension(Exception):
1410 """Unsupported repository extension."""
1412 def __init__(self, extension: str) -> None:
1413 """Initialize UnsupportedExtension exception.
1415 Args:
1416 extension: The unsupported repository extension
1417 """
1418 self.extension = extension
1421class Repo(BaseRepo):
1422 """A git repository backed by local disk.
1424 To open an existing repository, call the constructor with
1425 the path of the repository.
1427 To create a new repository, use the Repo.init class method.
1429 Note that a repository object may hold on to resources such
1430 as file handles for performance reasons; call .close() to free
1431 up those resources.
1433 Attributes:
1434 path: Path to the working copy (if it exists) or repository control
1435 directory (if the repository is bare)
1436 bare: Whether this is a bare repository
1437 """
1439 path: str
1440 bare: bool
1441 object_store: DiskObjectStore
1442 filter_context: "FilterContext | None"
1444 def __init__(
1445 self,
1446 root: str | bytes | os.PathLike[str],
1447 object_store: PackBasedObjectStore | None = None,
1448 bare: bool | None = None,
1449 ) -> None:
1450 """Open a repository on disk.
1452 Args:
1453 root: Path to the repository's root.
1454 object_store: ObjectStore to use; if omitted, we use the
1455 repository's default object store
1456 bare: True if this is a bare repository.
1457 """
1458 root = os.fspath(root)
1459 if isinstance(root, bytes):
1460 root = os.fsdecode(root)
1461 hidden_path = os.path.join(root, CONTROLDIR)
1462 if bare is None:
1463 if os.path.isfile(hidden_path) or os.path.isdir(
1464 os.path.join(hidden_path, OBJECTDIR)
1465 ):
1466 bare = False
1467 elif os.path.isdir(os.path.join(root, OBJECTDIR)) and os.path.isdir(
1468 os.path.join(root, REFSDIR)
1469 ):
1470 bare = True
1471 else:
1472 raise NotGitRepository(
1473 "No git repository was found at {path}".format(**dict(path=root))
1474 )
1476 self.bare = bare
1477 if bare is False:
1478 if os.path.isfile(hidden_path):
1479 with open(hidden_path, "rb") as f:
1480 path = read_gitfile(f)
1481 self._controldir = os.path.join(root, path)
1482 else:
1483 self._controldir = hidden_path
1484 else:
1485 self._controldir = root
1486 commondir = self.get_named_file(COMMONDIR)
1487 if commondir is not None:
1488 with commondir:
1489 self._commondir = os.path.join(
1490 self.controldir(),
1491 os.fsdecode(commondir.read().rstrip(b"\r\n")),
1492 )
1493 else:
1494 self._commondir = self._controldir
1495 self.path = root
1497 # Initialize refs early so they're available for config condition matchers
1498 self.refs = DiskRefsContainer(
1499 self.commondir(), self._controldir, logger=self._write_reflog
1500 )
1502 # Initialize worktrees container
1503 from .worktree import WorkTreeContainer
1505 self.worktrees = WorkTreeContainer(self)
1507 config = self.get_config()
1508 try:
1509 repository_format_version = config.get("core", "repositoryformatversion")
1510 format_version = (
1511 0
1512 if repository_format_version is None
1513 else int(repository_format_version)
1514 )
1515 except KeyError:
1516 format_version = 0
1518 if format_version not in (0, 1):
1519 raise UnsupportedVersion(format_version)
1521 # Track extensions we encounter
1522 has_reftable_extension = False
1523 for extension, value in config.items((b"extensions",)):
1524 if extension.lower() == b"refstorage":
1525 if value == b"reftable":
1526 has_reftable_extension = True
1527 else:
1528 raise UnsupportedExtension(f"refStorage = {value.decode()}")
1529 elif extension.lower() not in (
1530 b"worktreeconfig",
1531 b"objectformat",
1532 b"relativeworktrees",
1533 ):
1534 raise UnsupportedExtension(extension.decode("utf-8"))
1536 if object_store is None:
1537 # Get shared repository permissions from config
1538 try:
1539 shared_value = config.get(("core",), "sharedRepository")
1540 file_mode, dir_mode = parse_shared_repository(shared_value)
1541 except KeyError:
1542 file_mode, dir_mode = None, None
1544 object_store = DiskObjectStore.from_config(
1545 os.path.join(self.commondir(), OBJECTDIR),
1546 config,
1547 file_mode=file_mode,
1548 dir_mode=dir_mode,
1549 )
1551 # Use reftable if extension is configured
1552 if has_reftable_extension:
1553 from .reftable import ReftableRefsContainer
1555 self.refs = ReftableRefsContainer(self.commondir())
1556 # Update worktrees container after refs change
1557 self.worktrees = WorkTreeContainer(self)
1558 BaseRepo.__init__(self, object_store, self.refs)
1560 # Determine hash algorithm from config if not already set
1561 if self.object_format is None:
1562 from .object_format import DEFAULT_OBJECT_FORMAT, get_object_format
1564 if format_version == 1:
1565 try:
1566 object_format = config.get((b"extensions",), b"objectformat")
1567 self.object_format = get_object_format(
1568 object_format.decode("ascii")
1569 )
1570 except KeyError:
1571 self.object_format = DEFAULT_OBJECT_FORMAT
1572 else:
1573 self.object_format = DEFAULT_OBJECT_FORMAT
1575 self._graftpoints = {}
1576 graft_file = self.get_named_file(
1577 os.path.join("info", "grafts"), basedir=self.commondir()
1578 )
1579 if graft_file:
1580 with graft_file:
1581 self._graftpoints.update(parse_graftpoints(graft_file))
1582 graft_file = self.get_named_file("shallow", basedir=self.commondir())
1583 if graft_file:
1584 with graft_file:
1585 self._graftpoints.update(parse_graftpoints(graft_file))
1587 self.hooks["pre-commit"] = PreCommitShellHook(self.path, self.controldir())
1588 self.hooks["commit-msg"] = CommitMsgShellHook(self.controldir())
1589 self.hooks["post-commit"] = PostCommitShellHook(self.controldir())
1590 self.hooks["pre-receive"] = PreReceiveShellHook(self.controldir())
1591 self.hooks["update"] = UpdateShellHook(self.controldir())
1592 self.hooks["post-receive"] = PostReceiveShellHook(self.controldir())
1594 # Initialize filter context as None, will be created lazily
1595 self.filter_context = None
1597 def get_worktree(self) -> "WorkTree":
1598 """Get the working tree for this repository.
1600 Returns:
1601 WorkTree instance for performing working tree operations
1602 """
1603 from .worktree import WorkTree
1605 return WorkTree(self, self.path)
1607 def _write_reflog(
1608 self,
1609 ref: bytes,
1610 old_sha: bytes,
1611 new_sha: bytes,
1612 committer: bytes | None,
1613 timestamp: int | None,
1614 timezone: int | None,
1615 message: bytes,
1616 ) -> None:
1617 from .reflog import format_reflog_line
1619 path = self._reflog_path(ref)
1621 # Get shared repository permissions
1622 file_mode, dir_mode = self._get_shared_repository_permissions()
1624 # Create directory with appropriate permissions
1625 parent_dir = os.path.dirname(path)
1626 # Create directory tree, setting permissions on each level if needed
1627 parts = []
1628 current = parent_dir
1629 while current and not os.path.exists(current):
1630 parts.append(current)
1631 current = os.path.dirname(current)
1632 parts.reverse()
1633 for part in parts:
1634 os.mkdir(part)
1635 if dir_mode is not None:
1636 os.chmod(part, dir_mode)
1637 if committer is None:
1638 config = self.get_config_stack()
1639 committer = get_user_identity(config)
1640 check_user_identity(committer)
1641 if timestamp is None:
1642 timestamp = int(time.time())
1643 if timezone is None:
1644 timezone = 0 # FIXME
1645 with open(path, "ab") as f:
1646 f.write(
1647 format_reflog_line(
1648 old_sha, new_sha, committer, timestamp, timezone, message
1649 )
1650 + b"\n"
1651 )
1653 # Set file permissions (open() respects umask, so we need chmod to set the actual mode)
1654 # Always chmod to ensure correct permissions even if file already existed
1655 if file_mode is not None:
1656 os.chmod(path, file_mode)
1658 def _reflog_path(self, ref: bytes) -> str:
1659 if ref.startswith((b"main-worktree/", b"worktrees/")):
1660 raise NotImplementedError(f"refs {ref.decode()} are not supported")
1662 base = self.controldir() if is_per_worktree_ref(ref) else self.commondir()
1663 return os.path.join(base, "logs", os.fsdecode(ref))
1665 def read_reflog(self, ref: bytes) -> Generator[reflog.Entry, None, None]:
1666 """Read reflog entries for a reference.
1668 Args:
1669 ref: Reference name (e.g. b'HEAD', b'refs/heads/master')
1671 Yields:
1672 reflog.Entry objects in chronological order (oldest first)
1673 """
1674 from .reflog import read_reflog
1676 path = self._reflog_path(ref)
1677 try:
1678 with open(path, "rb") as f:
1679 yield from read_reflog(f)
1680 except FileNotFoundError:
1681 return
1683 @classmethod
1684 def discover(cls, start: str | bytes | os.PathLike[str] = ".") -> "Repo":
1685 """Iterate parent directories to discover a repository.
1687 Return a Repo object for the first parent directory that looks like a
1688 Git repository.
1690 Args:
1691 start: The directory to start discovery from (defaults to '.')
1692 """
1693 path = os.path.abspath(start)
1694 while True:
1695 try:
1696 return cls(path)
1697 except NotGitRepository:
1698 new_path, _tail = os.path.split(path)
1699 if new_path == path: # Root reached
1700 break
1701 path = new_path
1702 start_str = os.fspath(start)
1703 if isinstance(start_str, bytes):
1704 start_str = start_str.decode("utf-8")
1705 raise NotGitRepository(f"No git repository was found at {start_str}")
1707 def controldir(self) -> str:
1708 """Return the path of the control directory."""
1709 return self._controldir
1711 def commondir(self) -> str:
1712 """Return the path of the common directory.
1714 For a main working tree, it is identical to controldir().
1716 For a linked working tree, it is the control directory of the
1717 main working tree.
1718 """
1719 return self._commondir
1721 def _determine_file_mode(self) -> bool:
1722 """Probe the file-system to determine whether permissions can be trusted.
1724 Returns: True if permissions can be trusted, False otherwise.
1725 """
1726 fname = os.path.join(self.path, ".probe-permissions")
1727 with open(fname, "w") as f:
1728 f.write("")
1730 st1 = os.lstat(fname)
1731 try:
1732 os.chmod(fname, st1.st_mode ^ stat.S_IXUSR)
1733 except PermissionError:
1734 return False
1735 st2 = os.lstat(fname)
1737 os.unlink(fname)
1739 mode_differs = st1.st_mode != st2.st_mode
1740 st2_has_exec = (st2.st_mode & stat.S_IXUSR) != 0
1742 return mode_differs and st2_has_exec
1744 def _determine_symlinks(self) -> bool:
1745 """Probe the filesystem to determine whether symlinks can be created.
1747 Returns: True if symlinks can be created, False otherwise.
1748 """
1749 # TODO(jelmer): Actually probe disk / look at filesystem
1750 return sys.platform != "win32"
1752 def _get_shared_repository_permissions(
1753 self,
1754 ) -> tuple[int | None, int | None]:
1755 """Get shared repository file and directory permissions from config.
1757 Returns:
1758 tuple of (file_mask, directory_mask) or (None, None) if not shared
1759 """
1760 try:
1761 config = self.get_config()
1762 value = config.get(("core",), "sharedRepository")
1763 return parse_shared_repository(value)
1764 except KeyError:
1765 return (None, None)
1767 def _put_named_file(self, path: str, contents: bytes) -> None:
1768 """Write a file to the control dir with the given name and contents.
1770 Args:
1771 path: The path to the file, relative to the control dir.
1772 contents: A string to write to the file.
1773 """
1774 path = path.lstrip(os.path.sep)
1776 # Get shared repository permissions
1777 file_mode, _ = self._get_shared_repository_permissions()
1779 # Create file with appropriate permissions
1780 if file_mode is not None:
1781 with GitFile(
1782 os.path.join(self.controldir(), path), "wb", mask=file_mode
1783 ) as f:
1784 f.write(contents)
1785 else:
1786 with GitFile(os.path.join(self.controldir(), path), "wb") as f:
1787 f.write(contents)
1789 def _del_named_file(self, path: str) -> None:
1790 try:
1791 os.unlink(os.path.join(self.controldir(), path))
1792 except FileNotFoundError:
1793 return
1795 def get_named_file(
1796 self,
1797 path: str | bytes,
1798 basedir: str | None = None,
1799 ) -> BinaryIO | None:
1800 """Get a file from the control dir with a specific name.
1802 Although the filename should be interpreted as a filename relative to
1803 the control dir in a disk-based Repo, the object returned need not be
1804 pointing to a file in that location.
1806 Args:
1807 path: The path to the file, relative to the control dir.
1808 basedir: Optional argument that specifies an alternative to the
1809 control dir.
1810 Returns: An open file object, or None if the file does not exist.
1811 """
1812 # TODO(dborowitz): sanitize filenames, since this is used directly by
1813 # the dumb web serving code.
1814 if basedir is None:
1815 basedir = self.controldir()
1816 if isinstance(path, bytes):
1817 path = path.decode("utf-8")
1818 path = path.lstrip(os.path.sep)
1819 try:
1820 return open(os.path.join(basedir, path), "rb")
1821 except FileNotFoundError:
1822 return None
1824 def index_path(self) -> str:
1825 """Return path to the index file."""
1826 return os.path.join(self.controldir(), INDEX_FILENAME)
1828 def open_index(self, config: "Config | None" = None) -> "Index":
1829 """Open the index for this repository.
1831 Args:
1832 config: Configuration to consult for index settings. If None,
1833 falls back to ``self.get_config_stack()``.
1835 Raises:
1836 NoIndexPresent: If no index is present
1837 Returns: The matching `Index`
1838 """
1839 from .index import Index, make_path_normalizer
1841 if not self.has_index():
1842 raise NoIndexPresent
1844 if config is None:
1845 config = self.get_config_stack()
1846 many_files = config.get_boolean(b"feature", b"manyFiles", False)
1847 skip_hash = False
1848 index_version = None
1850 if many_files:
1851 # When feature.manyFiles is enabled, set index.version=4 and index.skipHash=true
1852 try:
1853 index_version_str = config.get(b"index", b"version")
1854 index_version = int(index_version_str)
1855 except KeyError:
1856 index_version = 4 # Default to version 4 for manyFiles
1857 skip_hash = config.get_boolean(b"index", b"skipHash", True)
1858 else:
1859 # Check for explicit index settings
1860 try:
1861 index_version_str = config.get(b"index", b"version")
1862 index_version = int(index_version_str)
1863 except KeyError:
1864 index_version = None
1865 skip_hash = config.get_boolean(b"index", b"skipHash", False)
1867 # Get shared repository permissions for index file
1868 file_mode, _ = self._get_shared_repository_permissions()
1870 return Index(
1871 self.index_path(),
1872 skip_hash=skip_hash,
1873 version=index_version,
1874 file_mode=file_mode,
1875 path_normalizer=make_path_normalizer(config),
1876 )
1878 def has_index(self) -> bool:
1879 """Check if an index is present."""
1880 # Bare repos must never have index files; non-bare repos may have a
1881 # missing index file, which is treated as empty.
1882 return not self.bare
1884 def clone(
1885 self,
1886 target_path: str | bytes | os.PathLike[str],
1887 *,
1888 mkdir: bool = True,
1889 bare: bool = False,
1890 origin: bytes = b"origin",
1891 checkout: bool | None = None,
1892 branch: bytes | None = None,
1893 progress: Callable[[str], None] | None = None,
1894 depth: int | None = None,
1895 symlinks: bool | None = None,
1896 ) -> "Repo":
1897 """Clone this repository.
1899 Args:
1900 target_path: Target path
1901 mkdir: Create the target directory
1902 bare: Whether to create a bare repository
1903 checkout: Whether or not to check-out HEAD after cloning
1904 origin: Base name for refs in target repository
1905 cloned from this repository
1906 branch: Optional branch or tag to be used as HEAD in the new repository
1907 instead of this repository's HEAD.
1908 progress: Optional progress function
1909 depth: Depth at which to fetch
1910 symlinks: Symlinks setting (default to autodetect)
1911 Returns: Created repository as `Repo`
1912 """
1913 encoded_path = os.fsencode(self.path)
1915 if mkdir:
1916 os.mkdir(target_path)
1918 try:
1919 if not bare:
1920 target = Repo.init(target_path, symlinks=symlinks)
1921 if checkout is None:
1922 checkout = True
1923 else:
1924 if checkout:
1925 raise ValueError("checkout and bare are incompatible")
1926 target = Repo.init_bare(target_path)
1928 try:
1929 target_config = target.get_config()
1930 target_config.set((b"remote", origin), b"url", encoded_path)
1931 target_config.set(
1932 (b"remote", origin),
1933 b"fetch",
1934 b"+refs/heads/*:refs/remotes/" + origin + b"/*",
1935 )
1936 target_config.write_to_path()
1938 ref_message = b"clone: from " + encoded_path
1939 self.fetch(target, depth=depth)
1940 target.refs.import_refs(
1941 Ref(b"refs/remotes/" + origin),
1942 self.refs.as_dict(Ref(b"refs/heads")),
1943 message=ref_message,
1944 )
1945 target.refs.import_refs(
1946 Ref(b"refs/tags"),
1947 self.refs.as_dict(Ref(b"refs/tags")),
1948 message=ref_message,
1949 )
1951 head_chain, origin_sha = self.refs.follow(HEADREF)
1952 origin_head = head_chain[-1] if head_chain else None
1953 head: ObjectID | None = None
1954 if origin_sha and not origin_head:
1955 # set detached HEAD
1956 target.refs[HEADREF] = origin_sha
1957 head = origin_sha
1958 else:
1959 _set_origin_head(target.refs, origin, origin_head)
1960 head_ref = _set_default_branch(
1961 target.refs, origin, origin_head, branch, ref_message
1962 )
1964 # Update target head
1965 if head_ref:
1966 head = _set_head(target.refs, head_ref, ref_message)
1967 else:
1968 head = None
1970 if checkout and head is not None:
1971 target.get_worktree().reset_index(config=target.get_config_stack())
1972 except BaseException:
1973 target.close()
1974 raise
1975 except BaseException:
1976 if mkdir:
1977 import shutil
1979 shutil.rmtree(target_path)
1980 raise
1981 return target
1983 def _get_config_condition_matchers(self) -> dict[str, "ConditionMatcher"]:
1984 """Get condition matchers for includeIf conditions.
1986 Returns a dict of condition prefix to matcher function.
1987 """
1988 from pathlib import Path
1990 from .config import ConditionMatcher, match_glob_pattern
1992 # Add gitdir matchers
1993 def match_gitdir(pattern: str, case_sensitive: bool = True) -> bool:
1994 """Match gitdir against a pattern.
1996 Args:
1997 pattern: Pattern to match against
1998 case_sensitive: Whether to match case-sensitively
2000 Returns:
2001 True if gitdir matches pattern
2002 """
2003 # Handle relative patterns (starting with ./)
2004 if pattern.startswith("./"):
2005 # Can't handle relative patterns without config directory context
2006 return False
2008 # Normalize repository path
2009 try:
2010 repo_path = str(Path(self._controldir).resolve())
2011 except (OSError, ValueError):
2012 return False
2014 # Expand ~ in pattern and normalize
2015 pattern = os.path.expanduser(pattern)
2017 # Normalize pattern following Git's rules
2018 pattern = pattern.replace("\\", "/")
2019 if not pattern.startswith(("~/", "./", "/", "**")):
2020 # Check for Windows absolute path
2021 if len(pattern) >= 2 and pattern[1] == ":":
2022 pass
2023 else:
2024 pattern = "**/" + pattern
2025 if pattern.endswith("/"):
2026 pattern = pattern + "**"
2028 # Use the existing _match_gitdir_pattern function
2029 from .config import _match_gitdir_pattern
2031 pattern_bytes = pattern.encode("utf-8", errors="replace")
2032 repo_path_bytes = repo_path.encode("utf-8", errors="replace")
2034 return _match_gitdir_pattern(
2035 repo_path_bytes, pattern_bytes, ignorecase=not case_sensitive
2036 )
2038 # Add onbranch matcher
2039 def match_onbranch(pattern: str) -> bool:
2040 """Match current branch against a pattern.
2042 Args:
2043 pattern: Pattern to match against
2045 Returns:
2046 True if current branch matches pattern
2047 """
2048 try:
2049 # Get the current branch using refs
2050 ref_chain, _ = self.refs.follow(HEADREF)
2051 head_ref = ref_chain[-1] # Get the final resolved ref
2052 except KeyError:
2053 pass
2054 else:
2055 if head_ref and head_ref.startswith(b"refs/heads/"):
2056 # Extract branch name from ref
2057 branch = extract_branch_name(head_ref).decode(
2058 "utf-8", errors="replace"
2059 )
2060 return match_glob_pattern(branch, pattern)
2061 return False
2063 matchers: dict[str, ConditionMatcher] = {
2064 "onbranch:": match_onbranch,
2065 "gitdir:": lambda pattern: match_gitdir(pattern, True),
2066 "gitdir/i:": lambda pattern: match_gitdir(pattern, False),
2067 }
2069 return matchers
2071 def get_worktree_config(self) -> "ConfigFile":
2072 """Get the worktree-specific config.
2074 Returns:
2075 ConfigFile object for the worktree config
2076 """
2077 from .config import ConfigFile
2079 path = os.path.join(self.commondir(), "config.worktree")
2080 try:
2081 # Pass condition matchers for includeIf evaluation
2082 condition_matchers = self._get_config_condition_matchers()
2083 return ConfigFile.from_path(path, condition_matchers=condition_matchers)
2084 except FileNotFoundError:
2085 cf = ConfigFile()
2086 cf.path = path
2087 return cf
2089 def get_config(self) -> "ConfigFile":
2090 """Retrieve the config object.
2092 Returns: `ConfigFile` object for the ``.git/config`` file.
2093 """
2094 from .config import ConfigFile
2096 path = os.path.join(self._commondir, "config")
2097 try:
2098 # Pass condition matchers for includeIf evaluation
2099 condition_matchers = self._get_config_condition_matchers()
2100 return ConfigFile.from_path(path, condition_matchers=condition_matchers)
2101 except FileNotFoundError:
2102 ret = ConfigFile()
2103 ret.path = path
2104 return ret
2106 def get_rebase_state_manager(self) -> "RebaseStateManager":
2107 """Get the appropriate rebase state manager for this repository.
2109 Returns: DiskRebaseStateManager instance
2110 """
2111 import os
2113 from .rebase import DiskRebaseStateManager
2115 path = os.path.join(self.controldir(), "rebase-merge")
2116 return DiskRebaseStateManager(path)
2118 def get_description(self) -> bytes | None:
2119 """Retrieve the description of this repository.
2121 Returns: Description as bytes or None.
2122 """
2123 path = os.path.join(self._controldir, "description")
2124 try:
2125 with GitFile(path, "rb") as f:
2126 return f.read()
2127 except FileNotFoundError:
2128 return None
2130 def __repr__(self) -> str:
2131 """Return string representation of this repository."""
2132 return f"<Repo at {self.path!r}>"
2134 def set_description(self, description: bytes) -> None:
2135 """Set the description for this repository.
2137 Args:
2138 description: Text to set as description for this repository.
2139 """
2140 self._put_named_file("description", description)
2142 @classmethod
2143 def _init_maybe_bare(
2144 cls,
2145 path: str | bytes | os.PathLike[str],
2146 controldir: str | bytes | os.PathLike[str],
2147 bare: bool,
2148 object_store: PackBasedObjectStore | None = None,
2149 config: "StackedConfig | None" = None,
2150 default_branch: bytes | None = None,
2151 symlinks: bool | None = None,
2152 format: int | None = None,
2153 shared_repository: str | bool | None = None,
2154 object_format: str | None = None,
2155 ) -> "Repo":
2156 path = os.fspath(path)
2157 if isinstance(path, bytes):
2158 path = os.fsdecode(path)
2159 controldir = os.fspath(controldir)
2160 if isinstance(controldir, bytes):
2161 controldir = os.fsdecode(controldir)
2163 # Determine shared repository permissions early
2164 file_mode: int | None = None
2165 dir_mode: int | None = None
2166 if shared_repository is not None:
2167 file_mode, dir_mode = parse_shared_repository(shared_repository)
2169 # Create base directories with appropriate permissions
2170 for d in BASE_DIRECTORIES:
2171 dir_path = os.path.join(controldir, *d)
2172 os.mkdir(dir_path)
2173 if dir_mode is not None:
2174 os.chmod(dir_path, dir_mode)
2176 # Determine hash algorithm
2177 from .object_format import get_object_format
2179 hash_alg = get_object_format(object_format)
2181 if object_store is None:
2182 object_store = DiskObjectStore.init(
2183 os.path.join(controldir, OBJECTDIR),
2184 file_mode=file_mode,
2185 dir_mode=dir_mode,
2186 object_format=hash_alg,
2187 )
2188 ret = cls(path, bare=bare, object_store=object_store)
2189 if default_branch is None:
2190 if config is None:
2191 from .config import StackedConfig
2193 config = StackedConfig.default()
2194 try:
2195 default_branch = config.get("init", "defaultBranch")
2196 except KeyError:
2197 default_branch = DEFAULT_BRANCH
2198 ret.refs.set_symbolic_ref(HEADREF, local_branch_name(default_branch))
2199 ret._init_files(
2200 bare=bare,
2201 symlinks=symlinks,
2202 format=format,
2203 shared_repository=shared_repository,
2204 object_format=object_format,
2205 )
2206 return ret
2208 @classmethod
2209 def init(
2210 cls,
2211 path: str | bytes | os.PathLike[str],
2212 *,
2213 mkdir: bool = False,
2214 config: "StackedConfig | None" = None,
2215 default_branch: bytes | None = None,
2216 symlinks: bool | None = None,
2217 format: int | None = None,
2218 shared_repository: str | bool | None = None,
2219 object_format: str | None = None,
2220 ) -> "Repo":
2221 """Create a new repository.
2223 Args:
2224 path: Path in which to create the repository
2225 mkdir: Whether to create the directory
2226 config: Configuration object
2227 default_branch: Default branch name
2228 symlinks: Whether to support symlinks
2229 format: Repository format version (defaults to 0)
2230 shared_repository: Shared repository setting (group, all, umask, or octal)
2231 object_format: Object format to use ("sha1" or "sha256", defaults to "sha1")
2232 Returns: `Repo` instance
2233 """
2234 path = os.fspath(path)
2235 if isinstance(path, bytes):
2236 path = os.fsdecode(path)
2237 if mkdir:
2238 os.mkdir(path)
2239 controldir = os.path.join(path, CONTROLDIR)
2240 os.mkdir(controldir)
2241 _set_filesystem_hidden(controldir)
2242 return cls._init_maybe_bare(
2243 path,
2244 controldir,
2245 False,
2246 config=config,
2247 default_branch=default_branch,
2248 symlinks=symlinks,
2249 format=format,
2250 shared_repository=shared_repository,
2251 object_format=object_format,
2252 )
2254 @classmethod
2255 def _init_new_working_directory(
2256 cls,
2257 path: str | bytes | os.PathLike[str],
2258 main_repo: "Repo",
2259 identifier: str | None = None,
2260 mkdir: bool = False,
2261 relative_paths: bool = False,
2262 ) -> "Repo":
2263 """Create a new working directory linked to a repository.
2265 Args:
2266 path: Path in which to create the working tree.
2267 main_repo: Main repository to reference
2268 identifier: Worktree identifier
2269 mkdir: Whether to create the directory
2270 relative_paths: Whether to use relative paths for gitdir references
2271 Returns: `Repo` instance
2272 """
2273 path = os.fspath(path)
2274 if isinstance(path, bytes):
2275 path = os.fsdecode(path)
2276 if mkdir:
2277 os.mkdir(path)
2278 if identifier is None:
2279 identifier = os.path.basename(path)
2280 # Ensure we use absolute path for the worktree control directory
2281 main_controldir = os.path.abspath(main_repo.controldir())
2282 main_worktreesdir = os.path.join(main_controldir, WORKTREES)
2283 worktree_controldir = os.path.join(main_worktreesdir, identifier)
2284 gitdirfile_abs = os.path.abspath(os.path.join(path, CONTROLDIR))
2286 # Write gitdir reference in .git file (can be relative)
2287 # Import helper from worktree module to avoid duplication
2288 from .worktree import _compute_gitdir_path
2290 gitdir_ref = _compute_gitdir_path(
2291 main_repo,
2292 worktree_controldir,
2293 os.path.dirname(gitdirfile_abs),
2294 relative_paths,
2295 )
2297 with open(gitdirfile_abs, "wb") as f:
2298 f.write(b"gitdir: " + os.fsencode(gitdir_ref) + b"\n")
2300 # Get shared repository permissions from main repository
2301 _, dir_mode = main_repo._get_shared_repository_permissions()
2303 # Create directories with appropriate permissions
2304 try:
2305 os.mkdir(main_worktreesdir)
2306 if dir_mode is not None:
2307 os.chmod(main_worktreesdir, dir_mode)
2308 except FileExistsError:
2309 pass
2310 try:
2311 os.mkdir(worktree_controldir)
2312 if dir_mode is not None:
2313 os.chmod(worktree_controldir, dir_mode)
2314 except FileExistsError:
2315 pass
2317 # Write gitdir path in control directory (can be relative)
2318 gitdir_path = _compute_gitdir_path(
2319 main_repo, gitdirfile_abs, worktree_controldir, relative_paths
2320 )
2322 with open(os.path.join(worktree_controldir, GITDIR), "wb") as f:
2323 f.write(os.fsencode(gitdir_path) + b"\n")
2324 with open(os.path.join(worktree_controldir, COMMONDIR), "wb") as f:
2325 f.write(b"../..\n")
2326 with open(os.path.join(worktree_controldir, "HEAD"), "wb") as f:
2327 f.write(main_repo.head() + b"\n")
2328 r = cls(os.path.normpath(path))
2329 r.get_worktree().reset_index(config=r.get_config_stack())
2330 return r
2332 @classmethod
2333 def init_bare(
2334 cls,
2335 path: str | bytes | os.PathLike[str],
2336 *,
2337 mkdir: bool = False,
2338 object_store: PackBasedObjectStore | None = None,
2339 config: "StackedConfig | None" = None,
2340 default_branch: bytes | None = None,
2341 format: int | None = None,
2342 shared_repository: str | bool | None = None,
2343 object_format: str | None = None,
2344 ) -> "Repo":
2345 """Create a new bare repository.
2347 ``path`` should already exist and be an empty directory.
2349 Args:
2350 path: Path to create bare repository in
2351 mkdir: Whether to create the directory
2352 object_store: Object store to use
2353 config: Configuration object
2354 default_branch: Default branch name
2355 format: Repository format version (defaults to 0)
2356 shared_repository: Shared repository setting (group, all, umask, or octal)
2357 object_format: Object format to use ("sha1" or "sha256", defaults to "sha1")
2358 Returns: a `Repo` instance
2359 """
2360 path = os.fspath(path)
2361 if isinstance(path, bytes):
2362 path = os.fsdecode(path)
2363 if mkdir:
2364 os.mkdir(path)
2365 return cls._init_maybe_bare(
2366 path,
2367 path,
2368 True,
2369 object_store=object_store,
2370 config=config,
2371 default_branch=default_branch,
2372 format=format,
2373 shared_repository=shared_repository,
2374 object_format=object_format,
2375 )
2377 create = init_bare
2379 def close(self) -> None:
2380 """Close any files opened by this repository."""
2381 self.object_store.close()
2382 # Clean up filter context if it was created
2383 if self.filter_context is not None:
2384 self.filter_context.close()
2385 self.filter_context = None
2387 def __enter__(self) -> Self:
2388 """Enter context manager."""
2389 return self
2391 def __exit__(
2392 self,
2393 exc_type: type[BaseException] | None,
2394 exc_val: BaseException | None,
2395 exc_tb: TracebackType | None,
2396 ) -> None:
2397 """Exit context manager and close repository."""
2398 self.close()
2400 def _read_gitattributes(self) -> dict[bytes, dict[bytes, bytes]]:
2401 """Read .gitattributes file from working tree.
2403 Returns:
2404 Dictionary mapping file patterns to attributes
2405 """
2406 gitattributes = {}
2407 gitattributes_path = os.path.join(self.path, ".gitattributes")
2409 if os.path.exists(gitattributes_path):
2410 with open(gitattributes_path, "rb") as f:
2411 for line in f:
2412 line = line.strip()
2413 if not line or line.startswith(b"#"):
2414 continue
2416 parts = line.split()
2417 if len(parts) < 2:
2418 continue
2420 pattern = parts[0]
2421 attrs = {}
2423 for attr in parts[1:]:
2424 if attr.startswith(b"-"):
2425 # Unset attribute
2426 attrs[attr[1:]] = b"false"
2427 elif b"=" in attr:
2428 # Set to value
2429 key, value = attr.split(b"=", 1)
2430 attrs[key] = value
2431 else:
2432 # Set attribute
2433 attrs[attr] = b"true"
2435 gitattributes[pattern] = attrs
2437 return gitattributes
2439 def get_blob_normalizer(
2440 self, config: "Config | None" = None
2441 ) -> "FilterBlobNormalizer":
2442 """Return a BlobNormalizer object.
2444 Args:
2445 config: Configuration to consult for filter setup. If None,
2446 falls back to ``self.get_config_stack()``.
2447 """
2448 from .filters import FilterBlobNormalizer, FilterContext, FilterRegistry
2450 if config is None:
2451 config = self.get_config_stack()
2452 git_attributes = self.get_gitattributes()
2454 # Lazily create FilterContext if needed
2455 if self.filter_context is None:
2456 filter_registry = FilterRegistry(config, self)
2457 self.filter_context = FilterContext(filter_registry)
2458 else:
2459 # Refresh the context with current config to handle config changes
2460 self.filter_context.refresh_config(config)
2462 return FilterBlobNormalizer(
2463 config, git_attributes, filter_context=self.filter_context
2464 )
2466 def get_gitattributes(self, tree: bytes | None = None) -> "GitAttributes":
2467 """Read gitattributes for the repository.
2469 Args:
2470 tree: Tree SHA to read .gitattributes from (defaults to HEAD)
2472 Returns:
2473 GitAttributes object that can be used to match paths
2474 """
2475 from .attrs import (
2476 GitAttributes,
2477 Pattern,
2478 parse_git_attributes,
2479 )
2481 patterns = []
2483 # Read system gitattributes (TODO: implement this)
2484 # Read global gitattributes (TODO: implement this)
2486 # Read repository .gitattributes from index/tree
2487 if tree is None:
2488 try:
2489 # Try to get from HEAD
2490 head = self[b"HEAD"]
2491 # Peel tags to get to the underlying commit
2492 while isinstance(head, Tag):
2493 _cls, obj = head.object
2494 head = self.get_object(obj)
2495 if not isinstance(head, Commit):
2496 raise ValueError(
2497 f"Expected HEAD to point to a Commit, got {type(head).__name__}. "
2498 f"This usually means HEAD points to a {type(head).__name__} object "
2499 f"instead of a Commit."
2500 )
2501 tree = head.tree
2502 except KeyError:
2503 # No HEAD, no attributes from tree
2504 pass
2506 if tree is not None:
2507 try:
2508 tree_obj = self[tree]
2509 assert isinstance(tree_obj, Tree)
2510 if b".gitattributes" in tree_obj:
2511 _, attrs_sha = tree_obj[b".gitattributes"]
2512 attrs_blob = self[attrs_sha]
2513 if isinstance(attrs_blob, Blob):
2514 attrs_data = BytesIO(attrs_blob.data)
2515 for pattern_bytes, attrs in parse_git_attributes(attrs_data):
2516 pattern = Pattern(pattern_bytes)
2517 patterns.append((pattern, attrs))
2518 except (KeyError, NotTreeError):
2519 pass
2521 # Read .git/info/attributes
2522 info_attrs_path = os.path.join(self.controldir(), "info", "attributes")
2523 if os.path.exists(info_attrs_path):
2524 with open(info_attrs_path, "rb") as f:
2525 for pattern_bytes, attrs in parse_git_attributes(f):
2526 pattern = Pattern(pattern_bytes)
2527 patterns.append((pattern, attrs))
2529 # Read .gitattributes from working directory (if it exists)
2530 working_attrs_path = os.path.join(self.path, ".gitattributes")
2531 if os.path.exists(working_attrs_path):
2532 with open(working_attrs_path, "rb") as f:
2533 for pattern_bytes, attrs in parse_git_attributes(f):
2534 pattern = Pattern(pattern_bytes)
2535 patterns.append((pattern, attrs))
2537 return GitAttributes(patterns)
2540class MemoryRepo(BaseRepo):
2541 """Repo that stores refs, objects, and named files in memory.
2543 MemoryRepos are always bare: they have no working tree and no index, since
2544 those have a stronger dependency on the filesystem.
2545 """
2547 filter_context: "FilterContext | None"
2549 def __init__(self) -> None:
2550 """Create a new repository in memory."""
2551 from .config import ConfigFile
2552 from .object_format import DEFAULT_OBJECT_FORMAT
2554 self._reflog: list[Any] = []
2555 refs_container = DictRefsContainer({}, logger=self._append_reflog)
2556 BaseRepo.__init__(self, MemoryObjectStore(), refs_container)
2557 self._named_files: dict[str, bytes] = {}
2558 self.bare = True
2559 self._config = ConfigFile()
2560 self._description: bytes | None = None
2561 self.filter_context = None
2562 # MemoryRepo defaults to default object format
2563 self.object_format = DEFAULT_OBJECT_FORMAT
2565 def _append_reflog(
2566 self,
2567 ref: bytes,
2568 old_sha: bytes | None,
2569 new_sha: bytes | None,
2570 committer: bytes | None,
2571 timestamp: int | None,
2572 timezone: int | None,
2573 message: bytes | None,
2574 ) -> None:
2575 self._reflog.append(
2576 (ref, old_sha, new_sha, committer, timestamp, timezone, message)
2577 )
2579 def set_description(self, description: bytes) -> None:
2580 """Set the description for this repository.
2582 Args:
2583 description: Text to set as description
2584 """
2585 self._description = description
2587 def get_description(self) -> bytes | None:
2588 """Get the description of this repository.
2590 Returns:
2591 Repository description as bytes
2592 """
2593 return self._description
2595 def _determine_file_mode(self) -> bool:
2596 """Probe the file-system to determine whether permissions can be trusted.
2598 Returns: True if permissions can be trusted, False otherwise.
2599 """
2600 return sys.platform != "win32"
2602 def _determine_symlinks(self) -> bool:
2603 """Probe the file-system to determine whether permissions can be trusted.
2605 Returns: True if permissions can be trusted, False otherwise.
2606 """
2607 return sys.platform != "win32"
2609 def _put_named_file(self, path: str, contents: bytes) -> None:
2610 """Write a file to the control dir with the given name and contents.
2612 Args:
2613 path: The path to the file, relative to the control dir.
2614 contents: A string to write to the file.
2615 """
2616 self._named_files[path] = contents
2618 def _del_named_file(self, path: str) -> None:
2619 try:
2620 del self._named_files[path]
2621 except KeyError:
2622 pass
2624 def get_named_file(
2625 self,
2626 path: str | bytes,
2627 basedir: str | None = None,
2628 ) -> BytesIO | None:
2629 """Get a file from the control dir with a specific name.
2631 Although the filename should be interpreted as a filename relative to
2632 the control dir in a disk-baked Repo, the object returned need not be
2633 pointing to a file in that location.
2635 Args:
2636 path: The path to the file, relative to the control dir.
2637 basedir: Optional base directory for the path
2638 Returns: An open file object, or None if the file does not exist.
2639 """
2640 path_str = path.decode() if isinstance(path, bytes) else path
2641 contents = self._named_files.get(path_str, None)
2642 if contents is None:
2643 return None
2644 return BytesIO(contents)
2646 def open_index(self, config: "Config | None" = None) -> "Index":
2647 """Fail to open index for this repo, since it is bare.
2649 Args:
2650 config: Unused; kept for signature compatibility with ``BaseRepo``.
2652 Raises:
2653 NoIndexPresent: Raised when no index is present
2654 """
2655 raise NoIndexPresent
2657 def _init_config(self, config: "ConfigFile") -> None:
2658 """Initialize repository configuration for MemoryRepo."""
2659 self._config = config
2661 def get_config(self) -> "ConfigFile":
2662 """Retrieve the config object.
2664 Returns: `ConfigFile` object.
2665 """
2666 return self._config
2668 def get_rebase_state_manager(self) -> "RebaseStateManager":
2669 """Get the appropriate rebase state manager for this repository.
2671 Returns: MemoryRebaseStateManager instance
2672 """
2673 from .rebase import MemoryRebaseStateManager
2675 return MemoryRebaseStateManager(self)
2677 def get_blob_normalizer(
2678 self, config: "Config | None" = None
2679 ) -> "FilterBlobNormalizer":
2680 """Return a BlobNormalizer object for checkin/checkout operations.
2682 Args:
2683 config: Configuration to consult for filter setup. If None,
2684 falls back to ``self.get_config_stack()``.
2685 """
2686 from .filters import FilterBlobNormalizer, FilterContext, FilterRegistry
2688 if config is None:
2689 config = self.get_config_stack()
2690 git_attributes = self.get_gitattributes()
2692 if self.filter_context is None:
2693 filter_registry = FilterRegistry(config, self)
2694 self.filter_context = FilterContext(filter_registry)
2695 else:
2696 self.filter_context.refresh_config(config)
2698 return FilterBlobNormalizer(
2699 config, git_attributes, filter_context=self.filter_context
2700 )
2702 def get_gitattributes(self, tree: bytes | None = None) -> "GitAttributes":
2703 """Read gitattributes for the repository."""
2704 from .attrs import GitAttributes
2706 # Memory repos don't have working trees or gitattributes files
2707 # Return empty GitAttributes
2708 return GitAttributes([])
2710 def close(self) -> None:
2711 """Close any resources opened by this repository."""
2712 # Clean up filter context if it was created
2713 if self.filter_context is not None:
2714 self.filter_context.close()
2715 self.filter_context = None
2716 # Close object store to release pack files
2717 self.object_store.close()
2719 def do_commit(
2720 self,
2721 message: bytes | None = None,
2722 committer: bytes | None = None,
2723 author: bytes | None = None,
2724 commit_timestamp: float | None = None,
2725 commit_timezone: int | None = None,
2726 author_timestamp: float | None = None,
2727 author_timezone: int | None = None,
2728 tree: ObjectID | None = None,
2729 encoding: bytes | None = None,
2730 ref: Ref | None = HEADREF,
2731 merge_heads: list[ObjectID] | None = None,
2732 no_verify: bool = False,
2733 sign: bool = False,
2734 config: "Config | None" = None,
2735 ) -> bytes:
2736 """Create a new commit.
2738 This is a simplified implementation for in-memory repositories that
2739 doesn't support worktree operations or hooks.
2741 Args:
2742 message: Commit message
2743 committer: Committer fullname
2744 author: Author fullname
2745 commit_timestamp: Commit timestamp (defaults to now)
2746 commit_timezone: Commit timestamp timezone (defaults to GMT)
2747 author_timestamp: Author timestamp (defaults to commit timestamp)
2748 author_timezone: Author timestamp timezone (defaults to commit timezone)
2749 tree: SHA1 of the tree root to use
2750 encoding: Encoding
2751 ref: Optional ref to commit to (defaults to current branch).
2752 If None, creates a dangling commit without updating any ref.
2753 merge_heads: Merge heads
2754 no_verify: Skip pre-commit and commit-msg hooks (ignored for MemoryRepo)
2755 sign: GPG Sign the commit (ignored for MemoryRepo)
2756 config: Configuration to consult for committer/author identity. If
2757 None, falls back to ``self.get_config_stack()``.
2759 Returns:
2760 New commit SHA1
2761 """
2762 import time
2764 from .objects import Commit
2766 if tree is None:
2767 raise ValueError("tree must be specified for MemoryRepo")
2769 c = Commit()
2770 if len(tree) != self.object_format.hex_length:
2771 raise ValueError(
2772 f"tree must be a {self.object_format.hex_length}-character hex sha string"
2773 )
2774 c.tree = tree
2776 if config is None:
2777 config = self.get_config_stack()
2778 if merge_heads is None:
2779 merge_heads = []
2780 if committer is None:
2781 committer = get_user_identity(config, kind="COMMITTER")
2782 check_user_identity(committer)
2783 c.committer = committer
2784 if commit_timestamp is None:
2785 commit_timestamp = time.time()
2786 c.commit_time = int(commit_timestamp)
2787 if commit_timezone is None:
2788 commit_timezone = 0
2789 c.commit_timezone = commit_timezone
2790 if author is None:
2791 author = get_user_identity(config, kind="AUTHOR")
2792 c.author = author
2793 check_user_identity(author)
2794 if author_timestamp is None:
2795 author_timestamp = commit_timestamp
2796 c.author_time = int(author_timestamp)
2797 if author_timezone is None:
2798 author_timezone = commit_timezone
2799 c.author_timezone = author_timezone
2800 if encoding is None:
2801 try:
2802 encoding = config.get(("i18n",), "commitEncoding")
2803 except KeyError:
2804 pass
2805 if encoding is not None:
2806 c.encoding = encoding
2808 # Handle message (for MemoryRepo, we don't support callable messages)
2809 if callable(message):
2810 message = message(self, c)
2811 if message is None:
2812 raise ValueError("Message callback returned None")
2814 if message is None:
2815 raise ValueError("No commit message specified")
2817 c.message = message
2819 if ref is None:
2820 # Create a dangling commit
2821 c.parents = merge_heads
2822 self.object_store.add_object(c)
2823 else:
2824 try:
2825 old_head = self.refs[ref]
2826 c.parents = [old_head, *merge_heads]
2827 self.object_store.add_object(c)
2828 ok = self.refs.set_if_equals(
2829 ref,
2830 old_head,
2831 c.id,
2832 message=b"commit: " + message,
2833 committer=committer,
2834 timestamp=int(commit_timestamp),
2835 timezone=commit_timezone,
2836 )
2837 except KeyError:
2838 c.parents = merge_heads
2839 self.object_store.add_object(c)
2840 ok = self.refs.add_if_new(
2841 ref,
2842 c.id,
2843 message=b"commit: " + message,
2844 committer=committer,
2845 timestamp=int(commit_timestamp),
2846 timezone=commit_timezone,
2847 )
2848 if not ok:
2849 from .errors import CommitError
2851 raise CommitError(f"{ref!r} changed during commit")
2853 return c.id
2855 @classmethod
2856 def init_bare(
2857 cls,
2858 objects: Iterable[ShaFile],
2859 refs: Mapping[Ref, ObjectID],
2860 format: int | None = None,
2861 object_format: str | None = None,
2862 ) -> "MemoryRepo":
2863 """Create a new bare repository in memory.
2865 Args:
2866 objects: Objects for the new repository,
2867 as iterable
2868 refs: Refs as dictionary, mapping names
2869 to object SHA1s
2870 format: Repository format version (defaults to 0)
2871 object_format: Object format to use ("sha1" or "sha256", defaults to "sha1")
2872 """
2873 ret = cls()
2874 for obj in objects:
2875 ret.object_store.add_object(obj)
2876 for refname, sha in refs.items():
2877 ret.refs.add_if_new(refname, sha)
2878 ret._init_files(bare=True, format=format, object_format=object_format)
2879 return ret