# Copyright (C) 2008, 2009 Michael Trier ([email protected]) and contributors
#
# This module is part of GitPython and is released under the
# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/
"""Module implementing a remote object allowing easy access to git remotes."""
__all__ = ["RemoteProgress", "PushInfo", "FetchInfo", "Remote"]
import contextlib
import logging
import re
from git.cmd import Git, handle_process_output
from git.compat import defenc, force_text
from git.config import GitConfigParser, SectionConstraint, cp
from git.exc import GitCommandError
from git.refs import Head, Reference, RemoteReference, SymbolicReference, TagReference
from git.util import (
CallableRemoteProgress,
IterableList,
IterableObj,
LazyMixin,
RemoteProgress,
join_path,
)
# typing-------------------------------------------------------
from typing import (
Any,
Callable,
Dict,
Iterator,
List,
NoReturn,
Optional,
Sequence,
TYPE_CHECKING,
Type,
Union,
cast,
overload,
)
from git.types import AnyGitObject, Literal, PathLike
if TYPE_CHECKING:
from git.objects.commit import Commit
from git.objects.submodule.base import UpdateProgress
from git.repo.base import Repo
flagKeyLiteral = Literal[" ", "!", "+", "-", "*", "=", "t", "?"]
# -------------------------------------------------------------
_logger = logging.getLogger(__name__)
# { Utilities
def add_progress(
kwargs: Any,
git: Git,
progress: Union[RemoteProgress, "UpdateProgress", Callable[..., RemoteProgress], None],
) -> Any:
"""Add the ``--progress`` flag to the given `kwargs` dict if supported by the git
command.
:note:
If the actual progress in the given progress instance is not given, we do not
request any progress.
:return:
Possibly altered `kwargs`
"""
if progress is not None:
v = git.version_info[:2]
if v >= (1, 7):
kwargs["progress"] = True
# END handle --progress
# END handle progress
return kwargs
# } END utilities
@overload
def to_progress_instance(progress: None) -> RemoteProgress: ...
@overload
def to_progress_instance(progress: Callable[..., Any]) -> CallableRemoteProgress: ...
@overload
def to_progress_instance(progress: RemoteProgress) -> RemoteProgress: ...
def to_progress_instance(
progress: Union[Callable[..., Any], RemoteProgress, None],
) -> Union[RemoteProgress, CallableRemoteProgress]:
"""Given the `progress` return a suitable object derived from
:class:`~git.util.RemoteProgress`."""
# New API only needs progress as a function.
if callable(progress):
return CallableRemoteProgress(progress)
# Where None is passed create a parser that eats the progress.
elif progress is None:
return RemoteProgress()
# Assume its the old API with an instance of RemoteProgress.
return progress
class PushInfo(IterableObj):
"""
Carries information about the result of a push operation of a single head::
info = remote.push()[0]
info.flags # bitflags providing more information about the result
info.local_ref # Reference pointing to the local reference that was pushed
# It is None if the ref was deleted.
info.remote_ref_string # path to the remote reference located on the remote side
info.remote_ref # Remote Reference on the local side corresponding to
# the remote_ref_string. It can be a TagReference as well.
info.old_commit # commit at which the remote_ref was standing before we pushed
# it to local_ref.commit. Will be None if an error was indicated
info.summary # summary line providing human readable english text about the push
"""
__slots__ = (
"local_ref",
"remote_ref_string",
"flags",
"_old_commit_sha",
"_remote",
"summary",
)
_id_attribute_ = "pushinfo"
(
NEW_TAG,
NEW_HEAD,
NO_MATCH,
REJECTED,
REMOTE_REJECTED,
REMOTE_FAILURE,
DELETED,
FORCED_UPDATE,
FAST_FORWARD,
UP_TO_DATE,
ERROR,
) = [1 << x for x in range(11)]
_flag_map = {
"X": NO_MATCH,
"-": DELETED,
"*": 0,
"+": FORCED_UPDATE,
" ": FAST_FORWARD,
"=": UP_TO_DATE,
"!": ERROR,
}
def __init__(
self,
flags: int,
local_ref: Union[SymbolicReference, None],
remote_ref_string: str,
remote: "Remote",
old_commit: Optional[str] = None,
summary: str = "",
) -> None:
"""Initialize a new instance.
local_ref: HEAD | Head | RemoteReference | TagReference | Reference | SymbolicReference | None
"""
self.flags = flags
self.local_ref = local_ref
self.remote_ref_string = remote_ref_string
self._remote = remote
self._old_commit_sha = old_commit
self.summary = summary
@property
def old_commit(self) -> Union["Commit", None]:
return self._old_commit_sha and self._remote.repo.commit(self._old_commit_sha) or None
@property
def remote_ref(self) -> Union[RemoteReference, TagReference]:
"""
:return:
Remote :class:`~git.refs.reference.Reference` or
:class:`~git.refs.tag.TagReference` in the local repository corresponding to
the :attr:`remote_ref_string` kept in this instance.
"""
# Translate heads to a local remote. Tags stay as they are.
if self.remote_ref_string.startswith("refs/tags"):
return TagReference(self._remote.repo, self.remote_ref_string)
elif self.remote_ref_string.startswith("refs/heads"):
remote_ref = Reference(self._remote.repo, self.remote_ref_string)
return RemoteReference(
self._remote.repo,
"refs/remotes/%s/%s" % (str(self._remote), remote_ref.name),
)
else:
raise ValueError("Could not handle remote ref: %r" % self.remote_ref_string)
# END
@classmethod
def _from_line(cls, remote: "Remote", line: str) -> "PushInfo":
"""Create a new :class:`PushInfo` instance as parsed from line which is expected
to be like refs/heads/master:refs/heads/master 05d2687..1d0568e as bytes."""
control_character, from_to, summary = line.split("\t", 3)
flags = 0
# Control character handling
try:
flags |= cls._flag_map[control_character]
except KeyError as e:
raise ValueError("Control character %r unknown as parsed from line %r" % (control_character, line)) from e
# END handle control character
# from_to handling
from_ref_string, to_ref_string = from_to.split(":")
if flags & cls.DELETED:
from_ref: Union[SymbolicReference, None] = None
else:
if from_ref_string == "(delete)":
from_ref = None
else:
from_ref = Reference.from_path(remote.repo, from_ref_string)
# Commit handling, could be message or commit info
old_commit: Optional[str] = None
if summary.startswith("["):
if "[rejected]" in summary:
flags |= cls.REJECTED
elif "[remote rejected]" in summary:
flags |= cls.REMOTE_REJECTED
elif "[remote failure]" in summary:
flags |= cls.REMOTE_FAILURE
elif "[no match]" in summary:
flags |= cls.ERROR
elif "[new tag]" in summary:
flags |= cls.NEW_TAG
elif "[new branch]" in summary:
flags |= cls.NEW_HEAD
# `uptodate` encoded in control character
else:
# Fast-forward or forced update - was encoded in control character,
# but we parse the old and new commit.
split_token = "..."
if control_character == " ":
split_token = ".."
old_sha, _new_sha = summary.split(" ")[0].split(split_token)
# Have to use constructor here as the sha usually is abbreviated.
old_commit = old_sha
# END message handling
return PushInfo(flags, from_ref, to_ref_string, remote, old_commit, summary)
@classmethod
def iter_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> NoReturn: # -> Iterator['PushInfo']:
raise NotImplementedError
class PushInfoList(IterableList[PushInfo]):
""":class:`~git.util.IterableList` of :class:`PushInfo` objects."""
def __new__(cls) -> "PushInfoList":
return cast(PushInfoList, IterableList.__new__(cls, "push_infos"))
def __init__(self) -> None:
super().__init__("push_infos")
self.error: Optional[Exception] = None
def raise_if_error(self) -> None:
"""Raise an exception if any ref failed to push."""
if self.error:
raise self.error
class FetchInfo(IterableObj):
"""
Carries information about the results of a fetch operation of a single head::
info = remote.fetch()[0]
info.ref # Symbolic Reference or RemoteReference to the changed
# remote head or FETCH_HEAD
info.flags # additional flags to be & with enumeration members,
# i.e. info.flags & info.REJECTED
# is 0 if ref is SymbolicReference
info.note # additional notes given by git-fetch intended for the user
info.old_commit # if info.flags & info.FORCED_UPDATE|info.FAST_FORWARD,
# field is set to the previous location of ref, otherwise None
info.remote_ref_path # The path from which we fetched on the remote. It's the remote's version of our info.ref
"""
__slots__ = ("ref", "old_commit", "flags", "note", "remote_ref_path")
_id_attribute_ = "fetchinfo"
(
NEW_TAG,
NEW_HEAD,
HEAD_UPTODATE,
TAG_UPDATE,
REJECTED,
FORCED_UPDATE,
FAST_FORWARD,
ERROR,
) = [1 << x for x in range(8)]
_re_fetch_result = re.compile(r"^ *(?:.{0,3})(.) (\[[\w \.$@]+\]|[\w\.$@]+) +(.+) -> ([^ ]+)( \(.*\)?$)?")
_flag_map: Dict[flagKeyLiteral, int] = {
"!": ERROR,
"+": FORCED_UPDATE,
"*": 0,
"=": HEAD_UPTODATE,
" ": FAST_FORWARD,
"-": TAG_UPDATE,
}
@classmethod
def refresh(cls) -> Literal[True]:
"""Update information about which :manpage:`git-fetch(1)` flags are supported
by the git executable being used.
Called by the :func:`git.refresh` function in the top level ``__init__``.
"""
# Clear the old values in _flag_map.
with contextlib.suppress(KeyError):
del cls._flag_map["t"]
with contextlib.suppress(KeyError):
del cls._flag_map["-"]
# Set the value given the git version.
if Git().version_info[:2] >= (2, 10):
cls._flag_map["t"] = cls.TAG_UPDATE
else:
cls._flag_map["-"] = cls.TAG_UPDATE
return True
def __init__(
self,
ref: SymbolicReference,
flags: int,
note: str = "",
old_commit: Union[AnyGitObject, None] = None,
remote_ref_path: Optional[PathLike] = None,
) -> None:
"""Initialize a new instance."""
self.ref = ref
self.flags = flags
self.note = note
self.old_commit = old_commit
self.remote_ref_path = remote_ref_path
def __str__(self) -> str:
return self.name
@property
def name(self) -> str:
""":return: Name of our remote ref"""
return self.ref.name
@property
def commit(self) -> "Commit":
""":return: Commit of our remote ref"""
return self.ref.commit
@classmethod
def _from_line(cls, repo: "Repo", line: str, fetch_line: str) -> "FetchInfo":
"""Parse information from the given line as returned by ``git-fetch -v`` and
return a new :class:`FetchInfo` object representing this information.
We can handle a line as follows::
%c %-*s %-*s -> %s%s
Where ``c`` is either a space, ``!``, ``+``, ``-``, ``*``, or ``=``:
- '!' means error
- '+' means success forcing update
- '-' means a tag was updated
- '*' means birth of new branch or tag
- '=' means the head was up to date (and not moved)
- ' ' means a fast-forward
`fetch_line` is the corresponding line from FETCH_HEAD, like::
acb0fa8b94ef421ad60c8507b634759a472cd56c not-for-merge branch '0.1.7RC' of /tmp/tmpya0vairemote_repo
"""
match = cls._re_fetch_result.match(line)
if match is None:
raise ValueError("Failed to parse line: %r" % line)
# Parse lines.
remote_local_ref_str: str
(
control_character,
operation,
local_remote_ref,
remote_local_ref_str,
note,
) = match.groups()
control_character = cast(flagKeyLiteral, control_character)
try:
_new_hex_sha, _fetch_operation, fetch_note = fetch_line.split("\t")
ref_type_name, fetch_note = fetch_note.split(" ", 1)
except ValueError as e: # unpack error
raise ValueError("Failed to parse FETCH_HEAD line: %r" % fetch_line) from e
# Parse flags from control_character.
flags = 0
try:
flags |= cls._flag_map[control_character]
except KeyError as e:
raise ValueError("Control character %r unknown as parsed from line %r" % (control_character, line)) from e
# END control char exception handling
# Parse operation string for more info.
# This makes no sense for symbolic refs, but we parse it anyway.
old_commit: Union[AnyGitObject, None] = None
is_tag_operation = False
if "rejected" in operation:
flags |= cls.REJECTED
if "new tag" in operation:
flags |= cls.NEW_TAG
is_tag_operation = True
if "tag update" in operation:
flags |= cls.TAG_UPDATE
is_tag_operation = True
if "new branch" in operation:
flags |= cls.NEW_HEAD
if "..." in operation or ".." in operation:
split_token = "..."
if control_character == " ":
split_token = split_token[:-1]
old_commit = repo.rev_parse(operation.split(split_token)[0])
# END handle refspec
# Handle FETCH_HEAD and figure out ref type.
# If we do not specify a target branch like master:refs/remotes/origin/master,
# the fetch result is stored in FETCH_HEAD which destroys the rule we usually
# have. In that case we use a symbolic reference which is detached.
ref_type: Optional[Type[SymbolicReference]] = None
if remote_local_ref_str == "FETCH_HEAD":
ref_type = SymbolicReference
elif ref_type_name == "tag" or is_tag_operation:
# The ref_type_name can be branch, whereas we are still seeing a tag
# operation. It happens during testing, which is based on actual git
# operations.
ref_type = TagReference
elif ref_type_name in ("remote-tracking", "branch"):
# Note: remote-tracking is just the first part of the
# 'remote-tracking branch' token. We don't parse it correctly, but it's
# enough to know what to do, and it's new in git 1.7something.
ref_type = RemoteReference
elif "/" in ref_type_name:
# If the fetch spec look something like '+refs/pull/*:refs/heads/pull/*',
# and is thus pretty much anything the user wants, we will have trouble
# determining what's going on. For now, we assume the local ref is a Head.
ref_type = Head
else:
raise TypeError("Cannot handle reference type: %r" % ref_type_name)
# END handle ref type
# Create ref instance.
if ref_type is SymbolicReference:
remote_local_ref = ref_type(repo, "FETCH_HEAD")
else:
# Determine prefix. Tags are usually pulled into refs/tags; they may have
# subdirectories. It is not clear sometimes where exactly the item is,
# unless we have an absolute path as indicated by the 'ref/' prefix.
# Otherwise even a tag could be in refs/remotes, which is when it will have
# the 'tags/' subdirectory in its path. We don't want to test for actual
# existence, but try to figure everything out analytically.
ref_path: Optional[PathLike] = None
remote_local_ref_str = remote_local_ref_str.strip()
if remote_local_ref_str.startswith(Reference._common_path_default + "/"):
# Always use actual type if we get absolute paths. This will always be
# the case if something is fetched outside of refs/remotes (if its not a
# tag).
ref_path = remote_local_ref_str
if ref_type is not TagReference and not remote_local_ref_str.startswith(
RemoteReference._common_path_default + "/"
):
ref_type = Reference
# END downgrade remote reference
elif ref_type is TagReference and "tags/" in remote_local_ref_str:
# Even though it's a tag, it is located in refs/remotes.
ref_path = join_path(RemoteReference._common_path_default, remote_local_ref_str)
else:
ref_path = join_path(ref_type._common_path_default, remote_local_ref_str)
# END obtain refpath
# Even though the path could be within the git conventions, we make sure we
# respect whatever the user wanted, and disabled path checking.
remote_local_ref = ref_type(repo, ref_path, check_path=False)
# END create ref instance
note = (note and note.strip()) or ""
return cls(remote_local_ref, flags, note, old_commit, local_remote_ref)
@classmethod
def iter_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> NoReturn: # -> Iterator['FetchInfo']:
raise NotImplementedError
class Remote(LazyMixin, IterableObj):
"""Provides easy read and write access to a git remote.
Everything not part of this interface is considered an option for the current
remote, allowing constructs like ``remote.pushurl`` to query the pushurl.
:note:
When querying configuration, the configuration accessor will be cached to speed
up subsequent accesses.
"""
__slots__ = ("repo", "name", "_config_reader")
_id_attribute_ = "name"
unsafe_git_fetch_options = [
# This option allows users to execute arbitrary commands.
# https://git-scm.com/docs/git-fetch#Documentation/git-fetch.txt---upload-packltupload-packgt
"--upload-pack",
]
unsafe_git_pull_options = [
# This option allows users to execute arbitrary commands.
# https://git-scm.com/docs/git-pull#Documentation/git-pull.txt---upload-packltupload-packgt
"--upload-pack"
]
unsafe_git_push_options = [
# This option allows users to execute arbitrary commands.
# https://git-scm.com/docs/git-push#Documentation/git-push.txt---execltgit-receive-packgt
"--receive-pack",
"--exec",
]
url: str # Obtained dynamically from _config_reader. See __getattr__ below.
"""The URL configured for the remote."""
def __init__(self, repo: "Repo", name: str) -> None:
"""Initialize a remote instance.
:param repo:
The repository we are a remote of.
:param name:
The name of the remote, e.g. ``origin``.
"""
self.repo = repo
self.name = name
def __getattr__(self, attr: str) -> Any:
"""Allows to call this instance like ``remote.special(*args, **kwargs)`` to
call ``git remote special self.name``."""
if attr == "_config_reader":
return super().__getattr__(attr)
# Sometimes, probably due to a bug in Python itself, we are being called even
# though a slot of the same name exists.
try:
return self._config_reader.get(attr)
except cp.NoOptionError:
return super().__getattr__(attr)
# END handle exception
def _config_section_name(self) -> str:
return 'remote "%s"' % self.name
def _set_cache_(self, attr: str) -> None:
if attr == "_config_reader":
# NOTE: This is cached as __getattr__ is overridden to return remote config
# values implicitly, such as in print(r.pushurl).
self._config_reader = SectionConstraint(
self.repo.config_reader("repository"),
self._config_section_name(),
)
else:
super()._set_cache_(attr)
def __str__(self) -> str:
return self.name
def __repr__(self) -> str:
return '