Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/repo.py: 37%

1# repo.py -- For dealing with git repositories.

5# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later

6# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU

7# General Public License as published by the Free Software Foundation; version 2.0

8# or (at your option) any later version. You can redistribute it and/or

9# modify it under the terms of either of these two licenses.

10#

11# Unless required by applicable law or agreed to in writing, software

12# distributed under the License is distributed on an "AS IS" BASIS,

13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

14# See the License for the specific language governing permissions and

15# limitations under the License.

16#

17# You should have received a copy of the licenses; if not, see

18# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License

19# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache

20# License, Version 2.0.

21#

24"""Repository access.

26This module contains the base class for git repositories

27(BaseRepo) and an implementation which uses a repository on

28local disk (Repo).

30"""

32__all__ = [

33 "BASE_DIRECTORIES",

34 "COMMONDIR",

35 "CONTROLDIR",

36 "DEFAULT_BRANCH",

37 "DEFAULT_OFS_DELTA",

38 "GITDIR",

39 "INDEX_FILENAME",

40 "OBJECTDIR",

41 "REFSDIR",

42 "REFSDIR_HEADS",

43 "REFSDIR_TAGS",

44 "WORKTREES",

45 "BaseRepo",

46 "DefaultIdentityNotFound",

47 "InvalidUserIdentity",

48 "MemoryRepo",

49 "ParentsProvider",

50 "Repo",

51 "UnsupportedExtension",

52 "UnsupportedVersion",

53 "check_user_identity",

54 "get_user_identity",

55 "parse_graftpoints",

56 "parse_shared_repository",

57 "read_gitfile",

58 "sanitize_user_identity",

59 "serialize_graftpoints",

60]

62import logging

63import os

64import stat

65import sys

66import time

67import warnings

68from collections.abc import Callable, Generator, Iterable, Iterator, Mapping, Sequence

69from io import BytesIO

70from types import TracebackType

71from typing import (

72 TYPE_CHECKING,

73 Any,

74 BinaryIO,

75 TypeVar,

76)

78if sys.version_info >= (3, 11):

79 from typing import Self

80else:

81 from typing_extensions import Self

83if TYPE_CHECKING:

84 # There are no circular imports here, but we try to defer imports as long

85 # as possible to reduce start-up time for anything that doesn't need

86 # these imports.

87 from .attrs import GitAttributes

88 from .config import ConditionMatcher, Config, ConfigFile, StackedConfig

89 from .diff_tree import RenameDetector

90 from .filters import FilterBlobNormalizer, FilterContext

91 from .index import Index

92 from .notes import Notes

93 from .object_format import ObjectFormat

94 from .object_store import BaseObjectStore, GraphWalker

95 from .pack import UnpackedObject

96 from .rebase import RebaseStateManager

97 from .walk import Walker

98 from .worktree import WorkTree

100from . import reflog

101from .errors import (

102 NoIndexPresent,

103 NotBlobError,

104 NotCommitError,

105 NotGitRepository,

106 NotTagError,

107 NotTreeError,

108 RefFormatError,

109)

110from .file import (

111 PERM_EVERYBODY,

112 PERM_GROUP,

113 GitFile,

114 SharedPerm,

115 adjust_shared_perm,

116)

117from .hooks import (

118 CommitMsgShellHook,

119 Hook,

120 PostCommitShellHook,

121 PostReceiveShellHook,

122 PreCommitShellHook,

123 PreReceiveShellHook,

124 UpdateShellHook,

125)

126from .object_store import (

127 DiskObjectStore,

128 MemoryObjectStore,

129 MissingObjectFinder,

130 ObjectStoreGraphWalker,

131 PackBasedObjectStore,

132 PackCapableObjectStore,

133 find_shallow,

134 peel_sha,

135)

136from .objects import (

137 Blob,

138 Commit,

139 ObjectID,

140 RawObjectID,

141 ShaFile,

142 Tag,

143 Tree,

144 check_hexsha,

145 valid_hexsha,

146)

147from .pack import generate_unpacked_objects

148from .refs import (

149 HEADREF,

150 LOCAL_TAG_PREFIX, # noqa: F401

151 SYMREF, # noqa: F401

152 DictRefsContainer,

153 DiskRefsContainer,

154 Ref,

155 RefsContainer,

156 _set_branch_tracking,

157 _set_default_branch,

158 _set_head,

159 _set_origin_head,

160 check_ref_format, # noqa: F401

161 extract_branch_name,

162 is_per_worktree_ref,

163 local_branch_name,

164 read_packed_refs, # noqa: F401

165 read_packed_refs_with_peeled, # noqa: F401

166 write_packed_refs, # noqa: F401

167)

168

169logger = logging.getLogger(__name__)

170

171CONTROLDIR = ".git"

172OBJECTDIR = "objects"

173DEFAULT_OFS_DELTA = True

174

175T = TypeVar("T", bound="ShaFile")

176REFSDIR = "refs"

177REFSDIR_TAGS = "tags"

178REFSDIR_HEADS = "heads"

179INDEX_FILENAME = "index"

180COMMONDIR = "commondir"

181GITDIR = "gitdir"

182WORKTREES = "worktrees"

183

184BASE_DIRECTORIES = [

185 ["branches"],

186 [REFSDIR],

187 [REFSDIR, REFSDIR_TAGS],

188 [REFSDIR, REFSDIR_HEADS],

189 ["hooks"],

190 ["info"],

191]

192

193DEFAULT_BRANCH = b"master"

194

195

196class InvalidUserIdentity(Exception):

197 """User identity is not of the format 'user <email>'."""

198

199 def __init__(self, identity: str) -> None:

200 """Initialize InvalidUserIdentity exception."""

201 self.identity = identity

202

203

204class DefaultIdentityNotFound(Exception):

205 """Default identity could not be determined."""

206

207

208# TODO(jelmer): Cache?

209def _get_default_identity(env: Mapping[str, str] | None = None) -> tuple[str, str]:

210 import socket

211

212 if env is None:

213 env = os.environ

214

215 for name in ("LOGNAME", "USER", "LNAME", "USERNAME"):

216 username = env.get(name)

217 if username:

218 break

219 else:

220 username = None

221

222 try:

223 import pwd

224 except ImportError:

225 fullname = None

226 else:

227 try:

228 entry = pwd.getpwuid(os.getuid()) # type: ignore[attr-defined,unused-ignore]

229 except KeyError:

230 fullname = None

231 else:

232 if getattr(entry, "gecos", None):

233 fullname = entry.pw_gecos.split(",")[0]

234 else:

235 fullname = None

236 if username is None:

237 username = entry.pw_name

238 if not fullname:

239 if username is None:

240 raise DefaultIdentityNotFound("no username found")

241 fullname = username

242 email = env.get("EMAIL")

243 if email is None:

244 if username is None:

245 raise DefaultIdentityNotFound("no username found")

246 email = f"{username}@{socket.gethostname()}"

247 return (fullname, email)

248

249

250def get_user_identity(config: "Config", kind: str | None = None) -> bytes:

251 """Determine the identity to use for new commits.

252

253 If kind is set, this first checks

254 GIT_${KIND}_NAME and GIT_${KIND}_EMAIL.

255

256 If those variables are not set, then it will fall back

257 to reading the user.name and user.email settings from

258 the specified configuration.

259

260 If that also fails, then it will fall back to using

261 the current users' identity as obtained from the host

262 system (e.g. the gecos field, $EMAIL, $USER@$(hostname -f).

263

264 Args:

265 config: Configuration stack to read from

266 kind: Optional kind to return identity for,

267 usually either "AUTHOR" or "COMMITTER".

268

269 Returns:

270 A user identity

271 """

272 user: bytes | None = None

273 email: bytes | None = None

274 if kind:

275 user_uc = os.environ.get("GIT_" + kind + "_NAME")

276 if user_uc is not None:

277 user = user_uc.encode("utf-8")

278 email_uc = os.environ.get("GIT_" + kind + "_EMAIL")

279 if email_uc is not None:

280 email = email_uc.encode("utf-8")

281 if user is None:

282 try:

283 user = config.get(("user",), "name")

284 except KeyError:

285 user = None

286 if email is None:

287 try:

288 email = config.get(("user",), "email")

289 except KeyError:

290 email = None

291 default_user, default_email = _get_default_identity()

292 if user is None:

293 user = default_user.encode("utf-8")

294 if email is None:

295 email = default_email.encode("utf-8")

296 if email.startswith(b"<") and email.endswith(b">"):

297 email = email[1:-1]

298 return user + b" <" + email + b">"

299

300

301def check_user_identity(identity: bytes) -> None:

302 """Verify that a user identity is formatted correctly.

303

304 Args:

305 identity: User identity bytestring

306 Raises:

307 InvalidUserIdentity: Raised when identity is invalid

308 """

309 try:

310 _fst, snd = identity.split(b" <", 1)

311 except ValueError as exc:

312 raise InvalidUserIdentity(identity.decode("utf-8", "replace")) from exc

313 if b">" not in snd:

314 raise InvalidUserIdentity(identity.decode("utf-8", "replace"))

315 if b"\0" in identity or b"\n" in identity:

316 raise InvalidUserIdentity(identity.decode("utf-8", "replace"))

317

318

319_IDENTITY_CRUD = bytes(range(33)) + b",:;<>\"\\'"

320

321

322def _strip_identity_crud(value: bytes) -> bytes:

323 r"""Strip characters using git's ``strbuf_addstr_without_crud`` rules.

324

325 Leading and trailing "crud" (bytes <= 32 as well as ``,:;<>"\'``) is

326 stripped, and the delimiter characters ``\n``, ``<`` and ``>`` are

327 dropped from the rest of the value.

328 """

329 # Git does not handle embedded NUL bytes here, but check_user_identity

330 # rejects them.

331 return value.strip(_IDENTITY_CRUD).translate(None, b"\0\n<>")

332

333

334def sanitize_user_identity(name: bytes, email: bytes) -> bytes:

335 r"""Build a user identity with a sanitized name and email.

336

337 Matches git's ``fmt_ident`` behavior: the name and the email are each

338 stripped of leading and trailing "crud" (bytes <= 32 as well as

339 ``,:;<>"\'``), and the delimiter characters ``\n``, ``<`` and ``>``

340 are dropped from the middle.

341

342 Unlike ``check_user_identity``, this function never rejects its input. It

343 is intended for identities outside the caller's control, such as those

344 used by an importer to build commits. The result always passes

345 ``check_user_identity``.

346

347 Args:

348 name: User name bytestring

349 email: Email bytestring

350

351 Returns:

352 Identity bytestring of the format ``name <email>``

353 """

354 return _strip_identity_crud(name) + b" <" + _strip_identity_crud(email) + b">"

355

356

357def parse_graftpoints(

358 graftpoints: Iterable[bytes],

359) -> dict[ObjectID, list[ObjectID]]:

360 """Convert a list of graftpoints into a dict.

361

362 Args:

363 graftpoints: Iterator of graftpoint lines

364

365 Each line is formatted as:

366 <commit sha1> <parent sha1> [<parent sha1>]*

367

368 Resulting dictionary is:

369 <commit sha1>: [<parent sha1>*]

370

371 https://git.wiki.kernel.org/index.php/GraftPoint

372 """

373 grafts: dict[ObjectID, list[ObjectID]] = {}

374 for line in graftpoints:

375 raw_graft = line.split(None, 1)

376

377 commit = ObjectID(raw_graft[0])

378 if len(raw_graft) == 2:

379 parents = [ObjectID(p) for p in raw_graft[1].split()]

380 else:

381 parents = []

382

383 for sha in [commit, *parents]:

384 check_hexsha(sha, "Invalid graftpoint")

385

386 grafts[commit] = parents

387 return grafts

388

389

390def serialize_graftpoints(graftpoints: Mapping[ObjectID, Sequence[ObjectID]]) -> bytes:

391 """Convert a dictionary of grafts into string.

392

393 The graft dictionary is:

394 <commit sha1>: [<parent sha1>*]

395

396 Each line is formatted as:

397 <commit sha1> <parent sha1> [<parent sha1>]*

398

399 https://git.wiki.kernel.org/index.php/GraftPoint

400

401 """

402 graft_lines = []

403 for commit, parents in graftpoints.items():

404 if parents:

405 graft_lines.append(commit + b" " + b" ".join(parents))

406 else:

407 graft_lines.append(commit)

408 return b"\n".join(graft_lines)

409

410

411def _set_filesystem_hidden(path: str) -> None:

412 """Mark path as to be hidden if supported by platform and filesystem.

413

414 On win32 uses SetFileAttributesW api:

415 <https://docs.microsoft.com/windows/desktop/api/fileapi/nf-fileapi-setfileattributesw>

416 """

417 if sys.platform == "win32":

418 import ctypes

419 from ctypes.wintypes import BOOL, DWORD, LPCWSTR

420

421 FILE_ATTRIBUTE_HIDDEN = 2

422 SetFileAttributesW = ctypes.WINFUNCTYPE(BOOL, LPCWSTR, DWORD)(

423 ("SetFileAttributesW", ctypes.windll.kernel32)

424 )

