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 "InvalidWorktreeConfiguration",

49 "MemoryRepo",

50 "ParentsProvider",

51 "Repo",

52 "UnsupportedExtension",

53 "UnsupportedVersion",

54 "check_user_identity",

55 "get_user_identity",

56 "parse_graftpoints",

57 "parse_shared_repository",

58 "read_gitfile",

59 "sanitize_user_identity",

60 "serialize_graftpoints",

61]

63import errno

64import logging

65import os

66import stat

67import sys

68import time

69import warnings

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

71from io import BytesIO

72from types import TracebackType

73from typing import (

74 TYPE_CHECKING,

75 Any,

76 BinaryIO,

77 TypeVar,

78)

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

81 from typing import Self

82else:

83 from typing_extensions import Self

85if TYPE_CHECKING:

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

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

88 # these imports.

89 from .attrs import AttributeValue, GitAttributes, Pattern

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

91 from .diff_tree import RenameDetector

92 from .filters import FilterBlobNormalizer, FilterContext

93 from .index import Index

94 from .notes import Notes

95 from .object_format import ObjectFormat

96 from .object_store import BaseObjectStore, GraphWalker

97 from .pack import UnpackedObject

98 from .rebase import RebaseStateManager

99 from .walk import Walker

100 from .worktree import WorkTree

101

102from . import _deprecated_aliases, reflog

103from .errors import (

104 NoIndexPresent,

105 NotBlobError,

106 NotCommitError,

107 NotGitRepository,

108 NotTagError,

109 NotTreeError,

110 RefFormatError,

111)

112from .file import (

113 PERM_EVERYBODY,

114 PERM_GROUP,

115 GitFile,

116 SharedPerm,

117 adjust_shared_perm,

118 open_nofollow_read,

119)

120from .hooks import (

121 CommitMsgShellHook,

122 Hook,

123 PostCommitShellHook,

124 PostReceiveShellHook,

125 PreCommitShellHook,

126 PreReceiveShellHook,

127 UpdateShellHook,

128)

129from .object_store import (

130 DiskObjectStore,

131 MemoryObjectStore,

132 MissingObjectFinder,

133 ObjectStoreGraphWalker,

134 PackBasedObjectStore,

135 PackCapableObjectStore,

136 find_shallow,

137)

138from .objects import (

139 Blob,

140 Commit,

141 ObjectID,

142 RawObjectID,

143 ShaFile,

144 Tag,

145 Tree,

146 check_hexsha,

147 valid_hexsha,

148)

149from .pack import generate_unpacked_objects

150from .refs import (

151 HEADREF,

152 DictRefsContainer,

153 DiskRefsContainer,

154 Ref,

155 RefsContainer,

156 _set_branch_tracking,

157 _set_default_branch,

158 _set_head,

159 _set_origin_head,

160 extract_branch_name,

161 is_per_worktree_ref,

162 local_branch_name,

163)

164

165logger = logging.getLogger(__name__)

166

167

168__getattr__ = _deprecated_aliases(

169 __name__,

170 {

171 "LOCAL_TAG_PREFIX": "dulwich.refs.LOCAL_TAG_PREFIX",

172 "SYMREF": "dulwich.refs.SYMREF",

173 "check_ref_format": "dulwich.refs.check_ref_format",

174 "read_packed_refs": "dulwich.refs.read_packed_refs",

175 "read_packed_refs_with_peeled": "dulwich.refs.read_packed_refs_with_peeled",

176 "write_packed_refs": "dulwich.refs.write_packed_refs",

177 },

178)

179

180

181CONTROLDIR = ".git"

182OBJECTDIR = "objects"

183DEFAULT_OFS_DELTA = True

184

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

186REFSDIR = "refs"

187REFSDIR_TAGS = "tags"

188REFSDIR_HEADS = "heads"

189INDEX_FILENAME = "index"

190COMMONDIR = "commondir"

191GITDIR = "gitdir"

192WORKTREES = "worktrees"

193

194BASE_DIRECTORIES = [

195 ["branches"],

196 [REFSDIR],

197 [REFSDIR, REFSDIR_TAGS],

198 [REFSDIR, REFSDIR_HEADS],

199 ["hooks"],

200 ["info"],

201]

202

203DEFAULT_BRANCH = b"master"

204

205

206class InvalidUserIdentity(Exception):

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

208

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

210 """Initialize InvalidUserIdentity exception."""

211 self.identity = identity

212

213

214class DefaultIdentityNotFound(Exception):

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

216

217

218# TODO(jelmer): Cache?

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

220 import socket

221

222 if env is None:

223 env = os.environ

224

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

226 username = env.get(name)

227 if username:

228 break

229 else:

230 username = None

231

232 try:

233 import pwd

234 except ImportError:

235 fullname = None

236 else:

237 try:

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

239 except KeyError:

240 fullname = None

241 else:

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

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

244 else:

245 fullname = None

246 if username is None:

247 username = entry.pw_name

248 if not fullname:

249 if username is None:

250 raise DefaultIdentityNotFound("no username found")

251 fullname = username

252 email = env.get("EMAIL")

253 if email is None:

254 if username is None:

255 raise DefaultIdentityNotFound("no username found")

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

257 return (fullname, email)

258

259

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

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

262

263 If kind is set, this first checks

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

265

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

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

268 the specified configuration.

269

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

271 the current users' identity as obtained from the host

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

273

274 Args:

275 config: Configuration stack to read from

276 kind: Optional kind to return identity for,

277 usually either "AUTHOR" or "COMMITTER".

278

279 Returns:

280 A user identity

281 """

282 user: bytes | None = None

283 email: bytes | None = None

284 if kind:

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

286 if user_uc is not None:

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

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

289 if email_uc is not None:

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

291 if user is None:

292 try:

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

294 except KeyError:

295 user = None

296 if email is None:

297 try:

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

299 except KeyError:

300 email = None

301 default_user, default_email = _get_default_identity()

302 if user is None:

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

304 if email is None:

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

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

307 email = email[1:-1]

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

309

310

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

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

313

314 Args:

315 identity: User identity bytestring

316 Raises:

317 InvalidUserIdentity: Raised when identity is invalid

318 """

319 try:

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

321 except ValueError as exc:

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

323 if b">" not in snd:

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

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

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

327

328

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

330

331

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

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

334

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

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

337 dropped from the rest of the value.

338 """

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

340 # rejects them.

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

342

343

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

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

346

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

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

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

350 are dropped from the middle.

351

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

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

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

355 ``check_user_identity``.

356

357 Args:

358 name: User name bytestring

359 email: Email bytestring

360

361 Returns:

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

363 """

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

365

366

