class documentation

Git smart server client.

Class Method from_parsedurl Create an instance of this client from a urlparse.parsed object.
Method __init__ Create a new GitClient instance.
Method archive Retrieve an archive of the specified tree.
Method clone Clone a repository.
Method close Close the client and release any resources.
Method fetch Fetch into a target repository.
Method fetch_pack Retrieve a pack from a git smart server.
Method fetch_with_bundle_uri Fetch into a target repository, using bundle URIs for bootstrap.
Method get_refs Retrieve the current refs from a git smart server.
Method get_url Retrieves full url to given path.
Method send_pack Upload a pack to a remote repository.
Instance Variable protocol_version Undocumented
Static Method _should_send_pack Undocumented
Static Method _warn_filter_objects Undocumented
Method _handle_receive_pack_tail Handle the tail of a 'git-receive-pack' request.
Method _negotiate_receive_pack_capabilities Undocumented
Method _negotiate_upload_pack_capabilities Undocumented
Instance Variable _fetch_capabilities Undocumented
Instance Variable _report_activity Undocumented
Instance Variable _report_status_parser Undocumented
Instance Variable _send_capabilities Undocumented
def from_parsedurl(cls, parsedurl: ParseResult, thin_packs: bool = True, report_activity: Callable[[int, str], None] | None = None, quiet: bool = False, include_tags: bool = False, dumb: bool = False, username: str | None = None, password: str | None = None, config: Config | None = None) -> GitClient:

Create an instance of this client from a urlparse.parsed object.

Parameters
parsedurl:ParseResultResult of urlparse()
thin_packs:boolWhether or not thin packs should be retrieved
report_activity:Callable[[int, str], None] | NoneOptional callback for reporting transport activity
quiet:boolWhether to suppress progress output
include_tags:boolWhether to include tags
dumb:boolWhether to use dumb HTTP transport (only for HTTP)
username:str | NoneOptional username for authentication (only for HTTP)
password:str | NoneOptional password for authentication (only for HTTP)
config:Config | NoneOptional configuration object
Returns
GitClientA GitClient object
def __init__(self, thin_packs: bool = True, report_activity: Callable[[int, str], None] | None = None, quiet: bool = False, include_tags: bool = False):

Create a new GitClient instance.

Parameters
thin_packs:boolWhether or not thin packs should be retrieved
report_activity:Callable[[int, str], None] | NoneOptional callback for reporting transport activity.
quiet:boolWhether to suppress output
include_tags:boolsend annotated tags when sending the objects they point to
def archive(self, path: bytes, committish: bytes, write_data: Callable[[bytes], None], progress: Callable[[bytes], None] | None = None, write_error: Callable[[bytes], None] | None = None, format: bytes | None = None, subdirs: Sequence[bytes] | None = None, prefix: bytes | None = None):

Retrieve an archive of the specified tree.

def clone(self, path: str, target_path: str, mkdir: bool = True, bare: bool = False, origin: str | None = 'origin', checkout: bool | None = None, branch: str | None = None, progress: Callable[[bytes], None] | None = None, depth: int | None = None, ref_prefix: Sequence[bytes] | None = None, filter_spec: bytes | None = None, protocol_version: int | None = None, bundle_uri: str | None = None) -> Repo:

Clone a repository.

Parameters
path:strPath to clone from
target_path:strLocal path to clone to
mkdir:boolWhether to create the target directory
bare:boolWhether to create a bare repository
origin:str | NoneName for the origin remote (default: "origin")
checkout:bool | NoneWhether to checkout HEAD after cloning
branch:str | NoneBranch to checkout (default: remote HEAD)
progress:Callable[[bytes], None] | NoneOptional callback for progress reporting
depth:int | NoneShallow clone depth
ref_prefix:Sequence[bytes] | NoneList of ref prefixes to fetch
filter_spec:bytes | NonePartial clone filter specification
protocol_version:int | NoneGit protocol version to use
bundle_uri:str | NoneOptional bundle URI to bootstrap the clone from. This can be a URL to a bundle file or a bundle list. Using a bundle URI can speed up the clone by downloading pre-computed pack data.
Returns
RepoThe newly created Repo object
def close(self):