425

426 if isinstance(path, bytes):

427 path = os.fsdecode(path)

428 if not SetFileAttributesW(path, FILE_ATTRIBUTE_HIDDEN):

429 pass # Could raise or log `ctypes.WinError()` here

430

431 # Could implement other platform specific filesystem hiding here

432

433

434def parse_shared_repository(value: str | bytes | bool) -> "SharedPerm | None":

435 """Parse core.sharedRepository configuration value.

436

437 Args:

438 value: Configuration value (string, bytes, or boolean)

439

440 Returns:

441 SharedPerm to apply, or None to leave permissions to the umask

442 """

443 if isinstance(value, bytes):

444 value = value.decode("utf-8", errors="replace")

445

446 # Handle boolean values

447 if isinstance(value, bool):

448 # true = group (same as "group"), false = umask

449 return PERM_GROUP if value else None

450

451 # Handle string values

452 value_lower = value.lower()

453

454 if value_lower in ("false", "0", "", "umask"):

455 # Use umask (no adjustment)

456 return None

457

458 if value_lower in ("true", "1", "group"):

459 return PERM_GROUP

460

461 if value_lower in ("all", "world", "everybody", "2"):

462 # Others gain read, and execute on directories, but never write.

463 return PERM_EVERYBODY

464

465 # Try to parse as octal. Unlike the named settings, this states the mode

466 # outright rather than loosening what the umask produced.

467 if value.startswith("0"):

468 try:

469 mode = int(value, 8)

470 except ValueError:

471 pass

472 else:

473 return SharedPerm(tweak=mode, replace=True)

474

475 # Default to umask for unrecognized values

476 return None

477

478

479def _enable_relative_worktrees_extension(repo: "Repo") -> None:

480 """Enable the relativeworktrees extension in repository config.

481

482 This sets core.repositoryformatversion to 1 (if not already) and

483 enables the extensions.relativeworktrees extension.

484

485 Args:

486 repo: The repository to configure

487 """

488 config = repo.get_config()

489

490 # Ensure repository format version is at least 1

491 try:

492 version = int(config.get(("core",), "repositoryformatversion"))

493 except KeyError:

494 version = 0

495

496 if version < 1:

497 config.set(("core",), "repositoryformatversion", "1")

498

499 # Enable the relativeworktrees extension

500 config.set(("extensions",), "relativeworktrees", True)

501 config.write_to_path()

502

503

504class ParentsProvider:

505 """Provider for commit parent information."""

506

507 def __init__(

508 self,

509 store: "BaseObjectStore",

510 grafts: dict[ObjectID, list[ObjectID]] = {},

511 shallows: Iterable[ObjectID] = [],

512 ) -> None:

513 """Initialize ParentsProvider.

514

515 Args:

516 store: Object store to use

517 grafts: Graft information

518 shallows: Shallow commit SHAs

519 """

520 self.store = store

521 self.grafts = grafts

522 self.shallows = set(shallows)

523

524 # Get commit graph once at initialization for performance

525 self.commit_graph = store.get_commit_graph()

526

527 def get_parents(

528 self, commit_id: ObjectID, commit: Commit | None = None

529 ) -> list[ObjectID]:

530 """Get parents for a commit using the parents provider."""

531 try:

532 return self.grafts[commit_id]

533 except KeyError:

534 pass

535 if commit_id in self.shallows:

536 return []

537

538 # Try to use commit graph for faster parent lookup

539 if self.commit_graph:

540 parents = self.commit_graph.get_parents(commit_id)

541 if parents is not None:

542 return parents

543

544 # Fallback to reading the commit object

545 if commit is None:

546 obj = self.store[commit_id]

547 if not isinstance(obj, Commit):

548 raise ValueError(

549 f"Expected Commit object for commit_id {commit_id.decode()}, "

550 f"got {type(obj).__name__}. This usually means a reference "

551 f"points to a {type(obj).__name__} object instead of a Commit."

552 )

553 commit = obj

554 result: list[ObjectID] = commit.parents

555 return result

556

557

558class BaseRepo:

559 """Base class for a git repository.

560

561 This base class is meant to be used for Repository implementations that e.g.

562 work on top of a different transport than a standard filesystem path.

563

564 Attributes:

565 object_store: Dictionary-like object for accessing

566 the objects

567 refs: Dictionary-like object with the refs in this

568 repository

569 """

570

571 def __init__(

572 self,

573 object_store: "PackCapableObjectStore",

574 refs: RefsContainer,

575 object_format: "ObjectFormat | None" = None,

576 ) -> None:

577 """Open a repository.

578

579 This shouldn't be called directly, but rather through one of the

580 base classes, such as MemoryRepo or Repo.

581

582 Args:

583 object_store: Object store to use

584 refs: Refs container to use

585 object_format: Hash algorithm to use (if None, will use object_store's format)

586 """

587 self.object_store = object_store

588 self.refs = refs

589

590 self._graftpoints: dict[ObjectID, list[ObjectID]] = {}

591 self.hooks: dict[str, Hook] = {}

592 if object_format is None:

593 self.object_format: ObjectFormat = object_store.object_format

594 else:

595 self.object_format = object_format

596

597 def _determine_file_mode(self) -> bool:

598 """Probe the file-system to determine whether permissions can be trusted.

599

600 Returns: True if permissions can be trusted, False otherwise.

601 """

602 raise NotImplementedError(self._determine_file_mode)

603

604 def _determine_symlinks(self) -> bool:

605 """Probe the filesystem to determine whether symlinks can be created.

606

607 Returns: True if symlinks can be created, False otherwise.

608 """

609 # For now, just mimic the old behaviour

610 return sys.platform != "win32"

611

612 def _init_files(

613 self,

614 bare: bool,

615 symlinks: bool | None = None,

616 format: int | None = None,

617 shared_repository: str | bool | None = None,

618 object_format: str | None = None,

619 ) -> None:

620 """Initialize a default set of named files."""

621 from .config import ConfigFile

622

623 self._put_named_file("description", b"Unnamed repository")

624 f = BytesIO()

625 cf = ConfigFile()

626

627 # Determine the appropriate format version

628 if object_format == "sha256":

629 # SHA256 requires format version 1

630 if format is None:

631 format = 1

632 elif format != 1:

633 raise ValueError(

634 "SHA256 object format requires repository format version 1"

635 )

636 else:

637 # SHA1 (default) can use format 0 or 1

638 if format is None:

639 format = 0

640

641 if format not in (0, 1):

642 raise ValueError(f"Unsupported repository format version: {format}")

643

644 cf.set("core", "repositoryformatversion", str(format))

645

646 # Set object format extension if using SHA256

647 if object_format == "sha256":

648 cf.set("extensions", "objectformat", "sha256")

649

650 # Set hash algorithm based on object format

651 from .object_format import get_object_format

652

653 self.object_format = get_object_format(object_format)

654

655 if self._determine_file_mode():

656 cf.set("core", "filemode", True)

657 else:

658 cf.set("core", "filemode", False)

659

660 if symlinks is None and not bare:

661 symlinks = self._determine_symlinks()

662

663 if symlinks is False:

664 cf.set("core", "symlinks", symlinks)

665

666 # On macOS, set precomposeunicode to true since HFS+/APFS

667 # returns filenames in NFD (decomposed) Unicode form

668 if sys.platform == "darwin":

669 cf.set("core", "precomposeunicode", True)

670

671 cf.set("core", "bare", bare)

672 cf.set("core", "logallrefupdates", True)

673

674 # Set shared repository if specified

675 if shared_repository is not None:

676 if isinstance(shared_repository, bool):

677 cf.set("core", "sharedRepository", shared_repository)

678 else:

679 cf.set("core", "sharedRepository", shared_repository)

680

681 cf.write_to_file(f)

682 self._put_named_file("config", f.getvalue())

683 self._put_named_file(os.path.join("info", "exclude"), b"")

684

685 # Allow subclasses to handle config initialization

686 self._init_config(cf)

687

688 def _init_config(self, config: "ConfigFile") -> None:

689 """Initialize repository configuration.

690

691 This method can be overridden by subclasses to handle config initialization.

692

693 Args:

694 config: The ConfigFile object that was just created

695 """

696 # Default implementation does nothing

697

698 def get_named_file(self, path: str) -> BinaryIO | None:

699 """Get a file from the control dir with a specific name.

700

701 Although the filename should be interpreted as a filename relative to

702 the control dir in a disk-based Repo, the object returned need not be

703 pointing to a file in that location.

704

705 Args:

706 path: The path to the file, relative to the control dir.

707 Returns: An open file object, or None if the file does not exist.

708 """

709 raise NotImplementedError(self.get_named_file)

710

711 def _put_named_file(self, path: str, contents: bytes) -> None:

712 """Write a file to the control dir with the given name and contents.

713

714 Args:

715 path: The path to the file, relative to the control dir.

716 contents: A string to write to the file.

717 """

718 raise NotImplementedError(self._put_named_file)

719

720 def _del_named_file(self, path: str) -> None:

721 """Delete a file in the control directory with the given name."""

722 raise NotImplementedError(self._del_named_file)

723

724 def open_index(self, config: "Config | None" = None) -> "Index":

725 """Open the index for this repository.

726

727 Args:

728 config: Configuration to consult for index settings. If None,

729 implementations may fall back to ``self.get_config_stack()``.

730

731 Raises:

732 NoIndexPresent: If no index is present

733 Returns: The matching `Index`

734 """

735 raise NotImplementedError(self.open_index)

736

737 def _change_object_format(self, object_format_name: str) -> None:

738 """Change the object format of this repository.

739

740 This can only be done if the object store is empty (no objects written yet).

741

742 Args:

743 object_format_name: Name of the new object format (e.g., "sha1", "sha256")

744

745 Raises:

746 AssertionError: If the object store is not empty

747 """

748 # Check if object store has any objects

749 for _ in self.object_store:

750 raise AssertionError(

751 "Cannot change object format: repository already contains objects"

752 )

753

754 # Update the object format

755 from .object_format import get_object_format

756

757 new_format = get_object_format(object_format_name)

758 self.object_format = new_format

759 self.object_store.object_format = new_format

760

761 # Update config file

762 config = self.get_config()

763

764 if object_format_name == "sha1":

765 # For SHA-1, explicitly remove objectformat extension if present

766 try:

767 config.remove("extensions", "objectformat")

768 except KeyError:

769 pass

770 else:

771 # For non-SHA-1 formats, set repositoryformatversion to 1 and objectformat extension

772 config.set("core", "repositoryformatversion", "1")

773 config.set("extensions", "objectformat", object_format_name)

774

775 config.write_to_path()

776

777 def fetch(

778 self,

779 target: "BaseRepo",

780 determine_wants: Callable[[Mapping[Ref, ObjectID], int | None], list[ObjectID]]

781 | None = None,

782 progress: Callable[..., None] | None = None,

783 depth: int | None = None,

784 ) -> dict[Ref, ObjectID]:

785 """Fetch objects into another repository.

786

787 Args:

788 target: The target repository

789 determine_wants: Optional function to determine what refs to

790 fetch.

791 progress: Optional progress function

792 depth: Optional shallow fetch depth

793 Returns: The local refs

794 """

795 # Fix object format if needed

796 if self.object_format != target.object_format:

797 # Change the target repo's format if it's empty

798 target._change_object_format(self.object_format.name)

799

800 if determine_wants is None:

801 determine_wants = target.object_store.determine_wants_all

802 count, pack_data = self.fetch_pack_data(

803 determine_wants,

804 target.get_graph_walker(),

805 progress=progress,

806 depth=depth,

807 )

808 target.object_store.add_pack_data(count, pack_data, progress)

809 return self.get_refs()

810

811 def fetch_pack_data(

812 self,

813 determine_wants: Callable[[Mapping[Ref, ObjectID], int | None], list[ObjectID]],

814 graph_walker: "GraphWalker",

815 progress: Callable[[bytes], None] | None,

816 *,

817 get_tagged: Callable[[], dict[ObjectID, ObjectID]] | None = None,

818 depth: int | None = None,

819 ) -> tuple[int, Iterator["UnpackedObject"]]:

820 """Fetch the pack data required for a set of revisions.

821

822 Args:

823 determine_wants: Function that takes a dictionary with heads

824 and returns the list of heads to fetch.

825 graph_walker: Object that can iterate over the list of revisions

826 to fetch and has an "ack" method that will be called to acknowledge

827 that a revision is present.

828 progress: Simple progress function that will be called with

829 updated progress strings.

830 get_tagged: Function that returns a dict of pointed-to sha ->

831 tag sha for including tags.

832 depth: Shallow fetch depth

833 Returns: count and iterator over pack data

834 """

835 missing_objects = self.find_missing_objects(

836 determine_wants, graph_walker, progress, get_tagged=get_tagged, depth=depth

837 )

838 if missing_objects is None:

839 return 0, iter([])

840 remote_has = missing_objects.get_remote_has()

841 object_ids = list(missing_objects)