367def parse_graftpoints(

368 graftpoints: Iterable[bytes],

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

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

371

372 Args:

373 graftpoints: Iterator of graftpoint lines

374

375 Each line is formatted as:

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

377

378 Resulting dictionary is:

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

380

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

382 """

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

384 for line in graftpoints:

385 raw_graft = line.split(None, 1)

386

387 commit = ObjectID(raw_graft[0])

388 if len(raw_graft) == 2:

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

390 else:

391 parents = []

392

393 for sha in [commit, *parents]:

394 check_hexsha(sha, "Invalid graftpoint")

395

396 grafts[commit] = parents

397 return grafts

398

399

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

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

402

403 The graft dictionary is:

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

405

406 Each line is formatted as:

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

408

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

410

411 """

412 graft_lines = []

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

414 if parents:

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

416 else:

417 graft_lines.append(commit)

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

419

420

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

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

423

424 On win32 uses SetFileAttributesW api:

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

426 """

427 if sys.platform == "win32":

428 import ctypes

429 from ctypes.wintypes import BOOL, DWORD, LPCWSTR

430

431 FILE_ATTRIBUTE_HIDDEN = 2

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

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

434 )

435

436 if isinstance(path, bytes):

437 path = os.fsdecode(path)

438 if not SetFileAttributesW(path, FILE_ATTRIBUTE_HIDDEN):

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

440

441 # Could implement other platform specific filesystem hiding here

442

443

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

445 """Parse core.sharedRepository configuration value.

446

447 Args:

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

449

450 Returns:

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

452 """

453 if isinstance(value, bytes):

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

455

456 # Handle boolean values

457 if isinstance(value, bool):

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

459 return PERM_GROUP if value else None

460

461 # Handle string values

462 value_lower = value.lower()

463

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

465 # Use umask (no adjustment)

466 return None

467

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

469 return PERM_GROUP

470

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

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

473 return PERM_EVERYBODY

474

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

476 # outright rather than loosening what the umask produced.

477 if value.startswith("0"):

478 try:

479 mode = int(value, 8)

480 except ValueError:

481 pass

482 else:

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

484

485 # Default to umask for unrecognized values

486 return None

487

488

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

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

491

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

493 enables the extensions.relativeworktrees extension.

494

495 Args:

496 repo: The repository to configure

497 """

498 config = repo.get_config()

499

500 # Ensure repository format version is at least 1

501 try:

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

503 except KeyError:

504 version = 0

505

506 if version < 1:

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

508

509 # Enable the relativeworktrees extension

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

511 config.write_to_path()

512

513

514class ParentsProvider:

515 """Provider for commit parent information."""

516

517 def __init__(

518 self,

519 store: "BaseObjectStore",

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

521 shallows: Iterable[ObjectID] = [],

522 ) -> None:

523 """Initialize ParentsProvider.

524

525 Args:

526 store: Object store to use

527 grafts: Graft information

528 shallows: Shallow commit SHAs

529 """

530 self.store = store

531 self.grafts = grafts

532 self.shallows = set(shallows)

533

534 # Get commit graph once at initialization for performance

535 self.commit_graph = store.get_commit_graph()

536

537 def get_parents(

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

539 ) -> list[ObjectID]:

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

541 try:

542 return self.grafts[commit_id]

543 except KeyError:

544 pass

545 if commit_id in self.shallows:

546 return []

547

548 # Try to use commit graph for faster parent lookup

549 if self.commit_graph:

550 parents = self.commit_graph.get_parents(commit_id)

551 if parents is not None:

552 return parents

553

554 # Fallback to reading the commit object

555 if commit is None:

556 obj = self.store[commit_id]

557 if not isinstance(obj, Commit):

558 raise ValueError(

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

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

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

562 )

563 commit = obj

564 result: list[ObjectID] = commit.parents

565 return result

566

567

568class BaseRepo:

569 """Base class for a git repository.

570

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

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

573

574 Attributes:

575 object_store: Dictionary-like object for accessing

576 the objects

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

578 repository

579 """

580

581 def __init__(

582 self,

583 object_store: "PackCapableObjectStore",

584 refs: RefsContainer,

585 object_format: "ObjectFormat | None" = None,

586 ) -> None:

587 """Open a repository.

588

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

590 base classes, such as MemoryRepo or Repo.

591

592 Args:

593 object_store: Object store to use

594 refs: Refs container to use

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

596 """

597 self.object_store = object_store

598 self.refs = refs

599

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

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

602 if object_format is None:

603 self.object_format: ObjectFormat = object_store.object_format

604 else:

605 self.object_format = object_format

606

607 def _determine_file_mode(self) -> bool:

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

609

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

611 """

612 raise NotImplementedError(self._determine_file_mode)

613

614 def _determine_symlinks(self) -> bool:

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

616

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

618 """

619 # For now, just mimic the old behaviour

620 return sys.platform != "win32"

621

622 def _init_files(

623 self,

624 bare: bool,

625 symlinks: bool | None = None,

626 format: int | None = None,

627 shared_repository: str | bool | None = None,

628 object_format: str | None = None,

629 ) -> None:

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

631 from .config import ConfigFile

632

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

634 f = BytesIO()

635 cf = ConfigFile()

636

637 # Determine the appropriate format version

638 if object_format == "sha256":

639 # SHA256 requires format version 1

640 if format is None:

641 format = 1

642 elif format != 1:

643 raise ValueError(

644 "SHA256 object format requires repository format version 1"

645 )

646 else:

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

648 if format is None:

649 format = 0

650

651 if format not in (0, 1):

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

653

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

655

656 # Set object format extension if using SHA256

657 if object_format == "sha256":

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

659

660 # Set hash algorithm based on object format

661 from .object_format import get_object_format

662

663 self.object_format = get_object_format(object_format)

664

665 if self._determine_file_mode():

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

667 else:

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

669

670 if symlinks is None and not bare:

671 symlinks = self._determine_symlinks()

672

673 if symlinks is False:

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

675

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

677 # returns filenames in NFD (decomposed) Unicode form

678 if sys.platform == "darwin":

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

680

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

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

683

684 # Set shared repository if specified

685 if shared_repository is not None:

686 if isinstance(shared_repository, bool):

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

688 else:

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

690

691 cf.write_to_file(f)

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

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

694

695 # Allow subclasses to handle config initialization

696 self._init_config(cf)

697

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

699 """Initialize repository configuration.

700

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

702

703 Args:

704 config: The ConfigFile object that was just created

705 """

706 # Default implementation does nothing

707

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

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

710

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

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

713 pointing to a file in that location.

714

715 Args:

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

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

718 """

719 raise NotImplementedError(self.get_named_file)

720

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

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

723

724 Args:

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

726 contents: A string to write to the file.

727 """

728 raise NotImplementedError(self._put_named_file)

729

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

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

732 raise NotImplementedError(self._del_named_file)

733

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

735 """Open the index for this repository.

736

737 Args:

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

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

740

741 Raises:

742 NoIndexPresent: If no index is present

743 Returns: The matching `Index`

744 """

745 raise NotImplementedError(self.open_index)

746

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

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

749

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

751

752 Args:

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

754

755 Raises:

756 AssertionError: If the object store is not empty

757 """

758 # Check if object store has any objects

759 for _ in self.object_store:

760 raise AssertionError(

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

762 )

763

764 # Update the object format

765 from .object_format import get_object_format

766

767 new_format = get_object_format(object_format_name)

768 self.object_format = new_format

769 self.object_store.object_format = new_format

770

771 # Update config file

772 config = self.get_config()

773

774 if object_format_name == "sha1":

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

776 try:

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

778 except KeyError:

779 pass

780 else:

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

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

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

784

785 config.write_to_path()

786

787 def fetch(

788 self,

789 target: "BaseRepo",

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

791 | None = None,

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

793 depth: int | None = None,

794 ) -> dict[Ref, ObjectID]:

795 """Fetch objects into another repository.

796

797 Args:

798 target: The target repository

799 determine_wants: Optional function to determine what refs to

800 fetch.

801 progress: Optional progress function

802 depth: Optional shallow fetch depth

803 Returns: The local refs

804 """

805 # Fix object format if needed

806 if self.object_format != target.object_format:

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

808 target._change_object_format(self.object_format.name)

809

810 if determine_wants is None:

811 determine_wants = target.object_store.determine_wants_all

812 count, pack_data = self.fetch_pack_data(

813 determine_wants,

814 target.get_graph_walker(),

815 progress=progress,

816 depth=depth,

817 )

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

819 return self.get_refs()

820

821 def fetch_pack_data(

822 self,

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

824 graph_walker: "GraphWalker",

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

826 *,

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

828 depth: int | None = None,

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

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

831

832 Args:

833 determine_wants: Function that takes a dictionary with heads

834 and returns the list of heads to fetch.

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

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

837 that a revision is present.

838 progress: Simple progress function that will be called with

839 updated progress strings.

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

841 tag sha for including tags.

842 depth: Shallow fetch depth

843 Returns: count and iterator over pack data

844 """

845 missing_objects = self.find_missing_objects(

846 determine_wants, graph_walker, progress, get_tagged=get_tagged, depth=depth

847 )

848 if missing_objects is None:

849 return 0, iter([])

850 remote_has = missing_objects.get_remote_has()

851 object_ids = list(missing_objects)

852 return len(object_ids), generate_unpacked_objects(

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

854 )

855

856 def find_missing_objects(

857 self,

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

859 graph_walker: "GraphWalker",

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

861 *,

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

863 depth: int | None = None,

864 ) -> MissingObjectFinder | None:

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

866

867 Args:

868 determine_wants: Function that takes a dictionary with heads

869 and returns the list of heads to fetch.

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

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

872 that a revision is present.

873 progress: Simple progress function that will be called with

874 updated progress strings.

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

876 tag sha for including tags.

877 depth: Shallow fetch depth

878 Returns: iterator over objects, with __len__ implemented

879 """

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

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

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

883 all_refs = self.get_refs()

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

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

886 if sha in self.object_store:

887 refs[ref] = sha

888 else:

889 logger.warning(

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

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

892 sha.decode("ascii"),

893 )

894

895 wants = determine_wants(refs, depth)

896 if not isinstance(wants, list):

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

898

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

900

901 unshallow: set[ObjectID] = set()

902 if depth not in (None, 0):

903 assert depth is not None

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

905 # Only update if graph_walker has shallow attribute

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

907 graph_walker, "shallow", None

908 )

909 if walker_shallow is not None:

910 unshallow = not_shallow & current_shallow

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

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

913 # parents stay unreachable and never get transferred.

914 walker_shallow.update(shallow - not_shallow)

915 walker_shallow.difference_update(unshallow)

916 new_shallow = walker_shallow - current_shallow

917 setattr(graph_walker, "unshallow", unshallow)

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

919 if update_shallow is not None:

920 update_shallow(new_shallow, unshallow)

921 else:

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

923

924 if wants == []:

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

926 # this interface.

927

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

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

930 return None

931

932 # Return an actual MissingObjectFinder with empty wants

933 return MissingObjectFinder(

934 self.object_store,

935 haves=[],

936 wants=[],

937 )

938

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

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

941 # this call as a side-effect.

942 haves = self.object_store.find_common_revisions(graph_walker)

943

944 # Deal with shallow requests separately because the haves do

945 # not reflect what objects are missing

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

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

948 # commits aren't missing.

949 haves = []

950

951 parents_provider = ParentsProvider(

952 self.object_store,

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

954 )

955

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

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

958

959 Args:

960 commit: Commit object

961

962 Returns:

963 List of parent commit SHAs

964 """

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

966

967 return MissingObjectFinder(

968 self.object_store,

969 haves=haves,

970 wants=wants,

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

972 progress=progress,

973 get_tagged=get_tagged,

974 get_parents=get_parents,

975 )

976

977 def generate_pack_data(

978 self,

979 have: set[ObjectID],

980 want: set[ObjectID],

981 *,

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

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

984 ofs_delta: bool | None = None,

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

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

987

988 Args:

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

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

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

992 ofs_delta: Whether OFS deltas can be included

993 progress: Optional progress reporting method

994 """

995 if shallow is None:

996 shallow = self.get_shallow()

997 return self.object_store.generate_pack_data(

998 have,

999 want,

1000 shallow=shallow,

1001 progress=progress,

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

1003 )

1004

1005 def get_graph_walker(

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

1007 ) -> ObjectStoreGraphWalker:

1008 """Retrieve a graph walker.

1009

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

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

1012

1013 Args:

1014 heads: Repository heads to use (optional)

1015 Returns: A graph walker object

1016 """

1017 if heads is None:

1018 heads = [

1019 sha

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

1021 if sha in self.object_store

1022 ]

1023 parents_provider = ParentsProvider(self.object_store)

1024 return ObjectStoreGraphWalker(

1025 heads,

1026 parents_provider.get_parents,

1027 shallow=self.get_shallow(),

1028 update_shallow=self.update_shallow,

1029 )

1030

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

1032 """Get dictionary with all refs.

1033

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

1035 """

1036 return self.refs.as_dict()

1037

1038 def head(self) -> ObjectID:

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

1040 # TODO: move this method to WorkTree

1041 return self.refs[HEADREF]

1042

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

1044 assert len(sha) in (

1045 self.object_format.oid_length,

1046 self.object_format.hex_length,

1047 )

1048 ret = self.get_object(sha)

1049 if not isinstance(ret, cls):

1050 if cls is Commit:

1051 raise NotCommitError(ret.id)

1052 elif cls is Blob:

1053 raise NotBlobError(ret.id)

1054 elif cls is Tree:

1055 raise NotTreeError(ret.id)

1056 elif cls is Tag:

1057 raise NotTagError(ret.id)

1058 else:

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

1060 return ret

1061

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

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

1064

1065 Args:

1066 sha: SHA to retrieve

1067 Returns: A ShaFile object

1068 Raises:

1069 KeyError: when the object can not be found

1070 """

1071 return self.object_store[sha]

1072

1073 def parents_provider(self) -> ParentsProvider:

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

1075

1076 Returns:

1077 ParentsProvider instance configured with grafts and shallows

1078 """

1079 return ParentsProvider(

1080 self.object_store,

1081 grafts=self._graftpoints,

1082 shallows=self.get_shallow(),

1083 )

1084

1085 def get_parents(

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

1087 ) -> list[ObjectID]:

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

1089

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

1091 will be returned instead.

1092

1093 Args:

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

1095 commit: Optional commit matching the sha

1096 Returns: List of parents

1097 """

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

1099

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

1101 """Retrieve the config object.

1102

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

1104 """

1105 raise NotImplementedError(self.get_config)

1106

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

1108 """Retrieve the worktree config object."""

1109 raise NotImplementedError(self.get_worktree_config)

1110

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

1112 """Retrieve the description for this repository.

1113

1114 Returns: Bytes with the description of the repository

1115 as set by the user.

1116 """

1117 raise NotImplementedError(self.get_description)

1118

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

1120 """Set the description for this repository.

1121

1122 Args:

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

1124 """

1125 raise NotImplementedError(self.set_description)

1126

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

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

1129

1130 Returns: RebaseStateManager instance

1131 """

1132 raise NotImplementedError(self.get_rebase_state_manager)

1133

1134 def get_blob_normalizer(

1135 self, config: "Config | None" = None, tree: bytes | None = None

1136 ) -> "FilterBlobNormalizer":

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

1138

1139 Args:

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

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

1142 tree: Tree to read .gitattributes from, e.g. the tree being

1143 checked out. See ``get_gitattributes``.

1144

1145 Returns: BlobNormalizer instance

1146 """

1147 raise NotImplementedError(self.get_blob_normalizer)

1148

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

1150 """Read gitattributes for the repository.

1151

1152 Args:

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

1154

1155 Returns:

1156 GitAttributes object that can be used to match paths

1157 """

1158 raise NotImplementedError(self.get_gitattributes)

1159

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

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

1162

1163 This stack accesses the configuration for both this repository

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

1165 lives in ~/.gitconfig.

1166

1167 Returns: `Config` instance for this repository

1168 """

1169 from .config import ConfigFile, StackedConfig

1170

1171 local_config = self.get_config()

1172 backends: list[ConfigFile] = [local_config]

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

1174 backends.append(self.get_worktree_config())

1175

1176 backends += StackedConfig.default_backends()

1177 return StackedConfig(backends, writable=local_config)

1178

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

1180 """Get the set of shallow commits.

1181

1182 Returns: Set of shallow commits.

1183 """

1184 f = self.get_named_file("shallow")

1185 if f is None:

1186 return set()

1187 with f:

1188 shallow: set[ObjectID] = set()

1189 for line in f:

1190 sha = line.strip()

1191 if not sha:

1192 continue

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

1194 shallow.add(ObjectID(sha))

1195 return shallow

1196

1197 def update_shallow(

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

1199 ) -> None:

1200 """Update the list of shallow objects.

1201

1202 Args:

1203 new_shallow: Newly shallow objects

1204 new_unshallow: Newly no longer shallow objects

1205 """

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

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

1208 shallow = self.get_shallow()

1209 if new_shallow:

1210 shallow.update(new_shallow)

1211 if new_unshallow:

1212 shallow.difference_update(new_unshallow)

1213 if shallow:

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

1215 else:

1216 self._del_named_file("shallow")

1217

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

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

1220

1221 Args:

1222 ref: The refname to peel.

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

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

1225 this will equal the original SHA1.

1226 """

1227 cached = self.refs.get_peeled(ref)

1228 if cached is not None:

1229 return cached

1230 return self.object_store.peel(self.refs[ref])[1].id

1231

1232 @property

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

1234 """Access notes functionality for this repository.

1235

1236 Returns:

1237 Notes object for accessing notes

1238 """

1239 from .notes import Notes

1240

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

1242

1243 def get_walker(

1244 self,

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

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

1247 order: str = "date",

1248 reverse: bool = False,

1249 max_entries: int | None = None,

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

1251 rename_detector: "RenameDetector | None" = None,

1252 follow: bool = False,

1253 since: int | None = None,

1254 until: int | None = None,

1255 queue_cls: type | None = None,

1256 simplify_history: bool = False,

1257 ) -> "Walker":

1258 """Obtain a walker for this repository.

1259

1260 Args:

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

1262 ancestors. Defaults to [HEAD]

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

1264 ancestors, overriding includes.

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

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

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

1268 memory.

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

1270 no limit.

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

1272 rename_detector: diff.RenameDetector object for detecting

1273 renames.

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

1275 default rename_detector.

1276 since: Timestamp to list commits after.

1277 until: Timestamp to list commits before.

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

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

1280 simplify_history: If True and paths is set, perform git-style

1281 history simplification: at a merge commit that is TREESAME to

1282 at least one parent for the requested paths, follow only that

1283 parent (matches ``git log <path>``). The default of False

1284 matches ``git log --full-history <path>``.

1285

1286 Returns: A `Walker` object

1287 """

1288 from .walk import Walker, _CommitTimeQueue

1289

1290 if include is None:

1291 include = [self.head()]

1292

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

1294 return Walker(

1295 self.object_store,

1296 include,

1297 exclude=exclude,

1298 order=order,

1299 reverse=reverse,

1300 max_entries=max_entries,

1301 paths=paths,

1302 rename_detector=rename_detector,

1303 follow=follow,

1304 since=since,

1305 until=until,

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

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

1308 simplify_history=simplify_history,

1309 )

1310

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

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

1313

1314 Args:

1315 name: A Git object SHA1 or a ref name

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

1317 Raises:

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

1319 """

1320 if not isinstance(name, bytes):

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

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

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

1324 try:

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

1326 except (RefFormatError, KeyError):

1327 pass

1328 # Otherwise, try as object ID if length matches

1329 if len(name) in (

1330 self.object_store.object_format.oid_length,

1331 self.object_store.object_format.hex_length,

1332 ):

1333 try:

1334 return self.object_store[

1335 ObjectID(name)

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

1337 else RawObjectID(name)

1338 ]

1339 except (KeyError, ValueError):

1340 pass

1341 # If nothing worked, raise KeyError

1342 raise KeyError(name)

1343

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

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

1346

1347 Args:

1348 name: Git object SHA1/SHA256 or ref name

1349 """

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

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

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

1353 return Ref(name) in self.refs

1354

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

1356 """Set a ref.

1357

1358 Args:

1359 name: ref name

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

1361 """

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

1363 ref_name = Ref(name)

1364 if isinstance(value, ShaFile):

1365 self.refs[ref_name] = value.id

1366 elif isinstance(value, bytes):

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

1368 else:

1369 raise TypeError(value)

1370 else:

1371 raise ValueError(name)

1372

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

1374 """Remove a ref.

1375

1376 Args:

1377 name: Name of the ref to remove

1378 """

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

1380 del self.refs[Ref(name)]

1381 else:

1382 raise ValueError(name)

1383

1384 def _get_user_identity(

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

1386 ) -> bytes:

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

1388 warnings.warn(

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

1390 DeprecationWarning,

1391 )

1392 return get_user_identity(config)

1393

1394 def _add_graftpoints(

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

1396 ) -> None:

1397 """Add or modify graftpoints.

1398

1399 Args:

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

1401 """

1402 # Simple validation

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

1404 for sha in [commit, *parents]:

1405 check_hexsha(sha, "Invalid graftpoint")

1406

1407 self._graftpoints.update(updated_graftpoints)

1408

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

1410 """Remove graftpoints.

1411

1412 Args:

1413 to_remove: List of commit shas

1414 """

1415 for sha in to_remove:

1416 del self._graftpoints[sha]

1417

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

1419 f = self.get_named_file(name)

1420 if f is None:

1421 return []

1422 with f:

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

1424

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

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

1427

1428 Returns:

1429 WorkTree instance for performing working tree operations

1430

1431 Raises:

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

1433 """

1434 raise NotImplementedError(

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

1436 )

1437

1438

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

1440 """Read a ``.git`` file.

1441

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

1443

1444 Args:

1445 f: File-like object to read from

1446 Returns: A path

1447 """

1448 cs = f.read()

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

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

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

1452

1453

1454class UnsupportedVersion(Exception):

1455 """Unsupported repository version."""

1456

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

1458 """Initialize UnsupportedVersion exception.

1459

1460 Args:

1461 version: The unsupported repository version

1462 """

1463 self.version = version

1464

1465

1466class UnsupportedExtension(Exception):

1467 """Unsupported repository extension."""

1468

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

1470 """Initialize UnsupportedExtension exception.

1471

1472 Args:

1473 extension: The unsupported repository extension

1474 """

1475 self.extension = extension

1476

1477

1478class InvalidWorktreeConfiguration(Exception):

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

1480

1481

1482class _GitAttributesLoader:

1483 """Read the .gitattributes of a directory for GitAttributes.

1484

1485 The working tree file is used if there is one, otherwise the one in

1486 the tree, as git does for checkin.

1487 """

1488

1489 def __init__(

1490 self,

1491 object_store: "BaseObjectStore",

1492 worktree_path: str | None,

1493 tree: ObjectID | None,

1494 ) -> None:

1495 self._object_store = object_store

1496 self._worktree_path = worktree_path

1497 # Tree id of each directory looked up so far; None if absent.

1498 self._trees: dict[bytes, ObjectID | None] = {b"": tree}

1499

1500 def __call__(

1501 self, dirpath: bytes

1502 ) -> "list[tuple[Pattern, Mapping[bytes, AttributeValue]]]":

1503 from .attrs import compile_gitattributes_patterns, parse_git_attributes

1504

1505 if self._worktree_path is not None:

1506 path = os.path.join(

1507 self._worktree_path, *os.fsdecode(dirpath).split("/"), ".gitattributes"

1508 )

1509 try:

1510 f = open_nofollow_read(path)

1511 except (FileNotFoundError, NotADirectoryError):

1512 pass

1513 except OSError as e:

1514 if e.errno not in (errno.ELOOP, errno.EMLINK):

1515 raise

1516 logger.warning("Ignoring %s: it is a symbolic link", path)

1517 else:

1518 with f:

1519 return compile_gitattributes_patterns(

1520 parse_git_attributes(f), path, dirpath

1521 )

1522

1523 tree_id = self._lookup_tree(dirpath)

1524 if tree_id is None:

1525 return []

1526 tree = self._object_store[tree_id]

1527 if not isinstance(tree, Tree):

1528 raise NotTreeError(tree_id)

1529 try:

1530 mode, sha = tree[b".gitattributes"]

1531 except KeyError:

1532 return []

1533 if not stat.S_ISREG(mode):

1534 return []

1535 blob = self._object_store[sha]

1536 if not isinstance(blob, Blob):

1537 return []

1538 source = dirpath + b"/.gitattributes" if dirpath else b".gitattributes"

1539 return compile_gitattributes_patterns(

1540 parse_git_attributes(BytesIO(blob.data)), source, dirpath

1541 )

1542

1543 def _lookup_tree(self, dirpath: bytes) -> ObjectID | None:

1544 try:

1545 return self._trees[dirpath]

1546 except KeyError:

1547 pass

1548 parent, _, name = dirpath.rpartition(b"/")

1549 parent_id = self._lookup_tree(parent)

1550 tree_id = None

1551 if parent_id is not None:

1552 parent_tree = self._object_store[parent_id]

1553 if not isinstance(parent_tree, Tree):

1554 raise NotTreeError(parent_id)

1555 try:

1556 mode, sha = parent_tree[name]

1557 except KeyError:

1558 pass

1559 else:

1560 # Like git, don't look inside submodules or non-directories.

1561 if stat.S_ISDIR(mode):

1562 tree_id = sha

1563 self._trees[dirpath] = tree_id

1564 return tree_id

1565

1566

1567class Repo(BaseRepo):

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

1569

1570 To open an existing repository, call the constructor with

1571 the path of the repository.

1572

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

1574

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

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

1577 up those resources.

1578

1579 Attributes:

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

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

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

1583 bare: Whether this is a bare repository

1584 """

1585

1586 path: str

1587 bare: bool

1588 object_store: DiskObjectStore

1589 filter_context: "FilterContext | None"

1590 _index_file_override: "str | None"

1591

1592 def __init__(

1593 self,

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

1595 object_store: PackBasedObjectStore | None = None,

1596 bare: bool | None = None,

1597 *,

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

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

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

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

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

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

1604 ) -> None:

1605 """Open a repository on disk.

1606

1607 Args:

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

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

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

1611 config may still override it.

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

1613 repository's default object store

1614 bare: True if this is a bare repository.

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

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

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

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

1619 control directory.

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

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

1622 object_directory: Explicit path to the primary object store

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

1624 ``<common_dir>/objects``.

1625 alternates: Extra alternate object directory paths (analogous to

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

1627 in ``objects/info/alternates``.

1628 index_file: Path to the index file (analogous to

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

1630 """

1631 if controldir is None:

1632 if root is None:

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

1634 root = os.fspath(root)

1635 if isinstance(root, bytes):

1636 root = os.fsdecode(root)

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

1638 if worktree is not None:

1639 bare = False

1640 if bare is None:

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

1642 os.path.join(hidden_path, OBJECTDIR)

1643 ):

1644 bare = False

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

1646 os.path.join(root, REFSDIR)

1647 ):

