Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/pack.py: 22%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# pack.py -- For dealing with packed git objects.
2# Copyright (C) 2007 James Westby <jw+debian@jameswestby.net>
3# Copyright (C) 2008-2013 Jelmer Vernooij <jelmer@jelmer.uk>
4#
5# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
6# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
7# General Public License as published by the Free Software Foundation; version 2.0
8# or (at your option) any later version. You can redistribute it and/or
9# modify it under the terms of either of these two licenses.
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS,
13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14# See the License for the specific language governing permissions and
15# limitations under the License.
16#
17# You should have received a copy of the licenses; if not, see
18# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
19# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
20# License, Version 2.0.
21#
23"""Classes for dealing with packed git objects.
25A pack is a compact representation of a bunch of objects, stored
26using deltas where possible.
28They have two parts, the pack file, which stores the data, and an index
29that tells you where the data is.
31To find an object you look in all of the index files 'til you find a
32match for the object name. You then use the pointer got from this as
33a pointer in to the corresponding packfile.
34"""
36__all__ = [
37 "DEFAULT_PACK_DELTA_WINDOW_SIZE",
38 "DEFAULT_PACK_INDEX_VERSION",
39 "DELTA_TYPES",
40 "OFS_DELTA",
41 "PACK_SPOOL_FILE_MAX_SIZE",
42 "REF_DELTA",
43 "DeltaChainIterator",
44 "FilePackIndex",
45 "MemoryPackIndex",
46 "ObjectContainer",
47 "Pack",
48 "PackChunkGenerator",
49 "PackData",
50 "PackFileDisappeared",
51 "PackHint",
52 "PackIndex",
53 "PackIndex1",
54 "PackIndex2",
55 "PackIndex3",
56 "PackIndexEntry",
57 "PackIndexer",
58 "PackInflater",
59 "PackStreamCopier",
60 "PackStreamReader",
61 "PackedObjectContainer",
62 "SHA1Reader",
63 "SHA1Writer",
64 "UnpackedObject",
65 "UnpackedObjectIterator",
66 "UnpackedObjectStream",
67 "UnresolvedDeltas",
68 "apply_delta",
69 "bisect_find_sha",
70 "chunks_length",
71 "compute_file_sha",
72 "deltas_from_sorted_objects",
73 "deltify_pack_objects",
74 "extend_pack",
75 "find_reusable_deltas",
76 "full_unpacked_object",
77 "generate_unpacked_objects",
78 "iter_sha1",
79 "load_pack_index",
80 "load_pack_index_file",
81 "obj_sha",
82 "pack_header_chunks",
83 "pack_object_chunks",
84 "pack_object_header",
85 "pack_objects_to_data",
86 "read_pack_header",
87 "read_zlib_chunks",
88 "sort_objects_for_delta",
89 "take_msb_bytes",
90 "unpack_object",
91 "verify_and_read",
92 "write_pack",
93 "write_pack_data",
94 "write_pack_from_container",
95 "write_pack_header",
96 "write_pack_index",
97 "write_pack_object",
98 "write_pack_objects",
99]
101import binascii
102from collections import defaultdict, deque
103from contextlib import suppress
104from io import BytesIO, UnsupportedOperation
106try:
107 from cdifflib import CSequenceMatcher as SequenceMatcher
108except ModuleNotFoundError:
109 from difflib import SequenceMatcher
111import os
112import struct
113import sys
114import warnings
115import zlib
116from collections.abc import Callable, Iterable, Iterator, Sequence, Set
117from hashlib import sha1, sha256
118from itertools import chain
119from os import SEEK_CUR, SEEK_END
120from struct import unpack_from
121from types import TracebackType
122from typing import (
123 IO,
124 TYPE_CHECKING,
125 Any,
126 BinaryIO,
127 Generic,
128 Protocol,
129 TypeVar,
130)
132if sys.version_info >= (3, 11):
133 from typing import Self
134else:
135 from typing_extensions import Self
137import mmap
139from .errors import ApplyDeltaError, ChecksumMismatch
140from .file import GitFile, _GitFile
141from .lru_cache import LRUSizeCache
142from .object_format import OBJECT_FORMAT_TYPE_NUMS, SHA1, ObjectFormat
143from .objects import (
144 ObjectID,
145 RawObjectID,
146 ShaFile,
147 hex_to_sha,
148 object_header,
149 sha_to_hex,
150)
152if TYPE_CHECKING:
153 from _hashlib import HASH as HashObject
155 from .bitmap import PackBitmap
156 from .commit_graph import CommitGraph
157 from .object_store import BaseObjectStore
158 from .refs import Ref
160# Some platforms (e.g. plan9) don't support mmap properly
161has_mmap = sys.platform != "Plan9"
163OFS_DELTA = 6
164REF_DELTA = 7
166DELTA_TYPES = (OFS_DELTA, REF_DELTA)
169DEFAULT_PACK_DELTA_WINDOW_SIZE = 10
171# Keep pack files under 16Mb in memory, otherwise write them out to disk
172PACK_SPOOL_FILE_MAX_SIZE = 16 * 1024 * 1024
174# Default pack index version to use when none is specified
175DEFAULT_PACK_INDEX_VERSION = 2
178OldUnpackedObject = tuple[bytes | int, list[bytes]] | list[bytes] | bytes
179ResolveExtRefFn = Callable[[RawObjectID | ObjectID], tuple[int, bytes | list[bytes]]]
180ProgressFn = Callable[[int, str], None]
181PackHint = tuple[int, bytes | None]
184def verify_and_read(
185 read_func: Callable[[int], bytes],
186 expected_hash: bytes,
187 hash_algo: str,
188 progress: Callable[[bytes], None] | None = None,
189) -> Iterator[bytes]:
190 """Read from stream, verify hash, then yield verified chunks.
192 This function downloads data to a temporary file (in-memory for small files,
193 on-disk for large ones) while computing its hash. Only after the hash is
194 verified to match expected_hash will it yield any data. This prevents
195 corrupted or malicious data from reaching the caller.
197 Args:
198 read_func: Function to read bytes (like file.read or HTTP response reader)
199 expected_hash: Expected hash as hex string bytes (e.g., b'a3b2c1...')
200 hash_algo: Hash algorithm name ('sha1' or 'sha256')
201 progress: Optional progress callback
203 Yields:
204 Chunks of verified data (only after hash verification succeeds)
206 Raises:
207 ValueError: If hash doesn't match or algorithm unsupported
208 """
209 from tempfile import SpooledTemporaryFile
211 from .object_format import OBJECT_FORMATS
213 # Get the hash function for this algorithm
214 obj_format = OBJECT_FORMATS.get(hash_algo)
215 if obj_format is None:
216 raise ValueError(f"Unsupported hash algorithm: {hash_algo}")
218 hasher = obj_format.new_hash()
220 # Download to temporary file (memory or disk) while computing hash
221 with SpooledTemporaryFile(
222 max_size=PACK_SPOOL_FILE_MAX_SIZE, prefix="dulwich-verify-"
223 ) as temp_file:
224 # Read data, hash it, and write to temp file
225 while True:
226 chunk = read_func(65536) # Read in 64KB chunks
227 if not chunk:
228 break
229 hasher.update(chunk)
230 temp_file.write(chunk)
232 # Verify hash BEFORE yielding any data
233 computed_hash = hasher.hexdigest().encode("ascii")
234 if computed_hash != expected_hash:
235 raise ValueError(
236 f"hash mismatch: expected {expected_hash.decode('ascii')}, "
237 f"got {computed_hash.decode('ascii')}"
238 )
240 # Hash verified! Now read from temp file and yield chunks
241 if progress:
242 progress(b"Hash verified, processing data\n")
244 temp_file.seek(0)
245 while True:
246 chunk = temp_file.read(65536)
247 if not chunk:
248 break
249 yield chunk
252class UnresolvedDeltas(Exception):
253 """Delta objects could not be resolved."""
255 def __init__(self, shas: list[bytes]) -> None:
256 """Initialize UnresolvedDeltas exception.
258 Args:
259 shas: List of SHA hashes for unresolved delta objects
260 """
261 self.shas = shas
264class ObjectContainer(Protocol):
265 """Protocol for objects that can contain git objects."""
267 def add_object(self, obj: ShaFile) -> None:
268 """Add a single object to this object store."""
270 def add_objects(
271 self,
272 objects: Sequence[tuple[ShaFile, str | None]],
273 progress: Callable[..., None] | None = None,
274 ) -> "Pack | None":
275 """Add a set of objects to this object store.
277 Args:
278 objects: Iterable over a list of (object, path) tuples
279 progress: Progress callback for object insertion
280 Returns: Optional Pack object of the objects written.
281 """
283 def __contains__(self, sha1: "ObjectID") -> bool:
284 """Check if a hex sha is present."""
285 ...
287 def __getitem__(self, sha1: "ObjectID | RawObjectID") -> ShaFile:
288 """Retrieve an object."""
289 ...
291 def get_commit_graph(self) -> "CommitGraph | None":
292 """Get the commit graph for this object store.
294 Returns:
295 CommitGraph object if available, None otherwise
296 """
297 return None
300class PackedObjectContainer(ObjectContainer):
301 """Container for objects packed in a pack file."""
303 def get_unpacked_object(
304 self, sha1: "ObjectID | RawObjectID", *, include_comp: bool = False
305 ) -> "UnpackedObject":
306 """Get a raw unresolved object.
308 Args:
309 sha1: SHA-1 hash of the object
310 include_comp: Whether to include compressed data
312 Returns:
313 UnpackedObject instance
314 """
315 raise NotImplementedError(self.get_unpacked_object)
317 def iterobjects_subset(
318 self, shas: Iterable["ObjectID"], *, allow_missing: bool = False
319 ) -> Iterator[ShaFile]:
320 """Iterate over a subset of objects.
322 Args:
323 shas: Iterable of object SHAs to retrieve
324 allow_missing: If True, skip missing objects
326 Returns:
327 Iterator of ShaFile objects
328 """
329 raise NotImplementedError(self.iterobjects_subset)
331 def iter_unpacked_subset(
332 self,
333 shas: Iterable["ObjectID | RawObjectID"],
334 *,
335 include_comp: bool = False,
336 allow_missing: bool = False,
337 convert_ofs_delta: bool = True,
338 ) -> Iterator["UnpackedObject"]:
339 """Iterate over unpacked objects from a subset of SHAs.
341 Args:
342 shas: Set of object SHAs to retrieve
343 include_comp: Include compressed data if True
344 allow_missing: If True, skip missing objects
345 convert_ofs_delta: If True, convert offset deltas to ref deltas
347 Returns:
348 Iterator of UnpackedObject instances
349 """
350 raise NotImplementedError(self.iter_unpacked_subset)
353class UnpackedObjectStream:
354 """Abstract base class for a stream of unpacked objects."""
356 def __iter__(self) -> Iterator["UnpackedObject"]:
357 """Iterate over unpacked objects."""
358 raise NotImplementedError(self.__iter__)
360 def __len__(self) -> int:
361 """Return the number of objects in the stream."""
362 raise NotImplementedError(self.__len__)
365def take_msb_bytes(
366 read: Callable[[int], bytes], crc32: int | None = None
367) -> tuple[list[int], int | None]:
368 """Read bytes marked with most significant bit.
370 Args:
371 read: Read function
372 crc32: Optional CRC32 checksum to update
374 Returns:
375 Tuple of (list of bytes read, updated CRC32 or None)
376 """
377 ret: list[int] = []
378 while len(ret) == 0 or ret[-1] & 0x80:
379 b = read(1)
380 if crc32 is not None:
381 crc32 = binascii.crc32(b, crc32)
382 ret.append(ord(b[:1]))
383 return ret, crc32
386class PackFileDisappeared(Exception):
387 """Raised when a pack file unexpectedly disappears.
389 This typically happens when a concurrent operation (e.g. ``git repack``
390 or ``git gc --auto``) removes a pack file between the moment dulwich
391 snapshots the pack directory and the moment it actually opens the
392 pack's ``.idx`` or ``.pack`` file.
394 The ``obj`` attribute holds the :class:`Pack` (or :class:`FilePackIndex`)
395 whose backing file vanished, so the caller can evict the stale object
396 from its cache and rescan the pack directory.
397 """
399 obj: "Pack | FilePackIndex"
401 def __init__(self, obj: "Pack | FilePackIndex") -> None:
402 """Initialize PackFileDisappeared exception.
404 Args:
405 obj: The pack or pack index that disappeared.
406 """
407 self.obj = obj
410class UnpackedObject:
411 """Class encapsulating an object unpacked from a pack file.
413 These objects should only be created from within unpack_object. Most
414 members start out as empty and are filled in at various points by
415 read_zlib_chunks, unpack_object, DeltaChainIterator, etc.
417 End users of this object should take care that the function they're getting
418 this object from is guaranteed to set the members they need.
419 """
421 __slots__ = [
422 "_sha", # Cached binary SHA.
423 "comp_chunks", # Compressed object chunks.
424 "crc32", # CRC32.
425 "decomp_chunks", # Decompressed object chunks.
426 "decomp_len", # Decompressed length of this object.
427 "delta_base", # Delta base offset or SHA.
428 "hash_func", # Hash function to use for computing object IDs.
429 "obj_chunks", # Decompressed and delta-resolved chunks.
430 "obj_type_num", # Type of this object.
431 "offset", # Offset in its pack.
432 "pack_type_num", # Type of this object in the pack (may be a delta).
433 ]
435 obj_type_num: int | None
436 obj_chunks: list[bytes] | None
437 delta_base: None | bytes | int
438 decomp_chunks: list[bytes]
439 comp_chunks: list[bytes] | None
440 decomp_len: int | None
441 crc32: int | None
442 offset: int | None
443 pack_type_num: int
444 _sha: bytes | None
445 hash_func: Callable[[], "HashObject"]
447 # TODO(dborowitz): read_zlib_chunks and unpack_object could very well be
448 # methods of this object.
449 def __init__(
450 self,
451 pack_type_num: int,
452 *,
453 delta_base: None | bytes | int = None,
454 decomp_len: int | None = None,
455 crc32: int | None = None,
456 sha: bytes | None = None,
457 decomp_chunks: list[bytes] | None = None,
458 offset: int | None = None,
459 hash_func: Callable[[], "HashObject"] = sha1,
460 ) -> None:
461 """Initialize an UnpackedObject.
463 Args:
464 pack_type_num: Type number of this object in the pack
465 delta_base: Delta base (offset or SHA) if this is a delta object
466 decomp_len: Decompressed length of this object
467 crc32: CRC32 checksum
468 sha: SHA hash of the object
469 decomp_chunks: Decompressed chunks
470 offset: Offset in the pack file
471 hash_func: Hash function to use (defaults to sha1)
472 """
473 self.offset = offset
474 self._sha = sha
475 self.pack_type_num = pack_type_num
476 self.delta_base = delta_base
477 self.comp_chunks = None
478 self.decomp_chunks: list[bytes] = decomp_chunks or []
479 if decomp_chunks is not None and decomp_len is None:
480 self.decomp_len = sum(map(len, decomp_chunks))
481 else:
482 self.decomp_len = decomp_len
483 self.crc32 = crc32
484 self.hash_func = hash_func
486 if pack_type_num in DELTA_TYPES:
487 self.obj_type_num = None
488 self.obj_chunks = None
489 else:
490 self.obj_type_num = pack_type_num
491 self.obj_chunks = self.decomp_chunks
492 self.delta_base = delta_base
494 def sha(self) -> RawObjectID:
495 """Return the binary SHA of this object."""
496 if self._sha is None:
497 assert self.obj_type_num is not None and self.obj_chunks is not None
498 self._sha = obj_sha(self.obj_type_num, self.obj_chunks, self.hash_func)
499 return RawObjectID(self._sha)
501 def sha_file(self) -> ShaFile:
502 """Return a ShaFile from this object."""
503 assert self.obj_type_num is not None and self.obj_chunks is not None
504 return ShaFile.from_raw_chunks(self.obj_type_num, self.obj_chunks)
506 # Only provided for backwards compatibility with code that expects either
507 # chunks or a delta tuple.
508 def _obj(self) -> OldUnpackedObject:
509 """Return the decompressed chunks, or (delta base, delta chunks)."""
510 if self.pack_type_num in DELTA_TYPES:
511 assert isinstance(self.delta_base, bytes | int)
512 return (self.delta_base, self.decomp_chunks)
513 else:
514 return self.decomp_chunks
516 def __eq__(self, other: object) -> bool:
517 """Check equality with another UnpackedObject."""
518 if not isinstance(other, UnpackedObject):
519 return False
520 for slot in self.__slots__:
521 if getattr(self, slot) != getattr(other, slot):
522 return False
523 return True
525 def __ne__(self, other: object) -> bool:
526 """Check inequality with another UnpackedObject."""
527 return not (self == other)
529 def __repr__(self) -> str:
530 """Return string representation of this UnpackedObject."""
531 data = [f"{s}={getattr(self, s)!r}" for s in self.__slots__]
532 return "{}({})".format(self.__class__.__name__, ", ".join(data))
535_ZLIB_BUFSIZE = 65536 # 64KB buffer for better I/O performance
537# Default maximum memory for caching delta base objects (matches Git's default
538# for core.deltaBaseCacheLimit).
539DEFAULT_DELTA_BASE_CACHE_LIMIT = 96 * 1024 * 1024 # 96 MiB
542def read_zlib_chunks(
543 read_some: Callable[[int], bytes],
544 unpacked: UnpackedObject,
545 include_comp: bool = False,
546 buffer_size: int = _ZLIB_BUFSIZE,
547) -> bytes:
548 """Read zlib data from a buffer.
550 This function requires that the buffer have additional data following the
551 compressed data, which is guaranteed to be the case for git pack files.
553 Args:
554 read_some: Read function that returns at least one byte, but may
555 return less than the requested size.
556 unpacked: An UnpackedObject to write result data to. If its crc32
557 attr is not None, the CRC32 of the compressed bytes will be computed
558 using this starting CRC32.
559 After this function, will have the following attrs set:
560 * comp_chunks (if include_comp is True)
561 * decomp_chunks
562 * decomp_len
563 * crc32
564 include_comp: If True, include compressed data in the result.
565 buffer_size: Size of the read buffer.
566 Returns: Leftover unused data from the decompression.
568 Raises:
569 zlib.error: if a decompression error occurred.
570 """
571 if unpacked.decomp_len is None or unpacked.decomp_len <= -1:
572 raise ValueError("non-negative zlib data stream size expected")
573 decomp_obj = zlib.decompressobj()
575 comp_chunks = []
576 decomp_chunks = unpacked.decomp_chunks
577 decomp_len = 0
578 crc32 = unpacked.crc32
579 max_decomp = unpacked.decomp_len
581 while True:
582 add = read_some(buffer_size)
583 if not add:
584 raise zlib.error("EOF before end of zlib stream")
585 comp_chunks.append(add)
586 # +1 so overrun surfaces as unconsumed_tail rather than being truncated.
587 remaining = max_decomp - decomp_len + 1
588 decomp = decomp_obj.decompress(add, remaining)
589 if decomp_obj.unconsumed_tail:
590 raise zlib.error("decompressed data exceeds expected size")
591 decomp_len += len(decomp)
592 decomp_chunks.append(decomp)
593 unused = decomp_obj.unused_data
594 if unused:
595 left = len(unused)
596 if crc32 is not None:
597 crc32 = binascii.crc32(add[:-left], crc32)
598 if include_comp:
599 comp_chunks[-1] = add[:-left]
600 break
601 elif crc32 is not None:
602 crc32 = binascii.crc32(add, crc32)
603 if crc32 is not None:
604 crc32 &= 0xFFFFFFFF
606 if decomp_len != unpacked.decomp_len:
607 raise zlib.error("decompressed data does not match expected size")
609 unpacked.crc32 = crc32
610 if include_comp:
611 unpacked.comp_chunks = comp_chunks
612 return unused
615def iter_sha1(iter: Iterable[bytes]) -> bytes:
616 """Return the hexdigest of the SHA1 over a set of names.
618 Args:
619 iter: Iterator over string objects
620 Returns: 40-byte hex sha1 digest
621 """
622 sha = sha1()
623 for name in iter:
624 sha.update(name)
625 return sha.hexdigest().encode("ascii")
628def load_pack_index(
629 path: str | os.PathLike[str], object_format: ObjectFormat
630) -> "PackIndex":
631 """Load an index file by path.
633 Args:
634 path: Path to the index file
635 object_format: Hash algorithm used by the repository
636 Returns: A PackIndex loaded from the given path
637 """
638 # Ownership of the file is transferred to the returned index, which mmaps
639 # it and closes it in PackIndex.close(). It must not be closed here: on
640 # Windows an mmap keeps the file locked, so closing the handle out from
641 # under a live mapping leaves the .idx undeletable until the index is GCed.
642 f = GitFile(path, "rb")
643 try:
644 return load_pack_index_file(path, f, object_format)
645 except BaseException:
646 f.close()
647 raise
650def _load_file_contents(
651 f: IO[bytes] | _GitFile, size: int | None = None
652) -> tuple[bytes | Any, int]:
653 """Load contents from a file, preferring mmap when possible.
655 Args:
656 f: File-like object to load
657 size: Expected size, or None to determine from file
658 Returns: Tuple of (contents, size)
659 """
660 try:
661 fd = f.fileno()
662 except (UnsupportedOperation, AttributeError):
663 fd = None
664 # Attempt to use mmap if possible
665 if fd is not None:
666 if size is None:
667 size = os.fstat(fd).st_size
668 if has_mmap:
669 try:
670 contents = mmap.mmap(fd, size, access=mmap.ACCESS_READ)
671 except (OSError, ValueError):
672 # Can't mmap - perhaps a socket or invalid file descriptor
673 pass
674 else:
675 return contents, size
676 contents_bytes = f.read()
677 size = len(contents_bytes)
678 return contents_bytes, size
681def load_pack_index_file(
682 path: str | os.PathLike[str],
683 f: IO[bytes] | _GitFile,
684 object_format: ObjectFormat,
685) -> "PackIndex":
686 """Load an index file from a file-like object.
688 Args:
689 path: Path for the index file
690 f: File-like object
691 object_format: Hash algorithm used by the repository
692 Returns: A PackIndex loaded from the given file
693 """
694 contents, size = _load_file_contents(f)
695 if contents[:4] == b"\377tOc":
696 version = struct.unpack(b">L", contents[4:8])[0]
697 if version == 2:
698 return PackIndex2(
699 path,
700 object_format,
701 file=f,
702 contents=contents,
703 size=size,
704 )
705 elif version == 3:
706 return PackIndex3(path, object_format, file=f, contents=contents, size=size)
707 else:
708 raise KeyError(f"Unknown pack index format {version}")
709 else:
710 return PackIndex1(path, object_format, file=f, contents=contents, size=size)
713def bisect_find_sha(
714 start: int, end: int, sha: bytes, unpack_name: Callable[[int], bytes]
715) -> int | None:
716 """Find a SHA in a data blob with sorted SHAs.
718 Args:
719 start: Start index of range to search
720 end: End index of range to search
721 sha: Sha to find
722 unpack_name: Callback to retrieve SHA by index
723 Returns: Index of the SHA, or None if it wasn't found
724 """
725 assert start <= end
726 while start <= end:
727 i = (start + end) // 2
728 file_sha = unpack_name(i)
729 if file_sha < sha:
730 start = i + 1
731 elif file_sha > sha:
732 end = i - 1
733 else:
734 return i
735 return None
738PackIndexEntry = tuple[RawObjectID, int, int | None]
741class PackIndex:
742 """An index in to a packfile.
744 Given a sha id of an object a pack index can tell you the location in the
745 packfile of that object if it has it.
746 """
748 object_format: "ObjectFormat"
750 def __eq__(self, other: object) -> bool:
751 """Check equality with another PackIndex."""
752 if not isinstance(other, PackIndex):
753 return False
755 for (name1, _, _), (name2, _, _) in zip(
756 self.iterentries(), other.iterentries()
757 ):
758 if name1 != name2:
759 return False
760 return True
762 def __ne__(self, other: object) -> bool:
763 """Check if this pack index is not equal to another."""
764 return not self.__eq__(other)
766 def __len__(self) -> int:
767 """Return the number of entries in this pack index."""
768 raise NotImplementedError(self.__len__)
770 def __iter__(self) -> Iterator[ObjectID]:
771 """Iterate over the SHAs in this pack."""
772 return map(lambda sha: sha_to_hex(RawObjectID(sha)), self._itersha())
774 def iterentries(self) -> Iterator[PackIndexEntry]:
775 """Iterate over the entries in this pack index.
777 Returns: iterator over tuples with object name, offset in packfile and
778 crc32 checksum.
779 """
780 raise NotImplementedError(self.iterentries)
782 def get_pack_checksum(self) -> bytes | None:
783 """Return the SHA1 checksum stored for the corresponding packfile.
785 Returns: 20-byte binary digest, or None if not available
786 """
787 raise NotImplementedError(self.get_pack_checksum)
789 def object_offset(self, sha: ObjectID | RawObjectID) -> int:
790 """Return the offset in to the corresponding packfile for the object.
792 Given the name of an object it will return the offset that object
793 lives at within the corresponding pack file. If the pack file doesn't
794 have the object then None will be returned.
795 """
796 raise NotImplementedError(self.object_offset)
798 def object_sha1(self, index: int) -> bytes:
799 """Return the SHA1 corresponding to the index in the pack file."""
800 for name, offset, _crc32 in self.iterentries():
801 if offset == index:
802 return name
803 else:
804 raise KeyError(index)
806 def _object_offset(self, sha: bytes) -> int:
807 """See object_offset.
809 Args:
810 sha: A *binary* SHA string. (20 characters long)_
811 """
812 raise NotImplementedError(self._object_offset)
814 def objects_sha1(self) -> bytes:
815 """Return the hex SHA1 over all the shas of all objects in this pack.
817 Note: This is used for the filename of the pack.
818 """
819 return iter_sha1(self._itersha())
821 def _itersha(self) -> Iterator[bytes]:
822 """Yield all the SHA1's of the objects in the index, sorted."""
823 raise NotImplementedError(self._itersha)
825 def iter_prefix(self, prefix: bytes) -> Iterator[RawObjectID]:
826 """Iterate over all SHA1s with the given prefix.
828 Args:
829 prefix: Binary prefix to match
830 Returns: Iterator of matching SHA1s
831 """
832 # Default implementation for PackIndex classes that don't override
833 for sha, _, _ in self.iterentries():
834 if sha.startswith(prefix):
835 yield RawObjectID(sha)
837 def close(self) -> None:
838 """Close any open files."""
840 def check(self) -> None:
841 """Check the consistency of this pack index."""
844class MemoryPackIndex(PackIndex):
845 """Pack index that is stored entirely in memory."""
847 def __init__(
848 self,
849 entries: list[PackIndexEntry],
850 object_format: ObjectFormat,
851 pack_checksum: bytes | None = None,
852 ) -> None:
853 """Create a new MemoryPackIndex.
855 Args:
856 entries: Sequence of name, idx, crc32 (sorted)
857 object_format: Object format used by this index
858 pack_checksum: Optional pack checksum
859 """
860 self._by_sha = {}
861 self._by_offset = {}
862 for name, offset, _crc32 in entries:
863 self._by_sha[name] = offset
864 self._by_offset[offset] = name
865 self._entries = entries
866 self._pack_checksum = pack_checksum
867 self.object_format = object_format
869 def get_pack_checksum(self) -> bytes | None:
870 """Return the SHA checksum stored for the corresponding packfile."""
871 return self._pack_checksum
873 def __len__(self) -> int:
874 """Return the number of entries in this pack index."""
875 return len(self._entries)
877 def object_offset(self, sha: ObjectID | RawObjectID) -> int:
878 """Return the offset for the given SHA.
880 Args:
881 sha: SHA to look up (binary or hex)
882 Returns: Offset in the pack file
883 """
884 lookup_sha: RawObjectID
885 if len(sha) == self.object_format.hex_length:
886 lookup_sha = hex_to_sha(ObjectID(sha))
887 else:
888 lookup_sha = RawObjectID(sha)
889 return self._by_sha[lookup_sha]
891 def object_sha1(self, index: int) -> bytes:
892 """Return the SHA1 for the object at the given offset."""
893 return self._by_offset[index]
895 def _itersha(self) -> Iterator[bytes]:
896 """Iterate over all SHA1s in the index."""
897 return iter(self._by_sha)
899 def iterentries(self) -> Iterator[PackIndexEntry]:
900 """Iterate over all index entries."""
901 return iter(self._entries)
903 @classmethod
904 def for_pack(cls, pack_data: "PackData") -> "MemoryPackIndex":
905 """Create a MemoryPackIndex from a PackData object."""
906 return MemoryPackIndex(
907 list(pack_data.sorted_entries()),
908 pack_checksum=pack_data.get_stored_checksum(),
909 object_format=pack_data.object_format,
910 )
912 @classmethod
913 def clone(cls, other_index: "PackIndex") -> "MemoryPackIndex":
914 """Create a copy of another PackIndex in memory."""
915 return cls(
916 list(other_index.iterentries()),
917 other_index.object_format,
918 other_index.get_pack_checksum(),
919 )
922class FilePackIndex(PackIndex):
923 """Pack index that is based on a file.
925 To do the loop it opens the file, and indexes first 256 4 byte groups
926 with the first byte of the sha id. The value in the four byte group indexed
927 is the end of the group that shares the same starting byte. Subtract one
928 from the starting byte and index again to find the start of the group.
929 The values are sorted by sha id within the group, so do the math to find
930 the start and end offset and then bisect in to find if the value is
931 present.
932 """
934 _fan_out_table: list[int]
935 _file: IO[bytes] | _GitFile
937 def __init__(
938 self,
939 filename: str | os.PathLike[str],
940 file: IO[bytes] | _GitFile | None = None,
941 contents: "bytes | mmap.mmap | None" = None,
942 size: int | None = None,
943 ) -> None:
944 """Create a pack index object.
946 Provide it with the name of the index file to consider, and it will map
947 it whenever required.
948 """
949 self._filename = filename
950 # Take the size now, so it can be checked each time we map the file to
951 # ensure that it hasn't changed.
952 if file is None:
953 self._file = GitFile(filename, "rb")
954 else:
955 self._file = file
956 if contents is None:
957 self._contents, self._size = _load_file_contents(self._file, size)
958 else:
959 self._contents = contents
960 self._size = size if size is not None else len(contents)
962 @property
963 def path(self) -> str:
964 """Return the path to this index file."""
965 return os.fspath(self._filename)
967 def __eq__(self, other: object) -> bool:
968 """Check equality with another FilePackIndex."""
969 # Quick optimization:
970 if (
971 isinstance(other, FilePackIndex)
972 and self._fan_out_table != other._fan_out_table
973 ):
974 return False
976 return super().__eq__(other)
978 def close(self) -> None:
979 """Close the underlying file and any mmap."""
980 # Close the mmap before the file: on Windows the mapping holds a lock
981 # on the file, so the handle cannot be released while it is alive.
982 close_fn = getattr(self._contents, "close", None)
983 if close_fn is not None:
984 close_fn()
985 self._file.close()
987 def __del__(self) -> None:
988 """Ensure the file and mmap are closed when GCed."""
989 if not getattr(self._file, "closed", True):
990 import warnings
992 warnings.warn(
993 f"unclosed pack index {self!r}",
994 ResourceWarning,
995 stacklevel=2,
996 source=self,
997 )
998 try:
999 self.close()
1000 except Exception:
1001 # Ignore errors during cleanup
1002 pass
1004 def __enter__(self) -> Self:
1005 """Enter context manager."""
1006 return self
1008 def __exit__(
1009 self,
1010 type: type | None,
1011 value: BaseException | None,
1012 traceback: TracebackType | None,
1013 ) -> None:
1014 """Exit context manager."""
1015 self.close()
1017 def __len__(self) -> int:
1018 """Return the number of entries in this pack index."""
1019 return self._fan_out_table[-1]
1021 def _unpack_entry(self, i: int) -> PackIndexEntry:
1022 """Unpack the i-th entry in the index file.
1024 Returns: Tuple with object name (SHA), offset in pack file and CRC32
1025 checksum (if known).
1026 """
1027 raise NotImplementedError(self._unpack_entry)
1029 def _unpack_name(self, i: int) -> bytes:
1030 """Unpack the i-th name from the index file."""
1031 raise NotImplementedError(self._unpack_name)
1033 def _unpack_offset(self, i: int) -> int:
1034 """Unpack the i-th object offset from the index file."""
1035 raise NotImplementedError(self._unpack_offset)
1037 def _unpack_crc32_checksum(self, i: int) -> int | None:
1038 """Unpack the crc32 checksum for the i-th object from the index file."""
1039 raise NotImplementedError(self._unpack_crc32_checksum)
1041 def _itersha(self) -> Iterator[bytes]:
1042 """Iterate over all SHA1s in the index."""
1043 for i in range(len(self)):
1044 yield self._unpack_name(i)
1046 def iterentries(self) -> Iterator[PackIndexEntry]:
1047 """Iterate over the entries in this pack index.
1049 Returns: iterator over tuples with object name, offset in packfile and
1050 crc32 checksum.
1051 """
1052 for i in range(len(self)):
1053 yield self._unpack_entry(i)
1055 def _read_fan_out_table(self, start_offset: int) -> list[int]:
1056 """Read the fan-out table from the index.
1058 The fan-out table contains 256 entries mapping first byte values
1059 to the number of objects with SHA1s less than or equal to that byte.
1061 Args:
1062 start_offset: Offset in the file where the fan-out table starts
1063 Returns: List of 256 integers
1064 """
1065 ret = []
1066 for i in range(0x100):
1067 fanout_entry = self._contents[
1068 start_offset + i * 4 : start_offset + (i + 1) * 4
1069 ]
1070 ret.append(struct.unpack(">L", fanout_entry)[0])
1071 return ret
1073 def check(self) -> None:
1074 """Check that the stored checksum matches the actual checksum."""
1075 actual = self.calculate_checksum()
1076 stored = self.get_stored_checksum()
1077 if actual != stored:
1078 raise ChecksumMismatch(stored, actual)
1080 def calculate_checksum(self) -> bytes:
1081 """Calculate the SHA1 checksum over this pack index.
1083 Returns: This is a 20-byte binary digest
1084 """
1085 return sha1(self._contents[:-20]).digest()
1087 def get_pack_checksum(self) -> bytes:
1088 """Return the SHA1 checksum stored for the corresponding packfile.
1090 Returns: 20-byte binary digest
1091 """
1092 return bytes(self._contents[-40:-20])
1094 def get_stored_checksum(self) -> bytes:
1095 """Return the SHA1 checksum stored for this index.
1097 Returns: 20-byte binary digest
1098 """
1099 return bytes(self._contents[-20:])
1101 def object_offset(self, sha: ObjectID | RawObjectID) -> int:
1102 """Return the offset in to the corresponding packfile for the object.
1104 Given the name of an object it will return the offset that object
1105 lives at within the corresponding pack file. If the pack file doesn't
1106 have the object then None will be returned.
1107 """
1108 lookup_sha: RawObjectID
1109 if len(sha) == self.object_format.hex_length: # hex string
1110 lookup_sha = hex_to_sha(ObjectID(sha))
1111 else:
1112 lookup_sha = RawObjectID(sha)
1113 try:
1114 return self._object_offset(lookup_sha)
1115 except ValueError as exc:
1116 closed = getattr(self._contents, "closed", None)
1117 if closed in (None, True):
1118 raise PackFileDisappeared(self) from exc
1119 raise
1121 def _object_offset(self, sha: bytes) -> int:
1122 """See object_offset.
1124 Args:
1125 sha: A *binary* SHA string. (20 characters long)_
1126 """
1127 hash_size = getattr(self, "hash_size", 20) # Default to SHA1 for v1
1128 assert len(sha) == hash_size
1129 idx = ord(sha[:1])
1130 if idx == 0:
1131 start = 0
1132 else:
1133 start = self._fan_out_table[idx - 1]
1134 end = self._fan_out_table[idx]
1135 i = bisect_find_sha(start, end, sha, self._unpack_name)
1136 if i is None:
1137 raise KeyError(sha)
1138 return self._unpack_offset(i)
1140 def iter_prefix(self, prefix: bytes) -> Iterator[RawObjectID]:
1141 """Iterate over all SHA1s with the given prefix."""
1142 start = ord(prefix[:1])
1143 if start == 0:
1144 start = 0
1145 else:
1146 start = self._fan_out_table[start - 1]
1147 end = ord(prefix[:1]) + 1
1148 if end == 0x100:
1149 end = len(self)
1150 else:
1151 end = self._fan_out_table[end]
1152 assert start <= end
1153 started = False
1154 for i in range(start, end):
1155 name: bytes = self._unpack_name(i)
1156 if name.startswith(prefix):
1157 yield RawObjectID(name)
1158 started = True
1159 elif started:
1160 break
1163class PackIndex1(FilePackIndex):
1164 """Version 1 Pack Index file."""
1166 object_format = SHA1
1168 def __init__(
1169 self,
1170 filename: str | os.PathLike[str],
1171 object_format: ObjectFormat,
1172 file: IO[bytes] | _GitFile | None = None,
1173 contents: bytes | None = None,
1174 size: int | None = None,
1175 ) -> None:
1176 """Initialize a version 1 pack index.
1178 Args:
1179 filename: Path to the index file
1180 object_format: Object format used by the repository
1181 file: Optional file object
1182 contents: Optional mmap'd contents
1183 size: Optional size of the index
1184 """
1185 super().__init__(filename, file, contents, size)
1187 # PackIndex1 only supports SHA1
1188 if object_format != SHA1:
1189 raise AssertionError(
1190 f"PackIndex1 only supports SHA1, not {object_format.name}"
1191 )
1193 self.object_format = object_format
1194 self.version = 1
1195 self._fan_out_table = self._read_fan_out_table(0)
1196 self.hash_size = self.object_format.oid_length
1197 self._entry_size = 4 + self.hash_size
1199 def _unpack_entry(self, i: int) -> tuple[RawObjectID, int, None]:
1200 base_offset = (0x100 * 4) + (i * self._entry_size)
1201 offset = unpack_from(">L", self._contents, base_offset)[0]
1202 name = self._contents[base_offset + 4 : base_offset + 4 + self.hash_size]
1203 return (RawObjectID(name), offset, None)
1205 def _unpack_name(self, i: int) -> bytes:
1206 offset = (0x100 * 4) + (i * self._entry_size) + 4
1207 return self._contents[offset : offset + self.hash_size]
1209 def _unpack_offset(self, i: int) -> int:
1210 offset = (0x100 * 4) + (i * self._entry_size)
1211 return int(unpack_from(">L", self._contents, offset)[0])
1213 def _unpack_crc32_checksum(self, i: int) -> None:
1214 # Not stored in v1 index files
1215 return None
1218class PackIndex2(FilePackIndex):
1219 """Version 2 Pack Index file."""
1221 object_format = SHA1
1223 def __init__(
1224 self,
1225 filename: str | os.PathLike[str],
1226 object_format: ObjectFormat,
1227 file: IO[bytes] | _GitFile | None = None,
1228 contents: bytes | None = None,
1229 size: int | None = None,
1230 ) -> None:
1231 """Initialize a version 2 pack index.
1233 Args:
1234 filename: Path to the index file
1235 object_format: Object format used by the repository
1236 file: Optional file object
1237 contents: Optional mmap'd contents
1238 size: Optional size of the index
1239 """
1240 super().__init__(filename, file, contents, size)
1241 self.object_format = object_format
1242 if self._contents[:4] != b"\377tOc":
1243 raise AssertionError("Not a v2 pack index file")
1244 (self.version,) = unpack_from(b">L", self._contents, 4)
1245 if self.version != 2:
1246 raise AssertionError(f"Version was {self.version}")
1247 self._fan_out_table = self._read_fan_out_table(8)
1248 self.hash_size = self.object_format.oid_length
1249 self._name_table_offset = 8 + 0x100 * 4
1250 self._crc32_table_offset = self._name_table_offset + self.hash_size * len(self)
1251 self._pack_offset_table_offset = self._crc32_table_offset + 4 * len(self)
1252 self._pack_offset_largetable_offset = self._pack_offset_table_offset + 4 * len(
1253 self
1254 )
1256 def _unpack_entry(self, i: int) -> tuple[RawObjectID, int, int]:
1257 return (
1258 RawObjectID(self._unpack_name(i)),
1259 self._unpack_offset(i),
1260 self._unpack_crc32_checksum(i),
1261 )
1263 def _unpack_name(self, i: int) -> bytes:
1264 offset = self._name_table_offset + i * self.hash_size
1265 return self._contents[offset : offset + self.hash_size]
1267 def _unpack_offset(self, i: int) -> int:
1268 offset = self._pack_offset_table_offset + i * 4
1269 offset_val = int(unpack_from(">L", self._contents, offset)[0])
1270 if offset_val & (2**31):
1271 offset = (
1272 self._pack_offset_largetable_offset + (offset_val & (2**31 - 1)) * 8
1273 )
1274 offset_val = int(unpack_from(">Q", self._contents, offset)[0])
1275 return offset_val
1277 def _unpack_crc32_checksum(self, i: int) -> int:
1278 return int(
1279 unpack_from(">L", self._contents, self._crc32_table_offset + i * 4)[0]
1280 )
1282 def get_pack_checksum(self) -> bytes:
1283 """Return the checksum stored for the corresponding packfile.
1285 Returns: binary digest (size depends on hash algorithm)
1286 """
1287 # Index ends with: pack_checksum + index_checksum
1288 # Each checksum is hash_size bytes
1289 checksum_size = self.hash_size
1290 return bytes(self._contents[-2 * checksum_size : -checksum_size])
1292 def get_stored_checksum(self) -> bytes:
1293 """Return the checksum stored for this index.
1295 Returns: binary digest (size depends on hash algorithm)
1296 """
1297 checksum_size = self.hash_size
1298 return bytes(self._contents[-checksum_size:])
1300 def calculate_checksum(self) -> bytes:
1301 """Calculate the checksum over this pack index.
1303 Returns: binary digest (size depends on hash algorithm)
1304 """
1305 # Determine hash function based on hash_size
1306 if self.hash_size == 20:
1307 hash_func = sha1
1308 elif self.hash_size == 32:
1309 hash_func = sha256
1310 else:
1311 raise ValueError(f"Unsupported hash size: {self.hash_size}")
1313 return hash_func(self._contents[: -self.hash_size]).digest()
1316class PackIndex3(FilePackIndex):
1317 """Version 3 Pack Index file.
1319 Supports variable hash sizes for SHA-1 (20 bytes) and SHA-256 (32 bytes).
1320 """
1322 def __init__(
1323 self,
1324 filename: str | os.PathLike[str],
1325 object_format: ObjectFormat,
1326 file: IO[bytes] | _GitFile | None = None,
1327 contents: bytes | None = None,
1328 size: int | None = None,
1329 ) -> None:
1330 """Initialize a version 3 pack index.
1332 Args:
1333 filename: Path to the index file
1334 object_format: Object format used by the repository
1335 file: Optional file object
1336 contents: Optional mmap'd contents
1337 size: Optional size of the index
1338 """
1339 super().__init__(filename, file, contents, size)
1340 if self._contents[:4] != b"\377tOc":
1341 raise AssertionError("Not a v3 pack index file")
1342 (self.version,) = unpack_from(b">L", self._contents, 4)
1343 if self.version != 3:
1344 raise AssertionError(f"Version was {self.version}")
1346 # Read hash algorithm identifier (1 = SHA-1, 2 = SHA-256)
1347 (self.hash_format,) = unpack_from(b">L", self._contents, 8)
1348 file_object_format = OBJECT_FORMAT_TYPE_NUMS[self.hash_format]
1350 # Verify provided object_format matches what's in the file
1351 if object_format != file_object_format:
1352 raise AssertionError(
1353 f"Object format mismatch: provided {object_format.name}, "
1354 f"but file contains {file_object_format.name}"
1355 )
1357 self.object_format = object_format
1358 self.hash_size = self.object_format.oid_length
1360 # Read length of shortened object names
1361 (self.shortened_oid_len,) = unpack_from(b">L", self._contents, 12)
1363 # Calculate offsets based on variable hash size
1364 self._fan_out_table = self._read_fan_out_table(
1365 16
1366 ) # After header (4 + 4 + 4 + 4)
1367 self._name_table_offset = 16 + 0x100 * 4
1368 self._crc32_table_offset = self._name_table_offset + self.hash_size * len(self)
1369 self._pack_offset_table_offset = self._crc32_table_offset + 4 * len(self)
1370 self._pack_offset_largetable_offset = self._pack_offset_table_offset + 4 * len(
1371 self
1372 )
1374 def _unpack_entry(self, i: int) -> tuple[RawObjectID, int, int]:
1375 return (
1376 RawObjectID(self._unpack_name(i)),
1377 self._unpack_offset(i),
1378 self._unpack_crc32_checksum(i),
1379 )
1381 def _unpack_name(self, i: int) -> bytes:
1382 offset = self._name_table_offset + i * self.hash_size
1383 return self._contents[offset : offset + self.hash_size]
1385 def _unpack_offset(self, i: int) -> int:
1386 offset_pos = self._pack_offset_table_offset + i * 4
1387 offset = unpack_from(">L", self._contents, offset_pos)[0]
1388 assert isinstance(offset, int)
1389 if offset & (2**31):
1390 large_offset_pos = (
1391 self._pack_offset_largetable_offset + (offset & (2**31 - 1)) * 8
1392 )
1393 offset = unpack_from(">Q", self._contents, large_offset_pos)[0]
1394 assert isinstance(offset, int)
1395 return offset
1397 def _unpack_crc32_checksum(self, i: int) -> int:
1398 result = unpack_from(">L", self._contents, self._crc32_table_offset + i * 4)[0]
1399 assert isinstance(result, int)
1400 return result
1403def read_pack_header(read: Callable[[int], bytes]) -> tuple[int, int]:
1404 """Read the header of a pack file.
1406 Args:
1407 read: Read function
1408 Returns: Tuple of (pack version, number of objects). If no data is
1409 available to read, returns (None, None).
1410 """
1411 header = read(12)
1412 if not header:
1413 raise AssertionError("file too short to contain pack")
1414 if header[:4] != b"PACK":
1415 raise AssertionError(f"Invalid pack header {header!r}")
1416 (version,) = unpack_from(b">L", header, 4)
1417 if version not in (2, 3):
1418 raise AssertionError(f"Version was {version}")
1419 (num_objects,) = unpack_from(b">L", header, 8)
1420 return (version, num_objects)
1423def chunks_length(chunks: bytes | Iterable[bytes]) -> int:
1424 """Get the total length of a sequence of chunks.
1426 Args:
1427 chunks: Either a single bytes object or an iterable of bytes
1428 Returns: Total length in bytes
1429 """
1430 if isinstance(chunks, bytes):
1431 return len(chunks)
1432 else:
1433 return sum(map(len, chunks))
1436def unpack_object(
1437 read_all: Callable[[int], bytes],
1438 hash_func: Callable[[], "HashObject"],
1439 read_some: Callable[[int], bytes] | None = None,
1440 compute_crc32: bool = False,
1441 include_comp: bool = False,
1442 zlib_bufsize: int = _ZLIB_BUFSIZE,
1443) -> tuple[UnpackedObject, bytes]:
1444 """Unpack a Git object.
1446 Args:
1447 read_all: Read function that blocks until the number of requested
1448 bytes are read.
1449 hash_func: Hash function to use for computing object IDs.
1450 read_some: Read function that returns at least one byte, but may not
1451 return the number of bytes requested.
1452 compute_crc32: If True, compute the CRC32 of the compressed data. If
1453 False, the returned CRC32 will be None.
1454 include_comp: If True, include compressed data in the result.
1455 zlib_bufsize: An optional buffer size for zlib operations.
1456 Returns: A tuple of (unpacked, unused), where unused is the unused data
1457 leftover from decompression, and unpacked in an UnpackedObject with
1458 the following attrs set:
1460 * obj_chunks (for non-delta types)
1461 * pack_type_num
1462 * delta_base (for delta types)
1463 * comp_chunks (if include_comp is True)
1464 * decomp_chunks
1465 * decomp_len
1466 * crc32 (if compute_crc32 is True)
1467 """
1468 if read_some is None:
1469 read_some = read_all
1470 if compute_crc32:
1471 crc32 = 0
1472 else:
1473 crc32 = None
1475 raw, crc32 = take_msb_bytes(read_all, crc32=crc32)
1476 type_num = (raw[0] >> 4) & 0x07
1477 size = raw[0] & 0x0F
1478 for i, byte in enumerate(raw[1:]):
1479 size += (byte & 0x7F) << ((i * 7) + 4)
1481 delta_base: int | bytes | None
1482 raw_base = len(raw)
1483 if type_num == OFS_DELTA:
1484 raw, crc32 = take_msb_bytes(read_all, crc32=crc32)
1485 raw_base += len(raw)
1486 if raw[-1] & 0x80:
1487 raise AssertionError
1488 delta_base_offset = raw[0] & 0x7F
1489 for byte in raw[1:]:
1490 delta_base_offset += 1
1491 delta_base_offset <<= 7
1492 delta_base_offset += byte & 0x7F
1493 if delta_base_offset == 0:
1494 # A zero offset makes the delta reference itself, which would
1495 # loop forever in resolve_object. git's C client rejects this
1496 # with "delta offset == 0 is invalid".
1497 raise ApplyDeltaError("OFS_DELTA has delta_base_offset of 0")
1498 delta_base = delta_base_offset
1499 elif type_num == REF_DELTA:
1500 # Determine hash size from hash_func
1501 hash_size = len(hash_func().digest())
1502 delta_base_obj = read_all(hash_size)
1503 if crc32 is not None:
1504 crc32 = binascii.crc32(delta_base_obj, crc32)
1505 delta_base = delta_base_obj
1506 raw_base += hash_size
1507 else:
1508 delta_base = None
1510 unpacked = UnpackedObject(
1511 type_num,
1512 delta_base=delta_base,
1513 decomp_len=size,
1514 crc32=crc32,
1515 hash_func=hash_func,
1516 )
1517 unused = read_zlib_chunks(
1518 read_some,
1519 unpacked,
1520 buffer_size=zlib_bufsize,
1521 include_comp=include_comp,
1522 )
1523 return unpacked, unused
1526def _compute_object_size(value: tuple[int, Any]) -> int:
1527 """Compute the size of a unresolved object for use with LRUSizeCache."""
1528 (num, obj) = value
1529 if num in DELTA_TYPES:
1530 return chunks_length(obj[1])
1531 return chunks_length(obj)
1534class PackStreamReader:
1535 """Class to read a pack stream.
1537 The pack is read from a ReceivableProtocol using read() or recv() as
1538 appropriate.
1539 """
1541 def __init__(
1542 self,
1543 hash_func: Callable[[], "HashObject"],
1544 read_all: Callable[[int], bytes],
1545 read_some: Callable[[int], bytes] | None = None,
1546 zlib_bufsize: int = _ZLIB_BUFSIZE,
1547 ) -> None:
1548 """Initialize pack stream reader.
1550 Args:
1551 hash_func: Hash function to use for computing object IDs
1552 read_all: Function to read all requested bytes
1553 read_some: Function to read some bytes (optional)
1554 zlib_bufsize: Buffer size for zlib decompression
1555 """
1556 self.read_all = read_all
1557 if read_some is None:
1558 self.read_some = read_all
1559 else:
1560 self.read_some = read_some
1561 self.hash_func = hash_func
1562 self.sha = hash_func()
1563 self._hash_size = len(hash_func().digest())
1564 self._offset = 0
1565 self._rbuf = BytesIO()
1566 # trailer is a deque to avoid memory allocation on small reads
1567 self._trailer: deque[int] = deque()
1568 self._zlib_bufsize = zlib_bufsize
1570 def _read(self, read: Callable[[int], bytes], size: int) -> bytes:
1571 """Read up to size bytes using the given callback.
1573 As a side effect, update the verifier's hash (excluding the last
1574 hash_size bytes read, which is the pack checksum).
1576 Args:
1577 read: The read callback to read from.
1578 size: The maximum number of bytes to read; the particular
1579 behavior is callback-specific.
1580 Returns: Bytes read
1581 """
1582 data = read(size)
1584 # maintain a trailer of the last hash_size bytes we've read
1585 n = len(data)
1586 self._offset += n
1587 tn = len(self._trailer)
1588 if n >= self._hash_size:
1589 to_pop = tn
1590 to_add = self._hash_size
1591 else:
1592 to_pop = max(n + tn - self._hash_size, 0)
1593 to_add = n
1594 self.sha.update(
1595 bytes(bytearray([self._trailer.popleft() for _ in range(to_pop)]))
1596 )
1597 self._trailer.extend(data[-to_add:])
1599 # hash everything but the trailer
1600 self.sha.update(data[:-to_add])
1601 return data
1603 def _buf_len(self) -> int:
1604 buf = self._rbuf
1605 start = buf.tell()
1606 buf.seek(0, SEEK_END)
1607 end = buf.tell()
1608 buf.seek(start)
1609 return end - start
1611 @property
1612 def offset(self) -> int:
1613 """Return current offset in the stream."""
1614 return self._offset - self._buf_len()
1616 def read(self, size: int) -> bytes:
1617 """Read, blocking until size bytes are read."""
1618 buf_len = self._buf_len()
1619 if buf_len >= size:
1620 return self._rbuf.read(size)
1621 buf_data = self._rbuf.read()
1622 self._rbuf = BytesIO()
1623 return buf_data + self._read(self.read_all, size - buf_len)
1625 def recv(self, size: int) -> bytes:
1626 """Read up to size bytes, blocking until one byte is read."""
1627 buf_len = self._buf_len()
1628 if buf_len:
1629 data = self._rbuf.read(size)
1630 if size >= buf_len:
1631 self._rbuf = BytesIO()
1632 return data
1633 return self._read(self.read_some, size)
1635 def __len__(self) -> int:
1636 """Return the number of objects in this pack."""
1637 return self._num_objects
1639 def read_objects(self, compute_crc32: bool = False) -> Iterator[UnpackedObject]:
1640 """Read the objects in this pack file.
1642 Args:
1643 compute_crc32: If True, compute the CRC32 of the compressed
1644 data. If False, the returned CRC32 will be None.
1645 Returns: Iterator over UnpackedObjects with the following members set:
1646 offset
1647 obj_type_num
1648 obj_chunks (for non-delta types)
1649 delta_base (for delta types)
1650 decomp_chunks
1651 decomp_len
1652 crc32 (if compute_crc32 is True)
1654 Raises:
1655 ChecksumMismatch: if the checksum of the pack contents does not
1656 match the checksum in the pack trailer.
1657 zlib.error: if an error occurred during zlib decompression.
1658 IOError: if an error occurred writing to the output file.
1659 """
1660 _pack_version, self._num_objects = read_pack_header(self.read)
1662 for _ in range(self._num_objects):
1663 offset = self.offset
1664 unpacked, unused = unpack_object(
1665 self.read,
1666 self.hash_func,
1667 read_some=self.recv,
1668 compute_crc32=compute_crc32,
1669 zlib_bufsize=self._zlib_bufsize,
1670 )
1671 unpacked.offset = offset
1673 # prepend any unused data to current read buffer
1674 buf = BytesIO()
1675 buf.write(unused)
1676 buf.write(self._rbuf.read())
1677 buf.seek(0)
1678 self._rbuf = buf
1680 yield unpacked
1682 if self._buf_len() < self._hash_size:
1683 # If the read buffer is full, then the last read() got the whole
1684 # trailer off the wire. If not, it means there is still some of the
1685 # trailer to read. We need to read() all hash_size bytes; N come from the
1686 # read buffer and (hash_size - N) come from the wire.
1687 self.read(self._hash_size)
1689 pack_sha = bytearray(self._trailer)
1690 if pack_sha != self.sha.digest():
1691 raise ChecksumMismatch(
1692 sha_to_hex(RawObjectID(bytes(pack_sha))), self.sha.hexdigest()
1693 )
1696class PackStreamCopier(PackStreamReader):
1697 """Class to verify a pack stream as it is being read.
1699 The pack is read from a ReceivableProtocol using read() or recv() as
1700 appropriate and written out to the given file-like object.
1701 """
1703 def __init__(
1704 self,
1705 hash_func: Callable[[], "HashObject"],
1706 read_all: Callable[[int], bytes],
1707 read_some: Callable[[int], bytes] | None,
1708 outfile: IO[bytes],
1709 delta_iter: "DeltaChainIterator[UnpackedObject] | None" = None,
1710 ) -> None:
1711 """Initialize the copier.
1713 Args:
1714 hash_func: Hash function to use for computing object IDs
1715 read_all: Read function that blocks until the number of
1716 requested bytes are read.
1717 read_some: Read function that returns at least one byte, but may
1718 not return the number of bytes requested.
1719 outfile: File-like object to write output through.
1720 delta_iter: Optional DeltaChainIterator to record deltas as we
1721 read them.
1722 """
1723 super().__init__(hash_func, read_all, read_some=read_some)
1724 self.outfile = outfile
1725 self._delta_iter = delta_iter
1727 def _read(self, read: Callable[[int], bytes], size: int) -> bytes:
1728 """Read data from the read callback and write it to the file."""
1729 data = super()._read(read, size)
1730 self.outfile.write(data)
1731 return data
1733 def verify(self, progress: Callable[..., None] | None = None) -> None:
1734 """Verify a pack stream and write it to the output file.
1736 See PackStreamReader.iterobjects for a list of exceptions this may
1737 throw.
1738 """
1739 i = 0 # default count of entries if read_objects() is empty
1740 for i, unpacked in enumerate(self.read_objects()):
1741 if self._delta_iter:
1742 self._delta_iter.record(unpacked)
1743 if progress is not None:
1744 progress(f"copying pack entries: {i}/{len(self)}\r".encode("ascii"))
1745 if progress is not None:
1746 progress(f"copied {i} pack entries\n".encode("ascii"))
1749def obj_sha(
1750 type: int,
1751 chunks: bytes | Iterable[bytes],
1752 hash_func: Callable[[], "HashObject"] = sha1,
1753) -> bytes:
1754 """Compute the SHA for a numeric type and object chunks.
1756 Args:
1757 type: Object type number
1758 chunks: Object data chunks
1759 hash_func: Hash function to use (defaults to sha1)
1761 Returns:
1762 Binary hash digest
1763 """
1764 sha = hash_func()
1765 sha.update(object_header(type, chunks_length(chunks)))
1766 if isinstance(chunks, bytes):
1767 sha.update(chunks)
1768 else:
1769 for chunk in chunks:
1770 sha.update(chunk)
1771 return sha.digest()
1774def compute_file_sha(
1775 f: IO[bytes],
1776 hash_func: Callable[[], "HashObject"],
1777 start_ofs: int = 0,
1778 end_ofs: int = 0,
1779 buffer_size: int = 1 << 16,
1780) -> "HashObject":
1781 """Hash a portion of a file into a new SHA.
1783 Args:
1784 f: A file-like object to read from that supports seek().
1785 hash_func: A callable that returns a new HashObject.
1786 start_ofs: The offset in the file to start reading at.
1787 end_ofs: The offset in the file to end reading at, relative to the
1788 end of the file.
1789 buffer_size: A buffer size for reading.
1790 Returns: A new SHA object updated with data read from the file.
1791 """
1792 sha = hash_func()
1793 f.seek(0, SEEK_END)
1794 length = f.tell()
1795 if start_ofs < 0:
1796 raise AssertionError(f"start_ofs cannot be negative: {start_ofs}")
1797 if (end_ofs < 0 and length + end_ofs < start_ofs) or end_ofs > length:
1798 raise AssertionError(
1799 f"Attempt to read beyond file length. start_ofs: {start_ofs}, end_ofs: {end_ofs}, file length: {length}"
1800 )
1801 todo = length + end_ofs - start_ofs
1802 f.seek(start_ofs)
1803 while todo:
1804 data = f.read(min(todo, buffer_size))
1805 sha.update(data)
1806 todo -= len(data)
1807 return sha
1810class PackData:
1811 """The data contained in a packfile.
1813 Pack files can be accessed both sequentially for exploding a pack, and
1814 directly with the help of an index to retrieve a specific object.
1816 The objects within are either complete or a delta against another.
1818 The header is variable length. If the MSB of each byte is set then it
1819 indicates that the subsequent byte is still part of the header.
1820 For the first byte the next MS bits are the type, which tells you the type
1821 of object, and whether it is a delta. The LS byte is the lowest bits of the
1822 size. For each subsequent byte the LS 7 bits are the next MS bits of the
1823 size, i.e. the last byte of the header contains the MS bits of the size.
1825 For the complete objects the data is stored as zlib deflated data.
1826 The size in the header is the uncompressed object size, so to uncompress
1827 you need to just keep feeding data to zlib until you get an object back,
1828 or it errors on bad data. This is done here by just giving the complete
1829 buffer from the start of the deflated object on. This is bad, but until I
1830 get mmap sorted out it will have to do.
1832 Currently there are no integrity checks done. Also no attempt is made to
1833 try and detect the delta case, or a request for an object at the wrong
1834 position. It will all just throw a zlib or KeyError.
1835 """
1837 def __init__(
1838 self,
1839 filename: str | os.PathLike[str],
1840 object_format: ObjectFormat,
1841 file: IO[bytes] | None = None,
1842 size: int | None = None,
1843 *,
1844 delta_window_size: int | None = None,
1845 window_memory: int | None = None,
1846 delta_cache_size: int | None = None,
1847 depth: int | None = None,
1848 threads: int | None = None,
1849 big_file_threshold: int | None = None,
1850 delta_base_cache_limit: int | None = None,
1851 ) -> None:
1852 """Create a PackData object representing the pack in the given filename.
1854 The file must exist and stay readable until the object is disposed of.
1855 It must also stay the same size. It will be mapped whenever needed.
1857 Currently there is a restriction on the size of the pack as the python
1858 mmap implementation is flawed.
1859 """
1860 self._filename = filename
1861 self.object_format = object_format
1862 self._size = size
1863 self._header_size = 12
1864 self.delta_window_size = delta_window_size
1865 self.window_memory = window_memory
1866 self.delta_cache_size = delta_cache_size
1867 self.depth = depth
1868 self.threads = threads
1869 self.big_file_threshold = big_file_threshold
1870 self.delta_base_cache_limit = delta_base_cache_limit
1871 self._file: IO[bytes]
1873 if file is None:
1874 self._file = GitFile(self._filename, "rb")
1875 else:
1876 self._file = file
1877 (_version, self._num_objects) = read_pack_header(self._file.read)
1879 # Use delta_base_cache_limit, then delta_cache_size, then default
1880 cache_size = (
1881 delta_base_cache_limit or delta_cache_size or DEFAULT_DELTA_BASE_CACHE_LIMIT
1882 )
1883 self._offset_cache = LRUSizeCache[int, tuple[int, OldUnpackedObject]](
1884 cache_size, compute_size=_compute_object_size
1885 )
1887 @property
1888 def filename(self) -> str:
1889 """Get the filename of the pack file.
1891 Returns:
1892 Base filename without directory path
1893 """
1894 return os.path.basename(self._filename)
1896 @property
1897 def path(self) -> str | os.PathLike[str]:
1898 """Get the full path of the pack file.
1900 Returns:
1901 Full path to the pack file
1902 """
1903 return self._filename
1905 @classmethod
1906 def from_file(
1907 cls,
1908 file: IO[bytes],
1909 object_format: ObjectFormat,
1910 size: int | None = None,
1911 ) -> "PackData":
1912 """Create a PackData object from an open file.
1914 Args:
1915 file: Open file object
1916 object_format: Object format
1917 size: Optional file size
1919 Returns:
1920 PackData instance
1921 """
1922 return cls(str(file), object_format, file=file, size=size)
1924 @classmethod
1925 def from_path(
1926 cls,
1927 path: str | os.PathLike[str],
1928 object_format: ObjectFormat,
1929 ) -> "PackData":
1930 """Create a PackData object from a file path.
1932 Args:
1933 path: Path to the pack file
1934 object_format: Object format
1936 Returns:
1937 PackData instance
1938 """
1939 return cls(filename=path, object_format=object_format)
1941 def close(self) -> None:
1942 """Close the underlying pack file."""
1943 if self._file is not None:
1944 self._file.close()
1945 self._file = None # type: ignore
1947 def __del__(self) -> None:
1948 """Ensure pack file is closed when PackData is garbage collected."""
1949 if getattr(self, "_file", None) is not None:
1950 import warnings
1952 warnings.warn(
1953 f"unclosed PackData {self!r}",
1954 ResourceWarning,
1955 stacklevel=2,
1956 source=self,
1957 )
1958 try:
1959 self.close()
1960 except Exception:
1961 # Ignore errors during cleanup
1962 pass
1964 def __enter__(self) -> Self:
1965 """Enter context manager."""
1966 return self
1968 def __exit__(
1969 self,
1970 type: type | None,
1971 value: BaseException | None,
1972 traceback: TracebackType | None,
1973 ) -> None:
1974 """Exit context manager."""
1975 self.close()
1977 def __eq__(self, other: object) -> bool:
1978 """Check equality with another object."""
1979 if isinstance(other, PackData):
1980 return self.get_stored_checksum() == other.get_stored_checksum()
1981 return False
1983 def _get_size(self) -> int:
1984 if self._size is not None:
1985 return self._size
1986 self._size = os.path.getsize(self._filename)
1987 if self._size < self._header_size:
1988 errmsg = f"{self._filename} is too small for a packfile ({self._size} < {self._header_size})"
1989 raise AssertionError(errmsg)
1990 return self._size
1992 def __len__(self) -> int:
1993 """Returns the number of objects in this pack."""
1994 return self._num_objects
1996 def calculate_checksum(self) -> bytes:
1997 """Calculate the checksum for this pack.
1999 Returns: Binary digest (size depends on hash algorithm)
2000 """
2001 return compute_file_sha(
2002 self._file,
2003 hash_func=self.object_format.hash_func,
2004 end_ofs=-self.object_format.oid_length,
2005 ).digest()
2007 def iter_unpacked(self, *, include_comp: bool = False) -> Iterator[UnpackedObject]:
2008 """Iterate over unpacked objects in the pack."""
2009 self._file.seek(self._header_size)
2011 if self._num_objects is None:
2012 return
2014 for _ in range(self._num_objects):
2015 offset = self._file.tell()
2016 unpacked, unused = unpack_object(
2017 self._file.read,
2018 self.object_format.hash_func,
2019 compute_crc32=False,
2020 include_comp=include_comp,
2021 )
2022 unpacked.offset = offset
2023 yield unpacked
2024 # Back up over unused data.
2025 self._file.seek(-len(unused), SEEK_CUR)
2027 def iterentries(
2028 self,
2029 progress: Callable[[int, int], None] | None = None,
2030 resolve_ext_ref: ResolveExtRefFn | None = None,
2031 ) -> Iterator[PackIndexEntry]:
2032 """Yield entries summarizing the contents of this pack.
2034 Args:
2035 progress: Progress function, called with current and total
2036 object count.
2037 resolve_ext_ref: Optional function to resolve external references
2038 Returns: iterator of tuples with (sha, offset, crc32)
2039 """
2040 num_objects = self._num_objects
2041 indexer = PackIndexer.for_pack_data(self, resolve_ext_ref=resolve_ext_ref)
2042 for i, result in enumerate(indexer):
2043 if progress is not None:
2044 progress(i, num_objects)
2045 yield result
2047 def sorted_entries(
2048 self,
2049 progress: Callable[[int, int], None] | None = None,
2050 resolve_ext_ref: ResolveExtRefFn | None = None,
2051 ) -> list[tuple[RawObjectID, int, int]]:
2052 """Return entries in this pack, sorted by SHA.
2054 Args:
2055 progress: Progress function, called with current and total
2056 object count
2057 resolve_ext_ref: Optional function to resolve external references
2058 Returns: Iterator of tuples with (sha, offset, crc32)
2059 """
2060 return sorted(
2061 self.iterentries(progress=progress, resolve_ext_ref=resolve_ext_ref) # type: ignore
2062 )
2064 def create_index_v1(
2065 self,
2066 filename: str,
2067 progress: Callable[..., None] | None = None,
2068 resolve_ext_ref: ResolveExtRefFn | None = None,
2069 ) -> bytes:
2070 """Create a version 1 file for this data file.
2072 Args:
2073 filename: Index filename.
2074 progress: Progress report function
2075 resolve_ext_ref: Optional function to resolve external references
2076 Returns: Checksum of index file
2077 """
2078 entries = self.sorted_entries(
2079 progress=progress, resolve_ext_ref=resolve_ext_ref
2080 )
2081 checksum = self.calculate_checksum()
2082 with GitFile(filename, "wb") as f:
2083 write_pack_index_v1(
2084 f,
2085 entries,
2086 checksum,
2087 )
2088 return checksum
2090 def create_index_v2(
2091 self,
2092 filename: str,
2093 progress: Callable[..., None] | None = None,
2094 resolve_ext_ref: ResolveExtRefFn | None = None,
2095 ) -> bytes:
2096 """Create a version 2 index file for this data file.
2098 Args:
2099 filename: Index filename.
2100 progress: Progress report function
2101 resolve_ext_ref: Optional function to resolve external references
2102 Returns: Checksum of index file
2103 """
2104 entries = self.sorted_entries(
2105 progress=progress, resolve_ext_ref=resolve_ext_ref
2106 )
2107 with GitFile(filename, "wb") as f:
2108 return write_pack_index_v2(f, entries, self.calculate_checksum())
2110 def create_index_v3(
2111 self,
2112 filename: str,
2113 progress: Callable[..., None] | None = None,
2114 resolve_ext_ref: ResolveExtRefFn | None = None,
2115 hash_format: int | None = None,
2116 ) -> bytes:
2117 """Create a version 3 index file for this data file.
2119 Args:
2120 filename: Index filename.
2121 progress: Progress report function
2122 resolve_ext_ref: Function to resolve external references
2123 hash_format: Hash algorithm identifier (1 = SHA-1, 2 = SHA-256)
2124 Returns: Checksum of index file
2125 """
2126 entries = self.sorted_entries(
2127 progress=progress, resolve_ext_ref=resolve_ext_ref
2128 )
2129 with GitFile(filename, "wb") as f:
2130 if hash_format is None:
2131 hash_format = 1 # Default to SHA-1
2132 return write_pack_index_v3(
2133 f, entries, self.calculate_checksum(), hash_format=hash_format
2134 )
2136 def create_index(
2137 self,
2138 filename: str,
2139 progress: Callable[..., None] | None = None,
2140 version: int = 2,
2141 resolve_ext_ref: ResolveExtRefFn | None = None,
2142 hash_format: int | None = None,
2143 ) -> bytes:
2144 """Create an index file for this data file.
2146 Args:
2147 filename: Index filename.
2148 progress: Progress report function
2149 version: Index version (1, 2, or 3)
2150 resolve_ext_ref: Function to resolve external references
2151 hash_format: Hash algorithm identifier for v3 (1 = SHA-1, 2 = SHA-256)
2152 Returns: Checksum of index file
2153 """
2154 if version == 1:
2155 return self.create_index_v1(
2156 filename, progress, resolve_ext_ref=resolve_ext_ref
2157 )
2158 elif version == 2:
2159 return self.create_index_v2(
2160 filename, progress, resolve_ext_ref=resolve_ext_ref
2161 )
2162 elif version == 3:
2163 return self.create_index_v3(
2164 filename,
2165 progress,
2166 resolve_ext_ref=resolve_ext_ref,
2167 hash_format=hash_format,
2168 )
2169 else:
2170 raise ValueError(f"unknown index format {version}")
2172 def get_stored_checksum(self) -> bytes:
2173 """Return the expected checksum stored in this pack."""
2174 checksum_size = self.object_format.oid_length
2175 self._file.seek(-checksum_size, SEEK_END)
2176 return self._file.read(checksum_size)
2178 def check(self) -> None:
2179 """Check the consistency of this pack."""
2180 actual = self.calculate_checksum()
2181 stored = self.get_stored_checksum()
2182 if actual != stored:
2183 raise ChecksumMismatch(stored, actual)
2185 def get_unpacked_object_at(
2186 self, offset: int, *, include_comp: bool = False
2187 ) -> UnpackedObject:
2188 """Given offset in the packfile return a UnpackedObject."""
2189 assert offset >= self._header_size
2190 self._file.seek(offset)
2191 unpacked, _ = unpack_object(
2192 self._file.read, self.object_format.hash_func, include_comp=include_comp
2193 )
2194 unpacked.offset = offset
2195 return unpacked
2197 def get_object_at(self, offset: int) -> tuple[int, OldUnpackedObject]:
2198 """Given an offset in to the packfile return the object that is there.
2200 Using the associated index the location of an object can be looked up,
2201 and then the packfile can be asked directly for that object using this
2202 function.
2203 """
2204 try:
2205 return self._offset_cache[offset]
2206 except KeyError:
2207 pass
2208 unpacked = self.get_unpacked_object_at(offset, include_comp=False)
2209 return (unpacked.pack_type_num, unpacked._obj())
2212T = TypeVar("T")
2215class DeltaChainIterator(Generic[T]):
2216 """Abstract iterator over pack data based on delta chains.
2218 Each object in the pack is guaranteed to be inflated exactly once,
2219 regardless of how many objects reference it as a delta base. As a result,
2220 memory usage is proportional to the length of the longest delta chain.
2222 Subclasses can override _result to define the result type of the iterator.
2223 By default, results are UnpackedObjects with the following members set:
2225 * offset
2226 * obj_type_num
2227 * obj_chunks
2228 * pack_type_num
2229 * delta_base (for delta types)
2230 * comp_chunks (if _include_comp is True)
2231 * decomp_chunks
2232 * decomp_len
2233 * crc32 (if _compute_crc32 is True)
2234 """
2236 _compute_crc32 = False
2237 _include_comp = False
2239 def __init__(
2240 self,
2241 file_obj: IO[bytes] | None,
2242 hash_func: Callable[[], "HashObject"],
2243 *,
2244 resolve_ext_ref: ResolveExtRefFn | None = None,
2245 object_format: "ObjectFormat | None" = None,
2246 ) -> None:
2247 """Initialize DeltaChainIterator.
2249 Args:
2250 file_obj: File object to read pack data from
2251 hash_func: Hash function to use for computing object IDs
2252 resolve_ext_ref: Optional function to resolve external references
2253 object_format: Optional object format. Required by subclasses
2254 that materialise objects (e.g. PackInflater) when iterating
2255 packs in a non-default hash algorithm such as SHA-256.
2256 """
2257 self._file = file_obj
2258 self.hash_func = hash_func
2259 self._object_format = object_format
2260 self._resolve_ext_ref = resolve_ext_ref
2261 self._pending_ofs: dict[int, list[int]] = defaultdict(list)
2262 self._pending_ref: dict[bytes, list[int]] = defaultdict(list)
2263 self._full_ofs: list[tuple[int, int]] = []
2264 self._ext_refs: list[RawObjectID] = []
2266 @classmethod
2267 def for_pack_data(
2268 cls, pack_data: PackData, resolve_ext_ref: ResolveExtRefFn | None = None
2269 ) -> "DeltaChainIterator[T]":
2270 """Create a DeltaChainIterator from pack data.
2272 Args:
2273 pack_data: PackData object to iterate
2274 resolve_ext_ref: Optional function to resolve external refs
2276 Returns:
2277 DeltaChainIterator instance
2278 """
2279 walker = cls(
2280 None,
2281 pack_data.object_format.hash_func,
2282 resolve_ext_ref=resolve_ext_ref,
2283 object_format=pack_data.object_format,
2284 )
2285 walker.set_pack_data(pack_data)
2286 for unpacked in pack_data.iter_unpacked(include_comp=False):
2287 walker.record(unpacked)
2288 return walker
2290 @classmethod
2291 def for_pack_subset(
2292 cls,
2293 pack: "Pack",
2294 shas: Iterable[ObjectID | RawObjectID],
2295 *,
2296 allow_missing: bool = False,
2297 resolve_ext_ref: ResolveExtRefFn | None = None,
2298 ) -> "DeltaChainIterator[T]":
2299 """Create a DeltaChainIterator for a subset of objects.
2301 Args:
2302 pack: Pack object containing the data
2303 shas: Iterable of object SHAs to include
2304 allow_missing: If True, skip missing objects
2305 resolve_ext_ref: Optional function to resolve external refs
2307 Returns:
2308 DeltaChainIterator instance
2309 """
2310 walker = cls(
2311 None,
2312 pack.object_format.hash_func,
2313 resolve_ext_ref=resolve_ext_ref,
2314 object_format=pack.object_format,
2315 )
2316 walker.set_pack_data(pack.data)
2317 todo = set()
2318 for sha in shas:
2319 try:
2320 off = pack.index.object_offset(sha)
2321 except KeyError:
2322 if not allow_missing:
2323 raise
2324 else:
2325 todo.add(off)
2326 done = set()
2327 while todo:
2328 off = todo.pop()
2329 unpacked = pack.data.get_unpacked_object_at(off)
2330 walker.record(unpacked)
2331 done.add(off)
2332 base_ofs = None
2333 if unpacked.pack_type_num == OFS_DELTA:
2334 assert unpacked.offset is not None
2335 assert unpacked.delta_base is not None
2336 assert isinstance(unpacked.delta_base, int)
2337 base_ofs = unpacked.offset - unpacked.delta_base
2338 elif unpacked.pack_type_num == REF_DELTA:
2339 with suppress(KeyError):
2340 assert isinstance(unpacked.delta_base, bytes)
2341 base_ofs = pack.index.object_offset(
2342 RawObjectID(unpacked.delta_base)
2343 )
2344 if base_ofs is not None and base_ofs not in done:
2345 todo.add(base_ofs)
2346 return walker
2348 def record(self, unpacked: UnpackedObject) -> None:
2349 """Record an unpacked object for later processing.
2351 Args:
2352 unpacked: UnpackedObject to record
2353 """
2354 type_num = unpacked.pack_type_num
2355 offset = unpacked.offset
2356 assert offset is not None
2357 if type_num == OFS_DELTA:
2358 assert unpacked.delta_base is not None
2359 assert isinstance(unpacked.delta_base, int)
2360 base_offset = offset - unpacked.delta_base
2361 self._pending_ofs[base_offset].append(offset)
2362 elif type_num == REF_DELTA:
2363 assert isinstance(unpacked.delta_base, bytes)
2364 self._pending_ref[unpacked.delta_base].append(offset)
2365 else:
2366 self._full_ofs.append((offset, type_num))
2368 def set_pack_data(self, pack_data: PackData) -> None:
2369 """Set the pack data for iteration.
2371 Args:
2372 pack_data: PackData object to use
2373 """
2374 self._file = pack_data._file
2376 def _walk_all_chains(self) -> Iterator[T]:
2377 for offset, type_num in self._full_ofs:
2378 yield from self._follow_chain(offset, type_num, None)
2379 yield from self._walk_ref_chains()
2380 assert not self._pending_ofs, repr(self._pending_ofs)
2382 def _ensure_no_pending(self) -> None:
2383 if self._pending_ref:
2384 raise UnresolvedDeltas(
2385 [sha_to_hex(RawObjectID(s)) for s in self._pending_ref]
2386 )
2388 def _walk_ref_chains(self) -> Iterator[T]:
2389 if not self._resolve_ext_ref:
2390 self._ensure_no_pending()
2391 return
2393 for base_sha, pending in sorted(self._pending_ref.items()):
2394 if base_sha not in self._pending_ref:
2395 continue
2396 try:
2397 type_num, chunks = self._resolve_ext_ref(RawObjectID(base_sha))
2398 except KeyError:
2399 # Not an external ref, but may depend on one. Either it will
2400 # get popped via a _follow_chain call, or we will raise an
2401 # error below.
2402 continue
2403 self._ext_refs.append(RawObjectID(base_sha))
2404 self._pending_ref.pop(base_sha)
2405 for new_offset in pending:
2406 yield from self._follow_chain(new_offset, type_num, chunks)
2408 self._ensure_no_pending()
2410 def _result(self, unpacked: UnpackedObject) -> T:
2411 raise NotImplementedError
2413 def _resolve_object(
2414 self,
2415 offset: int,
2416 obj_type_num: int,
2417 base_chunks: bytes | list[bytes] | None,
2418 ) -> UnpackedObject:
2419 assert self._file is not None
2420 self._file.seek(offset)
2421 unpacked, _ = unpack_object(
2422 self._file.read,
2423 self.hash_func,
2424 read_some=None,
2425 compute_crc32=self._compute_crc32,
2426 include_comp=self._include_comp,
2427 )
2428 unpacked.offset = offset
2429 if base_chunks is None:
2430 assert unpacked.pack_type_num == obj_type_num
2431 else:
2432 assert unpacked.pack_type_num in DELTA_TYPES
2433 unpacked.obj_type_num = obj_type_num
2434 unpacked.obj_chunks = apply_delta(base_chunks, unpacked.decomp_chunks)
2435 # A delta that resolves to a zero-byte payload for a
2436 # commit/tree/tag is malformed: ``_parse_message`` /
2437 # ``parse_tree`` accept the empty input silently, so without
2438 # this guard a too-short delta could materialise an
2439 # otherwise-valid SHA pointing at an empty commit object
2440 # (which ``git fsck`` rejects). Only blobs may legitimately
2441 # be empty, and an empty blob would never be stored as a
2442 # delta in practice.
2443 # Blob.type_num == 3 (avoid the import cycle).
2444 if obj_type_num != 3 and chunks_length(unpacked.obj_chunks) == 0:
2445 raise ApplyDeltaError(
2446 f"delta resolved to empty payload for type {obj_type_num}"
2447 )
2448 return unpacked
2450 def _follow_chain(
2451 self,
2452 offset: int,
2453 obj_type_num: int,
2454 base_chunks: bytes | list[bytes] | None,
2455 ) -> Iterator[T]:
2456 # Unlike PackData.get_object_at, there is no need to cache offsets as
2457 # this approach by design inflates each object exactly once.
2458 todo = [(offset, obj_type_num, base_chunks)]
2459 while todo:
2460 (offset, obj_type_num, base_chunks) = todo.pop()
2461 unpacked = self._resolve_object(offset, obj_type_num, base_chunks)
2462 yield self._result(unpacked)
2464 assert unpacked.offset is not None
2465 unblocked = chain(
2466 self._pending_ofs.pop(unpacked.offset, []),
2467 self._pending_ref.pop(unpacked.sha(), []),
2468 )
2469 todo.extend(
2470 (new_offset, unpacked.obj_type_num, unpacked.obj_chunks) # type: ignore
2471 for new_offset in unblocked
2472 )
2474 def __iter__(self) -> Iterator[T]:
2475 """Iterate over objects in the pack."""
2476 return self._walk_all_chains()
2478 def ext_refs(self) -> list[RawObjectID]:
2479 """Return external references."""
2480 return self._ext_refs
2483class UnpackedObjectIterator(DeltaChainIterator[UnpackedObject]):
2484 """Delta chain iterator that yield unpacked objects."""
2486 def _result(self, unpacked: UnpackedObject) -> UnpackedObject:
2487 """Return the unpacked object.
2489 Args:
2490 unpacked: The unpacked object
2492 Returns:
2493 The unpacked object unchanged
2494 """
2495 return unpacked
2498class PackIndexer(DeltaChainIterator[PackIndexEntry]):
2499 """Delta chain iterator that yields index entries."""
2501 _compute_crc32 = True
2503 def _result(self, unpacked: UnpackedObject) -> PackIndexEntry:
2504 """Convert unpacked object to pack index entry.
2506 Args:
2507 unpacked: The unpacked object
2509 Returns:
2510 Tuple of (sha, offset, crc32) for index entry
2511 """
2512 assert unpacked.offset is not None
2513 return unpacked.sha(), unpacked.offset, unpacked.crc32
2516class PackInflater(DeltaChainIterator[ShaFile]):
2517 """Delta chain iterator that yields ShaFile objects."""
2519 def _result(self, unpacked: UnpackedObject) -> ShaFile:
2520 """Convert unpacked object to ShaFile.
2522 Args:
2523 unpacked: The unpacked object
2525 Returns:
2526 ShaFile object from the unpacked data
2527 """
2528 assert unpacked.obj_type_num is not None and unpacked.obj_chunks is not None
2529 return ShaFile.from_raw_chunks(
2530 unpacked.obj_type_num,
2531 unpacked.obj_chunks,
2532 object_format=self._object_format,
2533 )
2536class SHA1Reader(BinaryIO):
2537 """Wrapper for file-like object that remembers the SHA1 of its data."""
2539 def __init__(self, f: IO[bytes]) -> None:
2540 """Initialize SHA1Reader.
2542 Args:
2543 f: File-like object to wrap
2544 """
2545 self.f = f
2546 self.sha1 = sha1(b"")
2548 def read(self, size: int = -1) -> bytes:
2549 """Read bytes and update SHA1.
2551 Args:
2552 size: Number of bytes to read, -1 for all
2554 Returns:
2555 Bytes read from file
2556 """
2557 data = self.f.read(size)
2558 self.sha1.update(data)
2559 return data
2561 def check_sha(self, allow_empty: bool = False) -> None:
2562 """Check if the SHA1 matches the expected value.
2564 Args:
2565 allow_empty: Allow empty SHA1 hash
2567 Raises:
2568 ChecksumMismatch: If SHA1 doesn't match
2569 """
2570 stored = self.f.read(20)
2571 # If git option index.skipHash is set the index will be empty
2572 if stored != self.sha1.digest() and (
2573 not allow_empty
2574 or (
2575 len(stored) == 20
2576 and sha_to_hex(RawObjectID(stored))
2577 != b"0000000000000000000000000000000000000000"
2578 )
2579 ):
2580 raise ChecksumMismatch(
2581 self.sha1.hexdigest(),
2582 sha_to_hex(RawObjectID(stored)) if stored else b"",
2583 )
2585 def close(self) -> None:
2586 """Close the underlying file."""
2587 return self.f.close()
2589 def tell(self) -> int:
2590 """Return current file position."""
2591 return self.f.tell()
2593 # BinaryIO abstract methods
2594 def readable(self) -> bool:
2595 """Check if file is readable."""
2596 return True
2598 def writable(self) -> bool:
2599 """Check if file is writable."""
2600 return False
2602 def seekable(self) -> bool:
2603 """Check if file is seekable."""
2604 return getattr(self.f, "seekable", lambda: False)()
2606 def seek(self, offset: int, whence: int = 0) -> int:
2607 """Seek to position in file.
2609 Args:
2610 offset: Position offset
2611 whence: Reference point (0=start, 1=current, 2=end)
2613 Returns:
2614 New file position
2615 """
2616 return self.f.seek(offset, whence)
2618 def flush(self) -> None:
2619 """Flush the file buffer."""
2620 if hasattr(self.f, "flush"):
2621 self.f.flush()
2623 def readline(self, size: int = -1) -> bytes:
2624 """Read a line from the file.
2626 Args:
2627 size: Maximum bytes to read
2629 Returns:
2630 Line read from file
2631 """
2632 return self.f.readline(size)
2634 def readlines(self, hint: int = -1) -> list[bytes]:
2635 """Read all lines from the file.
2637 Args:
2638 hint: Approximate number of bytes to read
2640 Returns:
2641 List of lines
2642 """
2643 return self.f.readlines(hint)
2645 def writelines(self, lines: Iterable[bytes], /) -> None: # type: ignore[override]
2646 """Write multiple lines to the file (not supported)."""
2647 raise UnsupportedOperation("writelines")
2649 def write(self, data: bytes, /) -> int: # type: ignore[override]
2650 """Write data to the file (not supported)."""
2651 raise UnsupportedOperation("write")
2653 def __enter__(self) -> Self:
2654 """Enter context manager."""
2655 return self
2657 def __exit__(
2658 self,
2659 type: type | None,
2660 value: BaseException | None,
2661 traceback: TracebackType | None,
2662 ) -> None:
2663 """Exit context manager and close file."""
2664 self.close()
2666 def __iter__(self) -> "SHA1Reader":
2667 """Return iterator for reading file lines."""
2668 return self
2670 def __next__(self) -> bytes:
2671 """Get next line from file.
2673 Returns:
2674 Next line
2676 Raises:
2677 StopIteration: When no more lines
2678 """
2679 line = self.readline()
2680 if not line:
2681 raise StopIteration
2682 return line
2684 def fileno(self) -> int:
2685 """Return file descriptor number."""
2686 return self.f.fileno()
2688 def isatty(self) -> bool:
2689 """Check if file is a terminal."""
2690 return getattr(self.f, "isatty", lambda: False)()
2692 def truncate(self, size: int | None = None) -> int:
2693 """Not supported for read-only file.
2695 Raises:
2696 UnsupportedOperation: Always raised
2697 """
2698 raise UnsupportedOperation("truncate")
2701class SHA1Writer(BinaryIO):
2702 """Wrapper for file-like object that remembers the SHA1 of its data."""
2704 def __init__(self, f: BinaryIO | IO[bytes]) -> None:
2705 """Initialize SHA1Writer.
2707 Args:
2708 f: File-like object to wrap
2709 """
2710 self.f = f
2711 self.length = 0
2712 self.sha1 = sha1(b"")
2713 self.digest: bytes | None = None
2715 def write(self, data: bytes | bytearray | memoryview, /) -> int: # type: ignore[override]
2716 """Write data and update SHA1.
2718 Args:
2719 data: Data to write
2721 Returns:
2722 Number of bytes written
2723 """
2724 self.sha1.update(data)
2725 written = self.f.write(data)
2726 self.length += written
2727 return written
2729 def write_sha(self) -> bytes:
2730 """Write the SHA1 digest to the file.
2732 Returns:
2733 The SHA1 digest bytes
2734 """
2735 sha = self.sha1.digest()
2736 assert len(sha) == 20
2737 self.f.write(sha)
2738 self.length += len(sha)
2739 return sha
2741 def close(self) -> None:
2742 """Close the pack file and finalize the SHA."""
2743 self.digest = self.write_sha()
2744 self.f.close()
2746 def offset(self) -> int:
2747 """Get the total number of bytes written.
2749 Returns:
2750 Total bytes written
2751 """
2752 return self.length
2754 def tell(self) -> int:
2755 """Return current file position."""
2756 return self.f.tell()
2758 # BinaryIO abstract methods
2759 def readable(self) -> bool:
2760 """Check if file is readable."""
2761 return False
2763 def writable(self) -> bool:
2764 """Check if file is writable."""
2765 return True
2767 def seekable(self) -> bool:
2768 """Check if file is seekable."""
2769 return getattr(self.f, "seekable", lambda: False)()
2771 def seek(self, offset: int, whence: int = 0) -> int:
2772 """Seek to position in file.
2774 Args:
2775 offset: Position offset
2776 whence: Reference point (0=start, 1=current, 2=end)
2778 Returns:
2779 New file position
2780 """
2781 return self.f.seek(offset, whence)
2783 def flush(self) -> None:
2784 """Flush the file buffer."""
2785 if hasattr(self.f, "flush"):
2786 self.f.flush()
2788 def readline(self, size: int = -1) -> bytes:
2789 """Not supported for write-only file.
2791 Raises:
2792 UnsupportedOperation: Always raised
2793 """
2794 raise UnsupportedOperation("readline")
2796 def readlines(self, hint: int = -1) -> list[bytes]:
2797 """Not supported for write-only file.
2799 Raises:
2800 UnsupportedOperation: Always raised
2801 """
2802 raise UnsupportedOperation("readlines")
2804 def writelines(self, lines: Iterable[bytes], /) -> None: # type: ignore[override]
2805 """Write multiple lines to the file.
2807 Args:
2808 lines: Iterable of lines to write
2809 """
2810 for line in lines:
2811 self.write(line)
2813 def read(self, size: int = -1) -> bytes:
2814 """Not supported for write-only file.
2816 Raises:
2817 UnsupportedOperation: Always raised
2818 """
2819 raise UnsupportedOperation("read")
2821 def __enter__(self) -> Self:
2822 """Enter context manager."""
2823 return self
2825 def __exit__(
2826 self,
2827 type: type | None,
2828 value: BaseException | None,
2829 traceback: TracebackType | None,
2830 ) -> None:
2831 """Exit context manager and close file."""
2832 self.f.close()
2834 def __iter__(self) -> "SHA1Writer":
2835 """Return iterator."""
2836 return self
2838 def __next__(self) -> bytes:
2839 """Not supported for write-only file.
2841 Raises:
2842 UnsupportedOperation: Always raised
2843 """
2844 raise UnsupportedOperation("__next__")
2846 def fileno(self) -> int:
2847 """Return file descriptor number."""
2848 return self.f.fileno()
2850 def isatty(self) -> bool:
2851 """Check if file is a terminal."""
2852 return getattr(self.f, "isatty", lambda: False)()
2854 def truncate(self, size: int | None = None) -> int:
2855 """Not supported for write-only file.
2857 Raises:
2858 UnsupportedOperation: Always raised
2859 """
2860 raise UnsupportedOperation("truncate")
2863class HashWriter(BinaryIO):
2864 """Wrapper for file-like object that computes hash of its data.
2866 This is a generic version that works with any hash algorithm.
2867 """
2869 def __init__(
2870 self, f: BinaryIO | IO[bytes], hash_func: Callable[[], "HashObject"]
2871 ) -> None:
2872 """Initialize HashWriter.
2874 Args:
2875 f: File-like object to wrap
2876 hash_func: Hash function (e.g., sha1, sha256)
2877 """
2878 self.f = f
2879 self.length = 0
2880 self.hash_obj = hash_func()
2881 self.digest: bytes | None = None
2883 def write(self, data: bytes | bytearray | memoryview, /) -> int: # type: ignore[override]
2884 """Write data and update hash.
2886 Args:
2887 data: Data to write
2889 Returns:
2890 Number of bytes written
2891 """
2892 self.hash_obj.update(data)
2893 written = self.f.write(data)
2894 self.length += written
2895 return written
2897 def write_hash(self) -> bytes:
2898 """Write the hash digest to the file.
2900 Returns:
2901 The hash digest bytes
2902 """
2903 digest = self.hash_obj.digest()
2904 self.f.write(digest)
2905 self.length += len(digest)
2906 return digest
2908 def close(self) -> None:
2909 """Close the pack file and finalize the hash."""
2910 self.digest = self.write_hash()
2911 self.f.close()
2913 def offset(self) -> int:
2914 """Get the total number of bytes written.
2916 Returns:
2917 Total bytes written
2918 """
2919 return self.length
2921 def tell(self) -> int:
2922 """Return current file position."""
2923 return self.f.tell()
2925 # BinaryIO abstract methods
2926 def readable(self) -> bool:
2927 """Check if file is readable."""
2928 return False
2930 def writable(self) -> bool:
2931 """Check if file is writable."""
2932 return True
2934 def seekable(self) -> bool:
2935 """Check if file is seekable."""
2936 return getattr(self.f, "seekable", lambda: False)()
2938 def seek(self, offset: int, whence: int = 0) -> int:
2939 """Seek to position in file.
2941 Args:
2942 offset: Position offset
2943 whence: Reference point (0=start, 1=current, 2=end)
2945 Returns:
2946 New file position
2947 """
2948 return self.f.seek(offset, whence)
2950 def flush(self) -> None:
2951 """Flush the file buffer."""
2952 if hasattr(self.f, "flush"):
2953 self.f.flush()
2955 def readline(self, size: int = -1) -> bytes:
2956 """Not supported for write-only file.
2958 Raises:
2959 UnsupportedOperation: Always raised
2960 """
2961 raise UnsupportedOperation("readline")
2963 def readlines(self, hint: int = -1) -> list[bytes]:
2964 """Not supported for write-only file.
2966 Raises:
2967 UnsupportedOperation: Always raised
2968 """
2969 raise UnsupportedOperation("readlines")
2971 def writelines(self, lines: Iterable[bytes], /) -> None: # type: ignore[override]
2972 """Write multiple lines to the file.
2974 Args:
2975 lines: Iterable of lines to write
2976 """
2977 for line in lines:
2978 self.write(line)
2980 def read(self, size: int = -1) -> bytes:
2981 """Not supported for write-only file.
2983 Raises:
2984 UnsupportedOperation: Always raised
2985 """
2986 raise UnsupportedOperation("read")
2988 def __enter__(self) -> Self:
2989 """Enter context manager."""
2990 return self
2992 def __exit__(
2993 self,
2994 type: type | None,
2995 value: BaseException | None,
2996 traceback: TracebackType | None,
2997 ) -> None:
2998 """Exit context manager and close file."""
2999 self.close()
3001 def __iter__(self) -> "HashWriter":
3002 """Return iterator."""
3003 return self
3005 def __next__(self) -> bytes:
3006 """Not supported for write-only file.
3008 Raises:
3009 UnsupportedOperation: Always raised
3010 """
3011 raise UnsupportedOperation("__next__")
3013 def fileno(self) -> int:
3014 """Return file descriptor number."""
3015 return self.f.fileno()
3017 def isatty(self) -> bool:
3018 """Check if file is a terminal."""
3019 return getattr(self.f, "isatty", lambda: False)()
3021 def truncate(self, size: int | None = None) -> int:
3022 """Not supported for write-only file.
3024 Raises:
3025 UnsupportedOperation: Always raised
3026 """
3027 raise UnsupportedOperation("truncate")
3030def pack_object_header(
3031 type_num: int,
3032 delta_base: bytes | int | None,
3033 size: int,
3034 object_format: "ObjectFormat",
3035) -> bytearray:
3036 """Create a pack object header for the given object info.
3038 Args:
3039 type_num: Numeric type of the object.
3040 delta_base: Delta base offset or ref, or None for whole objects.
3041 size: Uncompressed object size.
3042 object_format: Object format (hash algorithm) to use.
3043 Returns: A header for a packed object.
3044 """
3045 header = []
3046 c = (type_num << 4) | (size & 15)
3047 size >>= 4
3048 while size:
3049 header.append(c | 0x80)
3050 c = size & 0x7F
3051 size >>= 7
3052 header.append(c)
3053 if type_num == OFS_DELTA:
3054 assert isinstance(delta_base, int)
3055 ret = [delta_base & 0x7F]
3056 delta_base >>= 7
3057 while delta_base:
3058 delta_base -= 1
3059 ret.insert(0, 0x80 | (delta_base & 0x7F))
3060 delta_base >>= 7
3061 header.extend(ret)
3062 elif type_num == REF_DELTA:
3063 assert isinstance(delta_base, bytes)
3064 assert len(delta_base) == object_format.oid_length
3065 header += delta_base
3066 return bytearray(header)
3069def pack_object_chunks(
3070 type: int,
3071 object: list[bytes] | tuple[bytes | int, list[bytes]],
3072 object_format: "ObjectFormat",
3073 *,
3074 compression_level: int = -1,
3075) -> Iterator[bytes]:
3076 """Generate chunks for a pack object.
3078 Args:
3079 type: Numeric type of the object
3080 object: Object to write
3081 object_format: Object format (hash algorithm) to use
3082 compression_level: the zlib compression level
3083 Returns: Chunks
3084 """
3085 if type in DELTA_TYPES:
3086 if isinstance(object, tuple):
3087 delta_base, object = object
3088 else:
3089 raise TypeError("Delta types require a tuple of (delta_base, object)")
3090 else:
3091 delta_base = None
3093 # Convert object to list of bytes chunks
3094 if isinstance(object, bytes):
3095 chunks = [object]
3096 elif isinstance(object, list):
3097 chunks = object
3098 elif isinstance(object, ShaFile):
3099 chunks = object.as_raw_chunks()
3100 else:
3101 # Shouldn't reach here with proper typing
3102 raise TypeError(f"Unexpected object type: {object.__class__.__name__}")
3104 yield bytes(
3105 pack_object_header(
3106 type, delta_base, sum(map(len, chunks)), object_format=object_format
3107 )
3108 )
3109 compressor = zlib.compressobj(level=compression_level)
3110 for data in chunks:
3111 yield compressor.compress(data)
3112 yield compressor.flush()
3115def write_pack_object(
3116 write: Callable[[bytes], int],
3117 type: int,
3118 object: list[bytes] | tuple[bytes | int, list[bytes]],
3119 object_format: "ObjectFormat",
3120 *,
3121 sha: "HashObject | None" = None,
3122 compression_level: int = -1,
3123) -> int:
3124 """Write pack object to a file.
3126 Args:
3127 write: Write function to use
3128 type: Numeric type of the object
3129 object: Object to write
3130 object_format: Object format (hash algorithm) to use
3131 sha: Optional SHA-1 hasher to update
3132 compression_level: the zlib compression level
3133 Returns: CRC32 checksum of the written object
3134 """
3135 crc32 = 0
3136 for chunk in pack_object_chunks(
3137 type, object, compression_level=compression_level, object_format=object_format
3138 ):
3139 write(chunk)
3140 if sha is not None:
3141 sha.update(chunk)
3142 crc32 = binascii.crc32(chunk, crc32)
3143 return crc32 & 0xFFFFFFFF
3146def write_pack(
3147 filename: str,
3148 objects: Sequence[ShaFile] | Sequence[tuple[ShaFile, bytes | None]],
3149 object_format: "ObjectFormat",
3150 *,
3151 deltify: bool | None = None,
3152 delta_window_size: int | None = None,
3153 compression_level: int = -1,
3154) -> tuple[bytes, bytes]:
3155 """Write a new pack data file.
3157 Args:
3158 filename: Path to the new pack file (without .pack extension)
3159 objects: Objects to write to the pack
3160 object_format: Object format
3161 delta_window_size: Delta window size
3162 deltify: Whether to deltify pack objects
3163 compression_level: the zlib compression level
3164 Returns: Tuple with checksum of pack file and index file
3165 """
3166 with GitFile(filename + ".pack", "wb") as f:
3167 entries, data_sum = write_pack_objects(
3168 f,
3169 objects,
3170 delta_window_size=delta_window_size,
3171 deltify=deltify,
3172 compression_level=compression_level,
3173 object_format=object_format,
3174 )
3175 entries_list = sorted([(k, v[0], v[1]) for (k, v) in entries.items()])
3176 with GitFile(filename + ".idx", "wb") as f:
3177 idx_sha = write_pack_index(f, entries_list, data_sum)
3178 return data_sum, idx_sha
3181def pack_header_chunks(num_objects: int) -> Iterator[bytes]:
3182 """Yield chunks for a pack header."""
3183 yield b"PACK" # Pack header
3184 yield struct.pack(b">L", 2) # Pack version
3185 yield struct.pack(b">L", num_objects) # Number of objects in pack
3188def write_pack_header(
3189 write: Callable[[bytes], int] | IO[bytes], num_objects: int
3190) -> None:
3191 """Write a pack header for the given number of objects."""
3192 if not callable(write):
3193 write_fn: Callable[[bytes], int] = write.write
3194 warnings.warn(
3195 "write_pack_header() now takes a write rather than file argument",
3196 DeprecationWarning,
3197 stacklevel=2,
3198 )
3199 else:
3200 write_fn = write
3201 for chunk in pack_header_chunks(num_objects):
3202 write_fn(chunk)
3205def find_reusable_deltas(
3206 container: PackedObjectContainer,
3207 object_ids: Set[ObjectID],
3208 *,
3209 other_haves: Set[ObjectID] | None = None,
3210 progress: Callable[..., None] | None = None,
3211) -> Iterator[UnpackedObject]:
3212 """Find deltas in a pack that can be reused.
3214 Args:
3215 container: Pack container to search for deltas
3216 object_ids: Set of object IDs to find deltas for
3217 other_haves: Set of other object IDs we have
3218 progress: Optional progress reporting callback
3220 Returns:
3221 Iterator of UnpackedObject entries that can be reused
3222 """
3223 if other_haves is None:
3224 other_haves = set()
3225 reused = 0
3226 for i, unpacked in enumerate(
3227 container.iter_unpacked_subset(
3228 object_ids, allow_missing=True, convert_ofs_delta=True
3229 )
3230 ):
3231 if progress is not None and i % 1000 == 0:
3232 progress(f"checking for reusable deltas: {i}/{len(object_ids)}\r".encode())
3233 if unpacked.pack_type_num == REF_DELTA:
3234 hexsha = sha_to_hex(unpacked.delta_base) # type: ignore
3235 if hexsha in object_ids or hexsha in other_haves:
3236 yield unpacked
3237 reused += 1
3238 if progress is not None:
3239 progress((f"found {reused} deltas to reuse\n").encode())
3242def deltify_pack_objects(
3243 objects: Iterator[ShaFile] | Iterator[tuple[ShaFile, bytes | None]],
3244 *,
3245 window_size: int | None = None,
3246 progress: Callable[..., None] | None = None,
3247) -> Iterator[UnpackedObject]:
3248 """Generate deltas for pack objects.
3250 Args:
3251 objects: An iterable of (object, path) tuples to deltify.
3252 window_size: Window size; None for default
3253 progress: Optional progress reporting callback
3254 Returns: Iterator over type_num, object id, delta_base, content
3255 delta_base is None for full text entries
3256 """
3258 def objects_with_hints() -> Iterator[tuple[ShaFile, tuple[int, bytes | None]]]:
3259 for e in objects:
3260 if isinstance(e, ShaFile):
3261 yield (e, (e.type_num, None))
3262 else:
3263 yield (e[0], (e[0].type_num, e[1]))
3265 sorted_objs = sort_objects_for_delta(objects_with_hints())
3266 yield from deltas_from_sorted_objects(
3267 sorted_objs,
3268 window_size=window_size,
3269 progress=progress,
3270 )
3273def sort_objects_for_delta(
3274 objects: Iterator[ShaFile] | Iterator[tuple[ShaFile, PackHint | None]],
3275) -> Iterator[tuple[ShaFile, bytes | None]]:
3276 """Sort objects for optimal delta compression.
3278 Args:
3279 objects: Iterator of objects or (object, hint) tuples
3281 Returns:
3282 Iterator of sorted (ShaFile, path) tuples
3283 """
3284 magic = []
3285 for entry in objects:
3286 if isinstance(entry, tuple):
3287 obj, hint = entry
3288 if hint is None:
3289 type_num = None
3290 path = None
3291 else:
3292 (type_num, path) = hint
3293 else:
3294 obj = entry
3295 type_num = None
3296 path = None
3297 magic.append((type_num, path, -obj.raw_length(), obj))
3298 # Build a list of objects ordered by the magic Linus heuristic
3299 # This helps us find good objects to diff against us
3300 magic.sort()
3301 return ((x[3], x[1]) for x in magic)
3304def deltas_from_sorted_objects(
3305 objects: Iterator[tuple[ShaFile, bytes | None]],
3306 window_size: int | None = None,
3307 progress: Callable[..., None] | None = None,
3308) -> Iterator[UnpackedObject]:
3309 """Create deltas from sorted objects.
3311 Args:
3312 objects: Iterator of sorted objects to deltify
3313 window_size: Delta window size; None for default
3314 progress: Optional progress reporting callback
3316 Returns:
3317 Iterator of UnpackedObject entries
3318 """
3319 # TODO(jelmer): Use threads
3320 if window_size is None:
3321 window_size = DEFAULT_PACK_DELTA_WINDOW_SIZE
3323 possible_bases: deque[tuple[bytes, int, bytes]] = deque()
3324 for i, (o, path) in enumerate(objects):
3325 if progress is not None and i % 1000 == 0:
3326 progress((f"generating deltas: {i}\r").encode())
3327 raw = o.as_raw_chunks()
3328 raw_bytes = b"".join(raw) # Join once for efficiency
3329 winner = raw
3330 winner_len = sum(map(len, winner))
3331 winner_base = None
3332 for base_id, base_type_num, base_bytes in possible_bases:
3333 if base_type_num != o.type_num:
3334 continue
3335 delta_len = 0
3336 delta = []
3337 for chunk in create_delta(base_bytes, raw_bytes):
3338 delta_len += len(chunk)
3339 if delta_len >= winner_len:
3340 break
3341 delta.append(chunk)
3342 else:
3343 winner_base = base_id
3344 winner = delta
3345 winner_len = sum(map(len, winner))
3346 yield UnpackedObject(
3347 o.type_num,
3348 sha=o.sha().digest(),
3349 delta_base=winner_base,
3350 decomp_len=winner_len,
3351 decomp_chunks=winner,
3352 )
3353 possible_bases.appendleft((o.sha().digest(), o.type_num, raw_bytes))
3354 while len(possible_bases) > window_size:
3355 possible_bases.pop()
3358def pack_objects_to_data(
3359 objects: Sequence[ShaFile]
3360 | Sequence[tuple[ShaFile, bytes | None]]
3361 | Sequence[tuple[ShaFile, PackHint | None]],
3362 *,
3363 deltify: bool | None = None,
3364 delta_window_size: int | None = None,
3365 ofs_delta: bool = True,
3366 progress: Callable[..., None] | None = None,
3367) -> tuple[int, Iterator[UnpackedObject]]:
3368 """Create pack data from objects.
3370 Args:
3371 objects: Pack objects
3372 deltify: Whether to deltify pack objects
3373 delta_window_size: Delta window size
3374 ofs_delta: Whether to use offset deltas
3375 progress: Optional progress reporting callback
3376 Returns: Tuples with (type_num, hexdigest, delta base, object chunks)
3377 """
3378 count = len(objects)
3379 if deltify is None:
3380 # PERFORMANCE/TODO(jelmer): This should be enabled but the python
3381 # implementation is *much* too slow at the moment.
3382 # Maybe consider enabling it just if the rust extension is available?
3383 deltify = False
3384 if deltify:
3385 return (
3386 count,
3387 deltify_pack_objects(
3388 iter(objects), # type: ignore
3389 window_size=delta_window_size,
3390 progress=progress,
3391 ),
3392 )
3393 else:
3395 def iter_without_path() -> Iterator[UnpackedObject]:
3396 for o in objects:
3397 if isinstance(o, tuple):
3398 yield full_unpacked_object(o[0])
3399 else:
3400 yield full_unpacked_object(o)
3402 return (count, iter_without_path())
3405def generate_unpacked_objects(
3406 container: PackedObjectContainer,
3407 object_ids: Sequence[tuple[ObjectID, PackHint | None]],
3408 delta_window_size: int | None = None,
3409 deltify: bool | None = None,
3410 reuse_deltas: bool = True,
3411 ofs_delta: bool = True,
3412 other_haves: set[ObjectID] | None = None,
3413 progress: Callable[..., None] | None = None,
3414) -> Iterator[UnpackedObject]:
3415 """Create pack data from objects.
3417 Returns: Tuples with (type_num, hexdigest, delta base, object chunks)
3418 """
3419 todo = dict(object_ids)
3420 if reuse_deltas:
3421 for unpack in find_reusable_deltas(
3422 container, set(todo), other_haves=other_haves, progress=progress
3423 ):
3424 del todo[sha_to_hex(RawObjectID(unpack.sha()))]
3425 yield unpack
3426 if deltify is None:
3427 # PERFORMANCE/TODO(jelmer): This should be enabled but is *much* too
3428 # slow at the moment.
3429 deltify = False
3430 if deltify:
3431 objects_to_delta = container.iterobjects_subset(
3432 todo.keys(), allow_missing=False
3433 )
3434 sorted_objs = sort_objects_for_delta((o, todo[o.id]) for o in objects_to_delta)
3435 yield from deltas_from_sorted_objects(
3436 sorted_objs,
3437 window_size=delta_window_size,
3438 progress=progress,
3439 )
3440 else:
3441 for oid in todo:
3442 yield full_unpacked_object(container[oid])
3445def full_unpacked_object(o: ShaFile) -> UnpackedObject:
3446 """Create an UnpackedObject from a ShaFile.
3448 Args:
3449 o: ShaFile object to convert
3451 Returns:
3452 UnpackedObject with full object data
3453 """
3454 return UnpackedObject(
3455 o.type_num,
3456 delta_base=None,
3457 crc32=None,
3458 decomp_chunks=o.as_raw_chunks(),
3459 sha=o.sha().digest(),
3460 )
3463def write_pack_from_container(
3464 write: Callable[[bytes], None]
3465 | Callable[[bytes | bytearray | memoryview], int]
3466 | IO[bytes],
3467 container: PackedObjectContainer,
3468 object_ids: Sequence[tuple[ObjectID, PackHint | None]],
3469 object_format: "ObjectFormat",
3470 *,
3471 delta_window_size: int | None = None,
3472 deltify: bool | None = None,
3473 reuse_deltas: bool = True,
3474 compression_level: int = -1,
3475 other_haves: set[ObjectID] | None = None,
3476) -> tuple[dict[bytes, tuple[int, int]], bytes]:
3477 """Write a new pack data file.
3479 Args:
3480 write: write function to use
3481 container: PackedObjectContainer
3482 object_ids: Sequence of (object_id, hint) tuples to write
3483 object_format: Object format (hash algorithm) to use
3484 delta_window_size: Sliding window size for searching for deltas;
3485 Set to None for default window size.
3486 deltify: Whether to deltify objects
3487 reuse_deltas: Whether to reuse existing deltas
3488 compression_level: the zlib compression level to use
3489 other_haves: Set of additional object IDs the receiver has
3490 Returns: Dict mapping id -> (offset, crc32 checksum), pack checksum
3491 """
3492 pack_contents_count = len(object_ids)
3493 pack_contents = generate_unpacked_objects(
3494 container,
3495 object_ids,
3496 delta_window_size=delta_window_size,
3497 deltify=deltify,
3498 reuse_deltas=reuse_deltas,
3499 other_haves=other_haves,
3500 )
3502 return write_pack_data(
3503 write,
3504 pack_contents,
3505 num_records=pack_contents_count,
3506 compression_level=compression_level,
3507 object_format=object_format,
3508 )
3511def write_pack_objects(
3512 write: Callable[[bytes], None] | IO[bytes],
3513 objects: Sequence[ShaFile] | Sequence[tuple[ShaFile, bytes | None]],
3514 object_format: "ObjectFormat",
3515 *,
3516 delta_window_size: int | None = None,
3517 deltify: bool | None = None,
3518 compression_level: int = -1,
3519) -> tuple[dict[bytes, tuple[int, int]], bytes]:
3520 """Write a new pack data file.
3522 Args:
3523 write: write function to use
3524 objects: Sequence of (object, path) tuples to write
3525 object_format: Object format (hash algorithm) to use
3526 delta_window_size: Sliding window size for searching for deltas;
3527 Set to None for default window size.
3528 deltify: Whether to deltify objects
3529 compression_level: the zlib compression level to use
3530 Returns: Dict mapping id -> (offset, crc32 checksum), pack checksum
3531 """
3532 pack_contents_count, pack_contents = pack_objects_to_data(objects, deltify=deltify)
3534 return write_pack_data(
3535 write,
3536 pack_contents,
3537 num_records=pack_contents_count,
3538 compression_level=compression_level,
3539 object_format=object_format,
3540 )
3543class PackChunkGenerator:
3544 """Generator for pack data chunks."""
3546 def __init__(
3547 self,
3548 object_format: "ObjectFormat",
3549 num_records: int | None = None,
3550 records: Iterator[UnpackedObject] | None = None,
3551 progress: Callable[..., None] | None = None,
3552 compression_level: int = -1,
3553 reuse_compressed: bool = True,
3554 ) -> None:
3555 """Initialize PackChunkGenerator.
3557 Args:
3558 num_records: Expected number of records
3559 records: Iterator of pack records
3560 progress: Optional progress callback
3561 compression_level: Compression level (-1 for default)
3562 reuse_compressed: Whether to reuse compressed chunks
3563 object_format: Object format (hash algorithm) to use
3564 """
3565 self.object_format = object_format
3566 self.cs = object_format.new_hash()
3567 self.entries: dict[bytes, tuple[int, int]] = {}
3568 if records is None:
3569 records = iter([]) # Empty iterator if None
3570 self._it = self._pack_data_chunks(
3571 records=records,
3572 num_records=num_records,
3573 progress=progress,
3574 compression_level=compression_level,
3575 reuse_compressed=reuse_compressed,
3576 )
3578 def sha1digest(self) -> bytes:
3579 """Return the SHA1 digest of the pack data."""
3580 return self.cs.digest()
3582 def __iter__(self) -> Iterator[bytes]:
3583 """Iterate over pack data chunks."""
3584 return self._it
3586 def _pack_data_chunks(
3587 self,
3588 records: Iterator[UnpackedObject],
3589 *,
3590 num_records: int | None = None,
3591 progress: Callable[..., None] | None = None,
3592 compression_level: int = -1,
3593 reuse_compressed: bool = True,
3594 ) -> Iterator[bytes]:
3595 """Iterate pack data file chunks.
3597 Args:
3598 records: Iterator over UnpackedObject
3599 num_records: Number of records (defaults to len(records) if not specified)
3600 progress: Function to report progress to
3601 compression_level: the zlib compression level
3602 reuse_compressed: Whether to reuse compressed chunks
3603 Returns: Dict mapping id -> (offset, crc32 checksum), pack checksum
3604 """
3605 # Write the pack
3606 if num_records is None:
3607 num_records = len(records) # type: ignore
3608 offset = 0
3609 for chunk in pack_header_chunks(num_records):
3610 yield chunk
3611 self.cs.update(chunk)
3612 offset += len(chunk)
3613 actual_num_records = 0
3614 for i, unpacked in enumerate(records):
3615 type_num = unpacked.pack_type_num
3616 if progress is not None and i % 1000 == 0:
3617 progress((f"writing pack data: {i}/{num_records}\r").encode("ascii"))
3618 raw: list[bytes] | tuple[int, list[bytes]] | tuple[bytes, list[bytes]]
3619 if unpacked.delta_base is not None:
3620 assert isinstance(unpacked.delta_base, bytes), (
3621 f"Expected bytes, got {type(unpacked.delta_base)}"
3622 )
3623 try:
3624 base_offset, _base_crc32 = self.entries[unpacked.delta_base]
3625 except KeyError:
3626 type_num = REF_DELTA
3627 assert isinstance(unpacked.delta_base, bytes)
3628 raw = (unpacked.delta_base, unpacked.decomp_chunks)
3629 else:
3630 type_num = OFS_DELTA
3631 raw = (offset - base_offset, unpacked.decomp_chunks)
3632 else:
3633 raw = unpacked.decomp_chunks
3634 chunks: list[bytes] | Iterator[bytes]
3635 if unpacked.comp_chunks is not None and reuse_compressed:
3636 chunks = unpacked.comp_chunks
3637 else:
3638 chunks = pack_object_chunks(
3639 type_num,
3640 raw,
3641 compression_level=compression_level,
3642 object_format=self.object_format,
3643 )
3644 crc32 = 0
3645 object_size = 0
3646 for chunk in chunks:
3647 yield chunk
3648 crc32 = binascii.crc32(chunk, crc32)
3649 self.cs.update(chunk)
3650 object_size += len(chunk)
3651 actual_num_records += 1
3652 self.entries[unpacked.sha()] = (offset, crc32)
3653 offset += object_size
3654 if actual_num_records != num_records:
3655 raise AssertionError(
3656 f"actual records written differs: {actual_num_records} != {num_records}"
3657 )
3659 yield self.cs.digest()
3662def write_pack_data(
3663 write: Callable[[bytes], None]
3664 | Callable[[bytes | bytearray | memoryview], int]
3665 | IO[bytes],
3666 records: Iterator[UnpackedObject],
3667 object_format: "ObjectFormat",
3668 *,
3669 num_records: int | None = None,
3670 progress: Callable[..., None] | None = None,
3671 compression_level: int = -1,
3672) -> tuple[dict[bytes, tuple[int, int]], bytes]:
3673 """Write a new pack data file.
3675 Args:
3676 write: Write function to use
3677 num_records: Number of records (defaults to len(records) if None)
3678 records: Iterator over type_num, object_id, delta_base, raw
3679 object_format: Object format (hash algorithm) to use
3680 progress: Function to report progress to
3681 compression_level: the zlib compression level
3682 Returns: Dict mapping id -> (offset, crc32 checksum), pack checksum
3683 """
3684 chunk_generator = PackChunkGenerator(
3685 num_records=num_records,
3686 records=records,
3687 progress=progress,
3688 compression_level=compression_level,
3689 object_format=object_format,
3690 )
3691 for chunk in chunk_generator:
3692 if callable(write):
3693 write(chunk)
3694 else:
3695 write.write(chunk)
3696 return chunk_generator.entries, chunk_generator.sha1digest()
3699def write_pack_index_v1(
3700 f: IO[bytes],
3701 entries: Iterable[tuple[bytes, int, int | None]],
3702 pack_checksum: bytes,
3703) -> bytes:
3704 """Write a new pack index file.
3706 Args:
3707 f: A file-like object to write to
3708 entries: List of tuples with object name (sha), offset_in_pack,
3709 and crc32_checksum.
3710 pack_checksum: Checksum of the pack file.
3711 Returns: The SHA of the written index file
3712 """
3713 f = SHA1Writer(f)
3714 fan_out_table: dict[int, int] = defaultdict(lambda: 0)
3715 for name, _offset, _entry_checksum in entries:
3716 fan_out_table[ord(name[:1])] += 1
3717 # Fan-out table
3718 for i in range(0x100):
3719 f.write(struct.pack(">L", fan_out_table[i]))
3720 fan_out_table[i + 1] += fan_out_table[i]
3721 for name, offset, _entry_checksum in entries:
3722 if len(name) != 20:
3723 raise TypeError("pack index v1 only supports SHA-1 names")
3724 if not (offset <= 0xFFFFFFFF):
3725 raise TypeError("pack format 1 only supports offsets < 2Gb")
3726 f.write(struct.pack(">L20s", offset, name))
3727 assert len(pack_checksum) == 20
3728 f.write(pack_checksum)
3729 return f.write_sha()
3732def _delta_encode_size(size: int) -> bytes:
3733 ret = bytearray()
3734 c = size & 0x7F
3735 size >>= 7
3736 while size:
3737 ret.append(c | 0x80)
3738 c = size & 0x7F
3739 size >>= 7
3740 ret.append(c)
3741 return bytes(ret)
3744# The length of delta compression copy operations in version 2 packs is limited
3745# to 64K. To copy more, we use several copy operations. Version 3 packs allow
3746# 24-bit lengths in copy operations, but we always make version 2 packs.
3747_MAX_COPY_LEN = 0xFFFF
3750def _encode_copy_operation(start: int, length: int) -> bytes:
3751 scratch = bytearray([0x80])
3752 for i in range(4):
3753 if start & 0xFF << i * 8:
3754 scratch.append((start >> i * 8) & 0xFF)
3755 scratch[0] |= 1 << i
3756 for i in range(2):
3757 if length & 0xFF << i * 8:
3758 scratch.append((length >> i * 8) & 0xFF)
3759 scratch[0] |= 1 << (4 + i)
3760 return bytes(scratch)
3763def _create_delta_py(
3764 base_buf: bytes | list[bytes], target_buf: bytes | list[bytes]
3765) -> Iterator[bytes]:
3766 """Use python difflib to work out how to transform base_buf to target_buf.
3768 Args:
3769 base_buf: Base buffer
3770 target_buf: Target buffer
3771 """
3772 if isinstance(base_buf, list):
3773 base_buf = b"".join(base_buf)
3774 if isinstance(target_buf, list):
3775 target_buf = b"".join(target_buf)
3776 # write delta header
3777 yield _delta_encode_size(len(base_buf))
3778 yield _delta_encode_size(len(target_buf))
3779 # write out delta opcodes
3780 seq = SequenceMatcher(isjunk=None, a=base_buf, b=target_buf)
3781 for opcode, i1, i2, j1, j2 in seq.get_opcodes():
3782 # Git patch opcodes don't care about deletes!
3783 # if opcode == 'replace' or opcode == 'delete':
3784 # pass
3785 if opcode == "equal":
3786 # If they are equal, unpacker will use data from base_buf
3787 # Write out an opcode that says what range to use
3788 copy_start = i1
3789 copy_len = i2 - i1
3790 while copy_len > 0:
3791 to_copy = min(copy_len, _MAX_COPY_LEN)
3792 yield _encode_copy_operation(copy_start, to_copy)
3793 copy_start += to_copy
3794 copy_len -= to_copy
3795 if opcode == "replace" or opcode == "insert":
3796 # If we are replacing a range or adding one, then we just
3797 # output it to the stream (prefixed by its size)
3798 s = j2 - j1
3799 o = j1
3800 while s > 127:
3801 yield bytes([127])
3802 yield bytes(memoryview(target_buf)[o : o + 127])
3803 s -= 127
3804 o += 127
3805 yield bytes([s])
3806 yield bytes(memoryview(target_buf)[o : o + s])
3809# Default to pure Python implementation
3810create_delta = _create_delta_py
3813def apply_delta(
3814 src_buf: bytes | list[bytes], delta: bytes | list[bytes]
3815) -> list[bytes]:
3816 """Based on the similar function in git's patch-delta.c.
3818 Args:
3819 src_buf: Source buffer
3820 delta: Delta instructions
3821 """
3822 if not isinstance(src_buf, bytes):
3823 src_buf = b"".join(src_buf)
3824 if not isinstance(delta, bytes):
3825 delta = b"".join(delta)
3826 out = []
3827 index = 0
3828 delta_length = len(delta)
3830 def get_delta_header_size(delta: bytes, index: int) -> tuple[int, int]:
3831 size = 0
3832 i = 0
3833 while True:
3834 # Bound-check explicitly: ``delta[index:index+1]`` silently
3835 # returns b"" past the end, which would crash with TypeError
3836 # in ``ord`` and leave the caller unable to distinguish a
3837 # truncated delta from a programming bug.
3838 if index >= delta_length:
3839 raise ApplyDeltaError("delta truncated in size header")
3840 cmd = ord(delta[index : index + 1])
3841 index += 1
3842 size |= (cmd & ~0x80) << i
3843 i += 7
3844 if not cmd & 0x80:
3845 break
3846 return size, index
3848 src_size, index = get_delta_header_size(delta, index)
3849 dest_size, index = get_delta_header_size(delta, index)
3850 if src_size != len(src_buf):
3851 raise ApplyDeltaError(
3852 f"Unexpected source buffer size: {src_size} vs {len(src_buf)}"
3853 )
3854 while index < delta_length:
3855 cmd = ord(delta[index : index + 1])
3856 index += 1
3857 if cmd & 0x80:
3858 cp_off = 0
3859 for i in range(4):
3860 if cmd & (1 << i):
3861 x = ord(delta[index : index + 1])
3862 index += 1
3863 cp_off |= x << (i * 8)
3864 cp_size = 0
3865 # Version 3 packs can contain copy sizes larger than 64K.
3866 for i in range(3):
3867 if cmd & (1 << (4 + i)):
3868 x = ord(delta[index : index + 1])
3869 index += 1
3870 cp_size |= x << (i * 8)
3871 if cp_size == 0:
3872 cp_size = 0x10000
3873 if (
3874 cp_off + cp_size < cp_size
3875 or cp_off + cp_size > src_size
3876 or cp_size > dest_size
3877 ):
3878 break
3879 out.append(src_buf[cp_off : cp_off + cp_size])
3880 elif cmd != 0:
3881 out.append(delta[index : index + cmd])
3882 index += cmd
3883 else:
3884 raise ApplyDeltaError("Invalid opcode 0")
3886 if index != delta_length:
3887 raise ApplyDeltaError(f"delta not empty: {delta[index:]!r}")
3889 if dest_size != chunks_length(out):
3890 raise ApplyDeltaError("dest size incorrect")
3892 return out
3895def write_pack_index_v2(
3896 f: IO[bytes],
3897 entries: Iterable[tuple[bytes, int, int | None]],
3898 pack_checksum: bytes,
3899) -> bytes:
3900 """Write a new pack index file.
3902 Args:
3903 f: File-like object to write to
3904 entries: List of tuples with object name (sha), offset_in_pack, and
3905 crc32_checksum.
3906 pack_checksum: Checksum of the pack file.
3907 Returns: The checksum of the index file written
3908 """
3909 # Determine hash algorithm from pack_checksum length
3910 if len(pack_checksum) == 20:
3911 hash_func = sha1
3912 elif len(pack_checksum) == 32:
3913 hash_func = sha256
3914 else:
3915 raise ValueError(f"Unsupported pack checksum length: {len(pack_checksum)}")
3917 f_writer = HashWriter(f, hash_func)
3918 f_writer.write(b"\377tOc") # Magic!
3919 f_writer.write(struct.pack(">L", 2))
3921 # Convert to list to allow multiple iterations
3922 entries_list = list(entries)
3924 fan_out_table: dict[int, int] = defaultdict(lambda: 0)
3925 for name, offset, entry_checksum in entries_list:
3926 fan_out_table[ord(name[:1])] += 1
3928 if entries_list:
3929 hash_size = len(entries_list[0][0])
3930 else:
3931 hash_size = len(pack_checksum) # Use pack_checksum length as hash size
3933 # Fan-out table
3934 largetable: list[int] = []
3935 for i in range(0x100):
3936 f_writer.write(struct.pack(b">L", fan_out_table[i]))
3937 fan_out_table[i + 1] += fan_out_table[i]
3938 for name, offset, entry_checksum in entries_list:
3939 if len(name) != hash_size:
3940 raise TypeError(
3941 f"Object name has wrong length: expected {hash_size}, got {len(name)}"
3942 )
3943 f_writer.write(name)
3944 for name, offset, entry_checksum in entries_list:
3945 f_writer.write(struct.pack(b">L", entry_checksum))
3946 for name, offset, entry_checksum in entries_list:
3947 if offset < 2**31:
3948 f_writer.write(struct.pack(b">L", offset))
3949 else:
3950 f_writer.write(struct.pack(b">L", 2**31 + len(largetable)))
3951 largetable.append(offset)
3952 for offset in largetable:
3953 f_writer.write(struct.pack(b">Q", offset))
3954 f_writer.write(pack_checksum)
3955 return f_writer.write_hash()
3958def write_pack_index_v3(
3959 f: IO[bytes],
3960 entries: Iterable[tuple[bytes, int, int | None]],
3961 pack_checksum: bytes,
3962 hash_format: int = 1,
3963) -> bytes:
3964 """Write a new pack index file in v3 format.
3966 Args:
3967 f: File-like object to write to
3968 entries: List of tuples with object name (sha), offset_in_pack, and
3969 crc32_checksum.
3970 pack_checksum: Checksum of the pack file.
3971 hash_format: Hash algorithm identifier (1 = SHA-1, 2 = SHA-256)
3972 Returns: The SHA of the index file written
3973 """
3974 if hash_format == 1:
3975 hash_size = 20 # SHA-1
3976 writer_cls = SHA1Writer
3977 elif hash_format == 2:
3978 hash_size = 32 # SHA-256
3979 # TODO: Add SHA256Writer when SHA-256 support is implemented
3980 raise NotImplementedError("SHA-256 support not yet implemented")
3981 else:
3982 raise ValueError(f"Unknown hash algorithm {hash_format}")
3984 # Convert entries to list to allow multiple iterations
3985 entries_list = list(entries)
3987 # Calculate shortest unambiguous prefix length for object names
3988 # For now, use full hash size (this could be optimized)
3989 shortened_oid_len = hash_size
3991 f = writer_cls(f)
3992 f.write(b"\377tOc") # Magic!
3993 f.write(struct.pack(">L", 3)) # Version 3
3994 f.write(struct.pack(">L", hash_format)) # Hash algorithm
3995 f.write(struct.pack(">L", shortened_oid_len)) # Shortened OID length
3997 fan_out_table: dict[int, int] = defaultdict(lambda: 0)
3998 for name, offset, entry_checksum in entries_list:
3999 if len(name) != hash_size:
4000 raise ValueError(
4001 f"Object name has wrong length: expected {hash_size}, got {len(name)}"
4002 )
4003 fan_out_table[ord(name[:1])] += 1
4005 # Fan-out table
4006 largetable: list[int] = []
4007 for i in range(0x100):
4008 f.write(struct.pack(b">L", fan_out_table[i]))
4009 fan_out_table[i + 1] += fan_out_table[i]
4011 # Object names table
4012 for name, offset, entry_checksum in entries_list:
4013 f.write(name)
4015 # CRC32 checksums table
4016 for name, offset, entry_checksum in entries_list:
4017 f.write(struct.pack(b">L", entry_checksum))
4019 # Offset table
4020 for name, offset, entry_checksum in entries_list:
4021 if offset < 2**31:
4022 f.write(struct.pack(b">L", offset))
4023 else:
4024 f.write(struct.pack(b">L", 2**31 + len(largetable)))
4025 largetable.append(offset)
4027 # Large offset table
4028 for offset in largetable:
4029 f.write(struct.pack(b">Q", offset))
4031 assert len(pack_checksum) == hash_size, (
4032 f"Pack checksum has wrong length: expected {hash_size}, got {len(pack_checksum)}"
4033 )
4034 f.write(pack_checksum)
4035 return f.write_sha()
4038def write_pack_index(
4039 f: IO[bytes],
4040 entries: Iterable[tuple[bytes, int, int | None]],
4041 pack_checksum: bytes,
4042 progress: Callable[..., None] | None = None,
4043 version: int | None = None,
4044) -> bytes:
4045 """Write a pack index file.
4047 Args:
4048 f: File-like object to write to.
4049 entries: List of (checksum, offset, crc32) tuples
4050 pack_checksum: Checksum of the pack file.
4051 progress: Progress function (not currently used)
4052 version: Pack index version to use (1, 2, or 3). If None, defaults to DEFAULT_PACK_INDEX_VERSION.
4054 Returns:
4055 SHA of the written index file
4057 Raises:
4058 ValueError: If an unsupported version is specified
4059 """
4060 if version is None:
4061 version = DEFAULT_PACK_INDEX_VERSION
4063 if version == 1:
4064 return write_pack_index_v1(f, entries, pack_checksum)
4065 elif version == 2:
4066 return write_pack_index_v2(f, entries, pack_checksum)
4067 elif version == 3:
4068 return write_pack_index_v3(f, entries, pack_checksum)
4069 else:
4070 raise ValueError(f"Unsupported pack index version: {version}")
4073class Pack:
4074 """A Git pack object."""
4076 _data_load: Callable[[], PackData] | None
4077 _idx_load: Callable[[], PackIndex] | None
4079 _data: PackData | None
4080 _idx: PackIndex | None
4081 _bitmap: "PackBitmap | None"
4083 def __init__(
4084 self,
4085 basename: str,
4086 *,
4087 object_format: ObjectFormat,
4088 resolve_ext_ref: ResolveExtRefFn | None = None,
4089 delta_window_size: int | None = None,
4090 window_memory: int | None = None,
4091 delta_cache_size: int | None = None,
4092 depth: int | None = None,
4093 threads: int | None = None,
4094 big_file_threshold: int | None = None,
4095 delta_base_cache_limit: int | None = None,
4096 ) -> None:
4097 """Initialize a Pack object.
4099 Args:
4100 basename: Base path for pack files (without .pack/.idx extension)
4101 object_format: Hash algorithm used by the repository
4102 resolve_ext_ref: Optional function to resolve external references
4103 delta_window_size: Size of the delta compression window
4104 window_memory: Memory limit for delta compression window
4105 delta_cache_size: Size of the delta cache
4106 depth: Maximum depth for delta chains
4107 threads: Number of threads to use for operations
4108 big_file_threshold: Size threshold for big file handling
4109 delta_base_cache_limit: Maximum bytes for delta base object cache
4110 """
4111 self._basename = basename
4112 self.object_format = object_format
4113 self._data = None
4114 self._idx = None
4115 self._bitmap = None
4116 self._idx_path = self._basename + ".idx"
4117 self._data_path = self._basename + ".pack"
4118 self._bitmap_path = self._basename + ".bitmap"
4119 self.delta_window_size = delta_window_size
4120 self.window_memory = window_memory
4121 self.delta_cache_size = delta_cache_size
4122 self.depth = depth
4123 self.threads = threads
4124 self.big_file_threshold = big_file_threshold
4125 self.delta_base_cache_limit = delta_base_cache_limit
4126 self._idx_load = lambda: load_pack_index(self._idx_path, object_format)
4127 self._data_load = lambda: PackData(
4128 self._data_path,
4129 delta_window_size=delta_window_size,
4130 window_memory=window_memory,
4131 delta_cache_size=delta_cache_size,
4132 depth=depth,
4133 threads=threads,
4134 big_file_threshold=big_file_threshold,
4135 delta_base_cache_limit=delta_base_cache_limit,
4136 object_format=object_format,
4137 )
4138 self.resolve_ext_ref = resolve_ext_ref
4140 @classmethod
4141 def from_lazy_objects(
4142 cls,
4143 data_fn: Callable[[], PackData],
4144 idx_fn: Callable[[], PackIndex],
4145 ) -> "Pack":
4146 """Create a new pack object from callables to load pack data and index objects."""
4147 # Load index to get object format
4148 idx = idx_fn()
4149 ret = cls("", object_format=idx.object_format)
4150 ret._data_load = data_fn
4151 ret._idx = idx
4152 ret._idx_load = None
4153 return ret
4155 @classmethod
4156 def from_objects(cls, data: PackData, idx: PackIndex) -> "Pack":
4157 """Create a new pack object from pack data and index objects."""
4158 ret = cls("", object_format=idx.object_format)
4159 ret._data = data
4160 ret._data_load = None
4161 ret._idx = idx
4162 ret._idx_load = None
4163 ret.check_length_and_checksum()
4164 return ret
4166 def name(self) -> bytes:
4167 """The SHA over the SHAs of the objects in this pack."""
4168 return self.index.objects_sha1()
4170 @property
4171 def data(self) -> PackData:
4172 """The pack data object being used."""
4173 if self._data is None:
4174 assert self._data_load
4175 try:
4176 self._data = self._data_load()
4177 except FileNotFoundError as exc:
4178 raise PackFileDisappeared(self) from exc
4179 self.check_length_and_checksum()
4180 return self._data
4182 @property
4183 def index(self) -> PackIndex:
4184 """The index being used.
4186 Note: This may be an in-memory index
4187 """
4188 if self._idx is None:
4189 assert self._idx_load
4190 try:
4191 self._idx = self._idx_load()
4192 except FileNotFoundError as exc:
4193 raise PackFileDisappeared(self) from exc
4194 return self._idx
4196 @property
4197 def bitmap(self) -> "PackBitmap | None":
4198 """The bitmap being used, if available.
4200 Returns:
4201 PackBitmap instance or None if no bitmap exists
4203 Raises:
4204 ValueError: If bitmap file is invalid or corrupt
4205 """
4206 if self._bitmap is None:
4207 from .bitmap import read_bitmap
4209 self._bitmap = read_bitmap(self._bitmap_path, pack_index=self.index)
4210 return self._bitmap
4212 def ensure_bitmap(
4213 self,
4214 object_store: "BaseObjectStore",
4215 refs: dict["Ref", "ObjectID"],
4216 commit_interval: int | None = None,
4217 progress: Callable[[str], None] | None = None,
4218 ) -> "PackBitmap":
4219 """Ensure a bitmap exists for this pack, generating one if needed.
4221 Args:
4222 object_store: Object store to read objects from
4223 refs: Dictionary of ref names to commit SHAs
4224 commit_interval: Include every Nth commit in bitmap index
4225 progress: Optional progress reporting callback
4227 Returns:
4228 PackBitmap instance (either existing or newly generated)
4229 """
4230 from .bitmap import generate_bitmap, write_bitmap
4232 # Check if bitmap already exists
4233 try:
4234 existing = self.bitmap
4235 if existing is not None:
4236 return existing
4237 except FileNotFoundError:
4238 pass # No bitmap, we'll generate one
4240 # Generate new bitmap
4241 if progress:
4242 progress(f"Generating bitmap for {self.name().decode('utf-8')}...\n")
4244 pack_bitmap = generate_bitmap(
4245 self.index,
4246 object_store,
4247 refs,
4248 self.get_stored_checksum(),
4249 commit_interval=commit_interval,
4250 progress=progress,
4251 )
4253 # Write bitmap file
4254 write_bitmap(self._bitmap_path, pack_bitmap)
4256 if progress:
4257 progress(f"Wrote {self._bitmap_path}\n")
4259 # Update cached bitmap
4260 self._bitmap = pack_bitmap
4262 return pack_bitmap
4264 @property
4265 def mmap_size(self) -> int:
4266 """Return the total mmapped memory usage of this pack.
4268 This includes the pack data file and index file sizes,
4269 but only for components that have been loaded (and thus mmapped).
4270 """
4271 total = 0
4272 if self._data is not None:
4273 total += self._data._get_size()
4274 if self._idx is not None and isinstance(self._idx, FilePackIndex):
4275 total += self._idx._size
4276 return total
4278 def close(self) -> None:
4279 """Close the pack file and index."""
4280 if self._data is not None:
4281 self._data.close()
4282 self._data = None
4283 if self._idx is not None:
4284 self._idx.close()
4285 self._idx = None
4287 def __del__(self) -> None:
4288 """Ensure pack file is closed when Pack is garbage collected."""
4289 if self._data is not None or self._idx is not None:
4290 import warnings
4292 warnings.warn(
4293 f"unclosed Pack {self!r}", ResourceWarning, stacklevel=2, source=self
4294 )
4295 try:
4296 self.close()
4297 except Exception:
4298 # Ignore errors during cleanup
4299 pass
4301 def __enter__(self) -> Self:
4302 """Enter context manager."""
4303 return self
4305 def __exit__(
4306 self,
4307 type: type | None,
4308 value: BaseException | None,
4309 traceback: TracebackType | None,
4310 ) -> None:
4311 """Exit context manager."""
4312 self.close()
4314 def __eq__(self, other: object) -> bool:
4315 """Check equality with another pack."""
4316 if not isinstance(other, Pack):
4317 return False
4318 return self.index == other.index
4320 def __len__(self) -> int:
4321 """Number of entries in this pack."""
4322 return len(self.index)
4324 def __repr__(self) -> str:
4325 """Return string representation of this pack."""
4326 return f"{self.__class__.__name__}({self._basename!r})"
4328 def __iter__(self) -> Iterator[ObjectID]:
4329 """Iterate over all the sha1s of the objects in this pack."""
4330 return iter(self.index)
4332 def check_length_and_checksum(self) -> None:
4333 """Sanity check the length and checksum of the pack index and data."""
4334 assert len(self.index) == len(self.data), (
4335 f"Length mismatch: {len(self.index)} (index) != {len(self.data)} (data)"
4336 )
4337 idx_stored_checksum = self.index.get_pack_checksum()
4338 data_stored_checksum = self.data.get_stored_checksum()
4339 if (
4340 idx_stored_checksum is not None
4341 and idx_stored_checksum != data_stored_checksum
4342 ):
4343 raise ChecksumMismatch(
4344 sha_to_hex(RawObjectID(idx_stored_checksum)),
4345 sha_to_hex(RawObjectID(data_stored_checksum)),
4346 )
4348 def check(self) -> None:
4349 """Check the integrity of this pack.
4351 Raises:
4352 ChecksumMismatch: if a checksum for the index or data is wrong
4353 """
4354 self.index.check()
4355 self.data.check()
4356 for obj in self.iterobjects():
4357 obj.check()
4358 # TODO: object connectivity checks
4360 def get_stored_checksum(self) -> bytes:
4361 """Return the stored checksum of the pack data."""
4362 return self.data.get_stored_checksum()
4364 def pack_tuples(self) -> list[tuple[ShaFile, None]]:
4365 """Return pack tuples for all objects in pack."""
4366 return [(o, None) for o in self.iterobjects()]
4368 def __contains__(self, sha1: ObjectID | RawObjectID) -> bool:
4369 """Check whether this pack contains a particular SHA1."""
4370 try:
4371 self.index.object_offset(sha1)
4372 return True
4373 except KeyError:
4374 return False
4376 def get_raw(self, sha1: RawObjectID | ObjectID) -> tuple[int, bytes]:
4377 """Get raw object data by SHA1."""
4378 offset = self.index.object_offset(sha1)
4379 obj_type, obj = self.data.get_object_at(offset)
4380 type_num, chunks = self.resolve_object(offset, obj_type, obj)
4381 return type_num, b"".join(chunks) # type: ignore[arg-type]
4383 def __getitem__(self, sha1: "ObjectID | RawObjectID") -> ShaFile:
4384 """Retrieve the specified SHA1."""
4385 type, uncomp = self.get_raw(sha1)
4386 return ShaFile.from_raw_string(type, uncomp, sha=sha1)
4388 def iterobjects(self) -> Iterator[ShaFile]:
4389 """Iterate over the objects in this pack."""
4390 return iter(
4391 PackInflater.for_pack_data(self.data, resolve_ext_ref=self.resolve_ext_ref)
4392 )
4394 def iterobjects_subset(
4395 self, shas: Iterable[ObjectID], *, allow_missing: bool = False
4396 ) -> Iterator[ShaFile]:
4397 """Iterate over a subset of objects in this pack."""
4398 return (
4399 uo
4400 for uo in PackInflater.for_pack_subset(
4401 self,
4402 shas,
4403 allow_missing=allow_missing,
4404 resolve_ext_ref=self.resolve_ext_ref,
4405 )
4406 if uo.id in shas
4407 )
4409 def iter_unpacked_subset(
4410 self,
4411 shas: Iterable[ObjectID | RawObjectID],
4412 *,
4413 include_comp: bool = False,
4414 allow_missing: bool = False,
4415 convert_ofs_delta: bool = False,
4416 ) -> Iterator[UnpackedObject]:
4417 """Iterate over unpacked objects in subset."""
4418 ofs_pending: dict[int, list[UnpackedObject]] = defaultdict(list)
4419 ofs: dict[int, bytes] = {}
4420 todo: set[ObjectID | RawObjectID] = set(shas)
4421 for unpacked in self.iter_unpacked(include_comp=include_comp):
4422 sha = unpacked.sha()
4423 if unpacked.offset is not None:
4424 ofs[unpacked.offset] = sha
4425 hexsha = sha_to_hex(RawObjectID(sha))
4426 if hexsha in todo:
4427 if unpacked.pack_type_num == OFS_DELTA:
4428 assert isinstance(unpacked.delta_base, int)
4429 assert unpacked.offset is not None
4430 base_offset = unpacked.offset - unpacked.delta_base
4431 try:
4432 unpacked.delta_base = ofs[base_offset]
4433 except KeyError:
4434 ofs_pending[base_offset].append(unpacked)
4435 continue
4436 else:
4437 unpacked.pack_type_num = REF_DELTA
4438 yield unpacked
4439 todo.remove(hexsha)
4440 if unpacked.offset is not None:
4441 for child in ofs_pending.pop(unpacked.offset, []):
4442 child.pack_type_num = REF_DELTA
4443 child.delta_base = sha
4444 yield child
4445 assert not ofs_pending
4446 if not allow_missing and todo:
4447 raise UnresolvedDeltas(list(todo))
4449 def iter_unpacked(self, include_comp: bool = False) -> Iterator[UnpackedObject]:
4450 """Iterate over all unpacked objects in this pack."""
4451 ofs_to_entries = {
4452 ofs: (sha, crc32) for (sha, ofs, crc32) in self.index.iterentries()
4453 }
4454 for unpacked in self.data.iter_unpacked(include_comp=include_comp):
4455 assert unpacked.offset is not None
4456 (sha, crc32) = ofs_to_entries[unpacked.offset]
4457 unpacked._sha = sha
4458 unpacked.crc32 = crc32
4459 yield unpacked
4461 def keep(self, msg: bytes | None = None) -> str:
4462 """Add a .keep file for the pack, preventing git from garbage collecting it.
4464 Args:
4465 msg: A message written inside the .keep file; can be used later
4466 to determine whether or not a .keep file is obsolete.
4467 Returns: The path of the .keep file, as a string.
4468 """
4469 keepfile_name = f"{self._basename}.keep"
4470 with GitFile(keepfile_name, "wb") as keepfile:
4471 if msg:
4472 keepfile.write(msg)
4473 keepfile.write(b"\n")
4474 return keepfile_name
4476 def get_ref(
4477 self, sha: RawObjectID | ObjectID
4478 ) -> tuple[int | None, int, OldUnpackedObject]:
4479 """Get the object for a ref SHA, only looking in this pack."""
4480 # TODO: cache these results
4481 try:
4482 offset = self.index.object_offset(sha)
4483 except KeyError:
4484 offset = None
4485 if offset:
4486 type, obj = self.data.get_object_at(offset)
4487 elif self.resolve_ext_ref:
4488 type, obj = self.resolve_ext_ref(sha)
4489 else:
4490 raise KeyError(sha)
4491 return offset, type, obj
4493 def resolve_object(
4494 self,
4495 offset: int,
4496 type: int,
4497 obj: OldUnpackedObject,
4498 get_ref: Callable[
4499 [RawObjectID | ObjectID], tuple[int | None, int, OldUnpackedObject]
4500 ]
4501 | None = None,
4502 ) -> tuple[int, OldUnpackedObject]:
4503 """Resolve an object, possibly resolving deltas when necessary.
4505 Returns: Tuple with object type and contents.
4506 """
4507 # Walk down the delta chain, building a stack of deltas to reach
4508 # the requested object.
4509 base_offset: int | None = offset
4510 base_type = type
4511 base_obj = obj
4512 delta_stack = []
4513 while base_type in DELTA_TYPES:
4514 prev_offset = base_offset
4515 if get_ref is None:
4516 get_ref = self.get_ref
4517 assert isinstance(base_obj, tuple), (
4518 f"Expected delta tuple, got {base_obj.__class__.__name__}"
4519 )
4520 if base_type == OFS_DELTA:
4521 (delta_offset, delta) = base_obj
4522 # TODO: clean up asserts and replace with nicer error messages
4523 assert isinstance(delta_offset, int), (
4524 f"Expected int, got {delta_offset.__class__}"
4525 )
4526 assert base_offset is not None
4527 base_offset = base_offset - delta_offset
4528 base_type, base_obj = self.data.get_object_at(base_offset)
4529 assert isinstance(base_type, int)
4530 elif base_type == REF_DELTA:
4531 (basename, delta) = base_obj
4532 assert (
4533 isinstance(basename, bytes)
4534 and len(basename) == self.object_format.oid_length
4535 )
4536 base_offset_temp, base_type, base_obj = get_ref(RawObjectID(basename))
4537 assert isinstance(base_type, int)
4538 # base_offset_temp can be None for thin packs (external references)
4539 base_offset = base_offset_temp
4540 if base_offset == prev_offset: # object is based on itself
4541 raise UnresolvedDeltas([basename])
4542 else:
4543 raise AssertionError(f"Unexpected delta type: {base_type}")
4544 delta_stack.append((prev_offset, base_type, delta))
4546 # Now grab the base object (mustn't be a delta) and apply the
4547 # deltas all the way up the stack.
4548 chunks = base_obj
4549 for prev_offset, _delta_type, delta in reversed(delta_stack):
4550 # Convert chunks to bytes for apply_delta if needed
4551 if isinstance(chunks, list):
4552 chunks_bytes = b"".join(chunks)
4553 elif isinstance(chunks, tuple):
4554 # For tuple type, second element is the actual data
4555 _, chunk_data = chunks
4556 if isinstance(chunk_data, list):
4557 chunks_bytes = b"".join(chunk_data)
4558 else:
4559 chunks_bytes = chunk_data
4560 else:
4561 chunks_bytes = chunks
4563 # Apply delta and get result as list
4564 chunks = apply_delta(chunks_bytes, delta)
4566 if prev_offset is not None:
4567 self.data._offset_cache[prev_offset] = base_type, chunks
4568 return base_type, chunks
4570 def entries(
4571 self, progress: Callable[[int, int], None] | None = None
4572 ) -> Iterator[PackIndexEntry]:
4573 """Yield entries summarizing the contents of this pack.
4575 Args:
4576 progress: Progress function, called with current and total
4577 object count.
4578 Returns: iterator of tuples with (sha, offset, crc32)
4579 """
4580 return self.data.iterentries(
4581 progress=progress, resolve_ext_ref=self.resolve_ext_ref
4582 )
4584 def sorted_entries(
4585 self, progress: Callable[[int, int], None] | None = None
4586 ) -> Iterator[PackIndexEntry]:
4587 """Return entries in this pack, sorted by SHA.
4589 Args:
4590 progress: Progress function, called with current and total
4591 object count
4592 Returns: Iterator of tuples with (sha, offset, crc32)
4593 """
4594 return iter(
4595 self.data.sorted_entries(
4596 progress=progress, resolve_ext_ref=self.resolve_ext_ref
4597 )
4598 )
4600 def get_unpacked_object(
4601 self,
4602 sha: ObjectID | RawObjectID,
4603 *,
4604 include_comp: bool = False,
4605 convert_ofs_delta: bool = True,
4606 ) -> UnpackedObject:
4607 """Get the unpacked object for a sha.
4609 Args:
4610 sha: SHA of object to fetch
4611 include_comp: Whether to include compression data in UnpackedObject
4612 convert_ofs_delta: Whether to convert offset deltas to ref deltas
4613 """
4614 offset = self.index.object_offset(sha)
4615 unpacked = self.data.get_unpacked_object_at(offset, include_comp=include_comp)
4616 if unpacked.pack_type_num == OFS_DELTA and convert_ofs_delta:
4617 assert isinstance(unpacked.delta_base, int)
4618 unpacked.delta_base = self.index.object_sha1(offset - unpacked.delta_base)
4619 unpacked.pack_type_num = REF_DELTA
4620 return unpacked
4623def extend_pack(
4624 f: BinaryIO,
4625 object_ids: Set["RawObjectID"],
4626 get_raw: Callable[["RawObjectID | ObjectID"], tuple[int, bytes]],
4627 object_format: "ObjectFormat",
4628 *,
4629 compression_level: int = -1,
4630 progress: Callable[[bytes], None] | None = None,
4631) -> tuple[bytes, list[tuple[RawObjectID, int, int]]]:
4632 """Extend a pack file with more objects.
4634 The caller should make sure that object_ids does not contain any objects
4635 that are already in the pack
4636 """
4637 # Update the header with the new number of objects.
4638 f.seek(0)
4639 _version, num_objects = read_pack_header(f.read)
4641 if object_ids:
4642 f.seek(0)
4643 write_pack_header(f.write, num_objects + len(object_ids))
4645 # Must flush before reading (http://bugs.python.org/issue3207)
4646 f.flush()
4648 # Rescan the rest of the pack, computing the SHA with the new header.
4649 new_sha = compute_file_sha(
4650 f, hash_func=object_format.hash_func, end_ofs=-object_format.oid_length
4651 )
4653 # Must reposition before writing (http://bugs.python.org/issue3207)
4654 f.seek(0, os.SEEK_CUR)
4656 extra_entries = []
4658 # Complete the pack.
4659 for i, object_id in enumerate(object_ids):
4660 if progress is not None:
4661 progress(
4662 (f"writing extra base objects: {i}/{len(object_ids)}\r").encode("ascii")
4663 )
4664 assert len(object_id) == object_format.oid_length
4665 type_num, data = get_raw(object_id)
4666 offset = f.tell()
4667 crc32 = write_pack_object(
4668 f.write,
4669 type_num,
4670 [data], # Convert bytes to list[bytes]
4671 sha=new_sha,
4672 compression_level=compression_level,
4673 object_format=object_format,
4674 )
4675 extra_entries.append((object_id, offset, crc32))
4676 pack_sha = new_sha.digest()
4677 f.write(pack_sha)
4678 return pack_sha, extra_entries
4681try:
4682 from dulwich._pack import ( # type: ignore
4683 apply_delta,
4684 bisect_find_sha,
4685 )
4686except ImportError:
4687 pass
4689# Try to import the Rust version of create_delta
4690try:
4691 from dulwich._pack import create_delta as _create_delta_rs
4692except ImportError:
4693 pass
4694else:
4695 # Wrap the Rust version to match the Python API (returns bytes instead of Iterator)
4696 def _create_delta_rs_wrapper(
4697 base_buf: bytes | list[bytes], target_buf: bytes | list[bytes]
4698 ) -> Iterator[bytes]:
4699 """Wrapper for Rust create_delta to match Python API."""
4700 if isinstance(base_buf, list):
4701 base_buf = b"".join(base_buf)
4702 if isinstance(target_buf, list):
4703 target_buf = b"".join(target_buf)
4704 yield _create_delta_rs(base_buf, target_buf)
4706 create_delta = _create_delta_rs_wrapper