842 return len(object_ids), generate_unpacked_objects(

843 self.object_store, object_ids, progress=progress, other_haves=remote_has

844 )

845

846 def find_missing_objects(

847 self,

848 determine_wants: Callable[[Mapping[Ref, ObjectID], int | None], list[ObjectID]],

849 graph_walker: "GraphWalker",

850 progress: Callable[[bytes], None] | None,

851 *,

852 get_tagged: Callable[[], dict[ObjectID, ObjectID]] | None = None,

853 depth: int | None = None,

854 ) -> MissingObjectFinder | None:

855 """Fetch the missing objects required for a set of revisions.

856

857 Args:

858 determine_wants: Function that takes a dictionary with heads

859 and returns the list of heads to fetch.

860 graph_walker: Object that can iterate over the list of revisions

861 to fetch and has an "ack" method that will be called to acknowledge

862 that a revision is present.

863 progress: Simple progress function that will be called with

864 updated progress strings.

865 get_tagged: Function that returns a dict of pointed-to sha ->

866 tag sha for including tags.

867 depth: Shallow fetch depth

868 Returns: iterator over objects, with __len__ implemented

869 """

870 # Filter out refs pointing to missing objects to avoid errors downstream.

871 # This makes Dulwich more robust when dealing with broken refs on disk.

872 # Previously serialize_refs() did this filtering as a side-effect.

873 all_refs = self.get_refs()

874 refs: dict[Ref, ObjectID] = {}

875 for ref, sha in all_refs.items():

876 if sha in self.object_store:

877 refs[ref] = sha

878 else:

879 logger.warning(

880 "ref %s points at non-present sha %s",

881 ref.decode("utf-8", "replace"),

882 sha.decode("ascii"),

883 )

884

885 wants = determine_wants(refs, depth)

886 if not isinstance(wants, list):

887 raise TypeError("determine_wants() did not return a list")

888

889 current_shallow = set(getattr(graph_walker, "shallow", set()))

890

891 unshallow: set[ObjectID] = set()

892 if depth not in (None, 0):

893 assert depth is not None

894 shallow, not_shallow = find_shallow(self.object_store, wants, depth)

895 # Only update if graph_walker has shallow attribute

896 walker_shallow: set[ObjectID] | None = getattr(

897 graph_walker, "shallow", None

898 )

899 if walker_shallow is not None:

900 unshallow = not_shallow & current_shallow

901 # Commits that were a shallow boundary but are now being

902 # deepened past must drop out of the boundary, otherwise their

903 # parents stay unreachable and never get transferred.

904 walker_shallow.update(shallow - not_shallow)

905 walker_shallow.difference_update(unshallow)

906 new_shallow = walker_shallow - current_shallow

907 setattr(graph_walker, "unshallow", unshallow)

908 update_shallow = getattr(graph_walker, "update_shallow", None)

909 if update_shallow is not None:

910 update_shallow(new_shallow, unshallow)

911 else:

912 unshallow = getattr(graph_walker, "unshallow", set())

913

914 if wants == []:

915 # TODO(dborowitz): find a way to short-circuit that doesn't change

916 # this interface.

917

918 if getattr(graph_walker, "shallow", set()) or unshallow:

919 # Do not send a pack in shallow short-circuit path

920 return None

921

922 # Return an actual MissingObjectFinder with empty wants

923 return MissingObjectFinder(

924 self.object_store,

925 haves=[],

926 wants=[],

927 )

928

929 # If the graph walker is set up with an implementation that can

930 # ACK/NAK to the wire, it will write data to the client through

931 # this call as a side-effect.

932 haves = self.object_store.find_common_revisions(graph_walker)

933

934 # Deal with shallow requests separately because the haves do

935 # not reflect what objects are missing

936 if getattr(graph_walker, "shallow", set()) or unshallow:

937 # TODO: filter the haves commits from iter_shas. the specific

938 # commits aren't missing.

939 haves = []

940

941 parents_provider = ParentsProvider(

942 self.object_store,

943 shallows=getattr(graph_walker, "shallow", current_shallow),

944 )

945

946 def get_parents(commit: Commit) -> list[ObjectID]:

947 """Get parents for a commit using the parents provider.

948

949 Args:

950 commit: Commit object

951

952 Returns:

953 List of parent commit SHAs

954 """

955 return parents_provider.get_parents(commit.id, commit)

956

957 return MissingObjectFinder(

958 self.object_store,

959 haves=haves,

960 wants=wants,

961 shallow=getattr(graph_walker, "shallow", set()),

962 progress=progress,

963 get_tagged=get_tagged,

964 get_parents=get_parents,

965 )

966

967 def generate_pack_data(

968 self,

969 have: set[ObjectID],

970 want: set[ObjectID],

971 *,

972 shallow: set[ObjectID] | None = None,

973 progress: Callable[[str], None] | None = None,

974 ofs_delta: bool | None = None,

975 ) -> tuple[int, Iterator["UnpackedObject"]]:

976 """Generate pack data objects for a set of wants/haves.

977

978 Args:

979 have: List of SHA1s of objects that should not be sent

980 want: List of SHA1s of objects that should be sent

981 shallow: Set of shallow commit SHA1s to skip (defaults to repo's shallow commits)

982 ofs_delta: Whether OFS deltas can be included

983 progress: Optional progress reporting method

984 """

985 if shallow is None:

986 shallow = self.get_shallow()

987 return self.object_store.generate_pack_data(

988 have,

989 want,

990 shallow=shallow,

991 progress=progress,

992 ofs_delta=ofs_delta if ofs_delta is not None else DEFAULT_OFS_DELTA,

993 )

994

995 def get_graph_walker(

996 self, heads: list[ObjectID] | None = None

997 ) -> ObjectStoreGraphWalker:

998 """Retrieve a graph walker.

999

1000 A graph walker is used by a remote repository (or proxy)

1001 to find out which objects are present in this repository.

1002

1003 Args:

1004 heads: Repository heads to use (optional)

1005 Returns: A graph walker object

1006 """

1007 if heads is None:

1008 heads = [

1009 sha

1010 for sha in self.refs.as_dict(Ref(b"refs/heads")).values()

1011 if sha in self.object_store

1012 ]

1013 parents_provider = ParentsProvider(self.object_store)

1014 return ObjectStoreGraphWalker(

1015 heads,

1016 parents_provider.get_parents,

1017 shallow=self.get_shallow(),

1018 update_shallow=self.update_shallow,

1019 )

1020

1021 def get_refs(self) -> dict[Ref, ObjectID]:

1022 """Get dictionary with all refs.

1023

1024 Returns: A ``dict`` mapping ref names to SHA1s

1025 """

1026 return self.refs.as_dict()

1027

1028 def head(self) -> ObjectID:

1029 """Return the SHA1 pointed at by HEAD."""

1030 # TODO: move this method to WorkTree

1031 return self.refs[HEADREF]

1032

1033 def _get_object(self, sha: ObjectID | RawObjectID, cls: type[T]) -> T:

1034 assert len(sha) in (

1035 self.object_format.oid_length,

1036 self.object_format.hex_length,

1037 )

1038 ret = self.get_object(sha)

1039 if not isinstance(ret, cls):

1040 if cls is Commit:

1041 raise NotCommitError(ret.id)

1042 elif cls is Blob:

1043 raise NotBlobError(ret.id)

1044 elif cls is Tree:

1045 raise NotTreeError(ret.id)

1046 elif cls is Tag:

1047 raise NotTagError(ret.id)

1048 else:

1049 raise Exception(f"Type invalid: {ret.type_name!r} != {cls.type_name!r}")

1050 return ret

1051

1052 def get_object(self, sha: ObjectID | RawObjectID) -> ShaFile:

1053 """Retrieve the object with the specified SHA.

1054

1055 Args:

1056 sha: SHA to retrieve

1057 Returns: A ShaFile object

1058 Raises:

1059 KeyError: when the object can not be found

1060 """

1061 return self.object_store[sha]

1062

1063 def parents_provider(self) -> ParentsProvider:

1064 """Get a parents provider for this repository.

1065

1066 Returns:

1067 ParentsProvider instance configured with grafts and shallows

1068 """

1069 return ParentsProvider(

1070 self.object_store,

1071 grafts=self._graftpoints,

1072 shallows=self.get_shallow(),

1073 )

1074

1075 def get_parents(

1076 self, sha: ObjectID, commit: Commit | None = None

1077 ) -> list[ObjectID]:

1078 """Retrieve the parents of a specific commit.

1079

1080 If the specific commit is a graftpoint, the graft parents

1081 will be returned instead.

1082

1083 Args:

1084 sha: SHA of the commit for which to retrieve the parents

1085 commit: Optional commit matching the sha

1086 Returns: List of parents

1087 """

1088 return self.parents_provider().get_parents(sha, commit)

1089

1090 def get_config(self) -> "ConfigFile":

1091 """Retrieve the config object.

1092

1093 Returns: `ConfigFile` object for the ``.git/config`` file.

1094 """

1095 raise NotImplementedError(self.get_config)

1096

1097 def get_worktree_config(self) -> "ConfigFile":

1098 """Retrieve the worktree config object."""

1099 raise NotImplementedError(self.get_worktree_config)

1100

1101 def get_description(self) -> bytes | None:

1102 """Retrieve the description for this repository.

1103

1104 Returns: Bytes with the description of the repository

1105 as set by the user.

1106 """

1107 raise NotImplementedError(self.get_description)

1108

1109 def set_description(self, description: bytes) -> None:

1110 """Set the description for this repository.

1111

1112 Args:

1113 description: Text to set as description for this repository.

1114 """

1115 raise NotImplementedError(self.set_description)

1116

1117 def get_rebase_state_manager(self) -> "RebaseStateManager":

1118 """Get the appropriate rebase state manager for this repository.

1119

1120 Returns: RebaseStateManager instance

1121 """

1122 raise NotImplementedError(self.get_rebase_state_manager)

1123

1124 def get_blob_normalizer(

1125 self, config: "Config | None" = None

1126 ) -> "FilterBlobNormalizer":

1127 """Return a BlobNormalizer object for checkin/checkout operations.

1128

1129 Args:

1130 config: Configuration to consult for filter setup. If None,

1131 implementations may fall back to ``self.get_config_stack()``.

1132

1133 Returns: BlobNormalizer instance

1134 """

1135 raise NotImplementedError(self.get_blob_normalizer)

1136

1137 def get_gitattributes(self, tree: bytes | None = None) -> "GitAttributes":

1138 """Read gitattributes for the repository.

1139

1140 Args:

1141 tree: Tree SHA to read .gitattributes from (defaults to HEAD)

1142

1143 Returns:

1144 GitAttributes object that can be used to match paths

1145 """

1146 raise NotImplementedError(self.get_gitattributes)

1147

1148 def get_config_stack(self) -> "StackedConfig":

1149 """Return a config stack for this repository.

1150

1151 This stack accesses the configuration for both this repository

1152 itself (.git/config) and the global configuration, which usually

1153 lives in ~/.gitconfig.

1154

1155 Returns: `Config` instance for this repository

1156 """

1157 from .config import ConfigFile, StackedConfig

1158

1159 local_config = self.get_config()

1160 backends: list[ConfigFile] = [local_config]

1161 if local_config.get_boolean((b"extensions",), b"worktreeconfig", False):

1162 backends.append(self.get_worktree_config())

1163

1164 backends += StackedConfig.default_backends()

1165 return StackedConfig(backends, writable=local_config)

1166

1167 def get_shallow(self) -> set[ObjectID]:

1168 """Get the set of shallow commits.

1169

1170 Returns: Set of shallow commits.

1171 """

1172 f = self.get_named_file("shallow")

1173 if f is None:

1174 return set()

1175 with f:

1176 shallow: set[ObjectID] = set()

1177 for line in f:

1178 sha = line.strip()

1179 if not sha:

1180 continue

1181 check_hexsha(sha, "invalid shallow object id")

1182 shallow.add(ObjectID(sha))

1183 return shallow

1184

1185 def update_shallow(

1186 self, new_shallow: set[ObjectID] | None, new_unshallow: set[ObjectID] | None

1187 ) -> None:

1188 """Update the list of shallow objects.

1189

1190 Args:

1191 new_shallow: Newly shallow objects

1192 new_unshallow: Newly no longer shallow objects

1193 """

1194 for sha in (*(new_shallow or ()), *(new_unshallow or ())):

1195 check_hexsha(sha, "invalid shallow object id")

1196 shallow = self.get_shallow()

1197 if new_shallow:

1198 shallow.update(new_shallow)

1199 if new_unshallow:

1200 shallow.difference_update(new_unshallow)

1201 if shallow:

1202 self._put_named_file("shallow", b"".join([sha + b"\n" for sha in shallow]))

1203 else:

1204 self._del_named_file("shallow")

1205

1206 def get_peeled(self, ref: Ref) -> ObjectID:

1207 """Get the peeled value of a ref.

1208

1209 Args:

1210 ref: The refname to peel.

1211 Returns: The fully-peeled SHA1 of a tag object, after peeling all

1212 intermediate tags; if the original ref does not point to a tag,

1213 this will equal the original SHA1.

1214 """