1648 bare = True

1649 else:

1650 raise NotGitRepository(

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

1652 **dict(path=root)

1653 )

1654 )

1655

1656 self.bare = bare

1657 if bare is False:

1658 if os.path.isfile(hidden_path):

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

1660 gitfile_path = read_gitfile(f)

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

1662 else:

1663 self._controldir = hidden_path

1664 else:

1665 self._controldir = root

1666 else:

1667 controldir = os.fspath(controldir)

1668 if isinstance(controldir, bytes):

1669 controldir = os.fsdecode(controldir)

1670 self._controldir = controldir

1671 if root is None:

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

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

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

1675 # should not silently latch onto the current directory.

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

1677 # semantics pass ``worktree`` explicitly.

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

1679 else:

1680 root = os.fspath(root)

1681 if isinstance(root, bytes):

1682 root = os.fsdecode(root)

1683 if worktree is not None:

1684 bare = False

1685 if bare is None:

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

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

1688 # config below may still flip this.

1689 bare = True

1690 self.bare = bare

1691 if commondir is not None:

1692 commondir_path = os.fspath(commondir)

1693 if isinstance(commondir_path, bytes):

1694 commondir_path = os.fsdecode(commondir_path)

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

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

1697 self._commondir = commondir_path

1698 else:

1699 commondir_file = self.get_named_file(COMMONDIR)