Close the client and release any resources.

Default implementation does nothing as most clients don't maintain persistent connections. Subclasses that hold resources should override this method to properly clean them up.

def fetch(self, path: bytes | str, target: BaseRepo, determine_wants: DetermineWantsFunc | None = None, progress: Callable[[bytes], None] | None = None, depth: int | None = None, ref_prefix: Sequence[bytes] | None = None, filter_spec: bytes | None = None, protocol_version: int | None = None, shallow_since: str | None = None, shallow_exclude: list[str] | None = None) -> FetchPackResult:

Fetch into a target repository.

Parameters
path:bytes | strPath to fetch from (as bytestring)
target:BaseRepoTarget repository to fetch into
determine_wants:DetermineWantsFunc | NoneOptional function to determine what refs to fetch. Receives dictionary of name->sha, should return list of shas to fetch. Defaults to all shas.
progress:Callable[[bytes], None] | NoneOptional progress function
depth:int | NoneDepth to fetch at
ref_prefix:Sequence[bytes] | NoneList of prefixes of desired references, as a list of bytestrings. Filtering is done by the server if supported, and client side otherwise.
filter_spec:bytes | NoneA git-rev-list-style object filter spec, as bytestring. Only used if the server supports the Git protocol-v2 'filter' feature, and ignored otherwise.
protocol_version:int | NoneDesired Git protocol version. By default the highest mutually supported protocol version will be used.
shallow_since:str | NoneDeepen the history to include commits after this date
shallow_exclude:list[str] | NoneDeepen the history to exclude commits reachable from these refs
Returns
FetchPackResultDictionary with all remote refs (not just those fetched)
def fetch_pack(self, path: bytes | str, determine_wants: DetermineWantsFunc, graph_walker: GraphWalker, pack_data: Callable[[bytes], int], *, progress: Callable[[bytes], None] | None = None, depth: int | None = None, ref_prefix: Sequence[bytes] | None = None, filter_spec: bytes | None = None, protocol_version: int | None = None, shallow_since: str | None = None, shallow_exclude: list[str] | None = None) -> FetchPackResult:

Retrieve a pack from a git smart server.

Parameters
path:bytes | strRemote path to fetch from
determine_wants:DetermineWantsFuncFunction determine what refs to fetch. Receives dictionary of name->sha, should return list of shas to fetch.
graph_walker:GraphWalkerObject with next() and ack().
pack_data:Callable[[bytes], int]Callback called for each bit of data in the pack
progress:Callable[[bytes], None] | NoneCallback for progress reports (strings)
depth:int | NoneShallow fetch depth
ref_prefix:Sequence[bytes] | NoneList of prefixes of desired references, as a list of bytestrings. Filtering is done by the server if supported, and client side otherwise.
filter_spec:bytes | NoneA git-rev-list-style object filter spec, as bytestring. Only used if the server supports the Git protocol-v2 'filter' feature, and ignored otherwise.
protocol_version:int | NoneDesired Git protocol version. By default the highest mutually supported protocol version will be used.
shallow_since:str | NoneDeepen the history to include commits after this date
shallow_exclude:list[str] | NoneDeepen the history to exclude commits reachable from these refs
Returns
FetchPackResultFetchPackResult object
def fetch_with_bundle_uri(self, path: bytes | str, target: BaseRepo, bundle_uri: str, determine_wants: DetermineWantsFunc | None = None, progress: Callable[[bytes], None] | None = None, depth: int | None = None, ref_prefix: Sequence[bytes] | None = None, filter_spec: bytes | None = None, protocol_version: int | None = None, shallow_since: str | None = None, shallow_exclude: list[str] | None = None, stored_creation_token: int | None = None) -> tuple[FetchPackResult, int | None]:

Fetch into a target repository, using bundle URIs for bootstrap.

This method first applies any available bundles from the bundle URI, then fetches remaining objects from the remote server. This can significantly speed up clones and fetches for large repositories.

If bundle URI fetch fails (e.g., network error, invalid bundle), the error is reported via the progress callback and the method continues with a regular fetch from the remote. The fetch will still succeed as long as the remote fetch completes successfully.