1215 cached = self.refs.get_peeled(ref)

1216 if cached is not None:

1217 return cached

1218 return peel_sha(self.object_store, self.refs[ref])[1].id

1219

1220 @property

1221 def notes(self) -> "Notes":

1222 """Access notes functionality for this repository.

1223

1224 Returns:

1225 Notes object for accessing notes

1226 """

1227 from .notes import Notes

1228

1229 return Notes(self.object_store, self.refs)

1230

1231 def get_walker(

1232 self,

1233 include: Sequence[ObjectID] | None = None,

1234 exclude: Sequence[ObjectID] | None = None,

1235 order: str = "date",

1236 reverse: bool = False,

1237 max_entries: int | None = None,

1238 paths: Sequence[bytes] | None = None,

1239 rename_detector: "RenameDetector | None" = None,

1240 follow: bool = False,

1241 since: int | None = None,

1242 until: int | None = None,

1243 queue_cls: type | None = None,

1244 ) -> "Walker":

1245 """Obtain a walker for this repository.

1246

1247 Args:

1248 include: Iterable of SHAs of commits to include along with their

1249 ancestors. Defaults to [HEAD]

1250 exclude: Iterable of SHAs of commits to exclude along with their

1251 ancestors, overriding includes.

1252 order: ORDER_* constant specifying the order of results.

1253 Anything other than ORDER_DATE may result in O(n) memory usage.

1254 reverse: If True, reverse the order of output, requiring O(n)

1255 memory.

1256 max_entries: The maximum number of entries to yield, or None for

1257 no limit.

1258 paths: Iterable of file or subtree paths to show entries for.

1259 rename_detector: diff.RenameDetector object for detecting

1260 renames.

1261 follow: If True, follow path across renames/copies. Forces a

1262 default rename_detector.

1263 since: Timestamp to list commits after.

1264 until: Timestamp to list commits before.

1265 queue_cls: A class to use for a queue of commits, supporting the

1266 iterator protocol. The constructor takes a single argument, the Walker.

1267

1268 Returns: A `Walker` object

1269 """

1270 from .walk import Walker, _CommitTimeQueue

1271

1272 if include is None:

1273 include = [self.head()]

1274

1275 # Pass all arguments to Walker explicitly to avoid type issues with **kwargs

1276 return Walker(

1277 self.object_store,

1278 include,

1279 exclude=exclude,

1280 order=order,

1281 reverse=reverse,

1282 max_entries=max_entries,

1283 paths=paths,

1284 rename_detector=rename_detector,

1285 follow=follow,

1286 since=since,

1287 until=until,

1288 get_parents=lambda commit: self.get_parents(commit.id, commit),

1289 queue_cls=queue_cls if queue_cls is not None else _CommitTimeQueue,

1290 )

1291

1292 def __getitem__(self, name: ObjectID | Ref | bytes) -> "ShaFile":

1293 """Retrieve a Git object by SHA1 or ref.

1294

1295 Args:

1296 name: A Git object SHA1 or a ref name

1297 Returns: A `ShaFile` object, such as a Commit or Blob

1298 Raises:

1299 KeyError: when the specified ref or object does not exist

1300 """

1301 if not isinstance(name, bytes):

1302 raise TypeError(f"'name' must be bytestring, not {type(name).__name__:.80}")

1303 # If it looks like a ref name, only try refs

1304 if name == b"HEAD" or name.startswith(b"refs/"):

1305 try:

1306 return self.object_store[self.refs[Ref(name)]]

1307 except (RefFormatError, KeyError):

1308 pass

1309 # Otherwise, try as object ID if length matches

1310 if len(name) in (

1311 self.object_store.object_format.oid_length,

1312 self.object_store.object_format.hex_length,

1313 ):

1314 try:

1315 return self.object_store[

1316 ObjectID(name)

1317 if len(name) == self.object_store.object_format.hex_length

1318 else RawObjectID(name)

1319 ]

1320 except (KeyError, ValueError):

1321 pass

1322 # If nothing worked, raise KeyError

1323 raise KeyError(name)

1324

1325 def __contains__(self, name: bytes) -> bool:

1326 """Check if a specific Git object or ref is present.

1327

1328 Args:

1329 name: Git object SHA1/SHA256 or ref name

1330 """

1331 # Check if it's a binary or hex SHA

1332 if len(name) == self.object_format.hex_length and valid_hexsha(name):

1333 return ObjectID(name) in self.object_store or Ref(name) in self.refs

1334 return Ref(name) in self.refs

1335

1336 def __setitem__(self, name: bytes, value: ShaFile | bytes) -> None:

1337 """Set a ref.

1338

1339 Args:

1340 name: ref name

1341 value: Ref value - either a ShaFile object, or a hex sha

1342 """

1343 if name.startswith(b"refs/") or name == HEADREF:

1344 ref_name = Ref(name)

1345 if isinstance(value, ShaFile):

1346 self.refs[ref_name] = value.id

1347 elif isinstance(value, bytes):

1348 self.refs[ref_name] = ObjectID(value)

1349 else:

1350 raise TypeError(value)

1351 else:

1352 raise ValueError(name)

1353

1354 def __delitem__(self, name: bytes) -> None:

1355 """Remove a ref.

1356

1357 Args:

1358 name: Name of the ref to remove

1359 """

1360 if name.startswith(b"refs/") or name == HEADREF:

1361 del self.refs[Ref(name)]

1362 else:

1363 raise ValueError(name)

1364

1365 def _get_user_identity(

1366 self, config: "StackedConfig", kind: str | None = None

1367 ) -> bytes:

1368 """Determine the identity to use for new commits."""

1369 warnings.warn(

1370 "use get_user_identity() rather than Repo._get_user_identity",

1371 DeprecationWarning,

1372 )

1373 return get_user_identity(config)

1374

1375 def _add_graftpoints(

1376 self, updated_graftpoints: dict[ObjectID, list[ObjectID]]

1377 ) -> None:

1378 """Add or modify graftpoints.

1379

1380 Args:

1381 updated_graftpoints: Dict of commit shas to list of parent shas

1382 """

1383 # Simple validation

1384 for commit, parents in updated_graftpoints.items():

1385 for sha in [commit, *parents]:

1386 check_hexsha(sha, "Invalid graftpoint")

1387

1388 self._graftpoints.update(updated_graftpoints)

1389

1390 def _remove_graftpoints(self, to_remove: Sequence[ObjectID] = ()) -> None:

1391 """Remove graftpoints.

1392

1393 Args:

1394 to_remove: List of commit shas

1395 """

1396 for sha in to_remove:

1397 del self._graftpoints[sha]

1398

1399 def _read_heads(self, name: str) -> list[ObjectID]:

1400 f = self.get_named_file(name)

1401 if f is None:

1402 return []

1403 with f:

1404 return [ObjectID(line.strip()) for line in f.readlines() if line.strip()]

1405

1406 def get_worktree(self) -> "WorkTree":

1407 """Get the working tree for this repository.

1408

1409 Returns:

1410 WorkTree instance for performing working tree operations

1411

1412 Raises:

1413 NotImplementedError: If the repository doesn't support working trees

1414 """

1415 raise NotImplementedError(

1416 "Working tree operations not supported by this repository type"

1417 )

1418

1419

1420def read_gitfile(f: BinaryIO) -> str:

1421 """Read a ``.git`` file.

1422

1423 The first line of the file should start with "gitdir: "

1424

1425 Args:

1426 f: File-like object to read from

1427 Returns: A path

1428 """

1429 cs = f.read()

1430 if not cs.startswith(b"gitdir: "):

1431 raise ValueError("Expected file to start with 'gitdir: '")

1432 return cs[len(b"gitdir: ") :].rstrip(b"\r\n").decode("utf-8")

1433

1434

1435class UnsupportedVersion(Exception):

1436 """Unsupported repository version."""

1437

1438 def __init__(self, version: int) -> None:

1439 """Initialize UnsupportedVersion exception.

1440

1441 Args:

1442 version: The unsupported repository version

1443 """

1444 self.version = version

1445

1446

1447class UnsupportedExtension(Exception):

1448 """Unsupported repository extension."""

1449

1450 def __init__(self, extension: str) -> None:

1451 """Initialize UnsupportedExtension exception.

1452

1453 Args:

1454 extension: The unsupported repository extension

1455 """

1456 self.extension = extension

1457

1458

1459class InvalidWorktreeConfiguration(Exception):

1460 """core.worktree is set on a bare repository."""

1461

1462

1463class Repo(BaseRepo):

1464 """A git repository backed by local disk.

1465

1466 To open an existing repository, call the constructor with

1467 the path of the repository.

1468

1469 To create a new repository, use the Repo.init class method.

1470

1471 Note that a repository object may hold on to resources such

1472 as file handles for performance reasons; call .close() to free

1473 up those resources.

1474

1475 Attributes:

1476 path: Path to the working copy (if it exists) or repository control

1477 directory (if the repository is bare). ``core.worktree`` overrides

1478 this, so it is not necessarily the parent of the control directory.

1479 bare: Whether this is a bare repository

1480 """

1481

1482 path: str

1483 bare: bool

1484 object_store: DiskObjectStore

1485 filter_context: "FilterContext | None"

1486 _index_file_override: "str | None"

1487

1488 def __init__(

1489 self,

1490 root: str | bytes | os.PathLike[str] | None = None,

1491 object_store: PackBasedObjectStore | None = None,

1492 bare: bool | None = None,

1493 *,

1494 controldir: str | bytes | os.PathLike[str] | None = None,

1495 commondir: str | bytes | os.PathLike[str] | None = None,

1496 worktree: str | bytes | os.PathLike[str] | None = None,

1497 object_directory: str | bytes | os.PathLike[str] | None = None,

1498 alternates: "Iterable[str | os.PathLike[str]] | None" = None,

1499 index_file: str | bytes | os.PathLike[str] | None = None,

1500 ) -> None:

1501 """Open a repository on disk.

1502

1503 Args:

1504 root: Path to the repository's root. Optional if ``controldir`` is

1505 given, in which case ``self.path`` defaults to ``worktree`` when

1506 provided and to ``controldir`` otherwise. ``core.worktree`` in

1507 config may still override it.

1508 object_store: ObjectStore to use; if omitted, we use the

1509 repository's default object store

1510 bare: True if this is a bare repository.

1511 controldir: Explicit path to the control directory (analogous to

1512 ``GIT_DIR``). When set, discovery based on ``root`` is skipped.

1513 commondir: Explicit path to the common directory (analogous to

1514 ``GIT_COMMON_DIR``). Relative paths are resolved against the

1515 control directory.

1516 worktree: Explicit worktree path (analogous to ``GIT_WORK_TREE``).

1517 Forces the repository to be treated as non-bare.

1518 object_directory: Explicit path to the primary object store

1519 (analogous to ``GIT_OBJECT_DIRECTORY``). Overrides

1520 ``<common_dir>/objects``.

1521 alternates: Extra alternate object directory paths (analogous to

1522 ``GIT_ALTERNATE_OBJECT_DIRECTORIES``). Appended to any listed

1523 in ``objects/info/alternates``.

1524 index_file: Path to the index file (analogous to

1525 ``GIT_INDEX_FILE``). Overrides ``<controldir>/index``.

1526 """

1527 if controldir is None:

1528 if root is None:

1529 raise TypeError("Repo() requires either root or controldir")

1530 root = os.fspath(root)

1531 if isinstance(root, bytes):

1532 root = os.fsdecode(root)

1533 hidden_path = os.path.join(root, CONTROLDIR)

1534 if worktree is not None:

1535 bare = False

1536 if bare is None:

1537 if os.path.isfile(hidden_path) or os.path.isdir(

1538 os.path.join(hidden_path, OBJECTDIR)

1539 ):

1540 bare = False

1541 elif os.path.isdir(os.path.join(root, OBJECTDIR)) and os.path.isdir(

1542 os.path.join(root, REFSDIR)

1543 ):

1544 bare = True

1545 else:

1546 raise NotGitRepository(

1547 "No git repository was found at {path}".format(

1548 **dict(path=root)

1549 )

1550 )

1551

1552 self.bare = bare

1553 if bare is False:

1554 if os.path.isfile(hidden_path):

1555 with open(hidden_path, "rb") as f:

1556 gitfile_path = read_gitfile(f)

1557 self._controldir = os.path.join(root, gitfile_path)

1558 else:

1559 self._controldir = hidden_path

1560 else:

1561 self._controldir = root

1562 else:

1563 controldir = os.fspath(controldir)

1564 if isinstance(controldir, bytes):

1565 controldir = os.fsdecode(controldir)

1566 self._controldir = controldir

1567 if root is None:

1568 # Without an explicit root, default to the worktree if given,

1569 # else to the control dir (bare layout). We deliberately do

1570 # not fall back to os.getcwd() here: a library-level ``Repo``

1571 # should not silently latch onto the current directory.