1700 if commondir_file is not None:

1701 with commondir_file:

1702 self._commondir = os.path.join(

1703 self.controldir(),

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

1705 )

1706 else:

1707 self._commondir = self._controldir

1708 self.path = root

1709 if worktree is not None:

1710 worktree_path = os.fspath(worktree)

1711 if isinstance(worktree_path, bytes):

1712 worktree_path = os.fsdecode(worktree_path)

1713 self.path = worktree_path

1714

1715 if index_file is not None:

1716 index_file_str = os.fspath(index_file)

1717 if isinstance(index_file_str, bytes):

1718 index_file_str = os.fsdecode(index_file_str)

1719 self._index_file_override = index_file_str

1720 else:

1721 self._index_file_override = None

1722

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

1724 self.refs = DiskRefsContainer(

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

1726 )

1727

1728 # Initialize worktrees container

1729 from .worktree import WorkTreeContainer

1730

1731 self.worktrees = WorkTreeContainer(self)

1732

1733 config = self.get_config()

1734 try:

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

1736 format_version = (

1737 0

1738 if repository_format_version is None

1739 else int(repository_format_version)

1740 )

1741 except KeyError:

1742 format_version = 0

1743

1744 if format_version not in (0, 1):

1745 raise UnsupportedVersion(format_version)

