dulwich.client

module documentation

Client side support for the Git protocol.

The Dulwich client supports the following capabilities:

thin-pack

multi_ack_detailed

multi_ack

side-band-64k

ofs-delta

quiet

report-status

delete-refs

shallow

Known capabilities that are not supported:

no-progress

include-tag

Class	`AbstractHttpGitClient`	Abstract base class for HTTP Git Clients.
Class	`AuthCallbackPoolManager`	Pool manager wrapper that handles authentication callbacks.
Class	`BundleClient`	Git Client that reads from a bundle file.
Class	`DetermineWantsFunc`	Protocol for determine_wants functions.
Class	`FetchPackResult`	Result of a fetch-pack operation.
Class	`GeneratePackDataFunc`	Protocol for generate_pack_data functions.
Class	`GitClient`	Git smart server client.
Class	`HTTPResponse`	Protocol for HTTP response objects (matches urllib3.response.HTTPResponse).
Class	`LocalGitClient`	Git Client that just uses a local on-disk repository.
Class	`LsRemoteResult`	Result of a ls-remote operation.
Class	`PackDataProgressWrapper`	Wrapper that reports progress during pack data reception.
Class	`PLinkSSHVendor`	SSH vendor that shells out to the local 'plink' command.
Class	`ReportStatusParser`	Handle status as reported by servers with 'report-status' capability.
Class	`SendPackResult`	Result of a upload-pack operation.
Class	`SSHGitClient`	Git client that connects over SSH.
Class	`SSHVendor`	A client side SSH implementation.
Class	`SubprocessGitClient`	Git client that talks to a server using a subprocess.
Class	`SubprocessSSHVendor`	SSH vendor that shells out to the local 'ssh' command.
Class	`SubprocessWrapper`	A socket-like object that talks to a subprocess via pipes.
Class	`TCPGitClient`	A Git Client that works over TCP directly (i.e. git://).
Class	`TraditionalGitClient`	Traditional Git client.
Class	`Urllib3HttpGitClient`	HTTP Git client using urllib3.
Exception	`HTTPProxyUnauthorized`	Raised when proxy authentication fails.
Exception	`HTTPUnauthorized`	Raised when authentication fails.
Exception	`InvalidWants`	Invalid wants.
Exception	`StrangeHostname`	Refusing to connect to strange SSH hostname.
Function	`apply_bundle_uri`	Apply bundles from a bundle URI to a repository.
Function	`build_fetch_request_v2`	Build fetch command packet list for protocol v2 (before delimiter).
Function	`build_ls_refs_request_v2`	Build ls-refs command packet lists for protocol v2.
Function	`check_for_proxy_bypass`	Check if proxy should be bypassed for the given URL.
Function	`check_wants`	Check that a set of wants is valid.
Function	`default_urllib3_manager`	Return urllib3 connection pool manager.
Function	`default_user_agent_string`	Return the default user agent string for Dulwich.
Function	`extract_object_format_from_capabilities`	Extract object format from server capabilities.
Function	`find_capability`	Find a capability with a specific key and value.
Function	`find_git_command`	Find command to run for system Git (usually C Git).
Function	`get_credentials_from_store`	Read credentials from a Git credential store.
Function	`get_transport_and_path`	Obtain a git client from a URL.
Function	`get_transport_and_path_from_url`	Obtain a git client from a URL.
Function	`negotiate_protocol_version`	Negotiate protocol version with the server.
Function	`parse_rsync_url`	Parse a rsync-style URL.
Function	`read_pkt_refs_v1`	Read references using protocol version 1.
Function	`read_pkt_refs_v2`	Read references using protocol version 2.
Function	`read_server_capabilities`	Read server capabilities from packet sequence.
Constant	`COMMON_CAPABILITIES`	Undocumented
Constant	`DEFAULT_GIT_CREDENTIALS_PATHS`	Undocumented
Constant	`DEFAULT_REF_PREFIX`	Undocumented
Constant	`RECEIVE_CAPABILITIES`	Undocumented
Constant	`UPLOAD_CAPABILITIES`	Undocumented
Variable	`logger`	Undocumented
Class	`_DeprecatedDictProxy`	Base class for result objects that provide deprecated dict-like interface.
Class	`_v1ReceivePackHeader`	No class docstring; 0/4 instance variable, 1/3 method documented
Function	`_download_packfile_from_uri`	Download a packfile from a URI and verify its hash.
Function	`_extract_symrefs_and_agent`	Extract symrefs and agent from capabilities.
Function	`_fileno_can_read`	Check if a file descriptor is readable.
Function	`_get_transport_and_path_from_url`	Undocumented
Function	`_handle_upload_pack_head`	Handle the head of a 'git-upload-pack' request.
Function	`_handle_upload_pack_tail`	Handle the tail of a 'git-upload-pack' request.
Function	`_read_shallow_updates`	Undocumented
Function	`_read_side_band64k_data`	Read per-channel data.
Function	`_remote_error_from_stderr`	Undocumented
Function	`_to_optional_dict`	Convert a dict[Ref, ObjectID] to dict[Ref, Optional[ObjectID]].
Function	`_urlmatch_http_sections`	Yield http config sections matching the given URL, ordered by specificity.
Function	`_win32_peek_avail`	Wrapper around PeekNamedPipe to check how many bytes are available.
Function	`_win32_url_to_path`	Convert a file: URL to a path.
Function	`_wrap_urllib3_exceptions`	Undocumented

def apply_bundle_uri(repo: BaseRepo, uri: str, filter_spec: str | None = None, stored_creation_token: int | None = None, progress: Callable[[bytes], None] | None = None) -> tuple[int | None, dict[bytes, bytes]]: ¶

Apply bundles from a bundle URI to a repository.

This function fetches bundles from the given URI and applies them to the repository, potentially speeding up subsequent fetches.

Parameters
repo:`BaseRepo`	The target repository to apply bundles to
uri:`str`	The bundle URI to fetch from
filter_spec:`str \| None`	Object filter to match (e.g., "blob:none")
stored_creation_token:`int \| None`	Previously stored creation token to skip bundles that have already been applied
progress:`Callable[[bytes], None] \| None`	Optional callback for progress reporting

Returns
`A tuple of (creation_token, refs) where`	`creation_token` is the highest creation token seen (for storing and skipping in future fetches), or None if not available `refs` is a dict mapping ref names to object IDs from the bundles

Raises
`BundleURIError`	If fetching or applying bundles fails

def build_fetch_request_v2(object_format: str | None) -> list[bytes]: ¶

Build fetch command packet list for protocol v2 (before delimiter).

Parameters
object_format:`str \| None`	Object format to use (e.g., "sha1", "sha256"), or None

Returns
`list[bytes]`	List of command packets to send before the delimiter

def build_ls_refs_request_v2(server_capabilities: set[bytes], object_format: str | None, ref_prefix: Sequence[bytes] | None = None) -> tuple[list[bytes], list[bytes]]: ¶

Build ls-refs command packet lists for protocol v2.

Parameters
server_capabilities:`set[bytes]`	Capabilities advertised by the server
object_format:`str \| None`	Object format to use (e.g., "sha1", "sha256"), or None
ref_prefix:`Sequence[bytes] \| None`	List of ref prefixes to request, or None for default

Returns
`tuple[list[bytes], list[bytes]]`	Tuple of (command packets before delimiter, argument packets after delimiter)

def check_for_proxy_bypass(base_url: str | None) -> bool: ¶

Check if proxy should be bypassed for the given URL.

def check_wants(wants: Set[bytes], refs: Mapping[bytes, bytes]): ¶

Check that a set of wants is valid.

Parameters
wants:`Set[bytes]`	Set of object SHAs to fetch
refs:`Mapping[bytes, bytes]`	Refs dictionary to check against

Return urllib3 connection pool manager.

Honour detected proxy configurations.

Parameters
config:`Config \| None`	`dulwich.config.ConfigDict` instance with Git configuration.
pool_manager_cls:`type \| None`	Pool manager class to use
proxy_manager_cls:`type \| None`	Proxy manager class to use
base_url:`str \| None`	Base URL for proxy bypass checks
timeout:`float \| None`	Timeout for HTTP requests in seconds
cert_reqs:`str \| None`	SSL certificate requirements (e.g. "CERT_REQUIRED", "CERT_NONE")
auth_callback:`Callable[[str, str, int], dict[str, str] \| None] \| None`	Optional callback for HTTP authentication
proxy_auth_callback:`Callable[[str, str, int], dict[str, str] \| None] \| None`	Optional callback for proxy authentication

Returns
`urllib3.ProxyManager \| urllib3.PoolManager \| AuthCallbackPoolManager`	Either pool_manager_cls (defaults to `urllib3.ProxyManager`) instance for proxy configurations, proxy_manager_cls (defaults to `urllib3.PoolManager`) instance otherwise. If auth callbacks are provided, returns an AuthCallbackPoolManager wrapper.

def default_user_agent_string() -> str: ¶

Return the default user agent string for Dulwich.

def extract_object_format_from_capabilities(capabilities: set[bytes]) -> str | None: ¶

Extract object format from server capabilities.

Parameters
capabilities:`set[bytes]`	Server capabilities

Returns
`str \| None`	Object format name as string (e.g., "sha1", "sha256"), or None if not specified

def find_capability(capabilities: Iterable[bytes], key: bytes, value: bytes | None) -> bytes | None: ¶

Find a capability with a specific key and value.

def find_git_command() -> list[str]: ¶

Find command to run for system Git (usually C Git).

def get_credentials_from_store(scheme: str, hostname: str, username: str | None = None, fnames: list[str] = DEFAULT_GIT_CREDENTIALS_PATHS) -> Iterator[tuple[str, str]]: ¶

Read credentials from a Git credential store.

def get_transport_and_path(location: str, config: Config | None = None, operation: str | None = None, thin_packs: bool = True, report_activity: Callable[[int, str], None] | None = None, quiet: bool = False, include_tags: bool = False, username: str | None = None, password: str | None = None, key_filename: str | None = None, ssh_command: str | None = None, pool_manager: urllib3.PoolManager | None = None) -> tuple[GitClient, str]: ¶

Obtain a git client from a URL.

Parameters
location:`str`	URL or path (a string)
config:`Config \| None`	Optional config object
operation:`str \| None`	Kind of operation that'll be performed; "pull" or "push"
thin_packs:`bool`	Whether or not thin packs should be retrieved
report_activity:`Callable[[int, str], None] \| None`	Optional callback for reporting transport activity
quiet:`bool`	Whether to suppress output
include_tags:`bool`	Send annotated tags when sending the objects they point to
username:`str \| None`	Optional username for authentication
password:`str \| None`	Optional password for authentication
key_filename:`str \| None`	Optional SSH key file
ssh_command:`str \| None`	Optional custom SSH command
pool_manager:`urllib3.PoolManager \| None`	Optional urllib3 PoolManager for HTTP(S) connections

Returns
`tuple[GitClient, str]`	Tuple with client instance and relative path.

def get_transport_and_path_from_url(url: str, config: Config | None = None, operation: str | None = None, thin_packs: bool = True, report_activity: Callable[[int, str], None] | None = None, quiet: bool = False, include_tags: bool = False, username: str | None = None, password: str | None = None, key_filename: str | None = None, ssh_command: str | None = None, pool_manager: urllib3.PoolManager | None = None) -> tuple[GitClient, str]: ¶

Obtain a git client from a URL.

Parameters
url:`str`	URL to open (a unicode string)
config:`Config \| None`	Optional config object
operation:`str \| None`	Kind of operation that'll be performed; "pull" or "push"
thin_packs:`bool`	Whether or not thin packs should be retrieved
report_activity:`Callable[[int, str], None] \| None`	Optional callback for reporting transport activity
quiet:`bool`	Whether to suppress output
include_tags:`bool`	Send annotated tags when sending the objects they point to
username:`str \| None`	Optional username for authentication
password:`str \| None`	Optional password for authentication
key_filename:`str \| None`	Optional SSH key file
ssh_command:`str \| None`	Optional custom SSH command
pool_manager:`urllib3.PoolManager \| None`	Optional urllib3 PoolManager for HTTP(S) connections

Returns
`tuple[GitClient, str]`	Tuple with client instance and relative path.

def negotiate_protocol_version(proto: Protocol) -> int: ¶

Negotiate protocol version with the server.

def parse_rsync_url(location: str) -> tuple[str | None, str, str]: ¶

Parse a rsync-style URL.

def read_pkt_refs_v1(pkt_seq: Iterable[bytes]) -> tuple[dict[Ref, ObjectID], set[bytes]]: ¶

Read references using protocol version 1.

def read_pkt_refs_v2(pkt_seq: Iterable[bytes]) -> tuple[dict[Ref, ObjectID | None], dict[Ref, Ref], dict[Ref, ObjectID]]: ¶

Read references using protocol version 2.

def read_server_capabilities(pkt_seq: Iterable[bytes]) -> set[bytes]: ¶

Read server capabilities from packet sequence.

COMMON_CAPABILITIES = ¶

Undocumented

Value

[CAPABILITY_OFS_DELTA, CAPABILITY_SIDE_BAND_64K]

DEFAULT_GIT_CREDENTIALS_PATHS = ¶

Undocumented

Value

[os.path.expanduser('~/.git-credentials'),
 get_xdg_config_home_path('git', 'credentials')]

DEFAULT_REF_PREFIX: list[bytes] = ¶

Undocumented

Value

[b'HEAD', b'refs/']

RECEIVE_CAPABILITIES = ¶

Undocumented

Value

[CAPABILITY_REPORT_STATUS,
 CAPABILITY_DELETE_REFS,
 CAPABILITY_ATOMIC,
 CAPABILITY_PUSH_OPTIONS,
 *COMMON_CAPABILITIES]

UPLOAD_CAPABILITIES = ¶

Undocumented

Value

[CAPABILITY_THIN_PACK,
 CAPABILITY_MULTI_ACK,
 CAPABILITY_MULTI_ACK_DETAILED,
 CAPABILITY_SHALLOW,
 *COMMON_CAPABILITIES]

logger = ¶

Undocumented

def _download_packfile_from_uri(uri: str, expected_hash: bytes, hash_algo: str, pack_data: Callable[[bytes], int], progress: Callable[[bytes], None] | None, http_request: Callable[[str], tuple[HTTPResponse, Callable[[int], bytes]]]): ¶

Download a packfile from a URI and verify its hash.

This function downloads data, verifies the hash matches expected_hash, and only then writes data to the repository. This prevents corrupted or malicious data from being written.

Parameters
uri:`str`	URI to download packfile from
expected_hash:`bytes`	Expected hash of the packfile
hash_algo:`str`	Hash algorithm to use (e.g., 'sha1', 'sha256')
pack_data:`Callable[[bytes], int]`	Callback to send pack data to
progress:`Callable[[bytes], None] \| None`	Optional progress callback
http_request:`Callable[[str], tuple[HTTPResponse, Callable[[int], bytes]]]`	Function to perform HTTP requests

Raises
`GitProtocolError`	If URI scheme is not HTTPS, hash doesn't match, or download fails

def _extract_symrefs_and_agent(capabilities: Iterable[bytes]) -> tuple[dict[Ref, Ref], bytes | None]: ¶

Extract symrefs and agent from capabilities.

Parameters
capabilities:`Iterable[bytes]`	List of capabilities

Returns
`tuple[dict[Ref, Ref], bytes \| None]`	(symrefs, agent) tuple

def _fileno_can_read(fileno: int) -> bool: ¶

Check if a file descriptor is readable.

def _get_transport_and_path_from_url(url: str, config: Config | None, operation: str | None, thin_packs: bool = True, report_activity: Callable[[int, str], None] | None = None, quiet: bool = False, include_tags: bool = False, username: str | None = None, password: str | None = None, key_filename: str | None = None, ssh_command: str | None = None, pool_manager: urllib3.PoolManager | None = None) -> tuple[GitClient, str]: ¶

Undocumented

def _handle_upload_pack_head(proto: Protocol, capabilities: Iterable[bytes], graph_walker: GraphWalker, wants: list[ObjectID], can_read: Callable[[], bool] | None, depth: int | None, protocol_version: int | None, shallow_since: str | None = None, shallow_exclude: list[str] | None = None) -> tuple[set[ObjectID] | None, set[ObjectID] | None]: ¶

Handle the head of a 'git-upload-pack' request.

whether there is extra graph data to read on proto: depth: Depth for request protocol_version: Neogiated Git protocol version. shallow_since: Deepen the history to include commits after this date shallow_exclude: Deepen the history to exclude commits reachable from these refs

Parameters
proto:`Protocol`	Protocol object to read from
capabilities:`Iterable[bytes]`	List of negotiated capabilities
graph_walker:`GraphWalker`	GraphWalker instance to call .ack() on
wants:`list[ObjectID]`	List of commits to fetch
can_read:`Callable[[], bool] \| None`	function that returns a boolean that indicates
depth:`int \| None`	Undocumented
protocol_version:`int \| None`	Undocumented
shallow_since:`str \| None`	Undocumented
shallow_exclude:`list[str] \| None`	Undocumented

Returns
`tuple[set[ObjectID] \| None, set[ObjectID] \| None]`	Undocumented

def _handle_upload_pack_tail(proto: Protocol, capabilities: Set[bytes], graph_walker: GraphWalker, pack_data: Callable[[bytes], int], progress: Callable[[bytes], None] | None = None, rbufsize: int = _RBUFSIZE, protocol_version: int = 0, http_request: Callable[[str], tuple[HTTPResponse, Callable[[int], bytes]]] | None = None): ¶

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

Parameters
proto:`Protocol`	Protocol object to read from
capabilities:`Set[bytes]`	List of negotiated capabilities
graph_walker:`GraphWalker`	GraphWalker instance to call .ack() on
pack_data:`Callable[[bytes], int]`	Function to call with pack data
progress:`Callable[[bytes], None] \| None`	Optional progress reporting function
rbufsize:`int`	Read buffer size
protocol_version:`int`	Neogiated Git protocol version.
http_request:`Callable[[str], tuple[HTTPResponse, Callable[[int], bytes]]] \| None`	Optional HTTP request function for downloading packfile URIs

def _read_shallow_updates(pkt_seq: Iterable[bytes]) -> tuple[set[ObjectID], set[ObjectID]]: ¶

Undocumented

def _read_side_band64k_data(pkt_seq: Iterable[bytes]) -> Iterator[tuple[int, bytes]]: ¶

Read per-channel data.

This requires the side-band-64k capability.

Parameters
pkt_seq:`Iterable[bytes]`	Sequence of packets to read

Returns
`Iterator[tuple[int, bytes]]`	Undocumented

def _remote_error_from_stderr(stderr: IO[bytes] | None) -> Exception: ¶

Undocumented

def _to_optional_dict(refs: Mapping[Ref, ObjectID]) -> dict[Ref, ObjectID | None]: ¶

Convert a dict[Ref, ObjectID] to dict[Ref, Optional[ObjectID]].

This is needed for compatibility with result types that expect Optional values.

def _urlmatch_http_sections(config: Config, url: str | None) -> Iterator[tuple[bytes, ...]]: ¶

Yield http config sections matching the given URL, ordered by specificity.

Yields sections from least specific to most specific, so callers can apply settings in order with more specific settings overriding less specific ones.

Parameters
config:`Config`	Git configuration object
url:`str \| None`	URL to match against config sections (if None, only yields global http section)

Returns
`Iterator[tuple[bytes, ...]]`	Undocumented

Yields
Config section tuples that match the URL, ordered by specificity

def _win32_peek_avail(handle: int) -> int: ¶

Wrapper around PeekNamedPipe to check how many bytes are available.

def _win32_url_to_path(parsed: ParseResult) -> str: ¶

Convert a file: URL to a path.

https://datatracker.ietf.org/doc/html/rfc8089

def _wrap_urllib3_exceptions(func: Callable[..., bytes]) -> Callable[..., bytes]: ¶

Undocumented