1572 # Callers that want git's ``GIT_DIR``-implies-cwd-worktree

1573 # semantics pass ``worktree`` explicitly.

1574 root = os.fspath(worktree) if worktree is not None else controldir

1575 else:

1576 root = os.fspath(root)

1577 if isinstance(root, bytes):

1578 root = os.fsdecode(root)

1579 if worktree is not None:

1580 bare = False

1581 if bare is None:

1582 # With an explicit control dir and no worktree override,

1583 # default to bare. core.bare / core.worktree parsed from

1584 # config below may still flip this.

1585 bare = True

1586 self.bare = bare

1587 if commondir is not None:

1588 commondir_path = os.fspath(commondir)

1589 if isinstance(commondir_path, bytes):

1590 commondir_path = os.fsdecode(commondir_path)

1591 if not os.path.isabs(commondir_path):

1592 commondir_path = os.path.join(self._controldir, commondir_path)

1593 self._commondir = commondir_path

1594 else:

1595 commondir_file = self.get_named_file(COMMONDIR)

1596 if commondir_file is not None:

1597 with commondir_file:

1598 self._commondir = os.path.join(

1599 self.controldir(),

1600 os.fsdecode(commondir_file.read().rstrip(b"\r\n")),

1601 )

1602 else:

1603 self._commondir = self._controldir

1604 self.path = root

1605 if worktree is not None:

1606 worktree_path = os.fspath(worktree)

1607 if isinstance(worktree_path, bytes):

1608 worktree_path = os.fsdecode(worktree_path)

1609 self.path = worktree_path

1610

1611 if index_file is not None:

1612 index_file_str = os.fspath(index_file)

1613 if isinstance(index_file_str, bytes):

1614 index_file_str = os.fsdecode(index_file_str)

1615 self._index_file_override = index_file_str

1616 else:

1617 self._index_file_override = None

1618

1619 # Initialize refs early so they're available for config condition matchers

1620 self.refs = DiskRefsContainer(

1621 self.commondir(), self._controldir, logger=self._write_reflog

1622 )

1623

1624 # Initialize worktrees container

1625 from .worktree import WorkTreeContainer

1626

1627 self.worktrees = WorkTreeContainer(self)

1628

1629 config = self.get_config()

1630 try:

1631 repository_format_version = config.get("core", "repositoryformatversion")

1632 format_version = (

1633 0

1634 if repository_format_version is None

1635 else int(repository_format_version)

1636 )

1637 except KeyError:

1638 format_version = 0

1639

1640 if format_version not in (0, 1):

1641 raise UnsupportedVersion(format_version)

1642

1643 try:

1644 configured_worktree = config.get((b"core",), b"worktree")

1645 except KeyError:

1646 configured_worktree = None

1647 if configured_worktree is not None:

1648 # A repository is bare because core.bare says so, not because it

1649 # was opened at its control directory: a submodule or a

1650 # --separate-git-dir repository is opened that way but still has a

1651 # working tree, named here.

1652 if config.get_boolean((b"core",), b"bare", False):

1653 raise InvalidWorktreeConfiguration(

1654 "core.bare and core.worktree are incompatible"

1655 )

1656 self.bare = False

1657 # An explicit worktree argument takes precedence over core.worktree.

1658 if worktree is None:

1659 # Relative paths are resolved against the control directory,

1660 # not against the current directory or the repository root.

1661 self.path = os.path.join(

1662 self._controldir, os.fsdecode(configured_worktree)

1663 )

1664

1665 # Track extensions we encounter

1666 has_reftable_extension = False

1667 for extension, value in config.items((b"extensions",)):

1668 if extension.lower() == b"refstorage":

1669 if value == b"reftable":

1670 has_reftable_extension = True

1671 else:

1672 raise UnsupportedExtension(f"refStorage = {value.decode()}")

1673 elif extension.lower() not in (

1674 b"worktreeconfig",

1675 b"objectformat",

1676 b"relativeworktrees",

1677 ):

1678 raise UnsupportedExtension(extension.decode("utf-8"))

1679

1680 if object_store is None:

1681 # Get shared repository permissions from config

1682 try:

1683 shared_value = config.get(("core",), "sharedRepository")

1684 shared_perm = parse_shared_repository(shared_value)

1685 except KeyError:

1686 shared_perm = None

1687

1688 if object_directory is not None:

1689 object_dir_path = os.fspath(object_directory)

1690 if isinstance(object_dir_path, bytes):

1691 object_dir_path = os.fsdecode(object_dir_path)

1692 else:

1693 object_dir_path = os.path.join(self.commondir(), OBJECTDIR)

1694 object_store = DiskObjectStore.from_config(

1695 object_dir_path,

1696 config,

1697 shared_perm=shared_perm,

1698 alternates=alternates,

1699 )

1700

1701 # Use reftable if extension is configured

1702 if has_reftable_extension:

1703 from .reftable import ReftableRefsContainer

1704

1705 self.refs = ReftableRefsContainer(self.commondir())

1706 # Update worktrees container after refs change

1707 self.worktrees = WorkTreeContainer(self)

1708 BaseRepo.__init__(self, object_store, self.refs)

1709

1710 # Determine hash algorithm from config if not already set

1711 if self.object_format is None:

1712 from .object_format import DEFAULT_OBJECT_FORMAT, get_object_format

1713

1714 if format_version == 1:

1715 try:

1716 object_format = config.get((b"extensions",), b"objectformat")

1717 self.object_format = get_object_format(

1718 object_format.decode("ascii")

1719 )

1720 except KeyError:

1721 self.object_format = DEFAULT_OBJECT_FORMAT

1722 else:

1723 self.object_format = DEFAULT_OBJECT_FORMAT

1724

1725 self._graftpoints = {}

1726 graft_file = self.get_named_file(

1727 os.path.join("info", "grafts"), basedir=self.commondir()

1728 )

1729 if graft_file:

1730 with graft_file:

1731 self._graftpoints.update(parse_graftpoints(graft_file))

1732 graft_file = self.get_named_file("shallow", basedir=self.commondir())

1733 if graft_file:

1734 with graft_file:

1735 self._graftpoints.update(parse_graftpoints(graft_file))

1736

1737 self.hooks["pre-commit"] = PreCommitShellHook(self.path, self.controldir())

1738 self.hooks["commit-msg"] = CommitMsgShellHook(self.controldir())

1739 self.hooks["post-commit"] = PostCommitShellHook(self.controldir())

1740 self.hooks["pre-receive"] = PreReceiveShellHook(self.controldir())

1741 self.hooks["update"] = UpdateShellHook(self.controldir())

1742 self.hooks["post-receive"] = PostReceiveShellHook(self.controldir())

1743

1744 # Initialize filter context as None, will be created lazily

1745 self.filter_context = None

1746

1747 def get_worktree(self) -> "WorkTree":

1748 """Get the working tree for this repository.

1749

1750 Returns:

1751 WorkTree instance for performing working tree operations

1752 """

1753 from .worktree import WorkTree

1754

1755 return WorkTree(self, self.path)

1756

1757 def _write_reflog(

1758 self,

1759 ref: bytes,

1760 old_sha: bytes,

1761 new_sha: bytes,

1762 committer: bytes | None,

1763 timestamp: int | None,

1764 timezone: int | None,

1765 message: bytes,

1766 ) -> None:

1767 from .reflog import format_reflog_line

1768

1769 path = self._reflog_path(ref)

1770

1771 # Get shared repository permissions

1772 shared_perm = self._get_shared_repository_permissions()

1773

1774 # Create directory with appropriate permissions

1775 parent_dir = os.path.dirname(path)

1776 # Create directory tree, setting permissions on each level if needed

1777 parts = []

1778 current = parent_dir

1779 while current and not os.path.exists(current):

1780 parts.append(current)

1781 current = os.path.dirname(current)

1782 parts.reverse()

1783 for part in parts:

1784 os.mkdir(part)

1785 adjust_shared_perm(part, shared_perm)

1786 if committer is None:

1787 config = self.get_config_stack()

1788 committer = get_user_identity(config)

1789 check_user_identity(committer)

1790 if timestamp is None:

1791 timestamp = int(time.time())

1792 if timezone is None:

1793 timezone = 0 # FIXME

1794 with open(path, "ab") as f:

1795 f.write(

1796 format_reflog_line(

1797 old_sha, new_sha, committer, timestamp, timezone, message

1798 )

1799 + b"\n"

1800 )

1801

1802 # Always adjust, so that permissions are right even if the file

1803 # already existed.

1804 adjust_shared_perm(path, shared_perm)

1805

1806 def _reflog_path(self, ref: bytes) -> str:

1807 if ref.startswith((b"main-worktree/", b"worktrees/")):

1808 raise NotImplementedError(f"refs {ref.decode()} are not supported")

1809

1810 base = self.controldir() if is_per_worktree_ref(ref) else self.commondir()

1811 return os.path.join(base, "logs", os.fsdecode(ref))

1812

1813 def read_reflog(self, ref: bytes) -> Generator[reflog.Entry, None, None]:

1814 """Read reflog entries for a reference.

1815

1816 Args:

1817 ref: Reference name (e.g. b'HEAD', b'refs/heads/master')

1818

1819 Yields:

1820 reflog.Entry objects in chronological order (oldest first)

1821 """

1822 from .reflog import read_reflog

1823

1824 path = self._reflog_path(ref)

1825 try:

1826 with open(path, "rb") as f:

1827 yield from read_reflog(f)

1828 except FileNotFoundError:

1829 return

1830

1831 @classmethod

1832 def discover(

1833 cls,

1834 start: str | bytes | os.PathLike[str] = ".",

1835 *,

1836 ceiling_dirs: "Iterable[str | os.PathLike[str]] | None" = None,

1837 across_filesystem: bool = True,

1838 ) -> "Repo":

1839 """Iterate parent directories to discover a repository.

1840

1841 Return a Repo object for the first parent directory that looks like a

1842 Git repository.

1843

1844 Args:

1845 start: The directory to start discovery from (defaults to '.')

1846 ceiling_dirs: Iterable of paths that discovery must not cross

1847 (analogous to ``GIT_CEILING_DIRECTORIES``). The ceiling

1848 directories themselves are not searched. As in git, a ceiling

1849 matching ``start`` itself is ignored. Entries are resolved the

1850 same way as the walking path, so passing either a symlinked or

1851 a resolved form works.

1852 across_filesystem: Whether to keep walking up past a filesystem

1853 boundary (analogous to ``GIT_DISCOVERY_ACROSS_FILESYSTEM``).

1854 When False, discovery stops before entering a parent directory

1855 that lives on a different device than ``start``.

1856 """

1857 # Both sides of the ceiling comparison are resolved so that they are

1858 # in the same form: os.getcwd() already returns a resolved path on

1859 # POSIX, and on Windows realpath() expands 8.3 short names that

1860 # abspath() leaves alone. This also mirrors git, which walks up from

1861 # the kernel-resolved cwd.

1862 ceilings = (

1863 {

1864 os.path.normcase(os.path.realpath(os.fsdecode(os.fspath(p))))

1865 for p in ceiling_dirs

1866 }

1867 if ceiling_dirs is not None

1868 else set()

1869 )

1870 path = os.path.realpath(start)

1871 # Device of the starting directory, only tracked when we must not

1872 # cross filesystem boundaries. Errors stat'ing it propagate: if the

1873 # start directory is unreadable, discovery from it is meaningless.

1874 start_dev = None if across_filesystem else os.stat(path).st_dev

1875 first = True

1876 while True:

1877 if not first and os.path.normcase(path) in ceilings:

1878 break

1879 first = False

1880 try:

1881 return cls(path)

1882 except NotGitRepository:

1883 new_path, _tail = os.path.split(path)

1884 if new_path == path: # Root reached

1885 break

1886 if start_dev is not None:

1887 try:

1888 parent_dev = os.stat(new_path).st_dev

1889 except FileNotFoundError:

1890 # Raced with the parent being removed; nothing left

1891 # to walk up into.

1892 break

1893 if parent_dev != start_dev:

1894 break

1895 path = new_path

1896 start_str = os.fspath(start)

1897 if isinstance(start_str, bytes):

1898 start_str = start_str.decode("utf-8")

1899 raise NotGitRepository(f"No git repository was found at {start_str}")

1900

1901 def controldir(self) -> str:

1902 """Return the path of the control directory."""

1903 return self._controldir

1904

1905 def commondir(self) -> str:

1906 """Return the path of the common directory.

1907

1908 For a main working tree, it is identical to controldir().

1909

1910 For a linked working tree, it is the control directory of the

1911 main working tree.

1912 """

1913 return self._commondir

1914

1915 def _determine_file_mode(self) -> bool:

1916 """Probe the file-system to determine whether permissions can be trusted.

1917

1918 Returns: True if permissions can be trusted, False otherwise.

1919 """

1920 fname = os.path.join(self.path, ".probe-permissions")

1921 with open(fname, "w") as f:

1922 f.write("")

1923

1924 st1 = os.lstat(fname)

1925 try:

1926 os.chmod(fname, st1.st_mode ^ stat.S_IXUSR)

1927 except PermissionError:

1928 return False

1929 st2 = os.lstat(fname)

1930

1931 os.unlink(fname)

1932

1933 mode_differs = st1.st_mode != st2.st_mode

1934 st2_has_exec = (st2.st_mode & stat.S_IXUSR) != 0

1935

1936 return mode_differs and st2_has_exec

1937

1938 def _determine_symlinks(self) -> bool:

1939 """Probe the filesystem to determine whether symlinks can be created.

1940

1941 Returns: True if symlinks can be created, False otherwise.

1942 """

1943 # TODO(jelmer): Actually probe disk / look at filesystem

1944 return sys.platform != "win32"

1945

1946 def _get_shared_repository_permissions(

1947 self,

1948 ) -> "SharedPerm | None":

1949 """Get the shared repository permission setting from config.

1950

1951 Returns:

1952 SharedPerm to apply, or None if not shared

1953 """

1954 try:

1955 config = self.get_config()

1956 value = config.get(("core",), "sharedRepository")

1957 return parse_shared_repository(value)

1958 except KeyError:

1959 return None

1960

1961 def _put_named_file(self, path: str, contents: bytes) -> None:

1962 """Write a file to the control dir with the given name and contents.

1963

1964 Args:

1965 path: The path to the file, relative to the control dir.

1966 contents: A string to write to the file.

1967 """

1968 path = path.lstrip(os.path.sep)

1969

1970 # Get shared repository permissions

1971 shared_perm = self._get_shared_repository_permissions()

1972

1973 # Create file with appropriate permissions

1974 full_path = os.path.join(self.controldir(), path)

1975 if shared_perm is not None:

1976 with GitFile(full_path, "wb", shared_perm=shared_perm) as f:

1977 f.write(contents)

1978 else:

1979 with GitFile(full_path, "wb") as f:

1980 f.write(contents)

1981

1982 def _del_named_file(self, path: str) -> None:

1983 try:

1984 os.unlink(os.path.join(self.controldir(), path))

1985 except FileNotFoundError:

1986 return

1987

1988 def get_named_file(

1989 self,

1990 path: str | bytes,

1991 basedir: str | None = None,

1992 ) -> BinaryIO | None:

1993 """Get a file from the control dir with a specific name.

1994

1995 Although the filename should be interpreted as a filename relative to

1996 the control dir in a disk-based Repo, the object returned need not be

1997 pointing to a file in that location.

1998

1999 Args:

2000 path: The path to the file, relative to the control dir.

2001 basedir: Optional argument that specifies an alternative to the

2002 control dir.

2003 Returns: An open file object, or None if the file does not exist.

2004 """

2005 # TODO(dborowitz): sanitize filenames, since this is used directly by

2006 # the dumb web serving code.

2007 if basedir is None:

2008 basedir = self.controldir()

2009 if isinstance(path, bytes):

2010 path = path.decode("utf-8")

2011 path = path.lstrip(os.path.sep)

2012 try:

2013 return open(os.path.join(basedir, path), "rb")

2014 except FileNotFoundError:

2015 return None

2016

2017 def index_path(self) -> str:

2018 """Return path to the index file."""

2019 if self._index_file_override is not None:

2020 return self._index_file_override

2021 return os.path.join(self.controldir(), INDEX_FILENAME)

2022

2023 def open_index(self, config: "Config | None" = None) -> "Index":

2024 """Open the index for this repository.

2025

2026 Args:

2027 config: Configuration to consult for index settings. If None,

2028 falls back to ``self.get_config_stack()``.

2029

2030 Raises:

2031 NoIndexPresent: If no index is present

2032 Returns: The matching `Index`

2033 """

2034 from .index import Index, make_path_normalizer

2035

2036 if not self.has_index():

2037 raise NoIndexPresent

2038

2039 if config is None:

2040 config = self.get_config_stack()

2041 many_files = config.get_boolean(b"feature", b"manyFiles", False)

2042 skip_hash = False

2043 index_version = None

2044

2045 if many_files:

2046 # When feature.manyFiles is enabled, set index.version=4 and index.skipHash=true

2047 try:

2048 index_version_str = config.get(b"index", b"version")

2049 index_version = int(index_version_str)

2050 except KeyError:

2051 index_version = 4 # Default to version 4 for manyFiles

2052 skip_hash = config.get_boolean(b"index", b"skipHash", True)

2053 else:

2054 # Check for explicit index settings

2055 try:

2056 index_version_str = config.get(b"index", b"version")

2057 index_version = int(index_version_str)

2058 except KeyError:

2059 index_version = None

2060 skip_hash = config.get_boolean(b"index", b"skipHash", False)

2061

2062 # Get shared repository permissions for index file

2063 shared_perm = self._get_shared_repository_permissions()

2064

2065 return Index(

2066 self.index_path(),

2067 skip_hash=skip_hash,

2068 version=index_version,

2069 shared_perm=shared_perm,

2070 path_normalizer=make_path_normalizer(config),

2071 )

2072

2073 def has_index(self) -> bool:

2074 """Check if an index is present."""

2075 # Bare repos must never have index files; non-bare repos may have a

2076 # missing index file, which is treated as empty.

2077 return not self.bare

2078

2079 def clone(

2080 self,

2081 target_path: str | bytes | os.PathLike[str],

2082 *,

2083 mkdir: bool = True,

2084 bare: bool = False,

2085 origin: bytes = b"origin",

2086 checkout: bool | None = None,

2087 branch: bytes | None = None,

2088 progress: Callable[[str], None] | None = None,

2089 depth: int | None = None,

2090 symlinks: bool | None = None,

2091 ) -> "Repo":

2092 """Clone this repository.

2093

2094 Args:

2095 target_path: Target path

2096 mkdir: Create the target directory

2097 bare: Whether to create a bare repository

2098 checkout: Whether or not to check-out HEAD after cloning

2099 origin: Base name for refs in target repository

2100 cloned from this repository

2101 branch: Optional branch or tag to be used as HEAD in the new repository

2102 instead of this repository's HEAD.

2103 progress: Optional progress function

2104 depth: Depth at which to fetch

2105 symlinks: Symlinks setting (default to autodetect)

2106 Returns: Created repository as `Repo`

2107 """

2108 encoded_path = os.fsencode(self.path)

2109

2110 if mkdir:

2111 os.mkdir(target_path)

2112

2113 try:

2114 if not bare:

2115 target = Repo.init(target_path, symlinks=symlinks)

2116 if checkout is None:

2117 checkout = True

2118 else:

2119 if checkout:

2120 raise ValueError("checkout and bare are incompatible")

2121 target = Repo.init_bare(target_path)

2122

2123 try:

2124 target_config = target.get_config()

2125 target_config.set((b"remote", origin), b"url", encoded_path)

2126 target_config.set(

2127 (b"remote", origin),

2128 b"fetch",

2129 b"+refs/heads/*:refs/remotes/" + origin + b"/*",

2130 )

2131 target_config.write_to_path()

2132

2133 ref_message = b"clone: from " + encoded_path

2134 self.fetch(target, depth=depth)

2135 target.refs.import_refs(

2136 Ref(b"refs/remotes/" + origin),

2137 self.refs.as_dict(Ref(b"refs/heads")),

2138 message=ref_message,

2139 )

2140 target.refs.import_refs(

2141 Ref(b"refs/tags"),

2142 self.refs.as_dict(Ref(b"refs/tags")),

2143 message=ref_message,

2144 )

2145

2146 head_chain, origin_sha = self.refs.follow(HEADREF)

2147 origin_head = head_chain[-1] if head_chain else None

2148 head: ObjectID | None = None

2149 if origin_sha and not origin_head:

2150 # set detached HEAD

2151 target.refs[HEADREF] = origin_sha

2152 head = origin_sha

2153 else:

2154 _set_origin_head(target.refs, origin, origin_head)

2155 head_ref = _set_default_branch(

2156 target.refs, origin, origin_head, branch, ref_message

2157 )

2158

2159 # Update target head

2160 if head_ref:

2161 head = _set_head(target.refs, head_ref, ref_message)

2162 if not bare:

2163 _set_branch_tracking(target.get_config(), head_ref, origin)

2164 else:

2165 head = None

2166

2167 if checkout and head is not None:

2168 target.get_worktree().reset_index(config=target.get_config_stack())

2169 except BaseException:

2170 target.close()

2171 raise

2172 except BaseException:

2173 if mkdir:

2174 import shutil

2175

2176 shutil.rmtree(target_path)

2177 raise

2178 return target

2179

2180 def _get_config_condition_matchers(self) -> dict[str, "ConditionMatcher"]:

2181 """Get condition matchers for includeIf conditions.

2182

2183 Returns a dict of condition prefix to matcher function.

2184 """

2185 from pathlib import Path

2186

2187 from .config import ConditionMatcher, match_glob_pattern

2188

2189 # Add gitdir matchers

2190 def match_gitdir(pattern: str, case_sensitive: bool = True) -> bool:

2191 """Match gitdir against a pattern.

2192

2193 Args:

2194 pattern: Pattern to match against

2195 case_sensitive: Whether to match case-sensitively

2196

2197 Returns:

2198 True if gitdir matches pattern

2199 """

2200 # Handle relative patterns (starting with ./)

2201 if pattern.startswith("./"):

2202 # Can't handle relative patterns without config directory context

2203 return False

2204

2205 # Normalize repository path

2206 try:

2207 repo_path = str(Path(self._controldir).resolve())

2208 except (OSError, ValueError):

2209 return False

2210

2211 # Expand ~ in pattern and normalize

2212 pattern = os.path.expanduser(pattern)

2213

2214 # Normalize pattern following Git's rules

2215 pattern = pattern.replace("\\", "/")

2216 if not pattern.startswith(("~/", "./", "/", "**")):

2217 # Check for Windows absolute path

2218 if len(pattern) >= 2 and pattern[1] == ":":

2219 pass

2220 else:

2221 pattern = "**/" + pattern

2222 if pattern.endswith("/"):

2223 pattern = pattern + "**"

2224

2225 # Use the existing _match_gitdir_pattern function

2226 from .config import _match_gitdir_pattern

2227

2228 pattern_bytes = pattern.encode("utf-8", errors="replace")

2229 repo_path_bytes = repo_path.encode("utf-8", errors="replace")

2230

2231 return _match_gitdir_pattern(

2232 repo_path_bytes, pattern_bytes, ignorecase=not case_sensitive

2233 )

2234

2235 # Add onbranch matcher

2236 def match_onbranch(pattern: str) -> bool:

2237 """Match current branch against a pattern.

2238

2239 Args:

2240 pattern: Pattern to match against

2241

2242 Returns:

2243 True if current branch matches pattern

2244 """

2245 try:

2246 # Get the current branch using refs

2247 ref_chain, _ = self.refs.follow(HEADREF)

2248 head_ref = ref_chain[-1] # Get the final resolved ref

2249 except KeyError:

2250 pass

2251 else:

2252 if head_ref and head_ref.startswith(b"refs/heads/"):

2253 # Extract branch name from ref

2254 branch = extract_branch_name(head_ref).decode(

2255 "utf-8", errors="replace"

2256 )

2257 return match_glob_pattern(branch, pattern)

2258 return False

2259

2260 matchers: dict[str, ConditionMatcher] = {

2261 "onbranch:": match_onbranch,

2262 "gitdir:": lambda pattern: match_gitdir(pattern, True),

2263 "gitdir/i:": lambda pattern: match_gitdir(pattern, False),

2264 }

2265

2266 return matchers

2267

2268 def get_worktree_config(self) -> "ConfigFile":

2269 """Get the worktree-specific config.

2270

2271 Returns:

2272 ConfigFile object for the worktree config

2273 """

2274 from .config import ConfigFile

2275

2276 path = os.path.join(self.commondir(), "config.worktree")

2277 try:

2278 # Pass condition matchers for includeIf evaluation

2279 condition_matchers = self._get_config_condition_matchers()

2280 return ConfigFile.from_path(path, condition_matchers=condition_matchers)

2281 except FileNotFoundError:

2282 cf = ConfigFile()

2283 cf.path = path

2284 return cf

2285

2286 def get_config(self) -> "ConfigFile":

2287 """Retrieve the config object.

2288

2289 Returns: `ConfigFile` object for the ``.git/config`` file.

2290 """

2291 from .config import ConfigFile

2292

2293 path = os.path.join(self._commondir, "config")

2294 try:

2295 # Pass condition matchers for includeIf evaluation

2296 condition_matchers = self._get_config_condition_matchers()

2297 return ConfigFile.from_path(path, condition_matchers=condition_matchers)

2298 except FileNotFoundError:

2299 ret = ConfigFile()

2300 ret.path = path

2301 return ret

2302

2303 def get_rebase_state_manager(self) -> "RebaseStateManager":

2304 """Get the appropriate rebase state manager for this repository.

2305

2306 Returns: DiskRebaseStateManager instance

2307 """

2308 import os

2309

2310 from .rebase import DiskRebaseStateManager