1746

1747 try:

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

1749 except KeyError:

1750 configured_worktree = None

1751 if configured_worktree is not None:

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

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

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

1755 # working tree, named here.

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

1757 raise InvalidWorktreeConfiguration(

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

1759 )

1760 self.bare = False

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

1762 if worktree is None:

1763 # Relative paths are resolved against the control directory,

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

1765 self.path = os.path.join(

1766 self._controldir, os.fsdecode(configured_worktree)

1767 )

1768

1769 # Track extensions we encounter

1770 has_reftable_extension = False

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

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

1773 if value == b"reftable":

1774 has_reftable_extension = True

1775 else:

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

1777 elif extension.lower() not in (

1778 b"worktreeconfig",

1779 b"objectformat",

1780 b"relativeworktrees",

1781 ):

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

1783

1784 if object_store is None:

1785 # Get shared repository permissions from config

1786 try:

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

1788 shared_perm = parse_shared_repository(shared_value)

1789 except KeyError:

1790 shared_perm = None

1791

1792 if object_directory is not None:

1793 object_dir_path = os.fspath(object_directory)

1794 if isinstance(object_dir_path, bytes):

1795 object_dir_path = os.fsdecode(object_dir_path)

1796 else:

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

1798 object_store = DiskObjectStore.from_config(

1799 object_dir_path,

1800 config,

1801 shared_perm=shared_perm,

1802 alternates=alternates,

1803 )

1804

1805 # Use reftable if extension is configured

1806 if has_reftable_extension:

1807 from .reftable import ReftableRefsContainer

1808

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

1810 # Update worktrees container after refs change

1811 self.worktrees = WorkTreeContainer(self)

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

1813

1814 # Determine hash algorithm from config if not already set

1815 if self.object_format is None:

1816 from .object_format import DEFAULT_OBJECT_FORMAT, get_object_format

1817

1818 if format_version == 1:

1819 try:

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

1821 self.object_format = get_object_format(

1822 object_format.decode("ascii")

1823 )

1824 except KeyError:

1825 self.object_format = DEFAULT_OBJECT_FORMAT

1826 else:

1827 self.object_format = DEFAULT_OBJECT_FORMAT

1828

1829 self._graftpoints = {}

1830 graft_file = self.get_named_file(

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

1832 )

1833 if graft_file:

1834 with graft_file:

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

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

1837 if graft_file:

1838 with graft_file:

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

1840

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

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

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

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

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

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

1847

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

1849 self.filter_context = None

1850

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

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

1853

1854 Returns:

1855 WorkTree instance for performing working tree operations

1856 """

1857 from .worktree import WorkTree

1858

1859 return WorkTree(self, self.path)

1860

1861 def _write_reflog(

1862 self,

1863 ref: bytes,

1864 old_sha: bytes,

1865 new_sha: bytes,

1866 committer: bytes | None,

1867 timestamp: int | None,

1868 timezone: int | None,

1869 message: bytes,

1870 ) -> None:

1871 from .reflog import format_reflog_line

1872

1873 path = self._reflog_path(ref)

1874

1875 # Get shared repository permissions

1876 shared_perm = self._get_shared_repository_permissions()

1877

1878 # Create directory with appropriate permissions

1879 parent_dir = os.path.dirname(path)

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

1881 parts = []

1882 current = parent_dir

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

1884 parts.append(current)

1885 current = os.path.dirname(current)

1886 parts.reverse()

1887 for part in parts:

1888 os.mkdir(part)

1889 adjust_shared_perm(part, shared_perm)

1890 if committer is None:

1891 config = self.get_config_stack()

1892 committer = get_user_identity(config)

1893 check_user_identity(committer)

1894 if timestamp is None:

1895 timestamp = int(time.time())

1896 if timezone is None:

1897 timezone = 0 # FIXME

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

1899 f.write(

1900 format_reflog_line(

1901 old_sha, new_sha, committer, timestamp, timezone, message

1902 )

1903 + b"\n"

1904 )

1905

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

1907 # already existed.

1908 adjust_shared_perm(path, shared_perm)

1909

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

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

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

1913

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

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

1916

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

1918 """Read reflog entries for a reference.

1919

1920 Args:

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

1922

1923 Yields:

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

1925 """

1926 from .reflog import read_reflog

1927

1928 path = self._reflog_path(ref)

1929 try:

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

1931 yield from read_reflog(f)

1932 except FileNotFoundError:

1933 return

1934

1935 @classmethod

1936 def discover(

1937 cls,

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

1939 *,

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

1941 across_filesystem: bool = True,

1942 ) -> "Repo":

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

1944

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

1946 Git repository.

1947

1948 Args:

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

1950 ceiling_dirs: Iterable of paths that discovery must not cross

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

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

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

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

1955 a resolved form works.

1956 across_filesystem: Whether to keep walking up past a filesystem

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

1958 When False, discovery stops before entering a parent directory

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

1960 """

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

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

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

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

1965 # the kernel-resolved cwd.

1966 ceilings = (

1967 {

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

1969 for p in ceiling_dirs

1970 }

1971 if ceiling_dirs is not None

1972 else set()

1973 )

1974 path = os.path.realpath(start)

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

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

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

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

1979 first = True

1980 while True:

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

1982 break

1983 first = False

1984 try:

1985 return cls(path)

1986 except NotGitRepository:

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

1988 if new_path == path: # Root reached

1989 break

1990 if start_dev is not None:

1991 try:

1992 parent_dev = os.stat(new_path).st_dev

1993 except FileNotFoundError:

1994 # Raced with the parent being removed; nothing left

1995 # to walk up into.

1996 break

1997 if parent_dev != start_dev:

1998 break

1999 path = new_path

2000 start_str = os.fspath(start)

2001 if isinstance(start_str, bytes):

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

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

2004

2005 def controldir(self) -> str:

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

2007 return self._controldir

2008

2009 def commondir(self) -> str:

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

2011

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

2013

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

2015 main working tree.

2016 """

2017 return self._commondir

2018

2019 def _determine_file_mode(self) -> bool:

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

2021

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

2023 """

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

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

2026 f.write("")

2027

2028 st1 = os.lstat(fname)

2029 try:

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

2031 except PermissionError:

2032 return False

2033 st2 = os.lstat(fname)

2034

2035 os.unlink(fname)

2036

2037 mode_differs = st1.st_mode != st2.st_mode

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

2039

2040 return mode_differs and st2_has_exec

2041

2042 def _determine_symlinks(self) -> bool:

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

2044

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

2046 """

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

2048 return sys.platform != "win32"

2049

2050 def _get_shared_repository_permissions(

2051 self,

2052 ) -> "SharedPerm | None":

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

2054

2055 Returns:

2056 SharedPerm to apply, or None if not shared

2057 """

2058 try:

2059 config = self.get_config()

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

2061 return parse_shared_repository(value)

2062 except KeyError:

2063 return None

2064

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

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

2067

2068 Args:

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

2070 contents: A string to write to the file.

2071 """

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

2073

2074 # Get shared repository permissions

2075 shared_perm = self._get_shared_repository_permissions()

2076

2077 # Create file with appropriate permissions

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

2079 if shared_perm is not None:

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

2081 f.write(contents)

2082 else:

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

2084 f.write(contents)

2085

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

2087 try:

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

2089 except FileNotFoundError:

2090 return

2091

2092 def get_named_file(

2093 self,

2094 path: str | bytes,

2095 basedir: str | None = None,

2096 ) -> BinaryIO | None:

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

2098

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

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

2101 pointing to a file in that location.

2102

2103 Args:

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

2105 basedir: Optional argument that specifies an alternative to the

2106 control dir.

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

2108 """

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

