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
references.mddocs/
References and Branches
Complete reference management including branches, tags, symbolic references, and packed refs with validation, atomic updates, and comprehensive locking support for safe concurrent operations.
Capabilities
Reference Container Classes
Container classes for managing Git references with different storage backends.
class RefsContainer:
"""Abstract container for Git references."""
def __init__(self, logger=None):
"""Initialize refs container with optional logger."""
def __getitem__(self, name: bytes) -> bytes:
"""Get reference value by name, following symbolic refs."""
def __setitem__(self, name: bytes, value: bytes) -> None:
"""Set reference value by name unconditionally."""
def __delitem__(self, name: bytes) -> None:
"""Remove reference by name unconditionally."""
def __contains__(self, refname: bytes) -> bool:
"""Check if reference exists in container."""
def __iter__(self) -> Iterator[bytes]:
"""Iterate over all reference names."""
def allkeys(self) -> Iterator[bytes]:
"""All refs present in this container."""
def keys(self, base: Optional[bytes] = None) -> Iterator[bytes]:
"""Refs present in container, optionally under a base."""
def subkeys(self, base: bytes) -> Set[bytes]:
"""Refs under a base with base prefix stripped."""
def as_dict(self, base: Optional[bytes] = None) -> Dict[bytes, bytes]:
"""Return contents of container as dictionary."""
def get_symrefs(self) -> Dict[bytes, bytes]:
"""Get symbolic references mapping source to target."""
def get_peeled(self, name: bytes) -> Optional[bytes]:
"""Get cached peeled value of a ref if available."""
def get_packed_refs(self) -> Dict[bytes, bytes]:
"""Get contents of packed-refs file."""
def add_packed_refs(self, new_refs: Dict[bytes, Optional[bytes]]) -> None:
"""Add given refs as packed refs (None means remove)."""
def read_ref(self, refname: bytes) -> Optional[bytes]:
"""Read reference without following symbolic refs."""
def read_loose_ref(self, name: bytes) -> Optional[bytes]:
"""Read loose reference and return its contents."""
def follow(self, name: bytes) -> Tuple[List[bytes], bytes]:
"""Follow reference name and return chain of refs and final SHA."""
def set_if_equals(
self,
name: bytes,
old_ref: Optional[bytes],
new_ref: bytes,
committer: Optional[bytes] = None,
timestamp: Optional[float] = None,
timezone: Optional[int] = None,
message: Optional[bytes] = None
) -> bool:
"""Set refname to new_ref only if it currently equals old_ref."""
def add_if_new(
self,
name: bytes,
ref: bytes,
committer: Optional[bytes] = None,
timestamp: Optional[float] = None,
timezone: Optional[int] = None,
message: Optional[bytes] = None
) -> bool:
"""Add new reference only if it doesn't already exist."""
def remove_if_equals(
self,
name: bytes,
old_ref: Optional[bytes],
committer: Optional[bytes] = None,
timestamp: Optional[float] = None,
timezone: Optional[int] = None,
message: Optional[bytes] = None
) -> bool:
"""Remove refname only if it currently equals old_ref."""
def set_symbolic_ref(
self,
name: bytes,
other: bytes,
committer: Optional[bytes] = None,
timestamp: Optional[float] = None,
timezone: Optional[int] = None,
message: Optional[bytes] = None
) -> None:
"""Make a ref point at another ref."""
def import_refs(
self,
base: bytes,
other: Dict[bytes, bytes],
committer: Optional[bytes] = None,
timestamp: Optional[float] = None,
timezone: Optional[int] = None,
message: Optional[bytes] = None,
prune: bool = False
) -> None:
"""Import refs from another container under base."""
def pack_refs(self, all: bool = False) -> None:
"""Pack loose refs into packed-refs file."""
class DictRefsContainer(RefsContainer):
"""Dictionary-based reference container.
This container does not support symbolic or packed references
and is not threadsafe.
"""
def __init__(
self,
refs: Dict[bytes, bytes],
logger=None
):
"""
Initialize dict-based refs container.
Args:
refs: Dictionary mapping ref names to SHA-1 values
logger: Optional reflog logger
"""
class InfoRefsContainer(RefsContainer):
"""Reference container based on info/refs file.
Read-only container that parses Git's info/refs format.
"""
def __init__(self, f):
"""
Initialize info/refs container.
Args:
f: File-like object containing info/refs data
"""
class DiskRefsContainer(RefsContainer):
"""Filesystem-based reference container.
Stores references as files in the Git repository's refs directory
and supports packed-refs files for efficient storage.
"""
def __init__(
self,
path: str,
worktree_path: Optional[str] = None,
logger=None
):
"""
Initialize filesystem refs container.
Args:
path: Path to Git repository
worktree_path: Optional path to worktree (for worktree support)
logger: Optional reflog logger
"""
def refpath(self, name: bytes) -> str:
"""
Get filesystem path for reference name.
Args:
name: Reference name
Returns:
Full filesystem path to reference file
"""
def get_packed_refs(self) -> Dict[bytes, bytes]:
"""
Get contents of packed-refs file.
Returns:
Dictionary mapping ref names to SHA-1 values
"""
def get_peeled(self, name: bytes) -> Optional[bytes]:
"""
Get peeled value from packed-refs file.
Args:
name: Reference name
Returns:
Peeled SHA-1 value if available, None otherwise
"""
def _write_packed_refs(self) -> None:
"""Write current packed refs to disk."""
def _read_packed_refs(self) -> Tuple[Dict[bytes, bytes], Dict[bytes, bytes]]:
"""Read packed refs from disk.
Returns:
Tuple of (packed_refs_dict, peeled_refs_dict)
"""Reference Validation Functions
Functions for validating and parsing Git references.
def check_ref_format(refname: bytes) -> bool:
"""
Check if reference name follows Git naming rules.
Args:
refname: Reference name to validate
Returns:
True if valid reference name
"""
def parse_symref_value(contents: bytes) -> bytes:
"""
Parse symbolic reference value from file contents.
Args:
contents: Raw file contents
Returns:
Target reference name
"""
def parse_remote_ref(ref: bytes) -> Tuple[bytes, bytes]:
"""
Parse remote reference into remote name and branch.
Args:
ref: Remote reference (e.g., b'refs/remotes/origin/main')
Returns:
Tuple of (remote_name, branch_name)
"""Packed References
Functions for reading and writing packed-refs files for efficient reference storage.
def read_packed_refs(f) -> Iterator[Tuple[bytes, bytes]]:
"""
Read packed references from file.
Args:
f: File-like object containing packed-refs data
Yields:
Tuples of (sha1, ref_name)
"""
def read_packed_refs_with_peeled(f) -> Iterator[Tuple[bytes, bytes, Optional[bytes]]]:
"""
Read packed refs file including peeled refs.
Assumes the "# pack-refs with: peeled" line was already read.
Args:
f: File-like object to read from, seek'ed to second line
Yields:
Tuples of (sha1, ref_name, peeled_sha1_or_none)
"""
def write_packed_refs(
f,
packed_refs: Dict[bytes, bytes],
peeled_refs: Optional[Dict[bytes, bytes]] = None
) -> None:
"""
Write packed references to file.
Args:
f: File-like object to write to
packed_refs: Dictionary of ref_name to SHA-1
peeled_refs: Optional dictionary of ref_name to peeled SHA-1
"""Reference Type Checking
Functions for identifying different types of references.
def is_local_branch(ref: bytes) -> bool:
"""
Check if reference is a local branch.
Args:
ref: Reference name
Returns:
True if local branch reference (starts with refs/heads/)
"""
def strip_peeled_refs(refs: Dict[bytes, bytes]) -> Dict[bytes, bytes]:
"""
Remove peeled reference entries from refs dictionary.
Args:
refs: Dictionary of references
Returns:
Dictionary with peeled refs (ending with ^{}) removed
"""
def split_peeled_refs(refs: Dict[bytes, bytes]) -> Tuple[Dict[bytes, bytes], Dict[bytes, bytes]]:
"""
Split peeled refs from regular refs.
Args:
refs: Dictionary of references
Returns:
Tuple of (regular_refs, peeled_refs) dictionaries
"""
def serialize_refs(
store: ObjectContainer,
refs: Dict[bytes, bytes]
) -> Iterator[bytes]:
"""
Serialize references to info/refs format.
Args:
store: Object store for peeling tags
refs: Dictionary of references
Yields:
Lines in info/refs format
"""
def read_info_refs(f) -> Dict[bytes, bytes]:
"""
Read info/refs file format.
Args:
f: File-like object containing info/refs data
Returns:
Dictionary mapping ref names to SHA-1 values
"""
def write_info_refs(
refs: Dict[bytes, bytes],
store: ObjectContainer
) -> Iterator[bytes]:
"""
Generate info/refs format with peeled values.
Args:
refs: Dictionary of references
store: Object store for peeling tags
Yields:
Lines in info/refs format including peeled values
"""Reference Locking
Context manager for atomic reference updates.
class locked_ref:
"""
Lock a ref while making modifications.
Works as a context manager to ensure atomic reference updates.
Locks the reference file to prevent concurrent modifications.
"""
def __init__(self, refs_container: DiskRefsContainer, refname: bytes):
"""
Initialize locked reference.
Args:
refs_container: DiskRefsContainer to operate on
refname: Name of reference to lock
"""
def __enter__(self) -> "locked_ref":
"""Enter context and acquire lock."""
def __exit__(self, exc_type, exc_value, traceback) -> None:
"""Exit context and release lock."""
def get(self) -> Optional[bytes]:
"""
Get current value of the ref.
Returns:
Current SHA-1 or symbolic ref value, None if not found
"""
def ensure_equals(self, expected_value: Optional[bytes]) -> bool:
"""
Ensure ref currently equals expected value.
Args:
expected_value: Expected current value
Returns:
True if ref equals expected value
"""
def set(self, new_ref: bytes) -> None:
"""
Set ref to new value.
Args:
new_ref: New SHA-1 or symbolic ref value
"""
def set_symbolic_ref(self, target: bytes) -> None:
"""
Make this ref point at another ref.
Args:
target: Name of ref to point at
"""
def delete(self) -> None:
"""Delete the ref file while holding the lock."""Exception Classes
class SymrefLoop(Exception):
"""Exception raised when symbolic reference creates a loop."""
def __init__(self, ref: bytes, depth: int):
"""
Initialize symref loop exception.
Args:
ref: Reference that caused the loop
depth: Depth at which loop was detected
"""
## Usage Examples
### Basic Reference Operations
```python
from dulwich.repo import Repo
from dulwich.refs import check_ref_format, parse_remote_ref
# Open repository and access references
repo = Repo('.')
refs = repo.refs
# List all references
for ref_name in refs.keys():
print(f"{ref_name.decode()}: {refs[ref_name].hex()}")
# Follow symbolic references
head_chain, head_sha = refs.follow(b'HEAD')
print(f"HEAD chain: {[r.decode() for r in head_chain]}")
print(f"HEAD SHA: {head_sha.decode()}")
# Get current branch safely
try:
head_target = refs[b'HEAD']
if b'refs/heads/' in head_chain[0]:
current_branch = head_chain[0][11:] # Remove 'refs/heads/' prefix
print(f"Current branch: {current_branch.decode()}")
except (KeyError, IndexError):
print("HEAD is detached")
# Create a new branch reference with validation
new_ref = b'refs/heads/feature-branch'
if check_ref_format(new_ref[5:]): # Remove 'refs/' prefix for validation
refs[new_ref] = repo.head()
# Parse remote references
remote_ref = b'refs/remotes/origin/main'
remote_name, branch_name = parse_remote_ref(remote_ref)
print(f"Remote: {remote_name.decode()}, Branch: {branch_name.decode()}")Working with Symbolic References
from dulwich.refs import parse_symref_value, locked_ref
# Check if HEAD is a symbolic reference
with open('.git/HEAD', 'rb') as f:
head_contents = f.read()
if head_contents.startswith(b'ref: '):
target_ref = parse_symref_value(head_contents)
print(f"HEAD points to: {target_ref.decode()}")
# Atomically update a reference
with locked_ref(repo.refs, b'refs/heads/feature') as ref_lock:
current_value = ref_lock.get()
if ref_lock.ensure_equals(current_value):
ref_lock.set(new_commit_sha)
print("Reference updated successfully")Working with Packed References
from dulwich.refs import read_packed_refs, write_packed_refs
# Read packed references
with open('.git/packed-refs', 'rb') as f:
for sha, ref_name in read_packed_refs(f):
print(f"{ref_name.decode()}: {sha.decode()}")
# Write packed references
packed_refs = {
b'refs/heads/main': b'abc123...',
b'refs/tags/v1.0': b'def456...'
}
with open('.git/packed-refs', 'wb') as f:
write_packed_refs(f, packed_refs)Advanced Reference Operations
from dulwich.refs import DiskRefsContainer, is_local_branch, strip_peeled_refs
# Work with disk-based references
refs_container = DiskRefsContainer('/path/to/repo')
# Filter for local branches only
all_refs = refs_container.as_dict()
local_branches = {
name: sha for name, sha in all_refs.items()
if is_local_branch(name)
}
# Remove peeled refs from dictionary
clean_refs = strip_peeled_refs(all_refs)
# Import refs from another repository
remote_refs = {'main': b'abc123...', 'develop': b'def456...'}
refs_container.import_refs(
b'refs/remotes/origin',
remote_refs,
message=b'fetch from origin'
)Install with Tessl CLI
npx tessl i tessl/pypi-dulwich