tessl/pypi-dulwich
Pure Python implementation of the Git version control system providing comprehensive access to Git repositories without requiring the Git command-line tool
—
Pending
object-storage.mddocs/
Object Storage
Flexible object storage backends supporting filesystem, memory, pack files, and cloud storage. Object stores provide efficient storage and retrieval of Git objects with various optimization strategies.
Capabilities
Base Object Store
Abstract interface for all object storage implementations.
class BaseObjectStore:
"""
Abstract base class for Git object storage.
Defines interface for storing and retrieving Git objects
with support for different storage backends.
"""
def __contains__(self, sha: bytes) -> bool:
"""
Check if object exists in store.
Parameters:
- sha: 20-byte SHA-1 hash
Returns:
True if object exists
"""
def __getitem__(self, sha: bytes) -> ShaFile:
"""
Retrieve object by SHA-1.
Parameters:
- sha: 20-byte SHA-1 hash
Returns:
ShaFile object (Blob, Tree, Commit, or Tag)
Raises:
KeyError: If object not found
"""
def __iter__(self) -> Iterator[bytes]:
"""
Iterate over all object SHA-1 hashes.
Yields:
20-byte SHA-1 hashes of all objects
"""
def add_object(self, obj: ShaFile) -> None:
"""
Add object to store.
Parameters:
- obj: ShaFile object to store
"""
def add_objects(self, objects: Iterator[ShaFile]) -> None:
"""
Add multiple objects to store.
Parameters:
- objects: Iterator of ShaFile objects
"""
def contains_loose(self, sha: bytes) -> bool:
"""
Check if object exists as loose object.
Parameters:
- sha: 20-byte SHA-1 hash
Returns:
True if loose object exists
"""
def contains_packed(self, sha: bytes) -> bool:
"""
Check if object exists in pack file.
Parameters:
- sha: 20-byte SHA-1 hash
Returns:
True if packed object exists
"""
def get_raw(self, sha: bytes) -> Tuple[int, bytes]:
"""
Get raw object data.
Parameters:
- sha: 20-byte SHA-1 hash
Returns:
Tuple of (object_type, raw_data)
"""
def find_missing_objects(
self,
haves: List[bytes],
wants: List[bytes],
shallow: Set[bytes] = None,
progress: Callable = None,
get_tagged: Callable = None,
get_parents: Callable = lambda commit: commit.parents
) -> Iterator[Tuple[bytes, Optional[bytes]]]:
"""
Find objects missing from store.
Parameters:
- haves: SHA-1 hashes of objects we have
- wants: SHA-1 hashes of objects we want
- shallow: Set of shallow commit SHA-1s to skip
- progress: Progress callback function
- get_tagged: Function to get tagged objects
- get_parents: Function for getting commit parents
Yields:
Tuples of (sha1_hash, path) for missing objects
"""
def find_common_revisions(
self,
graphwalker: GraphWalker
) -> List[bytes]:
"""
Find common revisions using graph walker.
Parameters:
- graphwalker: Graph walker for revision traversal
Returns:
List of common revision SHA-1 hashes
"""
def generate_pack_data(
self,
have: List[bytes],
want: List[bytes],
shallow: Set[bytes] = None,
progress: Callable = None,
ofs_delta: bool = True
) -> Tuple[int, Iterator[UnpackedObject]]:
"""
Generate pack data for a set of wants/haves.
Parameters:
- have: SHA-1s of objects that should not be sent
- want: SHA-1s of objects that should be sent
- shallow: Set of shallow commit SHA-1s to skip
- progress: Optional progress reporting function
- ofs_delta: Whether OFS deltas can be included
Returns:
Tuple of (object_count, unpacked_objects_iterator)
"""
def determine_wants_all(
self,
refs: Dict[bytes, bytes],
depth: Optional[int] = None
) -> List[bytes]:
"""
Determine which refs are wanted based on availability and depth.
Parameters:
- refs: Dictionary mapping ref names to SHA-1 hashes
- depth: Optional depth limit for shallow operations
Returns:
List of SHA-1 hashes that are wanted
"""
def tree_changes(
self,
source: bytes,
target: bytes,
want_unchanged: bool = False,
include_trees: bool = False,
change_type_same: bool = False,
rename_detector: Optional[RenameDetector] = None,
paths: Optional[List[bytes]] = None
) -> Iterator[Tuple[Tuple[bytes, bytes], Tuple[int, int], Tuple[bytes, bytes]]]:
"""
Find differences between contents of two trees.
Parameters:
- source: SHA-1 of source tree
- target: SHA-1 of target tree
- want_unchanged: Whether unchanged files should be reported
- include_trees: Whether to include trees in output
- change_type_same: Whether to report files changing type
- rename_detector: RenameDetector for detecting renames
- paths: Optional list of paths to filter (as bytes)
Yields:
Tuples of ((old_path, new_path), (old_mode, new_mode), (old_sha, new_sha))
"""
def iterobjects_subset(
self,
shas: Iterable[bytes],
*,
allow_missing: bool = False
) -> Iterator[ShaFile]:
"""
Iterate over subset of objects by SHA-1.
Parameters:
- shas: Iterable of SHA-1 hashes to retrieve
- allow_missing: If True, skip missing objects instead of raising KeyError
Yields:
ShaFile objects for requested SHA-1s
"""
def close(self) -> None:
"""
Close any files opened by this object store.
Default implementation is a NO-OP. Subclasses should override
to provide proper cleanup of resources.
"""
def prune(self, grace_period: Optional[int] = None) -> None:
"""
Prune/clean up this object store.
This includes removing orphaned temporary files and other
housekeeping tasks.
Parameters:
- grace_period: Grace period in seconds for removing temporary files.
If None, uses the default grace period.
"""
def iter_prefix(self, prefix: bytes) -> Iterator[bytes]:
"""
Iterate over SHA-1s that start with given prefix.
Parameters:
- prefix: SHA-1 prefix to match
Yields:
SHA-1 hashes starting with the prefix
"""
def get_commit_graph(self) -> Optional[CommitGraph]:
"""
Get commit graph for this object store.
Returns:
CommitGraph object if available, None otherwise
"""
def write_commit_graph(
self,
refs: Optional[List[bytes]] = None,
reachable: bool = True
) -> None:
"""
Write commit graph file for this object store.
Parameters:
- refs: List of refs to include. If None, includes all refs.
- reachable: If True, includes all reachable commits from refs.
If False, only includes direct ref targets.
"""
def get_object_mtime(self, sha: bytes) -> float:
"""
Get modification time of an object.
Parameters:
- sha: SHA-1 of the object
Returns:
Modification time as seconds since epoch
Raises:
KeyError: If object is not found
"""Filesystem Object Store
Standard Git object storage using filesystem layout.
class DiskObjectStore(BaseObjectStore):
"""
Object store using standard Git filesystem layout.
Stores objects as loose files and pack files in .git/objects
directory with standard Git directory structure.
"""
def __init__(self, path: str):
"""
Initialize disk object store.
Parameters:
- path: Path to objects directory (.git/objects)
"""
@property
def path(self) -> str:
"""
Objects directory path.
Returns:
Absolute path to objects directory
"""
def move_in_pack(self, path: str) -> None:
"""
Move pack file into objects directory.
Parameters:
- path: Path to pack file to move
"""
def add_pack(self) -> Tuple[BinaryIO, BinaryIO]:
"""
Create new pack files for writing.
Returns:
Tuple of (pack_file, index_file) streams
"""
def pack_loose_objects(self) -> None:
"""
Pack loose objects into pack file.
Combines loose objects into efficient pack format
to reduce disk usage and improve performance.
"""
def repack(self) -> None:
"""
Repack all objects for optimal storage.
Combines all objects into optimized pack files
and removes redundant loose objects.
"""
def _get_loose_object(self, sha: bytes) -> Optional[ShaFile]:
"""
Get loose object by SHA-1 hash.
Parameters:
- sha: 20-byte SHA-1 hash
Returns:
ShaFile object if found, None otherwise
"""
def _iter_loose_objects(self) -> Iterator[bytes]:
"""
Iterate over SHA-1s of all loose objects.
Yields:
20-byte SHA-1 hashes of loose objects in store
"""
def _remove_pack(self, name: str) -> None:
"""
Remove pack file from disk.
Parameters:
- name: Pack file name to remove
"""
def _get_pack(self, name: str) -> Pack:
"""
Get pack object by name.
Parameters:
- name: Pack file name
Returns:
Pack object
"""
def _iter_pack_names(self) -> Iterator[str]:
"""
Iterate over pack file names.
Yields:
Pack file names in objects directory
"""
def _update_pack_cache(self) -> List[Pack]:
"""
Update pack cache with new packs from disk.
Returns:
List of newly added packs
"""Memory Object Store
In-memory object storage for temporary operations.
class MemoryObjectStore(BaseObjectStore):
"""
Object store using in-memory storage.
Stores all objects in memory without persistence.
Useful for testing and temporary operations.
"""
def __init__(self):
"""Initialize empty memory object store."""
def add_pack_data(
self,
count: int,
unpacked_objects: Iterator[UnpackedObject],
progress: Callable = None
) -> None:
"""
Add pack data to memory store.
Parameters:
- count: Number of objects in pack
- unpacked_objects: Iterator of unpacked object data
- progress: Progress callback function
"""
def add_thin_pack(
self,
read_all: Callable[[], bytes],
read_some: Callable[[int], bytes],
progress: Callable = None
) -> None:
"""
Add thin pack to memory store.
Parameters:
- read_all: Function to read all available data
- read_some: Function to read specified number of bytes
- progress: Progress callback function
"""Pack-Based Object Store
Object store optimized for pack file operations with comprehensive pack management.
Provides base functionality for object stores that use Git pack files as their primary storage mechanism, including delta compression, streaming operations, and pack caching.
class PackBasedObjectStore(BaseObjectStore):
"""
Object store using pack files for storage.
Optimized for pack file operations with efficient
delta compression and streaming support.
"""
def __init__(
self,
pack_compression_level: int = -1,
pack_index_version: Optional[int] = None,
pack_delta_window_size: Optional[int] = None,
pack_window_memory: Optional[int] = None,
pack_delta_cache_size: Optional[int] = None,
pack_depth: Optional[int] = None,
pack_threads: Optional[int] = None,
pack_big_file_threshold: Optional[int] = None
):
"""
Initialize pack-based object store.
Parameters:
- pack_compression_level: Compression level for pack files (-1 to 9)
- pack_index_version: Pack index format version (1 or 2)
- pack_delta_window_size: Size of delta compression window
- pack_window_memory: Memory limit for delta window
- pack_delta_cache_size: Size of delta cache
- pack_depth: Maximum delta depth
- pack_threads: Number of threads for pack operations
- pack_big_file_threshold: Threshold for big file handling
"""
@property
def alternates(self) -> List[BaseObjectStore]:
"""
Get list of alternate object stores.
Returns:
List of alternate object stores to search
"""
def add_pack(self) -> Tuple[BinaryIO, Callable[[], None], Callable[[], None]]:
"""
Add new pack to this object store.
Returns:
Tuple of (pack_file_stream, commit_function, abort_function)
"""
def add_pack_data(
self,
count: int,
unpacked_objects: Iterator[UnpackedObject],
progress: Callable = None
) -> Optional[Pack]:
"""
Add pack data to store.
Parameters:
- count: Number of objects in pack
- unpacked_objects: Iterator of unpacked object data
- progress: Progress callback function
Returns:
Pack object if pack was created, None if empty
"""
def _add_cached_pack(self, base_name: str, pack: Pack) -> None:
"""
Add newly appeared pack to the cache by path.
Parameters:
- base_name: Base name of the pack file
- pack: Pack object to cache
"""Overlay Object Store
Layered object store with fallback to base store.
class OverlayObjectStore(BaseObjectStore):
"""
Object store with overlay and base layers.
Provides layered storage where objects are first checked
in overlay layer, falling back to base layer if not found.
"""
def __init__(
self,
bases: List[BaseObjectStore],
add_store: Optional[BaseObjectStore] = None
):
"""
Initialize overlay object store.
Parameters:
- bases: List of base object stores (checked in order)
- add_store: Store for adding new objects
"""
@property
def bases(self) -> List[BaseObjectStore]:
"""
Base object stores.
Returns:
List of base stores checked for objects
"""
@property
def packs(self) -> List[Pack]:
"""
All packs from base stores.
Returns:
Combined list of packs from all base stores
"""
def iter_unpacked_subset(
self,
shas: Iterable[bytes],
*,
include_comp: bool = False,
allow_missing: bool = False,
convert_ofs_delta: bool = True
) -> Iterator[ShaFile]:
"""
Iterate over unpacked subset of objects.
Parameters:
- shas: SHA-1 hashes to retrieve
- include_comp: Whether to include compressed objects
- allow_missing: Whether to skip missing objects
- convert_ofs_delta: Whether to convert OFS deltas
Yields:
ShaFile objects from base stores
"""Bucket-Based Object Store
Object store implementation for cloud storage backends like Amazon S3.
Optimized for bucket-based storage systems where objects are stored as pack files in cloud storage buckets. Does not support loose objects, making it ideal for distributed and cloud-native Git hosting scenarios.
class BucketBasedObjectStore(PackBasedObjectStore):
"""
Object store using bucket organization like S3.
Implementation that uses a bucket store like S3 as backend.
Optimized for cloud storage scenarios with no loose objects.
"""
def _iter_loose_objects(self) -> Iterator[bytes]:
"""
Iterate over SHA-1s of all loose objects.
Returns:
Empty iterator (bucket stores have no loose objects)
"""
def _get_loose_object(self, sha: bytes) -> None:
"""
Get loose object by SHA-1.
Parameters:
- sha: 20-byte SHA-1 hash
Returns:
None (bucket stores have no loose objects)
"""
def delete_loose_object(self, sha: bytes) -> None:
"""
Delete loose object by SHA-1.
Parameters:
- sha: 20-byte SHA-1 hash
Note:
No-op for bucket stores as there are no loose objects.
"""
def _remove_pack(self, name: str) -> None:
"""
Remove pack from bucket storage.
Parameters:
- name: Pack file name
Raises:
NotImplementedError: Must be implemented by subclasses
"""
def _iter_pack_names(self) -> Iterator[str]:
"""
Iterate over pack file names in bucket.
Yields:
Pack file names available in bucket
Raises:
NotImplementedError: Must be implemented by subclasses
"""
def _get_pack(self, name: str) -> Pack:
"""
Get pack object from bucket storage.
Parameters:
- name: Pack file name
Returns:
Pack object
Raises:
NotImplementedError: Must be implemented by subclasses
"""
def _update_pack_cache(self) -> List[Pack]:
"""
Update pack cache with packs from bucket storage.
Returns:
List of newly added packs
"""
def _upload_pack(
self,
basename: str,
pack_file: BinaryIO,
index_file: BinaryIO
) -> None:
"""
Upload pack and index files to bucket storage.
Parameters:
- basename: Base name for pack files
- pack_file: Pack file stream
- index_file: Index file stream
Raises:
NotImplementedError: Must be implemented by subclasses
"""Protocol Classes
Protocol definitions for object store interfaces.
class PackContainer(Protocol):
"""
Protocol for containers that can manage pack files.
Defines interface for adding new pack files to storage.
"""
def add_pack(self) -> Tuple[BinaryIO, Callable[[], None], Callable[[], None]]:
"""
Add new pack to container.
Returns:
Tuple of (pack_file_stream, commit_function, abort_function)
"""
class ObjectIterator(Protocol):
"""
Protocol for iterating over objects.
Defines interface for object iteration with filtering.
"""
def iterobjects_subset(
self,
shas: Iterable[bytes],
*,
allow_missing: bool = False
) -> Iterator[ShaFile]:
"""
Iterate over subset of objects by SHA-1.
Parameters:
- shas: SHA-1 hashes to retrieve
- allow_missing: Whether to skip missing objects
Yields:
ShaFile objects for requested SHA-1s
"""Utility Classes
Helper classes for object store operations.
class MissingObjectFinder:
"""
Finds missing objects between object stores.
Used for determining what objects need to be transferred
during fetch and push operations.
"""
def __init__(
self,
object_store: BaseObjectStore,
haves: Iterable[bytes],
wants: Iterable[bytes],
*,
shallow: Optional[Set[bytes]] = None,
progress: Optional[Callable] = None,
get_tagged: Optional[Callable] = None,
get_parents: Callable[[Commit], List[bytes]] = lambda commit: commit.parents
):
"""
Initialize missing object finder.
Parameters:
- object_store: Object store containing objects to be sent
- haves: SHA-1 hashes of commits not to send (already present)
- wants: SHA-1 hashes of commits to send
- shallow: Set of shallow commit SHA-1s to skip
- progress: Progress callback function
- get_tagged: Function returning dict of pointed-to sha -> tag sha
- get_parents: Function for getting parents of a commit
"""
def get_remote_has(self) -> Set[bytes]:
"""
Get set of objects the remote has.
Returns:
Set of SHA-1 hashes known to be on remote
"""
def add_todo(
self,
entries: Iterable[Tuple[bytes, Optional[bytes], Optional[int], bool]]
) -> None:
"""
Add objects to todo list for processing.
Parameters:
- entries: Iterable of (sha, name, type_num, leaf) tuples
"""
def __next__(self) -> Tuple[bytes, Optional[PackHint]]:
"""
Get next missing object.
Returns:
Tuple of (sha1_hash, pack_hint) for next missing object
Raises:
StopIteration: When no more missing objects
"""
def __iter__(self):
"""
Make this object iterable.
Returns:
Self as iterator
"""
class ObjectStoreGraphWalker:
"""
Graph walker that finds what commits are missing from an object store.
Walks commit and tag relationships to determine
object dependencies and ancestry for protocol operations.
"""
heads: Set[bytes]
"""Revisions without descendants in the local repo."""
get_parents: Callable[[bytes], List[bytes]]
"""Function to retrieve parents in the local repo."""
shallow: Set[bytes]
"""Set of shallow commit SHA-1s."""
def __init__(
self,
local_heads: Iterable[bytes],
get_parents: Callable[[bytes], List[bytes]],
shallow: Optional[Set[bytes]] = None,
update_shallow: Optional[Callable] = None
):
"""
Initialize graph walker.
Parameters:
- local_heads: Heads to start search with
- get_parents: Function for finding parents of a SHA-1
- shallow: Set of shallow commit SHA-1s
- update_shallow: Function to update shallow set
"""
def nak(self) -> None:
"""
Indicate nothing in common was found.
"""
def ack(self, sha: bytes) -> None:
"""
Acknowledge that a revision and its ancestors are present.
Parameters:
- sha: SHA-1 hash of acknowledged revision
"""
def __next__(self) -> Optional[bytes]:
"""
Get next revision to ask about.
Returns:
SHA-1 hash of next revision, or None when done
"""
def __iter__(self):
"""
Make this object iterable.
Returns:
Self as iterator
"""Utility Functions
Standalone functions for object store operations.
def find_shallow(
store: BaseObjectStore,
heads: Iterable[bytes],
depth: int
) -> Tuple[Set[bytes], Set[bytes]]:
"""
Find shallow commits at specified depth.
Parameters:
- store: Object store for looking up objects
- heads: Iterable of head SHA-1s to start walking from
- depth: Depth of ancestors to include (1 = heads only)
Returns:
Tuple of (shallow_commits, not_shallow_commits) sets
Note:
Sets may overlap if commit is reachable along multiple paths.
"""
def get_depth(
store: BaseObjectStore,
head: bytes,
get_parents: Callable[[Commit], List[bytes]] = lambda commit: commit.parents,
max_depth: Optional[int] = None
) -> int:
"""
Get actual depth from head commit.
For commits with multiple parents, the largest possible depth
will be returned.
Parameters:
- store: Object store containing commits
- head: HEAD commit SHA-1 hash
- get_parents: Function for getting parents of a commit
- max_depth: Maximum depth to search
Returns:
Current available depth for the given head
"""
def tree_lookup_path(
lookup_obj: Callable[[bytes], ShaFile],
root_sha: bytes,
path: bytes
) -> Tuple[int, bytes]:
"""
Look up object at path in tree.
Parameters:
- lookup_obj: Function to retrieve objects by SHA-1
- root_sha: Root tree SHA-1 hash
- path: Path within tree to lookup
Returns:
Tuple of (mode, sha1_hash) for object at path
Raises:
NotTreeError: If root_sha does not point to a tree object
"""
def commit_tree_changes(
object_store: BaseObjectStore,
tree: Tree,
changes: List[TreeChange]
) -> Tree:
"""
Apply changes to tree and return new tree.
Parameters:
- object_store: Object store for reading/writing objects
- tree: Original tree object to modify
- changes: List of TreeChange objects describing modifications
Returns:
Modified tree object with changes applied
"""
def read_packs_file(f: BinaryIO) -> Iterator[str]:
"""
Yield pack names listed in a packs file.
Parameters:
- f: File-like object containing packs data
Yields:
Pack file names from the packs file
"""
def iter_tree_contents(
store: ObjectContainer,
tree_id: Optional[bytes],
*,
include_trees: bool = False
) -> Iterator[TreeEntry]:
"""
Iterate recursively over tree contents.
Iteration is depth-first pre-order, as in e.g. os.walk.
Parameters:
- store: Object store containing objects
- tree_id: Tree SHA-1 hash (can be None)
- include_trees: Include tree objects in iteration
Yields:
TreeEntry namedtuples for each object in tree
"""
def peel_sha(
store: ObjectContainer,
sha: bytes
) -> Tuple[ShaFile, ShaFile]:
"""
Peel all tags from a SHA.
Follows tag references to underlying objects until
reaching a non-tag object.
Parameters:
- store: Object store containing objects
- sha: SHA-1 hash of object to peel
Returns:
Tuple of (original_object, peeled_object)
"""Usage Examples
Basic Object Store Operations
from dulwich.object_store import DiskObjectStore, MemoryObjectStore
from dulwich.objects import Blob
# Create disk object store
disk_store = DiskObjectStore("/path/to/repo/.git/objects")
# Check if object exists
blob_sha = b'...' # 20-byte SHA-1
if blob_sha in disk_store:
blob = disk_store[blob_sha]
print(f"Blob size: {len(blob.data)}")
# Add new object
new_blob = Blob(b"New file content")
disk_store.add_object(new_blob)
# Create memory store for temporary operations
memory_store = MemoryObjectStore()
memory_store.add_object(new_blob)Object Store Layering
from dulwich.object_store import OverlayObjectStore, DiskObjectStore, MemoryObjectStore
# Create layered store
base_store = DiskObjectStore("/path/to/repo/.git/objects")
overlay_store = MemoryObjectStore()
layered_store = OverlayObjectStore([base_store], overlay_store)
# Objects are first checked in overlay, then base
# New objects go to overlay
layered_store.add_object(Blob(b"Temporary content"))Finding Missing Objects
from dulwich.object_store import MissingObjectFinder
# Find objects needed for synchronization
haves = [b'...'] # SHA-1s of objects we have
wants = [b'...'] # SHA-1s of objects we want
finder = MissingObjectFinder(
object_store=disk_store,
haves=haves,
wants=wants
)
# Iterate over missing objects
missing_objects = []
for sha, pack_hint in finder:
missing_objects.append((sha, pack_hint))
print(f"Need {len(missing_objects)} objects")
# Alternative: use with graph walker
from dulwich.object_store import ObjectStoreGraphWalker
# Set up graph walker for protocol operations
local_heads = [repo.head()]
def get_parents(sha):
return disk_store[sha].parents
walker = ObjectStoreGraphWalker(local_heads, get_parents)
# Find common revisions
common_revs = disk_store.find_common_revisions(walker)
print(f"Found {len(common_revs)} common revisions")Install with Tessl CLI
npx tessl i tessl/pypi-dulwich