2110 # the dumb web serving code.

2111 if basedir is None:

2112 basedir = self.controldir()

2113 if isinstance(path, bytes):

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

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

2116 try:

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

2118 except FileNotFoundError:

2119 return None

2120

2121 def index_path(self) -> str:

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

2123 if self._index_file_override is not None:

2124 return self._index_file_override

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

2126

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

2128 """Open the index for this repository.

2129

2130 Args:

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

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

2133

2134 Raises:

2135 NoIndexPresent: If no index is present

2136 Returns: The matching `Index`

2137 """

2138 from .index import Index, make_path_normalizer

2139

2140 if not self.has_index():

2141 raise NoIndexPresent

2142

2143 if config is None:

2144 config = self.get_config_stack()

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

2146 skip_hash = False

2147 index_version = None

2148

2149 if many_files:

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

2151 try:

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

2153 index_version = int(index_version_str)

2154 except KeyError:

2155 index_version = 4 # Default to version 4 for manyFiles

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

2157 else:

2158 # Check for explicit index settings

2159 try:

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

2161 index_version = int(index_version_str)

2162 except KeyError:

2163 index_version = None

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

2165

2166 # Get shared repository permissions for index file

2167 shared_perm = self._get_shared_repository_permissions()

2168

2169 return Index(

2170 self.index_path(),

2171 skip_hash=skip_hash,

2172 version=index_version,

2173 shared_perm=shared_perm,

2174 path_normalizer=make_path_normalizer(config),

2175 object_format=self.object_format,

2176 )

2177

2178 def has_index(self) -> bool:

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

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

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

2182 return not self.bare

2183

2184 def clone(

2185 self,

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

2187 *,

2188 mkdir: bool = True,

2189 bare: bool = False,

2190 origin: bytes = b"origin",

2191 checkout: bool | None = None,

2192 branch: bytes | None = None,

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

2194 depth: int | None = None,

2195 symlinks: bool | None = None,

2196 ) -> "Repo":

2197 """Clone this repository.

2198

2199 Args:

2200 target_path: Target path

2201 mkdir: Create the target directory

2202 bare: Whether to create a bare repository

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

2204 origin: Base name for refs in target repository

2205 cloned from this repository

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

2207 instead of this repository's HEAD.

2208 progress: Optional progress function

2209 depth: Depth at which to fetch

2210 symlinks: Symlinks setting (default to autodetect)

2211 Returns: Created repository as `Repo`

2212 """

2213 encoded_path = os.fsencode(self.path)

2214

2215 if mkdir:

2216 os.mkdir(target_path)

2217

2218 try:

2219 if not bare:

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

2221 if checkout is None:

2222 checkout = True

2223 else:

2224 if checkout:

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

2226 target = Repo.init_bare(target_path)

2227

2228 try:

2229 target_config = target.get_config()

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

2231 target_config.set(

2232 (b"remote", origin),

2233 b"fetch",

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

2235 )

2236 target_config.write_to_path()

2237

2238 ref_message = b"clone: from " + encoded_path

2239 self.fetch(target, depth=depth)

2240 target.refs.import_refs(

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

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

2243 message=ref_message,

2244 )

2245 target.refs.import_refs(

2246 Ref(b"refs/tags"),

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

2248 message=ref_message,

2249 )

2250

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

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

2253 head: ObjectID | None = None

2254 if origin_sha and not origin_head:

2255 # set detached HEAD

2256 target.refs[HEADREF] = origin_sha

2257 head = origin_sha

2258 else:

2259 _set_origin_head(target.refs, origin, origin_head)

2260 head_ref = _set_default_branch(

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

2262 )

2263

2264 # Update target head

2265 if head_ref:

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

2267 if not bare:

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

2269 else:

2270 head = None

2271

2272 if checkout and head is not None:

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

2274 except BaseException:

2275 target.close()

2276 raise

2277 except BaseException:

2278 if mkdir:

2279 import shutil

2280

2281 shutil.rmtree(target_path)

2282 raise

2283 return target

2284

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

2286 """Get condition matchers for includeIf conditions.

2287

2288 Returns a dict of condition prefix to matcher function.

2289 """

2290 from pathlib import Path

2291

2292 from .config import ConditionMatcher, match_glob_pattern

2293

2294 # Add gitdir matchers

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

2296 """Match gitdir against a pattern.

2297

2298 Args:

2299 pattern: Pattern to match against

2300 case_sensitive: Whether to match case-sensitively

2301

2302 Returns:

2303 True if gitdir matches pattern

2304 """

2305 # Handle relative patterns (starting with ./)

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

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

2308 return False

2309

2310 # Normalize repository path

2311 try:

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

2313 except (OSError, ValueError):

2314 return False

2315

2316 # Expand ~ in pattern and normalize

2317 pattern = os.path.expanduser(pattern)

2318

2319 # Normalize pattern following Git's rules

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

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

2322 # Check for Windows absolute path

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

2324 pass

2325 else:

2326 pattern = "**/" + pattern

2327 if pattern.endswith("/"):

2328 pattern = pattern + "**"

2329

2330 # Use the existing _match_gitdir_pattern function

2331 from .config import _match_gitdir_pattern

2332

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

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

2335

2336 return _match_gitdir_pattern(

2337 repo_path_bytes, pattern_bytes, ignorecase=not case_sensitive

2338 )

2339

2340 # Add onbranch matcher

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

2342 """Match current branch against a pattern.

2343

2344 Args:

2345 pattern: Pattern to match against

2346

2347 Returns:

2348 True if current branch matches pattern

2349 """

2350 try:

2351 # Get the current branch using refs

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

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

2354 except KeyError:

2355 pass

2356 else:

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

2358 # Extract branch name from ref

2359 branch = extract_branch_name(head_ref).decode(

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

2361 )

2362 return match_glob_pattern(branch, pattern)

2363 return False

2364

2365 matchers: dict[str, ConditionMatcher] = {

2366 "onbranch:": match_onbranch,

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

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

2369 }

2370

2371 return matchers

2372

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

2374 """Get the worktree-specific config.

2375

2376 Returns:

2377 ConfigFile object for the worktree config

2378 """

2379 from .config import ConfigFile

2380

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

2382 try:

2383 # Pass condition matchers for includeIf evaluation

2384 condition_matchers = self._get_config_condition_matchers()

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

2386 except FileNotFoundError:

2387 cf = ConfigFile()

2388 cf.path = path

2389 return cf

2390

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

2392 """Retrieve the config object.

2393

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

2395 """

2396 from .config import ConfigFile

2397

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

2399 try:

2400 # Pass condition matchers for includeIf evaluation

2401 condition_matchers = self._get_config_condition_matchers()

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

2403 except FileNotFoundError:

2404 ret = ConfigFile()

2405 ret.path = path

2406 return ret

2407

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

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

2410

2411 Returns: DiskRebaseStateManager instance

2412 """

2413 import os

2414

2415 from .rebase import DiskRebaseStateManager

2416

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

2418 return DiskRebaseStateManager(path)

2419

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

2421 """Retrieve the description of this repository.

2422

2423 Returns: Description as bytes or None.

2424 """

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

2426 try:

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

2428 return f.read()

2429 except FileNotFoundError:

2430 return None

2431

2432 def __repr__(self) -> str:

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

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

2435

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

2437 """Set the description for this repository.

2438

2439 Args:

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

2441 """

2442 self._put_named_file("description", description)

2443

2444 @classmethod

2445 def _init_maybe_bare(

2446 cls,

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

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

2449 bare: bool,

2450 object_store: PackBasedObjectStore | None = None,

2451 config: "StackedConfig | None" = None,

2452 default_branch: bytes | None = None,

2453 symlinks: bool | None = None,

2454 format: int | None = None,

2455 shared_repository: str | bool | None = None,

2456 object_format: str | None = None,

2457 ) -> "Repo":

2458 path = os.fspath(path)

2459 if isinstance(path, bytes):

2460 path = os.fsdecode(path)

2461 controldir = os.fspath(controldir)

2462 if isinstance(controldir, bytes):

2463 controldir = os.fsdecode(controldir)

2464

2465 # Determine shared repository permissions early

2466 shared_perm: SharedPerm | None = None

2467 if shared_repository is not None:

2468 shared_perm = parse_shared_repository(shared_repository)

2469

2470 # Create base directories with appropriate permissions

2471 for d in BASE_DIRECTORIES:

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

2473 os.mkdir(dir_path)

2474 adjust_shared_perm(dir_path, shared_perm)

2475

2476 # Determine hash algorithm

2477 from .object_format import get_object_format

2478

2479 hash_alg = get_object_format(object_format)

2480

2481 if object_store is None:

2482 object_store = DiskObjectStore.init(

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

2484 shared_perm=shared_perm,

2485 object_format=hash_alg,

2486 )

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

2488 if default_branch is None:

2489 if config is None:

2490 from .config import StackedConfig

2491

2492 config = StackedConfig.default()

2493 try:

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

2495 except KeyError:

2496 default_branch = DEFAULT_BRANCH

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

2498 ret._init_files(

2499 bare=bare,

2500 symlinks=symlinks,

2501 format=format,

2502 shared_repository=shared_repository,

2503 object_format=object_format,

2504 )

2505 return ret

2506

2507 @classmethod

2508 def init(

2509 cls,

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

2511 *,

2512 mkdir: bool = False,

2513 config: "StackedConfig | None" = None,

2514 default_branch: bytes | None = None,

2515 symlinks: bool | None = None,

2516 format: int | None = None,

2517 shared_repository: str | bool | None = None,

2518 object_format: str | None = None,

2519 ) -> "Repo":

2520 """Create a new repository.

