Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/repo.py: 37%
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 "InvalidWorktreeConfiguration",
49 "MemoryRepo",
50 "ParentsProvider",
51 "Repo",
52 "UnsupportedExtension",
53 "UnsupportedVersion",
54 "check_user_identity",
55 "get_user_identity",
56 "parse_graftpoints",
57 "parse_shared_repository",
58 "read_gitfile",
59 "sanitize_user_identity",
60 "serialize_graftpoints",
61]
63import errno
64import logging
65import os
66import stat
67import sys
68import time
69import warnings
70from collections.abc import Callable, Generator, Iterable, Iterator, Mapping, Sequence
71from io import BytesIO
72from types import TracebackType
73from typing import (
74 TYPE_CHECKING,
75 Any,
76 BinaryIO,
77 TypeVar,
78)
80if sys.version_info >= (3, 11):
81 from typing import Self
82else:
83 from typing_extensions import Self
85if TYPE_CHECKING:
86 # There are no circular imports here, but we try to defer imports as long
87 # as possible to reduce start-up time for anything that doesn't need
88 # these imports.
89 from .attrs import GitAttributes
90 from .config import ConditionMatcher, Config, ConfigFile, StackedConfig
91 from .diff_tree import RenameDetector
92 from .filters import FilterBlobNormalizer, FilterContext
93 from .index import Index
94 from .notes import Notes
95 from .object_format import ObjectFormat
96 from .object_store import BaseObjectStore, GraphWalker
97 from .pack import UnpackedObject
98 from .rebase import RebaseStateManager
99 from .walk import Walker
100 from .worktree import WorkTree
102from . import _deprecated_aliases, reflog
103from .errors import (
104 NoIndexPresent,
105 NotBlobError,
106 NotCommitError,
107 NotGitRepository,
108 NotTagError,
109 NotTreeError,
110 RefFormatError,
111)
112from .file import (
113 PERM_EVERYBODY,
114 PERM_GROUP,
115 GitFile,
116 SharedPerm,
117 adjust_shared_perm,
118 open_nofollow_read,
119)
120from .hooks import (
121 CommitMsgShellHook,
122 Hook,
123 PostCommitShellHook,
124 PostReceiveShellHook,
125 PreCommitShellHook,
126 PreReceiveShellHook,
127 UpdateShellHook,
128)
129from .object_store import (
130 DiskObjectStore,
131 MemoryObjectStore,
132 MissingObjectFinder,
133 ObjectStoreGraphWalker,
134 PackBasedObjectStore,
135 PackCapableObjectStore,
136 find_shallow,
137)
138from .objects import (
139 Blob,
140 Commit,
141 ObjectID,
142 RawObjectID,
143 ShaFile,
144 Tag,
145 Tree,
146 check_hexsha,
147 valid_hexsha,
148)
149from .pack import generate_unpacked_objects
150from .refs import (
151 HEADREF,
152 DictRefsContainer,
153 DiskRefsContainer,
154 Ref,
155 RefsContainer,
156 _set_branch_tracking,
157 _set_default_branch,
158 _set_head,
159 _set_origin_head,
160 extract_branch_name,
161 is_per_worktree_ref,
162 local_branch_name,
163)
165logger = logging.getLogger(__name__)
168__getattr__ = _deprecated_aliases(
169 __name__,
170 {
171 "LOCAL_TAG_PREFIX": "dulwich.refs.LOCAL_TAG_PREFIX",
172 "SYMREF": "dulwich.refs.SYMREF",
173 "check_ref_format": "dulwich.refs.check_ref_format",
174 "read_packed_refs": "dulwich.refs.read_packed_refs",
175 "read_packed_refs_with_peeled": "dulwich.refs.read_packed_refs_with_peeled",
176 "write_packed_refs": "dulwich.refs.write_packed_refs",
177 },
178)
181CONTROLDIR = ".git"
182OBJECTDIR = "objects"
183DEFAULT_OFS_DELTA = True
185T = TypeVar("T", bound="ShaFile")
186REFSDIR = "refs"
187REFSDIR_TAGS = "tags"
188REFSDIR_HEADS = "heads"
189INDEX_FILENAME = "index"
190COMMONDIR = "commondir"
191GITDIR = "gitdir"
192WORKTREES = "worktrees"
194BASE_DIRECTORIES = [
195 ["branches"],
196 [REFSDIR],
197 [REFSDIR, REFSDIR_TAGS],
198 [REFSDIR, REFSDIR_HEADS],
199 ["hooks"],
200 ["info"],
201]
203DEFAULT_BRANCH = b"master"
206class InvalidUserIdentity(Exception):
207 """User identity is not of the format 'user <email>'."""
209 def __init__(self, identity: str) -> None:
210 """Initialize InvalidUserIdentity exception."""
211 self.identity = identity
214class DefaultIdentityNotFound(Exception):
215 """Default identity could not be determined."""
218# TODO(jelmer): Cache?
219def _get_default_identity(env: Mapping[str, str] | None = None) -> tuple[str, str]:
220 import socket
222 if env is None:
223 env = os.environ
225 for name in ("LOGNAME", "USER", "LNAME", "USERNAME"):
226 username = env.get(name)
227 if username:
228 break
229 else:
230 username = None
232 try:
233 import pwd
234 except ImportError:
235 fullname = None
236 else:
237 try:
238 entry = pwd.getpwuid(os.getuid()) # type: ignore[attr-defined,unused-ignore]
239 except KeyError:
240 fullname = None
241 else:
242 if getattr(entry, "gecos", None):
243 fullname = entry.pw_gecos.split(",")[0]
244 else:
245 fullname = None
246 if username is None:
247 username = entry.pw_name
248 if not fullname:
249 if username is None:
250 raise DefaultIdentityNotFound("no username found")
251 fullname = username
252 email = env.get("EMAIL")
253 if email is None:
254 if username is None:
255 raise DefaultIdentityNotFound("no username found")
256 email = f"{username}@{socket.gethostname()}"
257 return (fullname, email)
260def get_user_identity(config: "Config", kind: str | None = None) -> bytes:
261 """Determine the identity to use for new commits.
263 If kind is set, this first checks
264 GIT_${KIND}_NAME and GIT_${KIND}_EMAIL.
266 If those variables are not set, then it will fall back
267 to reading the user.name and user.email settings from
268 the specified configuration.
270 If that also fails, then it will fall back to using
271 the current users' identity as obtained from the host
272 system (e.g. the gecos field, $EMAIL, $USER@$(hostname -f).
274 Args:
275 config: Configuration stack to read from
276 kind: Optional kind to return identity for,
277 usually either "AUTHOR" or "COMMITTER".
279 Returns:
280 A user identity
281 """
282 user: bytes | None = None
283 email: bytes | None = None
284 if kind:
285 user_uc = os.environ.get("GIT_" + kind + "_NAME")
286 if user_uc is not None:
287 user = user_uc.encode("utf-8")
288 email_uc = os.environ.get("GIT_" + kind + "_EMAIL")
289 if email_uc is not None:
290 email = email_uc.encode("utf-8")
291 if user is None:
292 try:
293 user = config.get(("user",), "name")
294 except KeyError:
295 user = None
296 if email is None:
297 try:
298 email = config.get(("user",), "email")
299 except KeyError:
300 email = None
301 default_user, default_email = _get_default_identity()
302 if user is None:
303 user = default_user.encode("utf-8")
304 if email is None:
305 email = default_email.encode("utf-8")
306 if email.startswith(b"<") and email.endswith(b">"):
307 email = email[1:-1]
308 return user + b" <" + email + b">"
311def check_user_identity(identity: bytes) -> None:
312 """Verify that a user identity is formatted correctly.
314 Args:
315 identity: User identity bytestring
316 Raises:
317 InvalidUserIdentity: Raised when identity is invalid
318 """
319 try:
320 _fst, snd = identity.split(b" <", 1)
321 except ValueError as exc:
322 raise InvalidUserIdentity(identity.decode("utf-8", "replace")) from exc
323 if b">" not in snd:
324 raise InvalidUserIdentity(identity.decode("utf-8", "replace"))
325 if b"\0" in identity or b"\n" in identity:
326 raise InvalidUserIdentity(identity.decode("utf-8", "replace"))
329_IDENTITY_CRUD = bytes(range(33)) + b",:;<>\"\\'"
332def _strip_identity_crud(value: bytes) -> bytes:
333 r"""Strip characters using git's ``strbuf_addstr_without_crud`` rules.
335 Leading and trailing "crud" (bytes <= 32 as well as ``,:;<>"\'``) is
336 stripped, and the delimiter characters ``\n``, ``<`` and ``>`` are
337 dropped from the rest of the value.
338 """
339 # Git does not handle embedded NUL bytes here, but check_user_identity
340 # rejects them.
341 return value.strip(_IDENTITY_CRUD).translate(None, b"\0\n<>")
344def sanitize_user_identity(name: bytes, email: bytes) -> bytes:
345 r"""Build a user identity with a sanitized name and email.
347 Matches git's ``fmt_ident`` behavior: the name and the email are each
348 stripped of leading and trailing "crud" (bytes <= 32 as well as
349 ``,:;<>"\'``), and the delimiter characters ``\n``, ``<`` and ``>``
350 are dropped from the middle.
352 Unlike ``check_user_identity``, this function never rejects its input. It
353 is intended for identities outside the caller's control, such as those
354 used by an importer to build commits. The result always passes
355 ``check_user_identity``.
357 Args:
358 name: User name bytestring
359 email: Email bytestring
361 Returns:
362 Identity bytestring of the format ``name <email>``
363 """
364 return _strip_identity_crud(name) + b" <" + _strip_identity_crud(email) + b">"
367def parse_graftpoints(
368 graftpoints: Iterable[bytes],
369) -> dict[ObjectID, list[ObjectID]]:
370 """Convert a list of graftpoints into a dict.
372 Args:
373 graftpoints: Iterator of graftpoint lines
375 Each line is formatted as:
376 <commit sha1> <parent sha1> [<parent sha1>]*
378 Resulting dictionary is:
379 <commit sha1>: [<parent sha1>*]
381 https://git.wiki.kernel.org/index.php/GraftPoint
382 """
383 grafts: dict[ObjectID, list[ObjectID]] = {}
384 for line in graftpoints:
385 raw_graft = line.split(None, 1)
387 commit = ObjectID(raw_graft[0])
388 if len(raw_graft) == 2:
389 parents = [ObjectID(p) for p in raw_graft[1].split()]
390 else:
391 parents = []
393 for sha in [commit, *parents]:
394 check_hexsha(sha, "Invalid graftpoint")
396 grafts[commit] = parents
397 return grafts
400def serialize_graftpoints(graftpoints: Mapping[ObjectID, Sequence[ObjectID]]) -> bytes:
401 """Convert a dictionary of grafts into string.
403 The graft dictionary is:
404 <commit sha1>: [<parent sha1>*]
406 Each line is formatted as:
407 <commit sha1> <parent sha1> [<parent sha1>]*
409 https://git.wiki.kernel.org/index.php/GraftPoint
411 """
412 graft_lines = []
413 for commit, parents in graftpoints.items():
414 if parents:
415 graft_lines.append(commit + b" " + b" ".join(parents))
416 else:
417 graft_lines.append(commit)
418 return b"\n".join(graft_lines)
421def _set_filesystem_hidden(path: str) -> None:
422 """Mark path as to be hidden if supported by platform and filesystem.
424 On win32 uses SetFileAttributesW api:
425 <https://docs.microsoft.com/windows/desktop/api/fileapi/nf-fileapi-setfileattributesw>
426 """
427 if sys.platform == "win32":
428 import ctypes
429 from ctypes.wintypes import BOOL, DWORD, LPCWSTR
431 FILE_ATTRIBUTE_HIDDEN = 2
432 SetFileAttributesW = ctypes.WINFUNCTYPE(BOOL, LPCWSTR, DWORD)(
433 ("SetFileAttributesW", ctypes.windll.kernel32)
434 )
436 if isinstance(path, bytes):
437 path = os.fsdecode(path)
438 if not SetFileAttributesW(path, FILE_ATTRIBUTE_HIDDEN):
439 pass # Could raise or log `ctypes.WinError()` here
441 # Could implement other platform specific filesystem hiding here
444def parse_shared_repository(value: str | bytes | bool) -> "SharedPerm | None":
445 """Parse core.sharedRepository configuration value.
447 Args:
448 value: Configuration value (string, bytes, or boolean)
450 Returns:
451 SharedPerm to apply, or None to leave permissions to the umask
452 """
453 if isinstance(value, bytes):
454 value = value.decode("utf-8", errors="replace")
456 # Handle boolean values
457 if isinstance(value, bool):
458 # true = group (same as "group"), false = umask
459 return PERM_GROUP if value else None
461 # Handle string values
462 value_lower = value.lower()
464 if value_lower in ("false", "0", "", "umask"):
465 # Use umask (no adjustment)
466 return None
468 if value_lower in ("true", "1", "group"):
469 return PERM_GROUP
471 if value_lower in ("all", "world", "everybody", "2"):
472 # Others gain read, and execute on directories, but never write.
473 return PERM_EVERYBODY
475 # Try to parse as octal. Unlike the named settings, this states the mode
476 # outright rather than loosening what the umask produced.
477 if value.startswith("0"):
478 try:
479 mode = int(value, 8)
480 except ValueError:
481 pass
482 else:
483 return SharedPerm(tweak=mode, replace=True)
485 # Default to umask for unrecognized values
486 return None
489def _enable_relative_worktrees_extension(repo: "Repo") -> None:
490 """Enable the relativeworktrees extension in repository config.
492 This sets core.repositoryformatversion to 1 (if not already) and
493 enables the extensions.relativeworktrees extension.
495 Args:
496 repo: The repository to configure
497 """
498 config = repo.get_config()
500 # Ensure repository format version is at least 1
501 try:
502 version = int(config.get(("core",), "repositoryformatversion"))
503 except KeyError:
504 version = 0
506 if version < 1:
507 config.set(("core",), "repositoryformatversion", "1")
509 # Enable the relativeworktrees extension
510 config.set(("extensions",), "relativeworktrees", True)
511 config.write_to_path()
514class ParentsProvider:
515 """Provider for commit parent information."""
517 def __init__(
518 self,
519 store: "BaseObjectStore",
520 grafts: dict[ObjectID, list[ObjectID]] = {},
521 shallows: Iterable[ObjectID] = [],
522 ) -> None:
523 """Initialize ParentsProvider.
525 Args:
526 store: Object store to use
527 grafts: Graft information
528 shallows: Shallow commit SHAs
529 """
530 self.store = store
531 self.grafts = grafts
532 self.shallows = set(shallows)
534 # Get commit graph once at initialization for performance
535 self.commit_graph = store.get_commit_graph()
537 def get_parents(
538 self, commit_id: ObjectID, commit: Commit | None = None
539 ) -> list[ObjectID]:
540 """Get parents for a commit using the parents provider."""
541 try:
542 return self.grafts[commit_id]
543 except KeyError:
544 pass
545 if commit_id in self.shallows:
546 return []
548 # Try to use commit graph for faster parent lookup
549 if self.commit_graph:
550 parents = self.commit_graph.get_parents(commit_id)
551 if parents is not None:
552 return parents
554 # Fallback to reading the commit object
555 if commit is None:
556 obj = self.store[commit_id]
557 if not isinstance(obj, Commit):
558 raise ValueError(
559 f"Expected Commit object for commit_id {commit_id.decode()}, "
560 f"got {type(obj).__name__}. This usually means a reference "
561 f"points to a {type(obj).__name__} object instead of a Commit."
562 )
563 commit = obj
564 result: list[ObjectID] = commit.parents
565 return result
568class BaseRepo:
569 """Base class for a git repository.
571 This base class is meant to be used for Repository implementations that e.g.
572 work on top of a different transport than a standard filesystem path.
574 Attributes:
575 object_store: Dictionary-like object for accessing
576 the objects
577 refs: Dictionary-like object with the refs in this
578 repository
579 """
581 def __init__(
582 self,
583 object_store: "PackCapableObjectStore",
584 refs: RefsContainer,
585 object_format: "ObjectFormat | None" = None,
586 ) -> None:
587 """Open a repository.
589 This shouldn't be called directly, but rather through one of the
590 base classes, such as MemoryRepo or Repo.
592 Args:
593 object_store: Object store to use
594 refs: Refs container to use
595 object_format: Hash algorithm to use (if None, will use object_store's format)
596 """
597 self.object_store = object_store
598 self.refs = refs
600 self._graftpoints: dict[ObjectID, list[ObjectID]] = {}
601 self.hooks: dict[str, Hook] = {}
602 if object_format is None:
603 self.object_format: ObjectFormat = object_store.object_format
604 else:
605 self.object_format = object_format
607 def _determine_file_mode(self) -> bool:
608 """Probe the file-system to determine whether permissions can be trusted.
610 Returns: True if permissions can be trusted, False otherwise.
611 """
612 raise NotImplementedError(self._determine_file_mode)
614 def _determine_symlinks(self) -> bool:
615 """Probe the filesystem to determine whether symlinks can be created.
617 Returns: True if symlinks can be created, False otherwise.
618 """
619 # For now, just mimic the old behaviour
620 return sys.platform != "win32"
622 def _init_files(
623 self,
624 bare: bool,
625 symlinks: bool | None = None,
626 format: int | None = None,
627 shared_repository: str | bool | None = None,
628 object_format: str | None = None,
629 ) -> None:
630 """Initialize a default set of named files."""
631 from .config import ConfigFile
633 self._put_named_file("description", b"Unnamed repository")
634 f = BytesIO()
635 cf = ConfigFile()
637 # Determine the appropriate format version
638 if object_format == "sha256":
639 # SHA256 requires format version 1
640 if format is None:
641 format = 1
642 elif format != 1:
643 raise ValueError(
644 "SHA256 object format requires repository format version 1"
645 )
646 else:
647 # SHA1 (default) can use format 0 or 1
648 if format is None:
649 format = 0
651 if format not in (0, 1):
652 raise ValueError(f"Unsupported repository format version: {format}")
654 cf.set("core", "repositoryformatversion", str(format))
656 # Set object format extension if using SHA256
657 if object_format == "sha256":
658 cf.set("extensions", "objectformat", "sha256")
660 # Set hash algorithm based on object format
661 from .object_format import get_object_format
663 self.object_format = get_object_format(object_format)
665 if self._determine_file_mode():
666 cf.set("core", "filemode", True)
667 else:
668 cf.set("core", "filemode", False)
670 if symlinks is None and not bare:
671 symlinks = self._determine_symlinks()
673 if symlinks is False:
674 cf.set("core", "symlinks", symlinks)
676 # On macOS, set precomposeunicode to true since HFS+/APFS
677 # returns filenames in NFD (decomposed) Unicode form
678 if sys.platform == "darwin":
679 cf.set("core", "precomposeunicode", True)
681 cf.set("core", "bare", bare)
682 cf.set("core", "logallrefupdates", True)
684 # Set shared repository if specified
685 if shared_repository is not None:
686 if isinstance(shared_repository, bool):
687 cf.set("core", "sharedRepository", shared_repository)
688 else:
689 cf.set("core", "sharedRepository", shared_repository)
691 cf.write_to_file(f)
692 self._put_named_file("config", f.getvalue())
693 self._put_named_file(os.path.join("info", "exclude"), b"")
695 # Allow subclasses to handle config initialization
696 self._init_config(cf)
698 def _init_config(self, config: "ConfigFile") -> None:
699 """Initialize repository configuration.
701 This method can be overridden by subclasses to handle config initialization.
703 Args:
704 config: The ConfigFile object that was just created
705 """
706 # Default implementation does nothing
708 def get_named_file(self, path: str) -> BinaryIO | None:
709 """Get a file from the control dir with a specific name.
711 Although the filename should be interpreted as a filename relative to
712 the control dir in a disk-based Repo, the object returned need not be
713 pointing to a file in that location.
715 Args:
716 path: The path to the file, relative to the control dir.
717 Returns: An open file object, or None if the file does not exist.
718 """
719 raise NotImplementedError(self.get_named_file)
721 def _put_named_file(self, path: str, contents: bytes) -> None:
722 """Write a file to the control dir with the given name and contents.
724 Args:
725 path: The path to the file, relative to the control dir.
726 contents: A string to write to the file.
727 """
728 raise NotImplementedError(self._put_named_file)
730 def _del_named_file(self, path: str) -> None:
731 """Delete a file in the control directory with the given name."""
732 raise NotImplementedError(self._del_named_file)
734 def open_index(self, config: "Config | None" = None) -> "Index":
735 """Open the index for this repository.
737 Args:
738 config: Configuration to consult for index settings. If None,
739 implementations may fall back to ``self.get_config_stack()``.
741 Raises:
742 NoIndexPresent: If no index is present
743 Returns: The matching `Index`
744 """
745 raise NotImplementedError(self.open_index)
747 def _change_object_format(self, object_format_name: str) -> None:
748 """Change the object format of this repository.
750 This can only be done if the object store is empty (no objects written yet).
752 Args:
753 object_format_name: Name of the new object format (e.g., "sha1", "sha256")
755 Raises:
756 AssertionError: If the object store is not empty
757 """
758 # Check if object store has any objects
759 for _ in self.object_store:
760 raise AssertionError(
761 "Cannot change object format: repository already contains objects"
762 )
764 # Update the object format
765 from .object_format import get_object_format
767 new_format = get_object_format(object_format_name)
768 self.object_format = new_format
769 self.object_store.object_format = new_format
771 # Update config file
772 config = self.get_config()
774 if object_format_name == "sha1":
775 # For SHA-1, explicitly remove objectformat extension if present
776 try:
777 config.remove("extensions", "objectformat")
778 except KeyError:
779 pass
780 else:
781 # For non-SHA-1 formats, set repositoryformatversion to 1 and objectformat extension
782 config.set("core", "repositoryformatversion", "1")
783 config.set("extensions", "objectformat", object_format_name)
785 config.write_to_path()
787 def fetch(
788 self,
789 target: "BaseRepo",
790 determine_wants: Callable[[Mapping[Ref, ObjectID], int | None], list[ObjectID]]
791 | None = None,
792 progress: Callable[..., None] | None = None,
793 depth: int | None = None,
794 ) -> dict[Ref, ObjectID]:
795 """Fetch objects into another repository.
797 Args:
798 target: The target repository
799 determine_wants: Optional function to determine what refs to
800 fetch.
801 progress: Optional progress function
802 depth: Optional shallow fetch depth
803 Returns: The local refs
804 """
805 # Fix object format if needed
806 if self.object_format != target.object_format:
807 # Change the target repo's format if it's empty
808 target._change_object_format(self.object_format.name)
810 if determine_wants is None:
811 determine_wants = target.object_store.determine_wants_all
812 count, pack_data = self.fetch_pack_data(
813 determine_wants,
814 target.get_graph_walker(),
815 progress=progress,
816 depth=depth,
817 )
818 target.object_store.add_pack_data(count, pack_data, progress)
819 return self.get_refs()
821 def fetch_pack_data(
822 self,
823 determine_wants: Callable[[Mapping[Ref, ObjectID], int | None], list[ObjectID]],
824 graph_walker: "GraphWalker",
825 progress: Callable[[bytes], None] | None,
826 *,
827 get_tagged: Callable[[], dict[ObjectID, ObjectID]] | None = None,
828 depth: int | None = None,
829 ) -> tuple[int, Iterator["UnpackedObject"]]:
830 """Fetch the pack data required for a set of revisions.
832 Args:
833 determine_wants: Function that takes a dictionary with heads
834 and returns the list of heads to fetch.
835 graph_walker: Object that can iterate over the list of revisions
836 to fetch and has an "ack" method that will be called to acknowledge
837 that a revision is present.
838 progress: Simple progress function that will be called with
839 updated progress strings.
840 get_tagged: Function that returns a dict of pointed-to sha ->
841 tag sha for including tags.
842 depth: Shallow fetch depth
843 Returns: count and iterator over pack data
844 """
845 missing_objects = self.find_missing_objects(
846 determine_wants, graph_walker, progress, get_tagged=get_tagged, depth=depth
847 )
848 if missing_objects is None:
849 return 0, iter([])
850 remote_has = missing_objects.get_remote_has()
851 object_ids = list(missing_objects)
852 return len(object_ids), generate_unpacked_objects(
853 self.object_store, object_ids, progress=progress, other_haves=remote_has
854 )
856 def find_missing_objects(
857 self,
858 determine_wants: Callable[[Mapping[Ref, ObjectID], int | None], list[ObjectID]],
859 graph_walker: "GraphWalker",
860 progress: Callable[[bytes], None] | None,
861 *,
862 get_tagged: Callable[[], dict[ObjectID, ObjectID]] | None = None,
863 depth: int | None = None,
864 ) -> MissingObjectFinder | None:
865 """Fetch the missing objects required for a set of revisions.
867 Args:
868 determine_wants: Function that takes a dictionary with heads
869 and returns the list of heads to fetch.
870 graph_walker: Object that can iterate over the list of revisions
871 to fetch and has an "ack" method that will be called to acknowledge
872 that a revision is present.
873 progress: Simple progress function that will be called with
874 updated progress strings.
875 get_tagged: Function that returns a dict of pointed-to sha ->
876 tag sha for including tags.
877 depth: Shallow fetch depth
878 Returns: iterator over objects, with __len__ implemented
879 """
880 # Filter out refs pointing to missing objects to avoid errors downstream.
881 # This makes Dulwich more robust when dealing with broken refs on disk.
882 # Previously serialize_refs() did this filtering as a side-effect.
883 all_refs = self.get_refs()
884 refs: dict[Ref, ObjectID] = {}
885 for ref, sha in all_refs.items():
886 if sha in self.object_store:
887 refs[ref] = sha
888 else:
889 logger.warning(
890 "ref %s points at non-present sha %s",
891 ref.decode("utf-8", "replace"),
892 sha.decode("ascii"),
893 )
895 wants = determine_wants(refs, depth)
896 if not isinstance(wants, list):
897 raise TypeError("determine_wants() did not return a list")
899 current_shallow = set(getattr(graph_walker, "shallow", set()))
901 unshallow: set[ObjectID] = set()
902 if depth not in (None, 0):
903 assert depth is not None
904 shallow, not_shallow = find_shallow(self.object_store, wants, depth)
905 # Only update if graph_walker has shallow attribute
906 walker_shallow: set[ObjectID] | None = getattr(
907 graph_walker, "shallow", None
908 )
909 if walker_shallow is not None:
910 unshallow = not_shallow & current_shallow
911 # Commits that were a shallow boundary but are now being
912 # deepened past must drop out of the boundary, otherwise their
913 # parents stay unreachable and never get transferred.
914 walker_shallow.update(shallow - not_shallow)
915 walker_shallow.difference_update(unshallow)
916 new_shallow = walker_shallow - current_shallow
917 setattr(graph_walker, "unshallow", unshallow)
918 update_shallow = getattr(graph_walker, "update_shallow", None)
919 if update_shallow is not None:
920 update_shallow(new_shallow, unshallow)
921 else:
922 unshallow = getattr(graph_walker, "unshallow", set())
924 if wants == []:
925 # TODO(dborowitz): find a way to short-circuit that doesn't change
926 # this interface.
928 if getattr(graph_walker, "shallow", set()) or unshallow:
929 # Do not send a pack in shallow short-circuit path
930 return None
932 # Return an actual MissingObjectFinder with empty wants
933 return MissingObjectFinder(
934 self.object_store,
935 haves=[],
936 wants=[],
937 )
939 # If the graph walker is set up with an implementation that can
940 # ACK/NAK to the wire, it will write data to the client through
941 # this call as a side-effect.
942 haves = self.object_store.find_common_revisions(graph_walker)
944 # Deal with shallow requests separately because the haves do
945 # not reflect what objects are missing
946 if getattr(graph_walker, "shallow", set()) or unshallow:
947 # TODO: filter the haves commits from iter_shas. the specific
948 # commits aren't missing.
949 haves = []
951 parents_provider = ParentsProvider(
952 self.object_store,
953 shallows=getattr(graph_walker, "shallow", current_shallow),
954 )
956 def get_parents(commit: Commit) -> list[ObjectID]:
957 """Get parents for a commit using the parents provider.
959 Args:
960 commit: Commit object
962 Returns:
963 List of parent commit SHAs
964 """
965 return parents_provider.get_parents(commit.id, commit)
967 return MissingObjectFinder(
968 self.object_store,
969 haves=haves,
970 wants=wants,
971 shallow=getattr(graph_walker, "shallow", set()),
972 progress=progress,
973 get_tagged=get_tagged,
974 get_parents=get_parents,
975 )
977 def generate_pack_data(
978 self,
979 have: set[ObjectID],
980 want: set[ObjectID],
981 *,
982 shallow: set[ObjectID] | None = None,
983 progress: Callable[[str], None] | None = None,
984 ofs_delta: bool | None = None,
985 ) -> tuple[int, Iterator["UnpackedObject"]]:
986 """Generate pack data objects for a set of wants/haves.
988 Args:
989 have: List of SHA1s of objects that should not be sent
990 want: List of SHA1s of objects that should be sent
991 shallow: Set of shallow commit SHA1s to skip (defaults to repo's shallow commits)
992 ofs_delta: Whether OFS deltas can be included
993 progress: Optional progress reporting method
994 """
995 if shallow is None:
996 shallow = self.get_shallow()
997 return self.object_store.generate_pack_data(
998 have,
999 want,
1000 shallow=shallow,
1001 progress=progress,
1002 ofs_delta=ofs_delta if ofs_delta is not None else DEFAULT_OFS_DELTA,
1003 )
1005 def get_graph_walker(
1006 self, heads: list[ObjectID] | None = None
1007 ) -> ObjectStoreGraphWalker:
1008 """Retrieve a graph walker.
1010 A graph walker is used by a remote repository (or proxy)
1011 to find out which objects are present in this repository.
1013 Args:
1014 heads: Repository heads to use (optional)
1015 Returns: A graph walker object
1016 """
1017 if heads is None:
1018 heads = [
1019 sha
1020 for sha in self.refs.as_dict(Ref(b"refs/heads")).values()
1021 if sha in self.object_store
1022 ]
1023 parents_provider = ParentsProvider(self.object_store)
1024 return ObjectStoreGraphWalker(
1025 heads,
1026 parents_provider.get_parents,
1027 shallow=self.get_shallow(),
1028 update_shallow=self.update_shallow,
1029 )
1031 def get_refs(self) -> dict[Ref, ObjectID]:
1032 """Get dictionary with all refs.
1034 Returns: A ``dict`` mapping ref names to SHA1s
1035 """
1036 return self.refs.as_dict()
1038 def head(self) -> ObjectID:
1039 """Return the SHA1 pointed at by HEAD."""
1040 # TODO: move this method to WorkTree
1041 return self.refs[HEADREF]
1043 def _get_object(self, sha: ObjectID | RawObjectID, cls: type[T]) -> T:
1044 assert len(sha) in (
1045 self.object_format.oid_length,
1046 self.object_format.hex_length,
1047 )
1048 ret = self.get_object(sha)
1049 if not isinstance(ret, cls):
1050 if cls is Commit:
1051 raise NotCommitError(ret.id)
1052 elif cls is Blob:
1053 raise NotBlobError(ret.id)
1054 elif cls is Tree:
1055 raise NotTreeError(ret.id)
1056 elif cls is Tag:
1057 raise NotTagError(ret.id)
1058 else:
1059 raise Exception(f"Type invalid: {ret.type_name!r} != {cls.type_name!r}")
1060 return ret
1062 def get_object(self, sha: ObjectID | RawObjectID) -> ShaFile:
1063 """Retrieve the object with the specified SHA.
1065 Args:
1066 sha: SHA to retrieve
1067 Returns: A ShaFile object
1068 Raises:
1069 KeyError: when the object can not be found
1070 """
1071 return self.object_store[sha]
1073 def parents_provider(self) -> ParentsProvider:
1074 """Get a parents provider for this repository.
1076 Returns:
1077 ParentsProvider instance configured with grafts and shallows
1078 """
1079 return ParentsProvider(
1080 self.object_store,
1081 grafts=self._graftpoints,
1082 shallows=self.get_shallow(),
1083 )
1085 def get_parents(
1086 self, sha: ObjectID, commit: Commit | None = None
1087 ) -> list[ObjectID]:
1088 """Retrieve the parents of a specific commit.
1090 If the specific commit is a graftpoint, the graft parents
1091 will be returned instead.
1093 Args:
1094 sha: SHA of the commit for which to retrieve the parents
1095 commit: Optional commit matching the sha
1096 Returns: List of parents
1097 """
1098 return self.parents_provider().get_parents(sha, commit)
1100 def get_config(self) -> "ConfigFile":
1101 """Retrieve the config object.
1103 Returns: `ConfigFile` object for the ``.git/config`` file.
1104 """
1105 raise NotImplementedError(self.get_config)
1107 def get_worktree_config(self) -> "ConfigFile":
1108 """Retrieve the worktree config object."""
1109 raise NotImplementedError(self.get_worktree_config)
1111 def get_description(self) -> bytes | None:
1112 """Retrieve the description for this repository.
1114 Returns: Bytes with the description of the repository
1115 as set by the user.
1116 """
1117 raise NotImplementedError(self.get_description)
1119 def set_description(self, description: bytes) -> None:
1120 """Set the description for this repository.
1122 Args:
1123 description: Text to set as description for this repository.
1124 """
1125 raise NotImplementedError(self.set_description)
1127 def get_rebase_state_manager(self) -> "RebaseStateManager":
1128 """Get the appropriate rebase state manager for this repository.
1130 Returns: RebaseStateManager instance
1131 """
1132 raise NotImplementedError(self.get_rebase_state_manager)
1134 def get_blob_normalizer(
1135 self, config: "Config | None" = None
1136 ) -> "FilterBlobNormalizer":
1137 """Return a BlobNormalizer object for checkin/checkout operations.
1139 Args:
1140 config: Configuration to consult for filter setup. If None,
1141 implementations may fall back to ``self.get_config_stack()``.
1143 Returns: BlobNormalizer instance
1144 """
1145 raise NotImplementedError(self.get_blob_normalizer)
1147 def get_gitattributes(self, tree: bytes | None = None) -> "GitAttributes":
1148 """Read gitattributes for the repository.
1150 Args:
1151 tree: Tree SHA to read .gitattributes from (defaults to HEAD)
1153 Returns:
1154 GitAttributes object that can be used to match paths
1155 """
1156 raise NotImplementedError(self.get_gitattributes)
1158 def get_config_stack(self) -> "StackedConfig":
1159 """Return a config stack for this repository.
1161 This stack accesses the configuration for both this repository
1162 itself (.git/config) and the global configuration, which usually
1163 lives in ~/.gitconfig.
1165 Returns: `Config` instance for this repository
1166 """
1167 from .config import ConfigFile, StackedConfig
1169 local_config = self.get_config()
1170 backends: list[ConfigFile] = [local_config]
1171 if local_config.get_boolean((b"extensions",), b"worktreeconfig", False):
1172 backends.append(self.get_worktree_config())
1174 backends += StackedConfig.default_backends()
1175 return StackedConfig(backends, writable=local_config)
1177 def get_shallow(self) -> set[ObjectID]:
1178 """Get the set of shallow commits.
1180 Returns: Set of shallow commits.
1181 """
1182 f = self.get_named_file("shallow")
1183 if f is None:
1184 return set()
1185 with f:
1186 shallow: set[ObjectID] = set()
1187 for line in f:
1188 sha = line.strip()
1189 if not sha:
1190 continue
1191 check_hexsha(sha, "invalid shallow object id")
1192 shallow.add(ObjectID(sha))
1193 return shallow
1195 def update_shallow(
1196 self, new_shallow: set[ObjectID] | None, new_unshallow: set[ObjectID] | None
1197 ) -> None:
1198 """Update the list of shallow objects.
1200 Args:
1201 new_shallow: Newly shallow objects
1202 new_unshallow: Newly no longer shallow objects
1203 """
1204 for sha in (*(new_shallow or ()), *(new_unshallow or ())):
1205 check_hexsha(sha, "invalid shallow object id")
1206 shallow = self.get_shallow()
1207 if new_shallow:
1208 shallow.update(new_shallow)
1209 if new_unshallow:
1210 shallow.difference_update(new_unshallow)
1211 if shallow:
1212 self._put_named_file("shallow", b"".join([sha + b"\n" for sha in shallow]))
1213 else:
1214 self._del_named_file("shallow")
1216 def get_peeled(self, ref: Ref) -> ObjectID:
1217 """Get the peeled value of a ref.
1219 Args:
1220 ref: The refname to peel.
1221 Returns: The fully-peeled SHA1 of a tag object, after peeling all
1222 intermediate tags; if the original ref does not point to a tag,
1223 this will equal the original SHA1.
1224 """
1225 cached = self.refs.get_peeled(ref)
1226 if cached is not None:
1227 return cached
1228 return self.object_store.peel(self.refs[ref])[1].id
1230 @property
1231 def notes(self) -> "Notes":
1232 """Access notes functionality for this repository.
1234 Returns:
1235 Notes object for accessing notes
1236 """
1237 from .notes import Notes
1239 return Notes(self.object_store, self.refs)
1241 def get_walker(
1242 self,
1243 include: Sequence[ObjectID] | None = None,
1244 exclude: Sequence[ObjectID] | None = None,
1245 order: str = "date",
1246 reverse: bool = False,
1247 max_entries: int | None = None,
1248 paths: Sequence[bytes] | None = None,
1249 rename_detector: "RenameDetector | None" = None,
1250 follow: bool = False,
1251 since: int | None = None,
1252 until: int | None = None,
1253 queue_cls: type | None = None,
1254 simplify_history: bool = False,
1255 ) -> "Walker":
1256 """Obtain a walker for this repository.
1258 Args:
1259 include: Iterable of SHAs of commits to include along with their
1260 ancestors. Defaults to [HEAD]
1261 exclude: Iterable of SHAs of commits to exclude along with their
1262 ancestors, overriding includes.
1263 order: ORDER_* constant specifying the order of results.
1264 Anything other than ORDER_DATE may result in O(n) memory usage.
1265 reverse: If True, reverse the order of output, requiring O(n)
1266 memory.
1267 max_entries: The maximum number of entries to yield, or None for
1268 no limit.
1269 paths: Iterable of file or subtree paths to show entries for.
1270 rename_detector: diff.RenameDetector object for detecting
1271 renames.
1272 follow: If True, follow path across renames/copies. Forces a
1273 default rename_detector.
1274 since: Timestamp to list commits after.
1275 until: Timestamp to list commits before.
1276 queue_cls: A class to use for a queue of commits, supporting the
1277 iterator protocol. The constructor takes a single argument, the Walker.
1278 simplify_history: If True and paths is set, perform git-style
1279 history simplification: at a merge commit that is TREESAME to
1280 at least one parent for the requested paths, follow only that
1281 parent (matches ``git log <path>``). The default of False
1282 matches ``git log --full-history <path>``.
1284 Returns: A `Walker` object
1285 """
1286 from .walk import Walker, _CommitTimeQueue
1288 if include is None:
1289 include = [self.head()]
1291 # Pass all arguments to Walker explicitly to avoid type issues with **kwargs
1292 return Walker(
1293 self.object_store,
1294 include,
1295 exclude=exclude,
1296 order=order,
1297 reverse=reverse,
1298 max_entries=max_entries,
1299 paths=paths,
1300 rename_detector=rename_detector,
1301 follow=follow,
1302 since=since,
1303 until=until,
1304 get_parents=lambda commit: self.get_parents(commit.id, commit),
1305 queue_cls=queue_cls if queue_cls is not None else _CommitTimeQueue,
1306 simplify_history=simplify_history,
1307 )
1309 def __getitem__(self, name: ObjectID | Ref | bytes) -> "ShaFile":
1310 """Retrieve a Git object by SHA1 or ref.
1312 Args:
1313 name: A Git object SHA1 or a ref name
1314 Returns: A `ShaFile` object, such as a Commit or Blob
1315 Raises:
1316 KeyError: when the specified ref or object does not exist
1317 """
1318 if not isinstance(name, bytes):
1319 raise TypeError(f"'name' must be bytestring, not {type(name).__name__:.80}")
1320 # If it looks like a ref name, only try refs
1321 if name == b"HEAD" or name.startswith(b"refs/"):
1322 try:
1323 return self.object_store[self.refs[Ref(name)]]
1324 except (RefFormatError, KeyError):
1325 pass
1326 # Otherwise, try as object ID if length matches
1327 if len(name) in (
1328 self.object_store.object_format.oid_length,
1329 self.object_store.object_format.hex_length,
1330 ):
1331 try:
1332 return self.object_store[
1333 ObjectID(name)
1334 if len(name) == self.object_store.object_format.hex_length
1335 else RawObjectID(name)
1336 ]
1337 except (KeyError, ValueError):
1338 pass
1339 # If nothing worked, raise KeyError
1340 raise KeyError(name)
1342 def __contains__(self, name: bytes) -> bool:
1343 """Check if a specific Git object or ref is present.
1345 Args:
1346 name: Git object SHA1/SHA256 or ref name
1347 """
1348 # Check if it's a binary or hex SHA
1349 if len(name) == self.object_format.hex_length and valid_hexsha(name):
1350 return ObjectID(name) in self.object_store or Ref(name) in self.refs
1351 return Ref(name) in self.refs
1353 def __setitem__(self, name: bytes, value: ShaFile | bytes) -> None:
1354 """Set a ref.
1356 Args:
1357 name: ref name
1358 value: Ref value - either a ShaFile object, or a hex sha
1359 """
1360 if name.startswith(b"refs/") or name == HEADREF:
1361 ref_name = Ref(name)
1362 if isinstance(value, ShaFile):
1363 self.refs[ref_name] = value.id
1364 elif isinstance(value, bytes):
1365 self.refs[ref_name] = ObjectID(value)
1366 else:
1367 raise TypeError(value)
1368 else:
1369 raise ValueError(name)
1371 def __delitem__(self, name: bytes) -> None:
1372 """Remove a ref.
1374 Args:
1375 name: Name of the ref to remove
1376 """
1377 if name.startswith(b"refs/") or name == HEADREF:
1378 del self.refs[Ref(name)]
1379 else:
1380 raise ValueError(name)
1382 def _get_user_identity(
1383 self, config: "StackedConfig", kind: str | None = None
1384 ) -> bytes:
1385 """Determine the identity to use for new commits."""
1386 warnings.warn(
1387 "use get_user_identity() rather than Repo._get_user_identity",
1388 DeprecationWarning,
1389 )
1390 return get_user_identity(config)
1392 def _add_graftpoints(
1393 self, updated_graftpoints: dict[ObjectID, list[ObjectID]]
1394 ) -> None:
1395 """Add or modify graftpoints.
1397 Args:
1398 updated_graftpoints: Dict of commit shas to list of parent shas
1399 """
1400 # Simple validation
1401 for commit, parents in updated_graftpoints.items():
1402 for sha in [commit, *parents]:
1403 check_hexsha(sha, "Invalid graftpoint")
1405 self._graftpoints.update(updated_graftpoints)
1407 def _remove_graftpoints(self, to_remove: Sequence[ObjectID] = ()) -> None:
1408 """Remove graftpoints.
1410 Args:
1411 to_remove: List of commit shas
1412 """
1413 for sha in to_remove:
1414 del self._graftpoints[sha]
1416 def _read_heads(self, name: str) -> list[ObjectID]:
1417 f = self.get_named_file(name)
1418 if f is None:
1419 return []
1420 with f:
1421 return [ObjectID(line.strip()) for line in f.readlines() if line.strip()]
1423 def get_worktree(self) -> "WorkTree":
1424 """Get the working tree for this repository.
1426 Returns:
1427 WorkTree instance for performing working tree operations
1429 Raises:
1430 NotImplementedError: If the repository doesn't support working trees
1431 """
1432 raise NotImplementedError(
1433 "Working tree operations not supported by this repository type"
1434 )
1437def read_gitfile(f: BinaryIO) -> str:
1438 """Read a ``.git`` file.
1440 The first line of the file should start with "gitdir: "
1442 Args:
1443 f: File-like object to read from
1444 Returns: A path
1445 """
1446 cs = f.read()
1447 if not cs.startswith(b"gitdir: "):
1448 raise ValueError("Expected file to start with 'gitdir: '")
1449 return cs[len(b"gitdir: ") :].rstrip(b"\r\n").decode("utf-8")
1452class UnsupportedVersion(Exception):
1453 """Unsupported repository version."""
1455 def __init__(self, version: int) -> None:
1456 """Initialize UnsupportedVersion exception.
1458 Args:
1459 version: The unsupported repository version
1460 """
1461 self.version = version
1464class UnsupportedExtension(Exception):
1465 """Unsupported repository extension."""
1467 def __init__(self, extension: str) -> None:
1468 """Initialize UnsupportedExtension exception.
1470 Args:
1471 extension: The unsupported repository extension
1472 """
1473 self.extension = extension
1476class InvalidWorktreeConfiguration(Exception):
1477 """core.worktree is set on a bare repository."""
1480class Repo(BaseRepo):
1481 """A git repository backed by local disk.
1483 To open an existing repository, call the constructor with
1484 the path of the repository.
1486 To create a new repository, use the Repo.init class method.
1488 Note that a repository object may hold on to resources such
1489 as file handles for performance reasons; call .close() to free
1490 up those resources.
1492 Attributes:
1493 path: Path to the working copy (if it exists) or repository control
1494 directory (if the repository is bare). ``core.worktree`` overrides
1495 this, so it is not necessarily the parent of the control directory.
1496 bare: Whether this is a bare repository
1497 """
1499 path: str
1500 bare: bool
1501 object_store: DiskObjectStore
1502 filter_context: "FilterContext | None"
1503 _index_file_override: "str | None"
1505 def __init__(
1506 self,
1507 root: str | bytes | os.PathLike[str] | None = None,
1508 object_store: PackBasedObjectStore | None = None,
1509 bare: bool | None = None,
1510 *,
1511 controldir: str | bytes | os.PathLike[str] | None = None,
1512 commondir: str | bytes | os.PathLike[str] | None = None,
1513 worktree: str | bytes | os.PathLike[str] | None = None,
1514 object_directory: str | bytes | os.PathLike[str] | None = None,
1515 alternates: "Iterable[str | os.PathLike[str]] | None" = None,
1516 index_file: str | bytes | os.PathLike[str] | None = None,
1517 ) -> None:
1518 """Open a repository on disk.
1520 Args:
1521 root: Path to the repository's root. Optional if ``controldir`` is
1522 given, in which case ``self.path`` defaults to ``worktree`` when
1523 provided and to ``controldir`` otherwise. ``core.worktree`` in
1524 config may still override it.
1525 object_store: ObjectStore to use; if omitted, we use the
1526 repository's default object store
1527 bare: True if this is a bare repository.
1528 controldir: Explicit path to the control directory (analogous to
1529 ``GIT_DIR``). When set, discovery based on ``root`` is skipped.
1530 commondir: Explicit path to the common directory (analogous to
1531 ``GIT_COMMON_DIR``). Relative paths are resolved against the
1532 control directory.
1533 worktree: Explicit worktree path (analogous to ``GIT_WORK_TREE``).
1534 Forces the repository to be treated as non-bare.
1535 object_directory: Explicit path to the primary object store
1536 (analogous to ``GIT_OBJECT_DIRECTORY``). Overrides
1537 ``<common_dir>/objects``.
1538 alternates: Extra alternate object directory paths (analogous to
1539 ``GIT_ALTERNATE_OBJECT_DIRECTORIES``). Appended to any listed
1540 in ``objects/info/alternates``.
1541 index_file: Path to the index file (analogous to
1542 ``GIT_INDEX_FILE``). Overrides ``<controldir>/index``.
1543 """
1544 if controldir is None:
1545 if root is None:
1546 raise TypeError("Repo() requires either root or controldir")
1547 root = os.fspath(root)
1548 if isinstance(root, bytes):
1549 root = os.fsdecode(root)
1550 hidden_path = os.path.join(root, CONTROLDIR)
1551 if worktree is not None:
1552 bare = False
1553 if bare is None:
1554 if os.path.isfile(hidden_path) or os.path.isdir(
1555 os.path.join(hidden_path, OBJECTDIR)
1556 ):
1557 bare = False
1558 elif os.path.isdir(os.path.join(root, OBJECTDIR)) and os.path.isdir(
1559 os.path.join(root, REFSDIR)
1560 ):
1561 bare = True
1562 else:
1563 raise NotGitRepository(
1564 "No git repository was found at {path}".format(
1565 **dict(path=root)
1566 )
1567 )
1569 self.bare = bare
1570 if bare is False:
1571 if os.path.isfile(hidden_path):
1572 with open(hidden_path, "rb") as f:
1573 gitfile_path = read_gitfile(f)
1574 self._controldir = os.path.join(root, gitfile_path)
1575 else:
1576 self._controldir = hidden_path
1577 else:
1578 self._controldir = root
1579 else:
1580 controldir = os.fspath(controldir)
1581 if isinstance(controldir, bytes):
1582 controldir = os.fsdecode(controldir)
1583 self._controldir = controldir
1584 if root is None:
1585 # Without an explicit root, default to the worktree if given,
1586 # else to the control dir (bare layout). We deliberately do
1587 # not fall back to os.getcwd() here: a library-level ``Repo``
1588 # should not silently latch onto the current directory.
1589 # Callers that want git's ``GIT_DIR``-implies-cwd-worktree
1590 # semantics pass ``worktree`` explicitly.
1591 root = os.fspath(worktree) if worktree is not None else controldir
1592 else:
1593 root = os.fspath(root)
1594 if isinstance(root, bytes):
1595 root = os.fsdecode(root)
1596 if worktree is not None:
1597 bare = False
1598 if bare is None:
1599 # With an explicit control dir and no worktree override,
1600 # default to bare. core.bare / core.worktree parsed from
1601 # config below may still flip this.
1602 bare = True
1603 self.bare = bare
1604 if commondir is not None:
1605 commondir_path = os.fspath(commondir)
1606 if isinstance(commondir_path, bytes):
1607 commondir_path = os.fsdecode(commondir_path)
1608 if not os.path.isabs(commondir_path):
1609 commondir_path = os.path.join(self._controldir, commondir_path)
1610 self._commondir = commondir_path
1611 else:
1612 commondir_file = self.get_named_file(COMMONDIR)
1613 if commondir_file is not None:
1614 with commondir_file:
1615 self._commondir = os.path.join(
1616 self.controldir(),
1617 os.fsdecode(commondir_file.read().rstrip(b"\r\n")),
1618 )
1619 else:
1620 self._commondir = self._controldir
1621 self.path = root
1622 if worktree is not None:
1623 worktree_path = os.fspath(worktree)
1624 if isinstance(worktree_path, bytes):
1625 worktree_path = os.fsdecode(worktree_path)
1626 self.path = worktree_path
1628 if index_file is not None:
1629 index_file_str = os.fspath(index_file)
1630 if isinstance(index_file_str, bytes):
1631 index_file_str = os.fsdecode(index_file_str)
1632 self._index_file_override = index_file_str
1633 else:
1634 self._index_file_override = None
1636 # Initialize refs early so they're available for config condition matchers
1637 self.refs = DiskRefsContainer(
1638 self.commondir(), self._controldir, logger=self._write_reflog
1639 )
1641 # Initialize worktrees container
1642 from .worktree import WorkTreeContainer
1644 self.worktrees = WorkTreeContainer(self)
1646 config = self.get_config()
1647 try:
1648 repository_format_version = config.get("core", "repositoryformatversion")
1649 format_version = (
1650 0
1651 if repository_format_version is None
1652 else int(repository_format_version)
1653 )
1654 except KeyError:
1655 format_version = 0
1657 if format_version not in (0, 1):
1658 raise UnsupportedVersion(format_version)
1660 try:
1661 configured_worktree = config.get((b"core",), b"worktree")
1662 except KeyError:
1663 configured_worktree = None
1664 if configured_worktree is not None:
1665 # A repository is bare because core.bare says so, not because it
1666 # was opened at its control directory: a submodule or a
1667 # --separate-git-dir repository is opened that way but still has a
1668 # working tree, named here.
1669 if config.get_boolean((b"core",), b"bare", False):
1670 raise InvalidWorktreeConfiguration(
1671 "core.bare and core.worktree are incompatible"
1672 )
1673 self.bare = False
1674 # An explicit worktree argument takes precedence over core.worktree.
1675 if worktree is None:
1676 # Relative paths are resolved against the control directory,
1677 # not against the current directory or the repository root.
1678 self.path = os.path.join(
1679 self._controldir, os.fsdecode(configured_worktree)
1680 )
1682 # Track extensions we encounter
1683 has_reftable_extension = False
1684 for extension, value in config.items((b"extensions",)):
1685 if extension.lower() == b"refstorage":
1686 if value == b"reftable":
1687 has_reftable_extension = True
1688 else:
1689 raise UnsupportedExtension(f"refStorage = {value.decode()}")
1690 elif extension.lower() not in (
1691 b"worktreeconfig",
1692 b"objectformat",
1693 b"relativeworktrees",
1694 ):
1695 raise UnsupportedExtension(extension.decode("utf-8"))
1697 if object_store is None:
1698 # Get shared repository permissions from config
1699 try:
1700 shared_value = config.get(("core",), "sharedRepository")
1701 shared_perm = parse_shared_repository(shared_value)
1702 except KeyError:
1703 shared_perm = None
1705 if object_directory is not None:
1706 object_dir_path = os.fspath(object_directory)
1707 if isinstance(object_dir_path, bytes):
1708 object_dir_path = os.fsdecode(object_dir_path)
1709 else:
1710 object_dir_path = os.path.join(self.commondir(), OBJECTDIR)
1711 object_store = DiskObjectStore.from_config(
1712 object_dir_path,
1713 config,
1714 shared_perm=shared_perm,
1715 alternates=alternates,
1716 )
1718 # Use reftable if extension is configured
1719 if has_reftable_extension:
1720 from .reftable import ReftableRefsContainer
1722 self.refs = ReftableRefsContainer(self.commondir())
1723 # Update worktrees container after refs change
1724 self.worktrees = WorkTreeContainer(self)
1725 BaseRepo.__init__(self, object_store, self.refs)
1727 # Determine hash algorithm from config if not already set
1728 if self.object_format is None:
1729 from .object_format import DEFAULT_OBJECT_FORMAT, get_object_format
1731 if format_version == 1:
1732 try:
1733 object_format = config.get((b"extensions",), b"objectformat")
1734 self.object_format = get_object_format(
1735 object_format.decode("ascii")
1736 )
1737 except KeyError:
1738 self.object_format = DEFAULT_OBJECT_FORMAT
1739 else:
1740 self.object_format = DEFAULT_OBJECT_FORMAT
1742 self._graftpoints = {}
1743 graft_file = self.get_named_file(
1744 os.path.join("info", "grafts"), basedir=self.commondir()
1745 )
1746 if graft_file:
1747 with graft_file:
1748 self._graftpoints.update(parse_graftpoints(graft_file))
1749 graft_file = self.get_named_file("shallow", basedir=self.commondir())
1750 if graft_file:
1751 with graft_file:
1752 self._graftpoints.update(parse_graftpoints(graft_file))
1754 self.hooks["pre-commit"] = PreCommitShellHook(self.path, self.controldir())
1755 self.hooks["commit-msg"] = CommitMsgShellHook(self.controldir())
1756 self.hooks["post-commit"] = PostCommitShellHook(self.controldir())
1757 self.hooks["pre-receive"] = PreReceiveShellHook(self.controldir())
1758 self.hooks["update"] = UpdateShellHook(self.controldir())
1759 self.hooks["post-receive"] = PostReceiveShellHook(self.controldir())
1761 # Initialize filter context as None, will be created lazily
1762 self.filter_context = None
1764 def get_worktree(self) -> "WorkTree":
1765 """Get the working tree for this repository.
1767 Returns:
1768 WorkTree instance for performing working tree operations
1769 """
1770 from .worktree import WorkTree
1772 return WorkTree(self, self.path)
1774 def _write_reflog(
1775 self,
1776 ref: bytes,
1777 old_sha: bytes,
1778 new_sha: bytes,
1779 committer: bytes | None,
1780 timestamp: int | None,
1781 timezone: int | None,
1782 message: bytes,
1783 ) -> None:
1784 from .reflog import format_reflog_line
1786 path = self._reflog_path(ref)
1788 # Get shared repository permissions
1789 shared_perm = self._get_shared_repository_permissions()
1791 # Create directory with appropriate permissions
1792 parent_dir = os.path.dirname(path)
1793 # Create directory tree, setting permissions on each level if needed
1794 parts = []
1795 current = parent_dir
1796 while current and not os.path.exists(current):
1797 parts.append(current)
1798 current = os.path.dirname(current)
1799 parts.reverse()
1800 for part in parts:
1801 os.mkdir(part)
1802 adjust_shared_perm(part, shared_perm)
1803 if committer is None:
1804 config = self.get_config_stack()
1805 committer = get_user_identity(config)
1806 check_user_identity(committer)
1807 if timestamp is None:
1808 timestamp = int(time.time())
1809 if timezone is None:
1810 timezone = 0 # FIXME
1811 with open(path, "ab") as f:
1812 f.write(
1813 format_reflog_line(
1814 old_sha, new_sha, committer, timestamp, timezone, message
1815 )
1816 + b"\n"
1817 )
1819 # Always adjust, so that permissions are right even if the file
1820 # already existed.
1821 adjust_shared_perm(path, shared_perm)
1823 def _reflog_path(self, ref: bytes) -> str:
1824 if ref.startswith((b"main-worktree/", b"worktrees/")):
1825 raise NotImplementedError(f"refs {ref.decode()} are not supported")
1827 base = self.controldir() if is_per_worktree_ref(ref) else self.commondir()
1828 return os.path.join(base, "logs", os.fsdecode(ref))
1830 def read_reflog(self, ref: bytes) -> Generator[reflog.Entry, None, None]:
1831 """Read reflog entries for a reference.
1833 Args:
1834 ref: Reference name (e.g. b'HEAD', b'refs/heads/master')
1836 Yields:
1837 reflog.Entry objects in chronological order (oldest first)
1838 """
1839 from .reflog import read_reflog
1841 path = self._reflog_path(ref)
1842 try:
1843 with open(path, "rb") as f:
1844 yield from read_reflog(f)
1845 except FileNotFoundError:
1846 return
1848 @classmethod
1849 def discover(
1850 cls,
1851 start: str | bytes | os.PathLike[str] = ".",
1852 *,
1853 ceiling_dirs: "Iterable[str | os.PathLike[str]] | None" = None,
1854 across_filesystem: bool = True,
1855 ) -> "Repo":
1856 """Iterate parent directories to discover a repository.
1858 Return a Repo object for the first parent directory that looks like a
1859 Git repository.
1861 Args:
1862 start: The directory to start discovery from (defaults to '.')
1863 ceiling_dirs: Iterable of paths that discovery must not cross
1864 (analogous to ``GIT_CEILING_DIRECTORIES``). The ceiling
1865 directories themselves are not searched. As in git, a ceiling
1866 matching ``start`` itself is ignored. Entries are resolved the
1867 same way as the walking path, so passing either a symlinked or
1868 a resolved form works.
1869 across_filesystem: Whether to keep walking up past a filesystem
1870 boundary (analogous to ``GIT_DISCOVERY_ACROSS_FILESYSTEM``).
1871 When False, discovery stops before entering a parent directory
1872 that lives on a different device than ``start``.
1873 """
1874 # Both sides of the ceiling comparison are resolved so that they are
1875 # in the same form: os.getcwd() already returns a resolved path on
1876 # POSIX, and on Windows realpath() expands 8.3 short names that
1877 # abspath() leaves alone. This also mirrors git, which walks up from
1878 # the kernel-resolved cwd.
1879 ceilings = (
1880 {
1881 os.path.normcase(os.path.realpath(os.fsdecode(os.fspath(p))))
1882 for p in ceiling_dirs
1883 }
1884 if ceiling_dirs is not None
1885 else set()
1886 )
1887 path = os.path.realpath(start)
1888 # Device of the starting directory, only tracked when we must not
1889 # cross filesystem boundaries. Errors stat'ing it propagate: if the
1890 # start directory is unreadable, discovery from it is meaningless.
1891 start_dev = None if across_filesystem else os.stat(path).st_dev
1892 first = True
1893 while True:
1894 if not first and os.path.normcase(path) in ceilings:
1895 break
1896 first = False
1897 try:
1898 return cls(path)
1899 except NotGitRepository:
1900 new_path, _tail = os.path.split(path)
1901 if new_path == path: # Root reached
1902 break
1903 if start_dev is not None:
1904 try:
1905 parent_dev = os.stat(new_path).st_dev
1906 except FileNotFoundError:
1907 # Raced with the parent being removed; nothing left
1908 # to walk up into.
1909 break
1910 if parent_dev != start_dev:
1911 break
1912 path = new_path
1913 start_str = os.fspath(start)
1914 if isinstance(start_str, bytes):
1915 start_str = start_str.decode("utf-8")
1916 raise NotGitRepository(f"No git repository was found at {start_str}")
1918 def controldir(self) -> str:
1919 """Return the path of the control directory."""
1920 return self._controldir
1922 def commondir(self) -> str:
1923 """Return the path of the common directory.
1925 For a main working tree, it is identical to controldir().
1927 For a linked working tree, it is the control directory of the
1928 main working tree.
1929 """
1930 return self._commondir
1932 def _determine_file_mode(self) -> bool:
1933 """Probe the file-system to determine whether permissions can be trusted.
1935 Returns: True if permissions can be trusted, False otherwise.
1936 """
1937 fname = os.path.join(self.path, ".probe-permissions")
1938 with open(fname, "w") as f:
1939 f.write("")
1941 st1 = os.lstat(fname)
1942 try:
1943 os.chmod(fname, st1.st_mode ^ stat.S_IXUSR)
1944 except PermissionError:
1945 return False
1946 st2 = os.lstat(fname)
1948 os.unlink(fname)
1950 mode_differs = st1.st_mode != st2.st_mode
1951 st2_has_exec = (st2.st_mode & stat.S_IXUSR) != 0
1953 return mode_differs and st2_has_exec
1955 def _determine_symlinks(self) -> bool:
1956 """Probe the filesystem to determine whether symlinks can be created.
1958 Returns: True if symlinks can be created, False otherwise.
1959 """
1960 # TODO(jelmer): Actually probe disk / look at filesystem
1961 return sys.platform != "win32"
1963 def _get_shared_repository_permissions(
1964 self,
1965 ) -> "SharedPerm | None":
1966 """Get the shared repository permission setting from config.
1968 Returns:
1969 SharedPerm to apply, or None if not shared
1970 """
1971 try:
1972 config = self.get_config()
1973 value = config.get(("core",), "sharedRepository")
1974 return parse_shared_repository(value)
1975 except KeyError:
1976 return None
1978 def _put_named_file(self, path: str, contents: bytes) -> None:
1979 """Write a file to the control dir with the given name and contents.
1981 Args:
1982 path: The path to the file, relative to the control dir.
1983 contents: A string to write to the file.
1984 """
1985 path = path.lstrip(os.path.sep)
1987 # Get shared repository permissions
1988 shared_perm = self._get_shared_repository_permissions()
1990 # Create file with appropriate permissions
1991 full_path = os.path.join(self.controldir(), path)
1992 if shared_perm is not None:
1993 with GitFile(full_path, "wb", shared_perm=shared_perm) as f:
1994 f.write(contents)
1995 else:
1996 with GitFile(full_path, "wb") as f:
1997 f.write(contents)
1999 def _del_named_file(self, path: str) -> None:
2000 try:
2001 os.unlink(os.path.join(self.controldir(), path))
2002 except FileNotFoundError:
2003 return
2005 def get_named_file(
2006 self,
2007 path: str | bytes,
2008 basedir: str | None = None,
2009 ) -> BinaryIO | None:
2010 """Get a file from the control dir with a specific name.
2012 Although the filename should be interpreted as a filename relative to
2013 the control dir in a disk-based Repo, the object returned need not be
2014 pointing to a file in that location.
2016 Args:
2017 path: The path to the file, relative to the control dir.
2018 basedir: Optional argument that specifies an alternative to the
2019 control dir.
2020 Returns: An open file object, or None if the file does not exist.
2021 """
2022 # TODO(dborowitz): sanitize filenames, since this is used directly by
2023 # the dumb web serving code.
2024 if basedir is None:
2025 basedir = self.controldir()
2026 if isinstance(path, bytes):
2027 path = path.decode("utf-8")
2028 path = path.lstrip(os.path.sep)
2029 try:
2030 return open(os.path.join(basedir, path), "rb")
2031 except FileNotFoundError:
2032 return None
2034 def index_path(self) -> str:
2035 """Return path to the index file."""
2036 if self._index_file_override is not None:
2037 return self._index_file_override
2038 return os.path.join(self.controldir(), INDEX_FILENAME)
2040 def open_index(self, config: "Config | None" = None) -> "Index":
2041 """Open the index for this repository.
2043 Args:
2044 config: Configuration to consult for index settings. If None,
2045 falls back to ``self.get_config_stack()``.
2047 Raises:
2048 NoIndexPresent: If no index is present
2049 Returns: The matching `Index`
2050 """
2051 from .index import Index, make_path_normalizer
2053 if not self.has_index():
2054 raise NoIndexPresent
2056 if config is None:
2057 config = self.get_config_stack()
2058 many_files = config.get_boolean(b"feature", b"manyFiles", False)
2059 skip_hash = False
2060 index_version = None
2062 if many_files:
2063 # When feature.manyFiles is enabled, set index.version=4 and index.skipHash=true
2064 try:
2065 index_version_str = config.get(b"index", b"version")
2066 index_version = int(index_version_str)
2067 except KeyError:
2068 index_version = 4 # Default to version 4 for manyFiles
2069 skip_hash = config.get_boolean(b"index", b"skipHash", True)
2070 else:
2071 # Check for explicit index settings
2072 try:
2073 index_version_str = config.get(b"index", b"version")
2074 index_version = int(index_version_str)
2075 except KeyError:
2076 index_version = None
2077 skip_hash = config.get_boolean(b"index", b"skipHash", False)
2079 # Get shared repository permissions for index file
2080 shared_perm = self._get_shared_repository_permissions()
2082 return Index(
2083 self.index_path(),
2084 skip_hash=skip_hash,
2085 version=index_version,
2086 shared_perm=shared_perm,
2087 path_normalizer=make_path_normalizer(config),
2088 object_format=self.object_format,
2089 )
2091 def has_index(self) -> bool:
2092 """Check if an index is present."""
2093 # Bare repos must never have index files; non-bare repos may have a
2094 # missing index file, which is treated as empty.
2095 return not self.bare
2097 def clone(
2098 self,
2099 target_path: str | bytes | os.PathLike[str],
2100 *,
2101 mkdir: bool = True,
2102 bare: bool = False,
2103 origin: bytes = b"origin",
2104 checkout: bool | None = None,
2105 branch: bytes | None = None,
2106 progress: Callable[[str], None] | None = None,
2107 depth: int | None = None,
2108 symlinks: bool | None = None,
2109 ) -> "Repo":
2110 """Clone this repository.
2112 Args:
2113 target_path: Target path
2114 mkdir: Create the target directory
2115 bare: Whether to create a bare repository
2116 checkout: Whether or not to check-out HEAD after cloning
2117 origin: Base name for refs in target repository
2118 cloned from this repository
2119 branch: Optional branch or tag to be used as HEAD in the new repository
2120 instead of this repository's HEAD.
2121 progress: Optional progress function
2122 depth: Depth at which to fetch
2123 symlinks: Symlinks setting (default to autodetect)
2124 Returns: Created repository as `Repo`
2125 """
2126 encoded_path = os.fsencode(self.path)
2128 if mkdir:
2129 os.mkdir(target_path)
2131 try:
2132 if not bare:
2133 target = Repo.init(target_path, symlinks=symlinks)
2134 if checkout is None:
2135 checkout = True
2136 else:
2137 if checkout:
2138 raise ValueError("checkout and bare are incompatible")
2139 target = Repo.init_bare(target_path)
2141 try:
2142 target_config = target.get_config()
2143 target_config.set((b"remote", origin), b"url", encoded_path)
2144 target_config.set(
2145 (b"remote", origin),
2146 b"fetch",
2147 b"+refs/heads/*:refs/remotes/" + origin + b"/*",
2148 )
2149 target_config.write_to_path()
2151 ref_message = b"clone: from " + encoded_path
2152 self.fetch(target, depth=depth)
2153 target.refs.import_refs(
2154 Ref(b"refs/remotes/" + origin),
2155 self.refs.as_dict(Ref(b"refs/heads")),
2156 message=ref_message,
2157 )
2158 target.refs.import_refs(
2159 Ref(b"refs/tags"),
2160 self.refs.as_dict(Ref(b"refs/tags")),
2161 message=ref_message,
2162 )
2164 head_chain, origin_sha = self.refs.follow(HEADREF)
2165 origin_head = head_chain[-1] if head_chain else None
2166 head: ObjectID | None = None
2167 if origin_sha and not origin_head:
2168 # set detached HEAD
2169 target.refs[HEADREF] = origin_sha
2170 head = origin_sha
2171 else:
2172 _set_origin_head(target.refs, origin, origin_head)
2173 head_ref = _set_default_branch(
2174 target.refs, origin, origin_head, branch, ref_message
2175 )
2177 # Update target head
2178 if head_ref:
2179 head = _set_head(target.refs, head_ref, ref_message)
2180 if not bare:
2181 _set_branch_tracking(target.get_config(), head_ref, origin)
2182 else:
2183 head = None
2185 if checkout and head is not None:
2186 target.get_worktree().reset_index(config=target.get_config_stack())
2187 except BaseException:
2188 target.close()
2189 raise
2190 except BaseException:
2191 if mkdir:
2192 import shutil
2194 shutil.rmtree(target_path)
2195 raise
2196 return target
2198 def _get_config_condition_matchers(self) -> dict[str, "ConditionMatcher"]:
2199 """Get condition matchers for includeIf conditions.
2201 Returns a dict of condition prefix to matcher function.
2202 """
2203 from pathlib import Path
2205 from .config import ConditionMatcher, match_glob_pattern
2207 # Add gitdir matchers
2208 def match_gitdir(pattern: str, case_sensitive: bool = True) -> bool:
2209 """Match gitdir against a pattern.
2211 Args:
2212 pattern: Pattern to match against
2213 case_sensitive: Whether to match case-sensitively
2215 Returns:
2216 True if gitdir matches pattern
2217 """
2218 # Handle relative patterns (starting with ./)
2219 if pattern.startswith("./"):
2220 # Can't handle relative patterns without config directory context
2221 return False
2223 # Normalize repository path
2224 try:
2225 repo_path = str(Path(self._controldir).resolve())
2226 except (OSError, ValueError):
2227 return False
2229 # Expand ~ in pattern and normalize
2230 pattern = os.path.expanduser(pattern)
2232 # Normalize pattern following Git's rules
2233 pattern = pattern.replace("\\", "/")
2234 if not pattern.startswith(("~/", "./", "/", "**")):
2235 # Check for Windows absolute path
2236 if len(pattern) >= 2 and pattern[1] == ":":
2237 pass
2238 else:
2239 pattern = "**/" + pattern
2240 if pattern.endswith("/"):
2241 pattern = pattern + "**"
2243 # Use the existing _match_gitdir_pattern function
2244 from .config import _match_gitdir_pattern
2246 pattern_bytes = pattern.encode("utf-8", errors="replace")
2247 repo_path_bytes = repo_path.encode("utf-8", errors="replace")
2249 return _match_gitdir_pattern(
2250 repo_path_bytes, pattern_bytes, ignorecase=not case_sensitive
2251 )
2253 # Add onbranch matcher
2254 def match_onbranch(pattern: str) -> bool:
2255 """Match current branch against a pattern.
2257 Args:
2258 pattern: Pattern to match against
2260 Returns:
2261 True if current branch matches pattern
2262 """
2263 try:
2264 # Get the current branch using refs
2265 ref_chain, _ = self.refs.follow(HEADREF)
2266 head_ref = ref_chain[-1] # Get the final resolved ref
2267 except KeyError:
2268 pass
2269 else:
2270 if head_ref and head_ref.startswith(b"refs/heads/"):
2271 # Extract branch name from ref
2272 branch = extract_branch_name(head_ref).decode(
2273 "utf-8", errors="replace"
2274 )
2275 return match_glob_pattern(branch, pattern)
2276 return False
2278 matchers: dict[str, ConditionMatcher] = {
2279 "onbranch:": match_onbranch,
2280 "gitdir:": lambda pattern: match_gitdir(pattern, True),
2281 "gitdir/i:": lambda pattern: match_gitdir(pattern, False),
2282 }
2284 return matchers
2286 def get_worktree_config(self) -> "ConfigFile":
2287 """Get the worktree-specific config.
2289 Returns:
2290 ConfigFile object for the worktree config
2291 """
2292 from .config import ConfigFile
2294 path = os.path.join(self.commondir(), "config.worktree")
2295 try:
2296 # Pass condition matchers for includeIf evaluation
2297 condition_matchers = self._get_config_condition_matchers()
2298 return ConfigFile.from_path(path, condition_matchers=condition_matchers)
2299 except FileNotFoundError:
2300 cf = ConfigFile()
2301 cf.path = path
2302 return cf
2304 def get_config(self) -> "ConfigFile":
2305 """Retrieve the config object.
2307 Returns: `ConfigFile` object for the ``.git/config`` file.
2308 """
2309 from .config import ConfigFile
2311 path = os.path.join(self._commondir, "config")
2312 try:
2313 # Pass condition matchers for includeIf evaluation
2314 condition_matchers = self._get_config_condition_matchers()
2315 return ConfigFile.from_path(path, condition_matchers=condition_matchers)
2316 except FileNotFoundError:
2317 ret = ConfigFile()
2318 ret.path = path
2319 return ret
2321 def get_rebase_state_manager(self) -> "RebaseStateManager":
2322 """Get the appropriate rebase state manager for this repository.
2324 Returns: DiskRebaseStateManager instance
2325 """
2326 import os
2328 from .rebase import DiskRebaseStateManager
2330 path = os.path.join(self.controldir(), "rebase-merge")
2331 return DiskRebaseStateManager(path)
2333 def get_description(self) -> bytes | None:
2334 """Retrieve the description of this repository.
2336 Returns: Description as bytes or None.
2337 """
2338 path = os.path.join(self._controldir, "description")
2339 try:
2340 with GitFile(path, "rb") as f:
2341 return f.read()
2342 except FileNotFoundError:
2343 return None
2345 def __repr__(self) -> str:
2346 """Return string representation of this repository."""
2347 return f"<Repo at {self.path!r}>"
2349 def set_description(self, description: bytes) -> None:
2350 """Set the description for this repository.
2352 Args:
2353 description: Text to set as description for this repository.
2354 """
2355 self._put_named_file("description", description)
2357 @classmethod
2358 def _init_maybe_bare(
2359 cls,
2360 path: str | bytes | os.PathLike[str],
2361 controldir: str | bytes | os.PathLike[str],
2362 bare: bool,
2363 object_store: PackBasedObjectStore | None = None,
2364 config: "StackedConfig | None" = None,
2365 default_branch: bytes | None = None,
2366 symlinks: bool | None = None,
2367 format: int | None = None,
2368 shared_repository: str | bool | None = None,
2369 object_format: str | None = None,
2370 ) -> "Repo":
2371 path = os.fspath(path)
2372 if isinstance(path, bytes):
2373 path = os.fsdecode(path)
2374 controldir = os.fspath(controldir)
2375 if isinstance(controldir, bytes):
2376 controldir = os.fsdecode(controldir)
2378 # Determine shared repository permissions early
2379 shared_perm: SharedPerm | None = None
2380 if shared_repository is not None:
2381 shared_perm = parse_shared_repository(shared_repository)
2383 # Create base directories with appropriate permissions
2384 for d in BASE_DIRECTORIES:
2385 dir_path = os.path.join(controldir, *d)
2386 os.mkdir(dir_path)
2387 adjust_shared_perm(dir_path, shared_perm)
2389 # Determine hash algorithm
2390 from .object_format import get_object_format
2392 hash_alg = get_object_format(object_format)
2394 if object_store is None:
2395 object_store = DiskObjectStore.init(
2396 os.path.join(controldir, OBJECTDIR),
2397 shared_perm=shared_perm,
2398 object_format=hash_alg,
2399 )
2400 ret = cls(path, bare=bare, object_store=object_store)
2401 if default_branch is None:
2402 if config is None:
2403 from .config import StackedConfig
2405 config = StackedConfig.default()
2406 try:
2407 default_branch = config.get("init", "defaultBranch")
2408 except KeyError:
2409 default_branch = DEFAULT_BRANCH
2410 ret.refs.set_symbolic_ref(HEADREF, local_branch_name(default_branch))
2411 ret._init_files(
2412 bare=bare,
2413 symlinks=symlinks,
2414 format=format,
2415 shared_repository=shared_repository,
2416 object_format=object_format,
2417 )
2418 return ret
2420 @classmethod
2421 def init(
2422 cls,
2423 path: str | bytes | os.PathLike[str],
2424 *,
2425 mkdir: bool = False,
2426 config: "StackedConfig | None" = None,
2427 default_branch: bytes | None = None,
2428 symlinks: bool | None = None,
2429 format: int | None = None,
2430 shared_repository: str | bool | None = None,
2431 object_format: str | None = None,
2432 ) -> "Repo":
2433 """Create a new repository.
2435 Args:
2436 path: Path in which to create the repository
2437 mkdir: Whether to create the directory
2438 config: Configuration object
2439 default_branch: Default branch name
2440 symlinks: Whether to support symlinks
2441 format: Repository format version (defaults to 0)
2442 shared_repository: Shared repository setting (group, all, umask, or octal)
2443 object_format: Object format to use ("sha1" or "sha256", defaults to "sha1")
2444 Returns: `Repo` instance
2445 """
2446 path = os.fspath(path)
2447 if isinstance(path, bytes):
2448 path = os.fsdecode(path)
2449 if mkdir:
2450 os.mkdir(path)
2451 controldir = os.path.join(path, CONTROLDIR)
2452 os.mkdir(controldir)
2453 _set_filesystem_hidden(controldir)
2454 return cls._init_maybe_bare(
2455 path,
2456 controldir,
2457 False,
2458 config=config,
2459 default_branch=default_branch,
2460 symlinks=symlinks,
2461 format=format,
2462 shared_repository=shared_repository,
2463 object_format=object_format,
2464 )
2466 @classmethod
2467 def _init_new_working_directory(
2468 cls,
2469 path: str | bytes | os.PathLike[str],
2470 main_repo: "Repo",
2471 identifier: str | None = None,
2472 mkdir: bool = False,
2473 relative_paths: bool = False,
2474 ) -> "Repo":
2475 """Create a new working directory linked to a repository.
2477 Args:
2478 path: Path in which to create the working tree.
2479 main_repo: Main repository to reference
2480 identifier: Worktree identifier
2481 mkdir: Whether to create the directory
2482 relative_paths: Whether to use relative paths for gitdir references
2483 Returns: `Repo` instance
2484 """
2485 path = os.fspath(path)
2486 if isinstance(path, bytes):
2487 path = os.fsdecode(path)
2488 if mkdir:
2489 os.mkdir(path)
2490 if identifier is None:
2491 identifier = os.path.basename(path)
2492 # Ensure we use absolute path for the worktree control directory
2493 main_controldir = os.path.abspath(main_repo.controldir())
2494 main_worktreesdir = os.path.join(main_controldir, WORKTREES)
2495 worktree_controldir = os.path.join(main_worktreesdir, identifier)
2496 gitdirfile_abs = os.path.abspath(os.path.join(path, CONTROLDIR))
2498 # Write gitdir reference in .git file (can be relative)
2499 # Import helper from worktree module to avoid duplication
2500 from .worktree import _compute_gitdir_path
2502 gitdir_ref = _compute_gitdir_path(
2503 main_repo,
2504 worktree_controldir,
2505 os.path.dirname(gitdirfile_abs),
2506 relative_paths,
2507 )
2509 with open(gitdirfile_abs, "wb") as f:
2510 f.write(b"gitdir: " + os.fsencode(gitdir_ref) + b"\n")
2512 # Get shared repository permissions from main repository
2513 shared_perm = main_repo._get_shared_repository_permissions()
2515 # Create directories with appropriate permissions
2516 try:
2517 os.mkdir(main_worktreesdir)
2518 adjust_shared_perm(main_worktreesdir, shared_perm)
2519 except FileExistsError:
2520 pass
2521 try:
2522 os.mkdir(worktree_controldir)
2523 adjust_shared_perm(worktree_controldir, shared_perm)
2524 except FileExistsError:
2525 pass
2527 # Write gitdir path in control directory (can be relative)
2528 gitdir_path = _compute_gitdir_path(
2529 main_repo, gitdirfile_abs, worktree_controldir, relative_paths
2530 )
2532 with open(os.path.join(worktree_controldir, GITDIR), "wb") as f:
2533 f.write(os.fsencode(gitdir_path) + b"\n")
2534 with open(os.path.join(worktree_controldir, COMMONDIR), "wb") as f:
2535 f.write(b"../..\n")
2536 with open(os.path.join(worktree_controldir, "HEAD"), "wb") as f:
2537 f.write(main_repo.head() + b"\n")
2538 r = cls(os.path.normpath(path))
2539 r.get_worktree().reset_index(config=r.get_config_stack())
2540 return r
2542 @classmethod
2543 def init_bare(
2544 cls,
2545 path: str | bytes | os.PathLike[str],
2546 *,
2547 mkdir: bool = False,
2548 object_store: PackBasedObjectStore | None = None,
2549 config: "StackedConfig | None" = None,
2550 default_branch: bytes | None = None,
2551 format: int | None = None,
2552 shared_repository: str | bool | None = None,
2553 object_format: str | None = None,
2554 ) -> "Repo":
2555 """Create a new bare repository.
2557 ``path`` should already exist and be an empty directory.
2559 Args:
2560 path: Path to create bare repository in
2561 mkdir: Whether to create the directory
2562 object_store: Object store to use
2563 config: Configuration object
2564 default_branch: Default branch name
2565 format: Repository format version (defaults to 0)
2566 shared_repository: Shared repository setting (group, all, umask, or octal)
2567 object_format: Object format to use ("sha1" or "sha256", defaults to "sha1")
2568 Returns: a `Repo` instance
2569 """
2570 path = os.fspath(path)
2571 if isinstance(path, bytes):
2572 path = os.fsdecode(path)
2573 if mkdir:
2574 os.mkdir(path)
2575 return cls._init_maybe_bare(
2576 path,
2577 path,
2578 True,
2579 object_store=object_store,
2580 config=config,
2581 default_branch=default_branch,
2582 format=format,
2583 shared_repository=shared_repository,
2584 object_format=object_format,
2585 )
2587 create = init_bare
2589 def close(self) -> None:
2590 """Close any files opened by this repository."""
2591 self.object_store.close()
2592 # Clean up filter context if it was created
2593 if self.filter_context is not None:
2594 self.filter_context.close()
2595 self.filter_context = None
2597 def __enter__(self) -> Self:
2598 """Enter context manager."""
2599 return self
2601 def __exit__(
2602 self,
2603 exc_type: type[BaseException] | None,
2604 exc_val: BaseException | None,
2605 exc_tb: TracebackType | None,
2606 ) -> None:
2607 """Exit context manager and close repository."""
2608 self.close()
2610 def _read_gitattributes(self) -> dict[bytes, dict[bytes, bytes]]:
2611 """Read .gitattributes file from working tree.
2613 A ``.gitattributes`` that is a symlink is skipped with a warning
2614 rather than followed, as git does.
2616 Returns:
2617 Dictionary mapping file patterns to attributes
2618 """
2619 gitattributes: dict[bytes, dict[bytes, bytes]] = {}
2620 gitattributes_path = os.path.join(self.path, ".gitattributes")
2622 try:
2623 f = open_nofollow_read(gitattributes_path)
2624 except FileNotFoundError:
2625 return gitattributes
2626 except OSError as e:
2627 if e.errno not in (errno.ELOOP, errno.EMLINK):
2628 raise
2629 logger.warning("Ignoring %s: it is a symbolic link", gitattributes_path)
2630 return gitattributes
2631 with f:
2632 for line in f:
2633 line = line.strip()
2634 if not line or line.startswith(b"#"):
2635 continue
2637 parts = line.split()
2638 if len(parts) < 2:
2639 continue
2641 pattern = parts[0]
2642 attrs = {}
2644 for attr in parts[1:]:
2645 if attr.startswith(b"-"):
2646 # Unset attribute
2647 attrs[attr[1:]] = b"false"
2648 elif b"=" in attr:
2649 # Set to value
2650 key, value = attr.split(b"=", 1)
2651 attrs[key] = value
2652 else:
2653 # Set attribute
2654 attrs[attr] = b"true"
2656 gitattributes[pattern] = attrs
2658 return gitattributes
2660 def get_blob_normalizer(
2661 self, config: "Config | None" = None
2662 ) -> "FilterBlobNormalizer":
2663 """Return a BlobNormalizer object.
2665 Args:
2666 config: Configuration to consult for filter setup. If None,
2667 falls back to ``self.get_config_stack()``.
2668 """
2669 from .filters import FilterBlobNormalizer, FilterContext, FilterRegistry
2671 if config is None:
2672 config = self.get_config_stack()
2673 git_attributes = self.get_gitattributes()
2675 # Lazily create FilterContext if needed
2676 if self.filter_context is None:
2677 filter_registry = FilterRegistry(config, self)
2678 self.filter_context = FilterContext(filter_registry)
2679 else:
2680 # Refresh the context with current config to handle config changes
2681 self.filter_context.refresh_config(config)
2683 return FilterBlobNormalizer(
2684 config, git_attributes, filter_context=self.filter_context
2685 )
2687 def get_gitattributes(self, tree: bytes | None = None) -> "GitAttributes":
2688 """Read gitattributes for the repository.
2690 Args:
2691 tree: Tree SHA to read .gitattributes from (defaults to HEAD)
2693 Returns:
2694 GitAttributes object that can be used to match paths
2695 """
2696 from .attrs import (
2697 GitAttributes,
2698 compile_gitattributes_patterns,
2699 parse_git_attributes,
2700 )
2702 patterns = []
2704 # Read system gitattributes (TODO: implement this)
2705 # Read global gitattributes (TODO: implement this)
2707 # Read repository .gitattributes from index/tree
2708 if tree is None:
2709 try:
2710 # Try to get from HEAD
2711 head = self[b"HEAD"]
2712 # Peel tags to get to the underlying commit
2713 while isinstance(head, Tag):
2714 _cls, obj = head.object
2715 head = self.get_object(obj)
2716 if not isinstance(head, Commit):
2717 raise ValueError(
2718 f"Expected HEAD to point to a Commit, got {type(head).__name__}. "
2719 f"This usually means HEAD points to a {type(head).__name__} object "
2720 f"instead of a Commit."
2721 )
2722 tree = head.tree
2723 except KeyError:
2724 # No HEAD, no attributes from tree
2725 pass
2727 if tree is not None:
2728 try:
2729 tree_obj = self[tree]
2730 assert isinstance(tree_obj, Tree)
2731 if b".gitattributes" in tree_obj:
2732 _, attrs_sha = tree_obj[b".gitattributes"]
2733 attrs_blob = self[attrs_sha]
2734 if isinstance(attrs_blob, Blob):
2735 attrs_data = BytesIO(attrs_blob.data)
2736 patterns.extend(
2737 compile_gitattributes_patterns(
2738 parse_git_attributes(attrs_data), b".gitattributes"
2739 )
2740 )
2741 except (KeyError, NotTreeError):
2742 pass
2744 # Read .git/info/attributes
2745 info_attrs_path = os.path.join(self.controldir(), "info", "attributes")
2746 if os.path.exists(info_attrs_path):
2747 with open(info_attrs_path, "rb") as f:
2748 patterns.extend(
2749 compile_gitattributes_patterns(
2750 parse_git_attributes(f), info_attrs_path
2751 )
2752 )
2754 # Read .gitattributes from working directory (if it exists). Like git,
2755 # don't follow a symlink there.
2756 working_attrs_path = os.path.join(self.path, ".gitattributes")
2757 try:
2758 working_attrs_file = open_nofollow_read(working_attrs_path)
2759 except FileNotFoundError:
2760 pass
2761 except OSError as e:
2762 if e.errno not in (errno.ELOOP, errno.EMLINK):
2763 raise
2764 logger.warning("Ignoring %s: it is a symbolic link", working_attrs_path)
2765 else:
2766 with working_attrs_file as f:
2767 patterns.extend(
2768 compile_gitattributes_patterns(
2769 parse_git_attributes(f), working_attrs_path
2770 )
2771 )
2773 return GitAttributes(patterns)
2776class MemoryRepo(BaseRepo):
2777 """Repo that stores refs, objects, and named files in memory.
2779 MemoryRepos are always bare: they have no working tree and no index, since
2780 those have a stronger dependency on the filesystem.
2781 """
2783 filter_context: "FilterContext | None"
2785 def __init__(self) -> None:
2786 """Create a new repository in memory."""
2787 from .config import ConfigFile
2788 from .object_format import DEFAULT_OBJECT_FORMAT
2790 self._reflog: list[Any] = []
2791 refs_container = DictRefsContainer({}, logger=self._append_reflog)
2792 BaseRepo.__init__(self, MemoryObjectStore(), refs_container)
2793 self._named_files: dict[str, bytes] = {}
2794 self.bare = True
2795 self._config = ConfigFile()
2796 self._description: bytes | None = None
2797 self.filter_context = None
2798 # MemoryRepo defaults to default object format
2799 self.object_format = DEFAULT_OBJECT_FORMAT
2801 def _append_reflog(
2802 self,
2803 ref: bytes,
2804 old_sha: bytes | None,
2805 new_sha: bytes | None,
2806 committer: bytes | None,
2807 timestamp: int | None,
2808 timezone: int | None,
2809 message: bytes | None,
2810 ) -> None:
2811 self._reflog.append(
2812 (ref, old_sha, new_sha, committer, timestamp, timezone, message)
2813 )
2815 def set_description(self, description: bytes) -> None:
2816 """Set the description for this repository.
2818 Args:
2819 description: Text to set as description
2820 """
2821 self._description = description
2823 def get_description(self) -> bytes | None:
2824 """Get the description of this repository.
2826 Returns:
2827 Repository description as bytes
2828 """
2829 return self._description
2831 def _determine_file_mode(self) -> bool:
2832 """Probe the file-system to determine whether permissions can be trusted.
2834 Returns: True if permissions can be trusted, False otherwise.
2835 """
2836 return sys.platform != "win32"
2838 def _determine_symlinks(self) -> bool:
2839 """Probe the file-system to determine whether permissions can be trusted.
2841 Returns: True if permissions can be trusted, False otherwise.
2842 """
2843 return sys.platform != "win32"
2845 def _put_named_file(self, path: str, contents: bytes) -> None:
2846 """Write a file to the control dir with the given name and contents.
2848 Args:
2849 path: The path to the file, relative to the control dir.
2850 contents: A string to write to the file.
2851 """
2852 self._named_files[path] = contents
2854 def _del_named_file(self, path: str) -> None:
2855 try:
2856 del self._named_files[path]
2857 except KeyError:
2858 pass
2860 def get_named_file(
2861 self,
2862 path: str | bytes,
2863 basedir: str | None = None,
2864 ) -> BytesIO | None:
2865 """Get a file from the control dir with a specific name.
2867 Although the filename should be interpreted as a filename relative to
2868 the control dir in a disk-baked Repo, the object returned need not be
2869 pointing to a file in that location.
2871 Args:
2872 path: The path to the file, relative to the control dir.
2873 basedir: Optional base directory for the path
2874 Returns: An open file object, or None if the file does not exist.
2875 """
2876 path_str = path.decode() if isinstance(path, bytes) else path
2877 contents = self._named_files.get(path_str, None)
2878 if contents is None:
2879 return None
2880 return BytesIO(contents)
2882 def open_index(self, config: "Config | None" = None) -> "Index":
2883 """Fail to open index for this repo, since it is bare.
2885 Args:
2886 config: Unused; kept for signature compatibility with ``BaseRepo``.
2888 Raises:
2889 NoIndexPresent: Raised when no index is present
2890 """
2891 raise NoIndexPresent
2893 def _init_config(self, config: "ConfigFile") -> None:
2894 """Initialize repository configuration for MemoryRepo."""
2895 self._config = config
2897 def get_config(self) -> "ConfigFile":
2898 """Retrieve the config object.
2900 Returns: `ConfigFile` object.
2901 """
2902 return self._config
2904 def get_rebase_state_manager(self) -> "RebaseStateManager":
2905 """Get the appropriate rebase state manager for this repository.
2907 Returns: MemoryRebaseStateManager instance
2908 """
2909 from .rebase import MemoryRebaseStateManager
2911 return MemoryRebaseStateManager(self)
2913 def get_blob_normalizer(
2914 self, config: "Config | None" = None
2915 ) -> "FilterBlobNormalizer":
2916 """Return a BlobNormalizer object for checkin/checkout operations.
2918 Args:
2919 config: Configuration to consult for filter setup. If None,
2920 falls back to ``self.get_config_stack()``.
2921 """
2922 from .filters import FilterBlobNormalizer, FilterContext, FilterRegistry
2924 if config is None:
2925 config = self.get_config_stack()
2926 git_attributes = self.get_gitattributes()
2928 if self.filter_context is None:
2929 filter_registry = FilterRegistry(config, self)
2930 self.filter_context = FilterContext(filter_registry)
2931 else:
2932 self.filter_context.refresh_config(config)
2934 return FilterBlobNormalizer(
2935 config, git_attributes, filter_context=self.filter_context
2936 )
2938 def get_gitattributes(self, tree: bytes | None = None) -> "GitAttributes":
2939 """Read gitattributes for the repository."""
2940 from .attrs import GitAttributes
2942 # Memory repos don't have working trees or gitattributes files
2943 # Return empty GitAttributes
2944 return GitAttributes([])
2946 def close(self) -> None:
2947 """Close any resources opened by this repository."""
2948 # Clean up filter context if it was created
2949 if self.filter_context is not None:
2950 self.filter_context.close()
2951 self.filter_context = None
2952 # Close object store to release pack files
2953 self.object_store.close()
2955 def do_commit(
2956 self,
2957 message: bytes | None = None,
2958 committer: bytes | None = None,
2959 author: bytes | None = None,
2960 commit_timestamp: float | None = None,
2961 commit_timezone: int | None = None,
2962 author_timestamp: float | None = None,
2963 author_timezone: int | None = None,
2964 tree: ObjectID | None = None,
2965 encoding: bytes | None = None,
2966 ref: Ref | None = HEADREF,
2967 merge_heads: list[ObjectID] | None = None,
2968 no_verify: bool = False,
2969 sign: bool = False,
2970 config: "Config | None" = None,
2971 ) -> bytes:
2972 """Create a new commit.
2974 This is a simplified implementation for in-memory repositories that
2975 doesn't support worktree operations or hooks.
2977 Args:
2978 message: Commit message
2979 committer: Committer fullname
2980 author: Author fullname
2981 commit_timestamp: Commit timestamp (defaults to now)
2982 commit_timezone: Commit timestamp timezone (defaults to GMT)
2983 author_timestamp: Author timestamp (defaults to commit timestamp)
2984 author_timezone: Author timestamp timezone (defaults to commit timezone)
2985 tree: SHA1 of the tree root to use
2986 encoding: Encoding
2987 ref: Optional ref to commit to (defaults to current branch).
2988 If None, creates a dangling commit without updating any ref.
2989 merge_heads: Merge heads
2990 no_verify: Skip pre-commit and commit-msg hooks (ignored for MemoryRepo)
2991 sign: GPG Sign the commit (ignored for MemoryRepo)
2992 config: Configuration to consult for committer/author identity. If
2993 None, falls back to ``self.get_config_stack()``.
2995 Returns:
2996 New commit SHA1
2997 """
2998 import time
3000 from .objects import Commit
3002 if tree is None:
3003 raise ValueError("tree must be specified for MemoryRepo")
3005 c = Commit()
3006 if len(tree) != self.object_format.hex_length:
3007 raise ValueError(
3008 f"tree must be a {self.object_format.hex_length}-character hex sha string"
3009 )
3010 c.tree = tree
3012 if config is None:
3013 config = self.get_config_stack()
3014 if merge_heads is None:
3015 merge_heads = []
3016 if committer is None:
3017 committer = get_user_identity(config, kind="COMMITTER")
3018 check_user_identity(committer)
3019 c.committer = committer
3020 if commit_timestamp is None:
3021 commit_timestamp = time.time()
3022 c.commit_time = int(commit_timestamp)
3023 if commit_timezone is None:
3024 commit_timezone = 0
3025 c.commit_timezone = commit_timezone
3026 if author is None:
3027 author = get_user_identity(config, kind="AUTHOR")
3028 c.author = author
3029 check_user_identity(author)
3030 if author_timestamp is None:
3031 author_timestamp = commit_timestamp
3032 c.author_time = int(author_timestamp)
3033 if author_timezone is None:
3034 author_timezone = commit_timezone
3035 c.author_timezone = author_timezone
3036 if encoding is None:
3037 try:
3038 encoding = config.get(("i18n",), "commitEncoding")
3039 except KeyError:
3040 pass
3041 if encoding is not None:
3042 c.encoding = encoding
3044 # Handle message (for MemoryRepo, we don't support callable messages)
3045 if callable(message):
3046 message = message(self, c)
3047 if message is None:
3048 raise ValueError("Message callback returned None")
3050 if message is None:
3051 raise ValueError("No commit message specified")
3053 c.message = message
3055 if ref is None:
3056 # Create a dangling commit
3057 c.parents = merge_heads
3058 self.object_store.add_object(c)
3059 else:
3060 try:
3061 old_head = self.refs[ref]
3062 c.parents = [old_head, *merge_heads]
3063 self.object_store.add_object(c)
3064 ok = self.refs.set_if_equals(
3065 ref,
3066 old_head,
3067 c.id,
3068 message=b"commit: " + message,
3069 committer=committer,
3070 timestamp=int(commit_timestamp),
3071 timezone=commit_timezone,
3072 )
3073 except KeyError:
3074 c.parents = merge_heads
3075 self.object_store.add_object(c)
3076 ok = self.refs.add_if_new(
3077 ref,
3078 c.id,
3079 message=b"commit: " + message,
3080 committer=committer,
3081 timestamp=int(commit_timestamp),
3082 timezone=commit_timezone,
3083 )
3084 if not ok:
3085 from .errors import CommitError
3087 raise CommitError(f"{ref!r} changed during commit")
3089 return c.id
3091 @classmethod
3092 def init_bare(
3093 cls,
3094 objects: Iterable[ShaFile],
3095 refs: Mapping[Ref, ObjectID],
3096 format: int | None = None,
3097 object_format: str | None = None,
3098 ) -> "MemoryRepo":
3099 """Create a new bare repository in memory.
3101 Args:
3102 objects: Objects for the new repository,
3103 as iterable
3104 refs: Refs as dictionary, mapping names
3105 to object SHA1s
3106 format: Repository format version (defaults to 0)
3107 object_format: Object format to use ("sha1" or "sha256", defaults to "sha1")
3108 """
3109 ret = cls()
3110 for obj in objects:
3111 ret.object_store.add_object(obj)
3112 for refname, sha in refs.items():
3113 ret.refs.add_if_new(refname, sha)
3114 ret._init_files(bare=True, format=format, object_format=object_format)
3115 return ret