class documentation

class ReftableRefsContainer(RefsContainer):

Constructor: ReftableRefsContainer(path, logger)

View In Hierarchy

A refs container backed by the reftable format.

Method __getitem__ Get the SHA1 for a reference name.
Method __init__ Initialize a reftable refs container.
Method add_if_new Add a ref only if it doesn't exist.
Method add_packed_refs Add packed refs. Creates a new reftable file with all refs consolidated.
Method allkeys Return set of all ref names.
Method batch_update Context manager for batching multiple ref updates into a single reftable.
Method follow Follow a reference name.
Method get_packed_refs Get packed refs. Reftable doesn't distinguish packed/loose.
Method get_peeled Return the cached peeled value of a ref, if available.
Method read_loose_ref Read a reference value without following symbolic refs.
Method remove_if_equals Remove a ref if it equals old_ref.
Method remove_packed_ref Remove a packed ref. Creates a deletion record.
Method set_if_equals Atomically set a ref if it currently equals old_ref.
Method set_symbolic_ref Set a symbolic reference.
Instance Variable path Undocumented
Instance Variable reftable_dir Undocumented
Method _apply_batch_updates Apply batch updates to the refs dict and update indices.
Method _compact_tables_list Compact tables list to single file like Git does.
Method _ensure_refs_heads_marker Ensure refs/heads marker file exists for Git compatibility.
Method _flush_pending_updates Flush pending ref updates like Git does - consolidate all refs.
Method _generate_table_path Generate a new reftable file path.
Method _get_next_update_index Get the next available update index across all tables.
Method _get_table_files Get sorted list of reftable files from tables.list.
Method _load_ref_update_indices Load the update indices for all refs from existing reftable files.
Method _process_pending_updates Process pending updates and return (head_update, other_updates).
Method _read_all_tables Read all reftable files and merge results.
Method _read_table_file Context manager helper to open and read a reftable file.
Method _resolve_symref_chain Resolve a symbolic reference chain to its final target.
Method _update_tables_list Update the tables.list file with current table files.
Method _write_batch_file Write all refs to a single batch file and return created filenames.
Method _write_batch_updates Write multiple ref updates to a single reftable file.
Method _write_ref_update Write a single ref update immediately to its own reftable file.
Method _write_single_ref_update Write a single ref update to its own reftable file like Git does.
Instance Variable _batch_mode Undocumented
Instance Variable _pending_updates Undocumented
Instance Variable _ref_update_indices Undocumented

Inherited from RefsContainer:

Method __contains__ Check if a reference exists.
Method __delitem__ Remove a refname.
Method __iter__ Iterate over all reference keys.
Method __setitem__ Set a reference name to point to the given SHA1.
Method as_dict Return the contents of this container as a dictionary.
Method get_symrefs Get a dict with all symrefs in this container.
Method import_refs Import refs from another repository.
Method keys Refs present in this container.
Method pack_refs Pack loose refs into packed-refs file.
Method read_ref Read a reference without following any references.
Method subkeys Refs present in this container under a base.
Method _check_refname Ensure a refname is valid and lives in refs or is HEAD.
Method _log Undocumented
Instance Variable _logger Undocumented
def __getitem__(self, name: Ref) -> ObjectID:

Get the SHA1 for a reference name.

This method follows all symbolic references.

def __init__(self, path: str | bytes, logger: Callable[[bytes, bytes, bytes, (bytes | None), (int | None), (int | None), bytes], None] | None = None):

Initialize a reftable refs container.

Parameters
path:str | bytesPath to the reftable directory
logger:Callable[[bytes, bytes, bytes, (bytes | None), (int | None), (int | None), bytes], None] | NoneOptional logger for reflog
def add_if_new(self, name: Ref, ref: ObjectID, committer: bytes | None = None, timestamp: int | None = None, timezone: int | None = None, message: bytes | None = None) -> bool:

Add a ref only if it doesn't exist.

def add_packed_refs(self, new_refs: Mapping[Ref, ObjectID | None]):

Add packed refs. Creates a new reftable file with all refs consolidated.

def allkeys(self) -> set[Ref]:

Return set of all ref names.

def batch_update(self) -> _ReftableBatchContext:

Context manager for batching multiple ref updates into a single reftable.

def follow(self, name: Ref) -> tuple[list[Ref], ObjectID | None]:

Follow a reference name.