2521

2522 Args:

2523 path: Path in which to create the repository

2524 mkdir: Whether to create the directory

2525 config: Configuration object

2526 default_branch: Default branch name

2527 symlinks: Whether to support symlinks

2528 format: Repository format version (defaults to 0)

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

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

2531 Returns: `Repo` instance

2532 """

2533 path = os.fspath(path)

2534 if isinstance(path, bytes):

2535 path = os.fsdecode(path)

2536 if mkdir:

2537 os.mkdir(path)

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

2539 os.mkdir(controldir)

2540 _set_filesystem_hidden(controldir)

2541 return cls._init_maybe_bare(

2542 path,

2543 controldir,

2544 False,

2545 config=config,

2546 default_branch=default_branch,

2547 symlinks=symlinks,

2548 format=format,

2549 shared_repository=shared_repository,

2550 object_format=object_format,

2551 )

2552

2553 @classmethod

2554 def _init_new_working_directory(

2555 cls,

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

2557 main_repo: "Repo",

2558 identifier: str | None = None,

2559 mkdir: bool = False,

2560 relative_paths: bool = False,

2561 ) -> "Repo":

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

2563

2564 Args:

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

2566 main_repo: Main repository to reference

2567 identifier: Worktree identifier

2568 mkdir: Whether to create the directory

2569 relative_paths: Whether to use relative paths for gitdir references

2570 Returns: `Repo` instance

2571 """

2572 path = os.fspath(path)

2573 if isinstance(path, bytes):

2574 path = os.fsdecode(path)

2575 if mkdir:

2576 os.mkdir(path)

2577 if identifier is None:

2578 identifier = os.path.basename(path)

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

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

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

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

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

2584

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

2586 # Import helper from worktree module to avoid duplication

2587 from .worktree import _compute_gitdir_path

2588

2589 gitdir_ref = _compute_gitdir_path(

2590 main_repo,

2591 worktree_controldir,

2592 os.path.dirname(gitdirfile_abs),

2593 relative_paths,

2594 )

2595

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

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

2598

2599 # Get shared repository permissions from main repository

2600 shared_perm = main_repo._get_shared_repository_permissions()

2601

2602 # Create directories with appropriate permissions

2603 try:

2604 os.mkdir(main_worktreesdir)

2605 adjust_shared_perm(main_worktreesdir, shared_perm)

2606 except FileExistsError:

2607 pass

2608 try:

2609 os.mkdir(worktree_controldir)

2610 adjust_shared_perm(worktree_controldir, shared_perm)

2611 except FileExistsError:

2612 pass

2613

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

2615 gitdir_path = _compute_gitdir_path(

2616 main_repo, gitdirfile_abs, worktree_controldir, relative_paths

2617 )

2618

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

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

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

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

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

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

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

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

2627 return r

2628

2629 @classmethod

2630 def init_bare(

2631 cls,

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

2633 *,

2634 mkdir: bool = False,

2635 object_store: PackBasedObjectStore | None = None,

2636 config: "StackedConfig | None" = None,

2637 default_branch: bytes | None = None,

2638 format: int | None = None,

2639 shared_repository: str | bool | None = None,

2640 object_format: str | None = None,

2641 ) -> "Repo":

2642 """Create a new bare repository.

2643

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

2645

2646 Args:

2647 path: Path to create bare repository in

2648 mkdir: Whether to create the directory

2649 object_store: Object store to use

2650 config: Configuration object

2651 default_branch: Default branch name

2652 format: Repository format version (defaults to 0)

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

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

2655 Returns: a `Repo` instance

2656 """

2657 path = os.fspath(path)

2658 if isinstance(path, bytes):

2659 path = os.fsdecode(path)

2660 if mkdir:

2661 os.mkdir(path)

2662 return cls._init_maybe_bare(

2663 path,

2664 path,

2665 True,

2666 object_store=object_store,

2667 config=config,

2668 default_branch=default_branch,

2669 format=format,

2670 shared_repository=shared_repository,

2671 object_format=object_format,

2672 )

2673

2674 create = init_bare

2675

2676 def close(self) -> None:

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

2678 self.object_store.close()

2679 # Clean up filter context if it was created

2680 if self.filter_context is not None:

2681 self.filter_context.close()

2682 self.filter_context = None

2683

2684 def __enter__(self) -> Self:

2685 """Enter context manager."""

2686 return self

2687

2688 def __exit__(

2689 self,

2690 exc_type: type[BaseException] | None,

2691 exc_val: BaseException | None,

2692 exc_tb: TracebackType | None,

2693 ) -> None:

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

2695 self.close()

2696

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

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

2699

2700 A ``.gitattributes`` that is a symlink is skipped with a warning

2701 rather than followed, as git does.

2702

2703 Returns:

2704 Dictionary mapping file patterns to attributes

2705 """

2706 gitattributes: dict[bytes, dict[bytes, bytes]] = {}

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

2708

2709 try:

2710 f = open_nofollow_read(gitattributes_path)

2711 except FileNotFoundError:

2712 return gitattributes

2713 except OSError as e:

2714 if e.errno not in (errno.ELOOP, errno.EMLINK):

2715 raise

2716 logger.warning("Ignoring %s: it is a symbolic link", gitattributes_path)

2717 return gitattributes

2718 with f:

2719 for line in f:

2720 line = line.strip()

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

2722 continue

2723

2724 parts = line.split()

2725 if len(parts) < 2:

2726 continue

2727

2728 pattern = parts[0]

2729 attrs = {}

2730

2731 for attr in parts[1:]:

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

2733 # Unset attribute

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

2735 elif b"=" in attr:

2736 # Set to value

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

2738 attrs[key] = value

2739 else:

2740 # Set attribute

2741 attrs[attr] = b"true"

2742

2743 gitattributes[pattern] = attrs

2744

2745 return gitattributes

2746

2747 def get_blob_normalizer(

2748 self, config: "Config | None" = None, tree: bytes | None = None

2749 ) -> "FilterBlobNormalizer":

2750 """Return a BlobNormalizer object.

2751

2752 Args:

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

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

2755 tree: Tree to read .gitattributes from, e.g. the tree being

2756 checked out. See ``get_gitattributes``.

2757 """

2758 from .filters import FilterBlobNormalizer, FilterContext, FilterRegistry

2759

2760 if config is None:

2761 config = self.get_config_stack()

2762 git_attributes = self.get_gitattributes(tree)

2763

2764 # Lazily create FilterContext if needed

2765 if self.filter_context is None:

2766 filter_registry = FilterRegistry(config, self)

2767 self.filter_context = FilterContext(filter_registry)

2768 else:

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

2770 self.filter_context.refresh_config(config)

2771

2772 return FilterBlobNormalizer(

2773 config, git_attributes, repo=self, filter_context=self.filter_context

2774 )

2775

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

2777 """Read gitattributes for the repository.

2778

2779 As in git, the ``.gitattributes`` of every directory applies to the

2780 paths below it, with deeper files taking precedence and

2781 ``$GIT_DIR/info/attributes`` taking precedence over all of them.

2782 Subdirectory files are read when a path below them is first matched.

2783

2784 Args:

2785 tree: Tree SHA to read .gitattributes from. If not given, each

2786 ``.gitattributes`` is read from the working tree, falling back

2787 to the HEAD tree when the working tree has none.

2788

2789 Returns:

2790 GitAttributes object that can be used to match paths