Parameters
path:bytes | strPath to fetch from (as bytestring)
target:BaseRepoTarget repository to fetch into
bundle_uri:strBundle URI to fetch from (URL to bundle or bundle list)
determine_wants:DetermineWantsFunc | NoneOptional function to determine what refs to fetch. Receives dictionary of name->sha, should return list of shas to fetch. Defaults to all shas.
progress:Callable[[bytes], None] | NoneOptional progress function
depth:int | NoneDepth to fetch at
ref_prefix:Sequence[bytes] | NoneList of prefixes of desired references
filter_spec:bytes | NoneA git-rev-list-style object filter spec
protocol_version:int | NoneDesired Git protocol version
shallow_since:str | NoneDeepen the history to include commits after this date
shallow_exclude:list[str] | NoneDeepen the history to exclude commits reachable from these refs
stored_creation_token:int | NonePreviously stored creation token to skip already-applied bundles (for incremental fetches)
Returns
A tuple of (FetchPackResult, creation_token) where
  • FetchPackResult contains refs and other fetch metadata
  • creation_token is the highest creation token seen (for storing and skipping in future fetches), or None if not available or if bundle fetch failed
def get_refs(self, path: bytes, protocol_version: int | None = None, ref_prefix: Sequence[bytes] | None = None) -> LsRemoteResult:

Retrieve the current refs from a git smart server.

Parameters
path:bytesPath to the repo to fetch from. (as bytestring)
protocol_version:int | NoneDesired Git protocol version.
ref_prefix:Sequence[bytes] | NonePrefix filter for refs.
Returns
LsRemoteResultLsRemoteResult object with refs and symrefs
def get_url(self, path: str) -> str:

Retrieves full url to given path.

Parameters
path:strRepository path (as string)
Returns
strUrl to path (as string)
def send_pack(self, path: bytes, update_refs: Callable[[dict[Ref, ObjectID]], dict[Ref, ObjectID]], generate_pack_data: GeneratePackDataFunc, progress: Callable[[bytes], None] | None = None, push_options: Sequence[bytes] | None = None, atomic: bool = False) -> SendPackResult:

Upload a pack to a remote repository.

Parameters
path:bytesRepository path (as bytestring)
update_refs:Callable[[dict[Ref, ObjectID]], dict[Ref, ObjectID]]Function to determine changes to remote refs. Receive dict with existing remote refs, returns dict with changed refs (name -> sha, where sha=ZERO_SHA for deletions)
generate_pack_data:GeneratePackDataFuncFunction that can return a tuple with number of objects and list of pack data to include
progress:Callable[[bytes], None] | NoneOptional progress function
push_options:Sequence[bytes] | NoneOptional list of push options to send to the server
atomic:boolIf True, request atomic push (all refs update or none do)
Returns
SendPackResultSendPackResult object
Raises
SendPackErrorif server rejects the pack data
GitProtocolErrorif atomic push is requested but server doesn't support it
def _should_send_pack(new_refs: Mapping[Ref, ObjectID]) -> bool:

Undocumented

def _warn_filter_objects():

Undocumented

def _handle_receive_pack_tail(self, proto: Protocol, capabilities: Set[bytes], progress: Callable[[bytes], None] | None = None) -> dict[bytes, str | None] | None:

Handle the tail of a 'git-receive-pack' request.

Parameters
proto:ProtocolProtocol object to read from
capabilities:Set[bytes]List of negotiated capabilities
progress:Callable[[bytes], None] | NoneOptional progress reporting function
Returns
dict mapping ref name toerror message if the ref failed to update None if it was updated successfully
def _negotiate_receive_pack_capabilities(self, server_capabilities: set[bytes]) -> tuple[set[bytes], bytes | None]:

Undocumented

def _negotiate_upload_pack_capabilities(self, server_capabilities: set[bytes]) -> tuple[set[bytes], dict[Ref, Ref], bytes | None]:

Undocumented

_fetch_capabilities =

Undocumented

_report_activity =
_send_capabilities =

Undocumented