Returns: a tuple of (refnames, sha), where refnames are the names of
references in the chain
def get_packed_refs(self) -> dict[Ref, ObjectID]:

Get packed refs. Reftable doesn't distinguish packed/loose.

def get_peeled(self, name: Ref) -> ObjectID | None:

Return the cached peeled value of a ref, if available.

Returns: The peeled value of the ref. If the ref is known not point to
a tag, this will be the SHA the ref refers to. If the ref may point to a tag, but no cached information is available, None is returned.
Parameters
name:RefName of the ref to peel
Returns
ObjectID | NoneUndocumented
def read_loose_ref(self, name: Ref) -> bytes:

Read a reference value without following symbolic refs.

Returns: The contents of the ref file.
For symbolic refs, returns b"ref: <target>"
Parameters
name:Refthe refname to read
Returns
bytesUndocumented
Raises
KeyErrorif the ref does not exist
def remove_if_equals(self, name: Ref, old_ref: ObjectID | None, committer: bytes | None = None, timestamp: int | None = None, timezone: int | None = None, message: bytes | None = None) -> bool:

Remove a ref if it equals old_ref.

def remove_packed_ref(self, name: bytes):

Remove a packed ref. Creates a deletion record.

def set_if_equals(self, name: Ref, old_ref: ObjectID | None, new_ref: ObjectID, committer: bytes | None = None, timestamp: int | None = None, timezone: int | None = None, message: bytes | None = None) -> bool:

Atomically set a ref if it currently equals old_ref.

def set_symbolic_ref(self, name: bytes, other: bytes, committer: bytes | None = None, timestamp: int | None = None, timezone: int | None = None, message: bytes | None = None):

Set a symbolic reference.

path =

Undocumented

reftable_dir =

Undocumented

def _apply_batch_updates(self, all_refs: dict[bytes, tuple[int, bytes]], other_updates: list[tuple[bytes, int, bytes]], head_update: tuple[bytes, int, bytes] | None, batch_update_index: int):

Apply batch updates to the refs dict and update indices.

def _compact_tables_list(self, new_table_name: str):

Compact tables list to single file like Git does.

def _ensure_refs_heads_marker(self):

Ensure refs/heads marker file exists for Git compatibility.

Git expects a refs/heads file (not directory) to exist when using reftables. This file contains a marker message indicating that the repository uses the reftable format instead of loose refs.

def _flush_pending_updates(self):

Flush pending ref updates like Git does - consolidate all refs.

def _generate_table_path(self, min_update_index: int | None = None, max_update_index: int | None = None) -> str:

Generate a new reftable file path.

def _get_next_update_index(self) -> int:

Get the next available update index across all tables.

def _get_table_files(self) -> list[str]:

Get sorted list of reftable files from tables.list.

def _load_ref_update_indices(self):

Load the update indices for all refs from existing reftable files.

def _process_pending_updates(self) -> tuple[tuple[bytes, int, bytes] | None, list[tuple[bytes, int, bytes]]]:

Process pending updates and return (head_update, other_updates).

def _read_all_tables(self) -> dict[Ref, tuple[int, bytes]]:

Read all reftable files and merge results.

def _read_table_file(self, table_file: str) -> BinaryIO:

Context manager helper to open and read a reftable file.

def _resolve_symref_chain(self, target: bytes) -> bytes:

Resolve a symbolic reference chain to its final target.

Git limits symref chains to 5 levels to prevent infinite loops. This method follows the chain until it finds a non-symbolic ref or hits the depth limit.

Parameters
target:bytesThe initial target of a symbolic reference
Returns
bytesThe final target after resolving all symbolic references
def _update_tables_list(self):

Update the tables.list file with current table files.

def _write_batch_file(self, all_refs: dict[bytes, tuple[int, bytes]], batch_update_index: int) -> list[str]:

Write all refs to a single batch file and return created filenames.

def _write_batch_updates(self, updates: Mapping[bytes, bytes | None]):

Write multiple ref updates to a single reftable file.

def _write_ref_update(self, name: bytes, value_type: int, value: bytes):

Write a single ref update immediately to its own reftable file.

def _write_single_ref_update(self, name: bytes, value_type: int, value: bytes):

Write a single ref update to its own reftable file like Git does.

_batch_mode: bool =

Undocumented

_pending_updates: list[RefUpdate] =

Undocumented

_ref_update_indices: dict[bytes, int] =

Undocumented