2311

2312 path = os.path.join(self.controldir(), "rebase-merge")

2313 return DiskRebaseStateManager(path)

2314

2315 def get_description(self) -> bytes | None:

2316 """Retrieve the description of this repository.

2317

2318 Returns: Description as bytes or None.

2319 """

2320 path = os.path.join(self._controldir, "description")

2321 try:

2322 with GitFile(path, "rb") as f:

2323 return f.read()

2324 except FileNotFoundError:

2325 return None

2326

2327 def __repr__(self) -> str:

2328 """Return string representation of this repository."""

2329 return f"<Repo at {self.path!r}>"

2330

2331 def set_description(self, description: bytes) -> None:

2332 """Set the description for this repository.

2333

2334 Args:

2335 description: Text to set as description for this repository.

2336 """

2337 self._put_named_file("description", description)

2338

2339 @classmethod

2340 def _init_maybe_bare(

2341 cls,

2342 path: str | bytes | os.PathLike[str],

2343 controldir: str | bytes | os.PathLike[str],

2344 bare: bool,

2345 object_store: PackBasedObjectStore | None = None,

2346 config: "StackedConfig | None" = None,

2347 default_branch: bytes | None = None,

2348 symlinks: bool | None = None,

2349 format: int | None = None,

2350 shared_repository: str | bool | None = None,

2351 object_format: str | None = None,

2352 ) -> "Repo":

2353 path = os.fspath(path)

2354 if isinstance(path, bytes):

2355 path = os.fsdecode(path)

2356 controldir = os.fspath(controldir)

2357 if isinstance(controldir, bytes):

2358 controldir = os.fsdecode(controldir)

2359

2360 # Determine shared repository permissions early

2361 shared_perm: SharedPerm | None = None

2362 if shared_repository is not None:

2363 shared_perm = parse_shared_repository(shared_repository)

2364

2365 # Create base directories with appropriate permissions

2366 for d in BASE_DIRECTORIES:

2367 dir_path = os.path.join(controldir, *d)

2368 os.mkdir(dir_path)

2369 adjust_shared_perm(dir_path, shared_perm)

2370

2371 # Determine hash algorithm

2372 from .object_format import get_object_format

2373

2374 hash_alg = get_object_format(object_format)

2375

2376 if object_store is None:

2377 object_store = DiskObjectStore.init(

2378 os.path.join(controldir, OBJECTDIR),

2379 shared_perm=shared_perm,

2380 object_format=hash_alg,

2381 )

2382 ret = cls(path, bare=bare, object_store=object_store)

2383 if default_branch is None:

2384 if config is None:

2385 from .config import StackedConfig

2386

2387 config = StackedConfig.default()

2388 try:

2389 default_branch = config.get("init", "defaultBranch")

2390 except KeyError:

2391 default_branch = DEFAULT_BRANCH

2392 ret.refs.set_symbolic_ref(HEADREF, local_branch_name(default_branch))

2393 ret._init_files(

2394 bare=bare,

2395 symlinks=symlinks,

2396 format=format,

2397 shared_repository=shared_repository,

2398 object_format=object_format,

2399 )

2400 return ret

2401

2402 @classmethod

2403 def init(

2404 cls,

2405 path: str | bytes | os.PathLike[str],

2406 *,

2407 mkdir: bool = False,

2408 config: "StackedConfig | None" = None,

2409 default_branch: bytes | None = None,

2410 symlinks: bool | None = None,

2411 format: int | None = None,

2412 shared_repository: str | bool | None = None,

2413 object_format: str | None = None,

2414 ) -> "Repo":

2415 """Create a new repository.

2416

2417 Args:

2418 path: Path in which to create the repository

2419 mkdir: Whether to create the directory

2420 config: Configuration object

2421 default_branch: Default branch name

2422 symlinks: Whether to support symlinks

2423 format: Repository format version (defaults to 0)

2424 shared_repository: Shared repository setting (group, all, umask, or octal)

2425 object_format: Object format to use ("sha1" or "sha256", defaults to "sha1")

2426 Returns: `Repo` instance

2427 """

2428 path = os.fspath(path)

2429 if isinstance(path, bytes):

2430 path = os.fsdecode(path)

2431 if mkdir:

2432 os.mkdir(path)

2433 controldir = os.path.join(path, CONTROLDIR)

2434 os.mkdir(controldir)

2435 _set_filesystem_hidden(controldir)

2436 return cls._init_maybe_bare(

2437 path,

2438 controldir,

2439 False,

2440 config=config,

2441 default_branch=default_branch,

2442 symlinks=symlinks,

2443 format=format,

2444 shared_repository=shared_repository,

2445 object_format=object_format,

2446 )

2447

2448 @classmethod

2449 def _init_new_working_directory(

2450 cls,

2451 path: str | bytes | os.PathLike[str],

2452 main_repo: "Repo",

2453 identifier: str | None = None,

2454 mkdir: bool = False,

2455 relative_paths: bool = False,

2456 ) -> "Repo":

2457 """Create a new working directory linked to a repository.

2458

2459 Args:

2460 path: Path in which to create the working tree.

2461 main_repo: Main repository to reference

2462 identifier: Worktree identifier

2463 mkdir: Whether to create the directory

2464 relative_paths: Whether to use relative paths for gitdir references

2465 Returns: `Repo` instance

2466 """

2467 path = os.fspath(path)

2468 if isinstance(path, bytes):

2469 path = os.fsdecode(path)

2470 if mkdir:

2471 os.mkdir(path)

2472 if identifier is None:

2473 identifier = os.path.basename(path)

2474 # Ensure we use absolute path for the worktree control directory

2475 main_controldir = os.path.abspath(main_repo.controldir())

2476 main_worktreesdir = os.path.join(main_controldir, WORKTREES)

2477 worktree_controldir = os.path.join(main_worktreesdir, identifier)

2478 gitdirfile_abs = os.path.abspath(os.path.join(path, CONTROLDIR))

2479

2480 # Write gitdir reference in .git file (can be relative)

2481 # Import helper from worktree module to avoid duplication

2482 from .worktree import _compute_gitdir_path

2483

2484 gitdir_ref = _compute_gitdir_path(

2485 main_repo,

2486 worktree_controldir,

2487 os.path.dirname(gitdirfile_abs),

2488 relative_paths,

2489 )

2490

2491 with open(gitdirfile_abs, "wb") as f:

2492 f.write(b"gitdir: " + os.fsencode(gitdir_ref) + b"\n")

2493

2494 # Get shared repository permissions from main repository

2495 shared_perm = main_repo._get_shared_repository_permissions()

2496

2497 # Create directories with appropriate permissions

2498 try:

2499 os.mkdir(main_worktreesdir)

2500 adjust_shared_perm(main_worktreesdir, shared_perm)

2501 except FileExistsError:

2502 pass

2503 try:

2504 os.mkdir(worktree_controldir)

2505 adjust_shared_perm(worktree_controldir, shared_perm)

2506 except FileExistsError:

2507 pass

2508

2509 # Write gitdir path in control directory (can be relative)

2510 gitdir_path = _compute_gitdir_path(

2511 main_repo, gitdirfile_abs, worktree_controldir, relative_paths

2512 )

2513

2514 with open(os.path.join(worktree_controldir, GITDIR), "wb") as f:

2515 f.write(os.fsencode(gitdir_path) + b"\n")

2516 with open(os.path.join(worktree_controldir, COMMONDIR), "wb") as f:

2517 f.write(b"../..\n")

2518 with open(os.path.join(worktree_controldir, "HEAD"), "wb") as f:

2519 f.write(main_repo.head() + b"\n")

2520 r = cls(os.path.normpath(path))

2521 r.get_worktree().reset_index(config=r.get_config_stack())

2522 return r

2523

2524 @classmethod

2525 def init_bare(

2526 cls,

2527 path: str | bytes | os.PathLike[str],

2528 *,

2529 mkdir: bool = False,

2530 object_store: PackBasedObjectStore | None = None,

2531 config: "StackedConfig | None" = None,

2532 default_branch: bytes | None = None,

2533 format: int | None = None,

2534 shared_repository: str | bool | None = None,

2535 object_format: str | None = None,

2536 ) -> "Repo":

2537 """Create a new bare repository.

2538

2539 ``path`` should already exist and be an empty directory.

2540

2541 Args:

2542 path: Path to create bare repository in

2543 mkdir: Whether to create the directory

2544 object_store: Object store to use

2545 config: Configuration object

2546 default_branch: Default branch name

2547 format: Repository format version (defaults to 0)

2548 shared_repository: Shared repository setting (group, all, umask, or octal)

2549 object_format: Object format to use ("sha1" or "sha256", defaults to "sha1")

2550 Returns: a `Repo` instance

2551 """

2552 path = os.fspath(path)

2553 if isinstance(path, bytes):

2554 path = os.fsdecode(path)

2555 if mkdir:

2556 os.mkdir(path)

2557 return cls._init_maybe_bare(

2558 path,

2559 path,

2560 True,

2561 object_store=object_store,

2562 config=config,

2563 default_branch=default_branch,

2564 format=format,

2565 shared_repository=shared_repository,

2566 object_format=object_format,

2567 )

2568

2569 create = init_bare

2570

2571 def close(self) -> None:

2572 """Close any files opened by this repository."""

2573 self.object_store.close()

2574 # Clean up filter context if it was created

2575 if self.filter_context is not None:

2576 self.filter_context.close()

2577 self.filter_context = None

2578

2579 def __enter__(self) -> Self:

2580 """Enter context manager."""

2581 return self

2582

2583 def __exit__(

2584 self,

2585 exc_type: type[BaseException] | None,

2586 exc_val: BaseException | None,

2587 exc_tb: TracebackType | None,

2588 ) -> None:

2589 """Exit context manager and close repository."""

2590 self.close()

2591

2592 def _read_gitattributes(self) -> dict[bytes, dict[bytes, bytes]]:

2593 """Read .gitattributes file from working tree.

2594

2595 Returns:

2596 Dictionary mapping file patterns to attributes

2597 """

2598 gitattributes = {}

2599 gitattributes_path = os.path.join(self.path, ".gitattributes")

2600

2601 if os.path.exists(gitattributes_path):

2602 with open(gitattributes_path, "rb") as f:

2603 for line in f:

2604 line = line.strip()

2605 if not line or line.startswith(b"#"):

2606 continue

2607

2608 parts = line.split()

2609 if len(parts) < 2:

2610 continue

2611

2612 pattern = parts[0]

2613 attrs = {}

2614

2615 for attr in parts[1:]:

2616 if attr.startswith(b"-"):

2617 # Unset attribute

2618 attrs[attr[1:]] = b"false"

2619 elif b"=" in attr:

2620 # Set to value

2621 key, value = attr.split(b"=", 1)

2622 attrs[key] = value

2623 else:

2624 # Set attribute

2625 attrs[attr] = b"true"

2626

2627 gitattributes[pattern] = attrs

2628

2629 return gitattributes

2630

2631 def get_blob_normalizer(

2632 self, config: "Config | None" = None

2633 ) -> "FilterBlobNormalizer":

2634 """Return a BlobNormalizer object.

2635

2636 Args:

2637 config: Configuration to consult for filter setup. If None,

2638 falls back to ``self.get_config_stack()``.

2639 """

2640 from .filters import FilterBlobNormalizer, FilterContext, FilterRegistry

2641

2642 if config is None:

2643 config = self.get_config_stack()

2644 git_attributes = self.get_gitattributes()

2645

2646 # Lazily create FilterContext if needed

2647 if self.filter_context is None:

2648 filter_registry = FilterRegistry(config, self)

2649 self.filter_context = FilterContext(filter_registry)

2650 else:

2651 # Refresh the context with current config to handle config changes

2652 self.filter_context.refresh_config(config)

2653

2654 return FilterBlobNormalizer(

2655 config, git_attributes, filter_context=self.filter_context

2656 )

2657

2658 def get_gitattributes(self, tree: bytes | None = None) -> "GitAttributes":

2659 """Read gitattributes for the repository.

2660

2661 Args:

2662 tree: Tree SHA to read .gitattributes from (defaults to HEAD)

2663

2664 Returns:

2665 GitAttributes object that can be used to match paths

2666 """

2667 from .attrs import (

2668 GitAttributes,

2669 compile_gitattributes_patterns,

2670 parse_git_attributes,

2671 )

2672

2673 patterns = []

2674

2675 # Read system gitattributes (TODO: implement this)

2676 # Read global gitattributes (TODO: implement this)

2677

2678 # Read repository .gitattributes from index/tree

2679 if tree is None:

2680 try:

2681 # Try to get from HEAD

2682 head = self[b"HEAD"]

2683 # Peel tags to get to the underlying commit

2684 while isinstance(head, Tag):

2685 _cls, obj = head.object

2686 head = self.get_object(obj)

2687 if not isinstance(head, Commit):

2688 raise ValueError(

2689 f"Expected HEAD to point to a Commit, got {type(head).__name__}. "

2690 f"This usually means HEAD points to a {type(head).__name__} object "

2691 f"instead of a Commit."

2692 )