2791 """

2792 from .attrs import GitAttributes, parse_gitattributes_file

2793

2794 # Read system gitattributes (TODO: implement this)

2795 # Read global gitattributes (TODO: implement this)

2796

2797 read_worktree = tree is None and not self.bare

2798 if tree is None:

2799 try:

2800 head = self[b"HEAD"]

2801 # Peel tags to get to the underlying commit

2802 while isinstance(head, Tag):

2803 _cls, obj = head.object

2804 head = self.get_object(obj)

2805 if not isinstance(head, Commit):

2806 raise ValueError(

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

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

2809 f"instead of a Commit."

2810 )

2811 tree = head.tree

2812 except KeyError:

2813 # No HEAD, no attributes from tree

2814 pass

2815

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

2817 try:

2818 info_patterns = parse_gitattributes_file(info_attrs_path)

2819 except FileNotFoundError:

2820 info_patterns = []

2821

2822 loader = _GitAttributesLoader(

2823 self.object_store,

2824 self.path if read_worktree else None,

2825 None if tree is None else ObjectID(tree),

2826 )

2827 return GitAttributes(

2828 loader(b""), directory_loader=loader, info_patterns=info_patterns

2829 )

2830

2831

2832class MemoryRepo(BaseRepo):

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

2834

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

2836 those have a stronger dependency on the filesystem.

2837 """

2838

2839 filter_context: "FilterContext | None"

2840

2841 def __init__(self) -> None:

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

2843 from .config import ConfigFile

2844 from .object_format import DEFAULT_OBJECT_FORMAT

2845

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

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

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

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

2850 self.bare = True

2851 self._config = ConfigFile()

2852 self._description: bytes | None = None

2853 self.filter_context = None

2854 # MemoryRepo defaults to default object format

2855 self.object_format = DEFAULT_OBJECT_FORMAT

2856

2857 def _append_reflog(

2858 self,

2859 ref: bytes,

2860 old_sha: bytes | None,

2861 new_sha: bytes | None,

2862 committer: bytes | None,

2863 timestamp: int | None,

2864 timezone: int | None,

2865 message: bytes | None,

2866 ) -> None:

2867 self._reflog.append(

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

2869 )

2870

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

2872 """Set the description for this repository.

2873

2874 Args:

2875 description: Text to set as description

2876 """

2877 self._description = description

2878

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

2880 """Get the description of this repository.

2881

2882 Returns:

2883 Repository description as bytes

2884 """

2885 return self._description

2886

2887 def _determine_file_mode(self) -> bool:

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

2889

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

2891 """

2892 return sys.platform != "win32"

2893

2894 def _determine_symlinks(self) -> bool:

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

2896

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

2898 """

2899 return sys.platform != "win32"

2900

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

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

2903

2904 Args:

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

2906 contents: A string to write to the file.

2907 """

2908 self._named_files[path] = contents

2909

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

2911 try:

2912 del self._named_files[path]

2913 except KeyError:

2914 pass

2915

2916 def get_named_file(

2917 self,

2918 path: str | bytes,

2919 basedir: str | None = None,

2920 ) -> BytesIO | None:

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

2922

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

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

2925 pointing to a file in that location.

2926

2927 Args:

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

2929 basedir: Optional base directory for the path

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

2931 """

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

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

2934 if contents is None:

2935 return None

2936 return BytesIO(contents)

2937

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

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

2940

2941 Args:

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

2943

2944 Raises:

2945 NoIndexPresent: Raised when no index is present

2946 """

2947 raise NoIndexPresent

2948

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

2950 """Initialize repository configuration for MemoryRepo."""

2951 self._config = config

2952

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

2954 """Retrieve the config object.

2955

2956 Returns: `ConfigFile` object.

2957 """

2958 return self._config

2959

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

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

2962

2963 Returns: MemoryRebaseStateManager instance

2964 """

2965 from .rebase import MemoryRebaseStateManager

2966

2967 return MemoryRebaseStateManager(self)

2968

2969 def get_blob_normalizer(

2970 self, config: "Config | None" = None, tree: bytes | None = None

2971 ) -> "FilterBlobNormalizer":

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

2973

2974 Args:

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

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

2977 tree: Tree to read .gitattributes from

2978 """

2979 from .filters import FilterBlobNormalizer, FilterContext, FilterRegistry

2980

2981 if config is None:

2982 config = self.get_config_stack()

2983 git_attributes = self.get_gitattributes(tree)

2984

2985 if self.filter_context is None:

2986 filter_registry = FilterRegistry(config, self)

2987 self.filter_context = FilterContext(filter_registry)

2988 else:

2989 self.filter_context.refresh_config(config)

2990

2991 return FilterBlobNormalizer(

2992 config, git_attributes, repo=self, filter_context=self.filter_context

2993 )

2994

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

2996 """Read gitattributes for the repository."""

2997 from .attrs import GitAttributes

2998

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

3000 # Return empty GitAttributes

3001 return GitAttributes([])

3002

3003 def close(self) -> None:

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

3005 # Clean up filter context if it was created

3006 if self.filter_context is not None:

3007 self.filter_context.close()

3008 self.filter_context = None

3009 # Close object store to release pack files

3010 self.object_store.close()

3011

3012 def do_commit(

3013 self,

3014 message: bytes | None = None,

3015 committer: bytes | None = None,

3016 author: bytes | None = None,

3017 commit_timestamp: float | None = None,

3018 commit_timezone: int | None = None,

3019 author_timestamp: float | None = None,

3020 author_timezone: int | None = None,

3021 tree: ObjectID | None = None,

3022 encoding: bytes | None = None,

3023 ref: Ref | None = HEADREF,

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

3025 no_verify: bool = False,

3026 sign: bool = False,

3027 config: "Config | None" = None,

3028 ) -> bytes:

3029 """Create a new commit.

3030

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

3032 doesn't support worktree operations or hooks.

3033

3034 Args:

3035 message: Commit message

3036 committer: Committer fullname

3037 author: Author fullname

3038 commit_timestamp: Commit timestamp (defaults to now)

3039 commit_timezone: Commit timestamp timezone (defaults to GMT)

3040 author_timestamp: Author timestamp (defaults to commit timestamp)

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

3042 tree: SHA1 of the tree root to use

3043 encoding: Encoding

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

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

3046 merge_heads: Merge heads

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

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

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

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

3051

3052 Returns:

3053 New commit SHA1

3054 """

3055 import time

3056

3057 from .objects import Commit

3058

3059 if tree is None:

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

3061

3062 c = Commit()

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

3064 raise ValueError(

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

3066 )

3067 c.tree = tree

3068

3069 if config is None:

3070 config = self.get_config_stack()

3071 if merge_heads is None:

3072 merge_heads = []

3073 if committer is None:

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

3075 check_user_identity(committer)

3076 c.committer = committer

3077 if commit_timestamp is None:

3078 commit_timestamp = time.time()

3079 c.commit_time = int(commit_timestamp)

3080 if commit_timezone is None:

3081 commit_timezone = 0

3082 c.commit_timezone = commit_timezone

3083 if author is None:

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

3085 c.author = author

3086 check_user_identity(author)

3087 if author_timestamp is None:

3088 author_timestamp = commit_timestamp

3089 c.author_time = int(author_timestamp)

3090 if author_timezone is None:

3091 author_timezone = commit_timezone

3092 c.author_timezone = author_timezone

3093 if encoding is None:

3094 try:

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

3096 except KeyError:

3097 pass

3098 if encoding is not None:

3099 c.encoding = encoding

3100

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

3102 if callable(message):

3103 message = message(self, c)

3104 if message is None:

3105 raise ValueError("Message callback returned None")

3106

3107 if message is None:

3108 raise ValueError("No commit message specified")

3109

3110 c.message = message

3111

3112 if ref is None:

3113 # Create a dangling commit

3114 c.parents = merge_heads

3115 self.object_store.add_object(c)

3116 else:

3117 try:

3118 old_head = self.refs[ref]

3119 c.parents = [old_head, *merge_heads]

3120 self.object_store.add_object(c)

3121 ok = self.refs.set_if_equals(

3122 ref,

3123 old_head,

3124 c.id,

3125 message=b"commit: " + message,

3126 committer=committer,

3127 timestamp=int(commit_timestamp),

3128 timezone=commit_timezone,

3129 )

3130 except KeyError:

3131 c.parents = merge_heads

3132 self.object_store.add_object(c)

3133 ok = self.refs.add_if_new(

3134 ref,

3135 c.id,

3136 message=b"commit: " + message,

3137 committer=committer,

3138 timestamp=int(commit_timestamp),

3139 timezone=commit_timezone,

3140 )

3141 if not ok:

3142 from .errors import CommitError

3143

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

3145

3146 return c.id

3147

3148 @classmethod

3149 def init_bare(

3150 cls,

3151 objects: Iterable[ShaFile],

3152 refs: Mapping[Ref, ObjectID],

3153 format: int | None = None,

3154 object_format: str | None = None,

3155 ) -> "MemoryRepo":

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

3157

3158 Args:

3159 objects: Objects for the new repository,

3160 as iterable

3161 refs: Refs as dictionary, mapping names

3162 to object SHA1s

3163 format: Repository format version (defaults to 0)

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

3165 """

3166 ret = cls()

3167 for obj in objects:

3168 ret.object_store.add_object(obj)

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

3170 ret.refs.add_if_new(refname, sha)

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

3172 return ret