2693 tree = head.tree

2694 except KeyError:

2695 # No HEAD, no attributes from tree

2696 pass

2697

2698 if tree is not None:

2699 try:

2700 tree_obj = self[tree]

2701 assert isinstance(tree_obj, Tree)

2702 if b".gitattributes" in tree_obj:

2703 _, attrs_sha = tree_obj[b".gitattributes"]

2704 attrs_blob = self[attrs_sha]

2705 if isinstance(attrs_blob, Blob):

2706 attrs_data = BytesIO(attrs_blob.data)

2707 patterns.extend(

2708 compile_gitattributes_patterns(

2709 parse_git_attributes(attrs_data), b".gitattributes"

2710 )

2711 )

2712 except (KeyError, NotTreeError):

2713 pass

2714

2715 # Read .git/info/attributes

2716 info_attrs_path = os.path.join(self.controldir(), "info", "attributes")

2717 if os.path.exists(info_attrs_path):

2718 with open(info_attrs_path, "rb") as f:

2719 patterns.extend(

2720 compile_gitattributes_patterns(

2721 parse_git_attributes(f), info_attrs_path

2722 )

2723 )

2724

2725 # Read .gitattributes from working directory (if it exists)

2726 working_attrs_path = os.path.join(self.path, ".gitattributes")

2727 if os.path.exists(working_attrs_path):

2728 with open(working_attrs_path, "rb") as f:

2729 patterns.extend(

2730 compile_gitattributes_patterns(

2731 parse_git_attributes(f), working_attrs_path

2732 )

2733 )

2734

2735 return GitAttributes(patterns)

2736

2737

2738class MemoryRepo(BaseRepo):

2739 """Repo that stores refs, objects, and named files in memory.

2740

2741 MemoryRepos are always bare: they have no working tree and no index, since

2742 those have a stronger dependency on the filesystem.

2743 """

2744

2745 filter_context: "FilterContext | None"

2746

2747 def __init__(self) -> None:

2748 """Create a new repository in memory."""

2749 from .config import ConfigFile

2750 from .object_format import DEFAULT_OBJECT_FORMAT

2751

2752 self._reflog: list[Any] = []

2753 refs_container = DictRefsContainer({}, logger=self._append_reflog)

2754 BaseRepo.__init__(self, MemoryObjectStore(), refs_container)

2755 self._named_files: dict[str, bytes] = {}

2756 self.bare = True

2757 self._config = ConfigFile()

2758 self._description: bytes | None = None

2759 self.filter_context = None

2760 # MemoryRepo defaults to default object format

2761 self.object_format = DEFAULT_OBJECT_FORMAT

2762

2763 def _append_reflog(

2764 self,

2765 ref: bytes,

2766 old_sha: bytes | None,

2767 new_sha: bytes | None,

2768 committer: bytes | None,

2769 timestamp: int | None,

2770 timezone: int | None,

2771 message: bytes | None,

2772 ) -> None:

2773 self._reflog.append(

2774 (ref, old_sha, new_sha, committer, timestamp, timezone, message)

2775 )

2776

2777 def set_description(self, description: bytes) -> None:

2778 """Set the description for this repository.

2779

2780 Args:

2781 description: Text to set as description

2782 """

2783 self._description = description

2784

2785 def get_description(self) -> bytes | None:

2786 """Get the description of this repository.

2787

2788 Returns:

2789 Repository description as bytes

2790 """

2791 return self._description

2792

2793 def _determine_file_mode(self) -> bool:

2794 """Probe the file-system to determine whether permissions can be trusted.

2795

2796 Returns: True if permissions can be trusted, False otherwise.

2797 """

2798 return sys.platform != "win32"

2799

2800 def _determine_symlinks(self) -> bool:

2801 """Probe the file-system to determine whether permissions can be trusted.

2802

2803 Returns: True if permissions can be trusted, False otherwise.

2804 """

2805 return sys.platform != "win32"

2806

2807 def _put_named_file(self, path: str, contents: bytes) -> None:

2808 """Write a file to the control dir with the given name and contents.

2809

2810 Args:

2811 path: The path to the file, relative to the control dir.

2812 contents: A string to write to the file.

2813 """

2814 self._named_files[path] = contents

2815

2816 def _del_named_file(self, path: str) -> None:

2817 try:

2818 del self._named_files[path]

2819 except KeyError:

2820 pass

2821

2822 def get_named_file(

2823 self,

2824 path: str | bytes,

2825 basedir: str | None = None,

2826 ) -> BytesIO | None:

2827 """Get a file from the control dir with a specific name.

2828

2829 Although the filename should be interpreted as a filename relative to

2830 the control dir in a disk-baked Repo, the object returned need not be

2831 pointing to a file in that location.

2832

2833 Args:

2834 path: The path to the file, relative to the control dir.

2835 basedir: Optional base directory for the path

2836 Returns: An open file object, or None if the file does not exist.

2837 """

2838 path_str = path.decode() if isinstance(path, bytes) else path

2839 contents = self._named_files.get(path_str, None)

2840 if contents is None:

2841 return None

2842 return BytesIO(contents)

2843

2844 def open_index(self, config: "Config | None" = None) -> "Index":

2845 """Fail to open index for this repo, since it is bare.

2846

2847 Args:

2848 config: Unused; kept for signature compatibility with ``BaseRepo``.

2849

2850 Raises:

2851 NoIndexPresent: Raised when no index is present

2852 """

2853 raise NoIndexPresent

2854

2855 def _init_config(self, config: "ConfigFile") -> None:

2856 """Initialize repository configuration for MemoryRepo."""

2857 self._config = config

2858

2859 def get_config(self) -> "ConfigFile":

2860 """Retrieve the config object.

2861

2862 Returns: `ConfigFile` object.

2863 """

2864 return self._config

2865

2866 def get_rebase_state_manager(self) -> "RebaseStateManager":

2867 """Get the appropriate rebase state manager for this repository.

2868

2869 Returns: MemoryRebaseStateManager instance

2870 """

2871 from .rebase import MemoryRebaseStateManager

2872

2873 return MemoryRebaseStateManager(self)

2874

2875 def get_blob_normalizer(

2876 self, config: "Config | None" = None

2877 ) -> "FilterBlobNormalizer":

2878 """Return a BlobNormalizer object for checkin/checkout operations.

2879

2880 Args:

2881 config: Configuration to consult for filter setup. If None,

2882 falls back to ``self.get_config_stack()``.

2883 """

2884 from .filters import FilterBlobNormalizer, FilterContext, FilterRegistry

2885

2886 if config is None:

2887 config = self.get_config_stack()

2888 git_attributes = self.get_gitattributes()

2889

2890 if self.filter_context is None:

2891 filter_registry = FilterRegistry(config, self)

2892 self.filter_context = FilterContext(filter_registry)

2893 else:

2894 self.filter_context.refresh_config(config)

2895

2896 return FilterBlobNormalizer(

2897 config, git_attributes, filter_context=self.filter_context

2898 )

2899

2900 def get_gitattributes(self, tree: bytes | None = None) -> "GitAttributes":

2901 """Read gitattributes for the repository."""

2902 from .attrs import GitAttributes

2903

2904 # Memory repos don't have working trees or gitattributes files

2905 # Return empty GitAttributes

2906 return GitAttributes([])

2907

2908 def close(self) -> None:

2909 """Close any resources opened by this repository."""

2910 # Clean up filter context if it was created

2911 if self.filter_context is not None:

2912 self.filter_context.close()

2913 self.filter_context = None

2914 # Close object store to release pack files

2915 self.object_store.close()

2916

2917 def do_commit(

2918 self,

2919 message: bytes | None = None,

2920 committer: bytes | None = None,

2921 author: bytes | None = None,

2922 commit_timestamp: float | None = None,

2923 commit_timezone: int | None = None,

2924 author_timestamp: float | None = None,

2925 author_timezone: int | None = None,

2926 tree: ObjectID | None = None,

2927 encoding: bytes | None = None,

2928 ref: Ref | None = HEADREF,

2929 merge_heads: list[ObjectID] | None = None,

2930 no_verify: bool = False,

2931 sign: bool = False,

2932 config: "Config | None" = None,

2933 ) -> bytes:

2934 """Create a new commit.

2935

2936 This is a simplified implementation for in-memory repositories that

2937 doesn't support worktree operations or hooks.

2938

2939 Args:

2940 message: Commit message

2941 committer: Committer fullname

2942 author: Author fullname

2943 commit_timestamp: Commit timestamp (defaults to now)

2944 commit_timezone: Commit timestamp timezone (defaults to GMT)

2945 author_timestamp: Author timestamp (defaults to commit timestamp)

2946 author_timezone: Author timestamp timezone (defaults to commit timezone)

2947 tree: SHA1 of the tree root to use

2948 encoding: Encoding

2949 ref: Optional ref to commit to (defaults to current branch).

2950 If None, creates a dangling commit without updating any ref.

2951 merge_heads: Merge heads

2952 no_verify: Skip pre-commit and commit-msg hooks (ignored for MemoryRepo)

2953 sign: GPG Sign the commit (ignored for MemoryRepo)

2954 config: Configuration to consult for committer/author identity. If

2955 None, falls back to ``self.get_config_stack()``.

2956

2957 Returns:

2958 New commit SHA1

2959 """

2960 import time

2961

2962 from .objects import Commit

2963

2964 if tree is None:

2965 raise ValueError("tree must be specified for MemoryRepo")

2966

2967 c = Commit()

2968 if len(tree) != self.object_format.hex_length:

2969 raise ValueError(

2970 f"tree must be a {self.object_format.hex_length}-character hex sha string"

2971 )

2972 c.tree = tree

2973

2974 if config is None:

2975 config = self.get_config_stack()

2976 if merge_heads is None:

2977 merge_heads = []

2978 if committer is None:

2979 committer = get_user_identity(config, kind="COMMITTER")

2980 check_user_identity(committer)

2981 c.committer = committer

2982 if commit_timestamp is None:

2983 commit_timestamp = time.time()

2984 c.commit_time = int(commit_timestamp)

2985 if commit_timezone is None:

2986 commit_timezone = 0

2987 c.commit_timezone = commit_timezone

2988 if author is None:

2989 author = get_user_identity(config, kind="AUTHOR")

2990 c.author = author

2991 check_user_identity(author)

2992 if author_timestamp is None:

2993 author_timestamp = commit_timestamp

2994 c.author_time = int(author_timestamp)

2995 if author_timezone is None:

2996 author_timezone = commit_timezone

2997 c.author_timezone = author_timezone

2998 if encoding is None:

2999 try:

3000 encoding = config.get(("i18n",), "commitEncoding")

3001 except KeyError:

3002 pass

3003 if encoding is not None:

3004 c.encoding = encoding

3005

3006 # Handle message (for MemoryRepo, we don't support callable messages)

3007 if callable(message):

3008 message = message(self, c)

3009 if message is None:

3010 raise ValueError("Message callback returned None")

3011

3012 if message is None:

3013 raise ValueError("No commit message specified")

3014

3015 c.message = message

3016

3017 if ref is None:

3018 # Create a dangling commit

3019 c.parents = merge_heads

3020 self.object_store.add_object(c)

3021 else:

3022 try:

3023 old_head = self.refs[ref]

3024 c.parents = [old_head, *merge_heads]

3025 self.object_store.add_object(c)

3026 ok = self.refs.set_if_equals(

3027 ref,

3028 old_head,

3029 c.id,

3030 message=b"commit: " + message,

3031 committer=committer,

3032 timestamp=int(commit_timestamp),

3033 timezone=commit_timezone,

3034 )

3035 except KeyError:

3036 c.parents = merge_heads

3037 self.object_store.add_object(c)

3038 ok = self.refs.add_if_new(

3039 ref,

3040 c.id,

3041 message=b"commit: " + message,

3042 committer=committer,

3043 timestamp=int(commit_timestamp),

3044 timezone=commit_timezone,

3045 )

3046 if not ok:

3047 from .errors import CommitError

3048

3049 raise CommitError(f"{ref!r} changed during commit")

3050

3051 return c.id

3052

3053 @classmethod

3054 def init_bare(

3055 cls,

3056 objects: Iterable[ShaFile],

3057 refs: Mapping[Ref, ObjectID],

3058 format: int | None = None,

3059 object_format: str | None = None,

3060 ) -> "MemoryRepo":

3061 """Create a new bare repository in memory.

3062

3063 Args:

3064 objects: Objects for the new repository,

3065 as iterable

3066 refs: Refs as dictionary, mapping names

3067 to object SHA1s

3068 format: Repository format version (defaults to 0)

3069 object_format: Object format to use ("sha1" or "sha256", defaults to "sha1")

3070 """

3071 ret = cls()

3072 for obj in objects:

3073 ret.object_store.add_object(obj)

3074 for refname, sha in refs.items():

3075 ret.refs.add_if_new(refname, sha)

3076 ret._init_files(bare=True, format=format, object_format=object_format)

3077 return ret