summaryrefslogtreecommitdiff
path: root/git/remote.py
diff options
context:
space:
mode:
authorSebastian Thiel <sebastian.thiel@icloud.com>2022-05-18 07:43:53 +0800
committerSebastian Thiel <sebastian.thiel@icloud.com>2022-05-18 07:43:53 +0800
commit21ec529987d10e0010badd37f8da3274167d436f (patch)
treea3394cfe902ce7edd07c89420c21c13274a2d295 /git/remote.py
parentb30720ee4d9762a03eae4fa7cfa4b0190d81784d (diff)
downloadgitpython-21ec529987d10e0010badd37f8da3274167d436f.tar.gz
Run everything through 'black'
That way people who use it won't be deterred, while it unifies style everywhere.
Diffstat (limited to 'git/remote.py')
-rw-r--r--git/remote.py538
1 files changed, 344 insertions, 194 deletions
diff --git a/git/remote.py b/git/remote.py
index 56f3c5b3..8cd79057 100644
--- a/git/remote.py
+++ b/git/remote.py
@@ -9,7 +9,7 @@ import logging
import re
from git.cmd import handle_process_output, Git
-from git.compat import (defenc, force_text)
+from git.compat import defenc, force_text
from git.exc import GitCommandError
from git.util import (
LazyMixin,
@@ -27,28 +27,36 @@ from git.config import (
SectionConstraint,
cp,
)
-from git.refs import (
- Head,
- Reference,
- RemoteReference,
- SymbolicReference,
- TagReference
-)
+from git.refs import Head, Reference, RemoteReference, SymbolicReference, TagReference
# typing-------------------------------------------------------
-from typing import (Any, Callable, Dict, Iterator, List, NoReturn, Optional, Sequence,
- TYPE_CHECKING, Type, Union, cast, overload)
+from typing import (
+ Any,
+ Callable,
+ Dict,
+ Iterator,
+ List,
+ NoReturn,
+ Optional,
+ Sequence,
+ TYPE_CHECKING,
+ Type,
+ Union,
+ cast,
+ overload,
+)
from git.types import PathLike, Literal, Commit_ish
if TYPE_CHECKING:
from git.repo.base import Repo
from git.objects.submodule.base import UpdateProgress
+
# from git.objects.commit import Commit
# from git.objects import Blob, Tree, TagObject
-flagKeyLiteral = Literal[' ', '!', '+', '-', '*', '=', 't', '?']
+flagKeyLiteral = Literal[" ", "!", "+", "-", "*", "=", "t", "?"]
# def is_flagKeyLiteral(inp: str) -> TypeGuard[flagKeyLiteral]:
# return inp in [' ', '!', '+', '-', '=', '*', 't', '?']
@@ -57,18 +65,22 @@ flagKeyLiteral = Literal[' ', '!', '+', '-', '*', '=', 't', '?']
# -------------------------------------------------------------
-log = logging.getLogger('git.remote')
+log = logging.getLogger("git.remote")
log.addHandler(logging.NullHandler())
-__all__ = ('RemoteProgress', 'PushInfo', 'FetchInfo', 'Remote')
+__all__ = ("RemoteProgress", "PushInfo", "FetchInfo", "Remote")
-#{ Utilities
+# { Utilities
-def add_progress(kwargs: Any, git: Git,
- progress: Union[RemoteProgress, 'UpdateProgress', Callable[..., RemoteProgress], None]
- ) -> Any:
+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. If the actual progress in the given progress instance is not
given, we do not request any progress
@@ -76,31 +88,33 @@ def add_progress(kwargs: Any, git: Git,
if progress is not None:
v = git.version_info[:2]
if v >= (1, 7):
- kwargs['progress'] = True
+ kwargs["progress"] = True
# END handle --progress
# END handle progress
return kwargs
-#} END utilities
+
+# } END utilities
-@ overload
+@overload
def to_progress_instance(progress: None) -> RemoteProgress:
...
-@ overload
+@overload
def to_progress_instance(progress: Callable[..., Any]) -> CallableRemoteProgress:
...
-@ overload
+@overload
def to_progress_instance(progress: RemoteProgress) -> RemoteProgress:
...
-def to_progress_instance(progress: Union[Callable[..., Any], RemoteProgress, None]
- ) -> Union[RemoteProgress, CallableRemoteProgress]:
+def to_progress_instance(
+ progress: Union[Callable[..., Any], RemoteProgress, None]
+) -> Union[RemoteProgress, CallableRemoteProgress]:
"""Given the 'progress' return a suitable object derived from
RemoteProgress().
"""
@@ -130,25 +144,53 @@ class PushInfo(IterableObj, object):
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 """
+ """
+
+ __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
@@ -156,11 +198,15 @@ class PushInfo(IterableObj, object):
self._old_commit_sha = old_commit
self.summary = summary
- @ property
+ @property
def old_commit(self) -> Union[str, SymbolicReference, Commit_ish, None]:
- return self._old_commit_sha and self._remote.repo.commit(self._old_commit_sha) or None
+ return (
+ self._old_commit_sha
+ and self._remote.repo.commit(self._old_commit_sha)
+ or None
+ )
- @ property
+ @property
def remote_ref(self) -> Union[RemoteReference, TagReference]:
"""
:return:
@@ -171,27 +217,33 @@ class PushInfo(IterableObj, object):
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))
+ 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':
+ @classmethod
+ def _from_line(cls, remote: "Remote", line: str) -> "PushInfo":
"""Create a new 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)
+ 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
+ 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(':')
+ from_ref_string, to_ref_string = from_to.split(":")
if flags & cls.DELETED:
from_ref: Union[SymbolicReference, None] = None
else:
@@ -202,7 +254,7 @@ class PushInfo(IterableObj, object):
# commit handling, could be message or commit info
old_commit: Optional[str] = None
- if summary.startswith('['):
+ if summary.startswith("["):
if "[rejected]" in summary:
flags |= cls.REJECTED
elif "[remote rejected]" in summary:
@@ -222,25 +274,26 @@ class PushInfo(IterableObj, object):
split_token = "..."
if control_character == " ":
split_token = ".."
- old_sha, _new_sha = summary.split(' ')[0].split(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']:
+ @classmethod
+ def iter_items(
+ cls, repo: "Repo", *args: Any, **kwargs: Any
+ ) -> NoReturn: # -> Iterator['PushInfo']:
raise NotImplementedError
class PushInfoList(IterableList[PushInfo]):
- def __new__(cls) -> 'PushInfoList':
- return cast(PushInfoList, IterableList.__new__(cls, 'push_infos'))
+ def __new__(cls) -> "PushInfoList":
+ return cast(PushInfoList, IterableList.__new__(cls, "push_infos"))
def __init__(self) -> None:
- super().__init__('push_infos')
+ super().__init__("push_infos")
self.error: Optional[Exception] = None
def raise_if_error(self) -> None:
@@ -267,24 +320,35 @@ class FetchInfo(IterableObj, object):
# 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)]
+ __slots__ = ("ref", "old_commit", "flags", "note", "remote_ref_path")
+ _id_attribute_ = "fetchinfo"
- _re_fetch_result = re.compile(r'^\s*(.) (\[[\w\s\.$@]+\]|[\w\.$@]+)\s+(.+) -> ([^\s]+)( \(.*\)?$)?')
+ (
+ 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"^\s*(.) (\[[\w\s\.$@]+\]|[\w\.$@]+)\s+(.+) -> ([^\s]+)( \(.*\)?$)?"
+ )
_flag_map: Dict[flagKeyLiteral, int] = {
- '!': ERROR,
- '+': FORCED_UPDATE,
- '*': 0,
- '=': HEAD_UPTODATE,
- ' ': FAST_FORWARD,
- '-': TAG_UPDATE,
+ "!": ERROR,
+ "+": FORCED_UPDATE,
+ "*": 0,
+ "=": HEAD_UPTODATE,
+ " ": FAST_FORWARD,
+ "-": TAG_UPDATE,
}
- @ classmethod
+ @classmethod
def refresh(cls) -> Literal[True]:
"""This gets called by the refresh function (see the top level
__init__).
@@ -308,9 +372,14 @@ class FetchInfo(IterableObj, object):
return True
- def __init__(self, ref: SymbolicReference, flags: int, note: str = '',
- old_commit: Union[Commit_ish, None] = None,
- remote_ref_path: Optional[PathLike] = None) -> None:
+ def __init__(
+ self,
+ ref: SymbolicReference,
+ flags: int,
+ note: str = "",
+ old_commit: Union[Commit_ish, None] = None,
+ remote_ref_path: Optional[PathLike] = None,
+ ) -> None:
"""
Initialize a new instance
"""
@@ -323,18 +392,18 @@ class FetchInfo(IterableObj, object):
def __str__(self) -> str:
return self.name
- @ property
+ @property
def name(self) -> str:
""":return: Name of our remote ref"""
return self.ref.name
- @ property
+ @property
def commit(self) -> Commit_ish:
""":return: Commit of our remote ref"""
return self.ref.commit
- @ classmethod
- def _from_line(cls, repo: 'Repo', line: str, fetch_line: str) -> 'FetchInfo':
+ @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 FetchInfo object representing this information.
@@ -357,12 +426,18 @@ class FetchInfo(IterableObj, object):
# parse lines
remote_local_ref_str: str
- control_character, operation, local_remote_ref, remote_local_ref_str, note = match.groups()
+ (
+ control_character,
+ operation,
+ local_remote_ref,
+ remote_local_ref_str,
+ note,
+ ) = match.groups()
# assert is_flagKeyLiteral(control_character), f"{control_character}"
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)
+ 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
@@ -371,25 +446,28 @@ class FetchInfo(IterableObj, object):
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
+ 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 - makes no sense for symbolic refs, but we parse it anyway
old_commit: Union[Commit_ish, None] = None
is_tag_operation = False
- if 'rejected' in operation:
+ if "rejected" in operation:
flags |= cls.REJECTED
- if 'new tag' in operation:
+ if "new tag" in operation:
flags |= cls.NEW_TAG
is_tag_operation = True
- if 'tag update' in operation:
+ if "tag update" in operation:
flags |= cls.TAG_UPDATE
is_tag_operation = True
- if 'new branch' in operation:
+ if "new branch" in operation:
flags |= cls.NEW_HEAD
- if '...' in operation or '..' in operation:
- split_token = '...'
- if control_character == ' ':
+ 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
@@ -409,7 +487,7 @@ class FetchInfo(IterableObj, object):
# note: remote-tracking is just the first part of the 'remote-tracking branch' token.
# We don't parse it correctly, but its enough to know what to do, and its new in git 1.7something
ref_type = RemoteReference
- elif '/' in ref_type_name:
+ elif "/" in ref_type_name:
# If the fetch spec look something like this '+refs/pull/*:refs/heads/pull/*', and is thus pretty
# much anything the user wants, we will have trouble to determine what's going on
# For now, we assume the local ref is a Head
@@ -434,15 +512,23 @@ class FetchInfo(IterableObj, object):
# always use actual type if we get absolute paths
# 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 + "/"):
+ 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:
+ elif ref_type is TagReference and "tags/" in remote_local_ref_str:
# even though its a tag, it is located in refs/remotes
- ref_path = join_path(RemoteReference._common_path_default, remote_local_ref_str)
+ 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)
+ 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
@@ -450,13 +536,14 @@ class FetchInfo(IterableObj, object):
remote_local_ref = ref_type(repo, ref_path, check_path=False)
# END create ref instance
- note = (note and note.strip()) or ''
+ 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']:
+ @classmethod
+ def iter_items(
+ cls, repo: "Repo", *args: Any, **kwargs: Any
+ ) -> NoReturn: # -> Iterator['FetchInfo']:
raise NotImplementedError
@@ -473,7 +560,7 @@ class Remote(LazyMixin, IterableObj):
__slots__ = ("repo", "name", "_config_reader")
_id_attribute_ = "name"
- def __init__(self, repo: 'Repo', name: str) -> None:
+ def __init__(self, repo: "Repo", name: str) -> None:
"""Initialize a remote instance
:param repo: The repository we are a remote of
@@ -503,7 +590,9 @@ class Remote(LazyMixin, IterableObj):
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())
+ self._config_reader = SectionConstraint(
+ self.repo.config_reader("repository"), self._config_section_name()
+ )
else:
super(Remote, self)._set_cache_(attr)
@@ -527,7 +616,7 @@ class Remote(LazyMixin, IterableObj):
:return: True if this is a valid, existing remote.
Valid remotes have an entry in the repository's configuration"""
try:
- self.config_reader.get('url')
+ self.config_reader.get("url")
return True
except cp.NoOptionError:
# we have the section at least ...
@@ -536,20 +625,22 @@ class Remote(LazyMixin, IterableObj):
return False
# end
- @ classmethod
- def iter_items(cls, repo: 'Repo', *args: Any, **kwargs: Any) -> Iterator['Remote']:
+ @classmethod
+ def iter_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> Iterator["Remote"]:
""":return: Iterator yielding Remote objects of the given repository"""
for section in repo.config_reader("repository").sections():
- if not section.startswith('remote '):
+ if not section.startswith("remote "):
continue
lbound = section.find('"')
rbound = section.rfind('"')
if lbound == -1 or rbound == -1:
raise ValueError("Remote-Section has invalid format: %r" % section)
- yield Remote(repo, section[lbound + 1:rbound])
+ yield Remote(repo, section[lbound + 1 : rbound])
# END for each configuration section
- def set_url(self, new_url: str, old_url: Optional[str] = None, **kwargs: Any) -> 'Remote':
+ def set_url(
+ self, new_url: str, old_url: Optional[str] = None, **kwargs: Any
+ ) -> "Remote":
"""Configure URLs on current remote (cf command git remote set_url)
This command manages URLs on the remote.
@@ -558,15 +649,15 @@ class Remote(LazyMixin, IterableObj):
:param old_url: when set, replaces this URL with new_url for the remote
:return: self
"""
- scmd = 'set-url'
- kwargs['insert_kwargs_after'] = scmd
+ scmd = "set-url"
+ kwargs["insert_kwargs_after"] = scmd
if old_url:
self.repo.git.remote(scmd, self.name, new_url, old_url, **kwargs)
else:
self.repo.git.remote(scmd, self.name, new_url, **kwargs)
return self
- def add_url(self, url: str, **kwargs: Any) -> 'Remote':
+ def add_url(self, url: str, **kwargs: Any) -> "Remote":
"""Adds a new url on current remote (special case of git remote set_url)
This command adds new URLs to a given remote, making it possible to have
@@ -577,7 +668,7 @@ class Remote(LazyMixin, IterableObj):
"""
return self.set_url(url, add=True)
- def delete_url(self, url: str, **kwargs: Any) -> 'Remote':
+ def delete_url(self, url: str, **kwargs: Any) -> "Remote":
"""Deletes a new url on current remote (special case of git remote set_url)
This command deletes new URLs to a given remote, making it possible to have
@@ -588,13 +679,13 @@ class Remote(LazyMixin, IterableObj):
"""
return self.set_url(url, delete=True)
- @ property
+ @property
def urls(self) -> Iterator[str]:
""":return: Iterator yielding all configured URL targets on a remote as strings"""
try:
remote_details = self.repo.git.remote("get-url", "--all", self.name)
assert isinstance(remote_details, str)
- for line in remote_details.split('\n'):
+ for line in remote_details.split("\n"):
yield line
except GitCommandError as ex:
## We are on git < 2.7 (i.e TravisCI as of Oct-2016),
@@ -602,37 +693,44 @@ class Remote(LazyMixin, IterableObj):
# see: https://github.com/gitpython-developers/GitPython/pull/528#issuecomment-252976319
# and: http://stackoverflow.com/a/32991784/548792
#
- if 'Unknown subcommand: get-url' in str(ex):
+ if "Unknown subcommand: get-url" in str(ex):
try:
remote_details = self.repo.git.remote("show", self.name)
assert isinstance(remote_details, str)
- for line in remote_details.split('\n'):
- if ' Push URL:' in line:
- yield line.split(': ')[-1]
+ for line in remote_details.split("\n"):
+ if " Push URL:" in line:
+ yield line.split(": ")[-1]
except GitCommandError as _ex:
- if any(msg in str(_ex) for msg in ['correct access rights', 'cannot run ssh']):
+ if any(
+ msg in str(_ex)
+ for msg in ["correct access rights", "cannot run ssh"]
+ ):
# If ssh is not setup to access this repository, see issue 694
- remote_details = self.repo.git.config('--get-all', 'remote.%s.url' % self.name)
+ remote_details = self.repo.git.config(
+ "--get-all", "remote.%s.url" % self.name
+ )
assert isinstance(remote_details, str)
- for line in remote_details.split('\n'):
+ for line in remote_details.split("\n"):
yield line
else:
raise _ex
else:
raise ex
- @ property
+ @property
def refs(self) -> IterableList[RemoteReference]:
"""
:return:
IterableList of RemoteReference objects. It is prefixed, allowing
you to omit the remote path portion, i.e.::
remote.refs.master # yields RemoteReference('/refs/remotes/origin/master')"""
- out_refs: IterableList[RemoteReference] = IterableList(RemoteReference._id_attribute_, "%s/" % self.name)
+ out_refs: IterableList[RemoteReference] = IterableList(
+ RemoteReference._id_attribute_, "%s/" % self.name
+ )
out_refs.extend(RemoteReference.list_items(self.repo, remote=self.name))
return out_refs
- @ property
+ @property
def stale_refs(self) -> IterableList[Reference]:
"""
:return:
@@ -647,8 +745,10 @@ class Remote(LazyMixin, IterableObj):
other kinds of references, for example, tag references, if these are stale
as well. This is a fix for the issue described here:
https://github.com/gitpython-developers/GitPython/issues/260
- """
- out_refs: IterableList[Reference] = IterableList(RemoteReference._id_attribute_, "%s/" % self.name)
+ """
+ out_refs: IterableList[Reference] = IterableList(
+ RemoteReference._id_attribute_, "%s/" % self.name
+ )
for line in self.repo.git.remote("prune", "--dry-run", self).splitlines()[2:]:
# expecting
# * [would prune] origin/new_branch
@@ -657,7 +757,7 @@ class Remote(LazyMixin, IterableObj):
continue
ref_name = line.replace(token, "")
# sometimes, paths start with a full ref name, like refs/tags/foo, see #260
- if ref_name.startswith(Reference._common_path_default + '/'):
+ if ref_name.startswith(Reference._common_path_default + "/"):
out_refs.append(Reference.from_path(self.repo, ref_name))
else:
fqhn = "%s/%s" % (RemoteReference._common_path_default, ref_name)
@@ -666,8 +766,8 @@ class Remote(LazyMixin, IterableObj):
# END for each line
return out_refs
- @ classmethod
- def create(cls, repo: 'Repo', name: str, url: str, **kwargs: Any) -> 'Remote':
+ @classmethod
+ def create(cls, repo: "Repo", name: str, url: str, **kwargs: Any) -> "Remote":
"""Create a new remote to the given repository
:param repo: Repository instance that is to receive the new remote
:param name: Desired name of the remote
@@ -675,18 +775,18 @@ class Remote(LazyMixin, IterableObj):
:param kwargs: Additional arguments to be passed to the git-remote add command
:return: New Remote instance
:raise GitCommandError: in case an origin with that name already exists"""
- scmd = 'add'
- kwargs['insert_kwargs_after'] = scmd
+ scmd = "add"
+ kwargs["insert_kwargs_after"] = scmd
repo.git.remote(scmd, name, Git.polish_url(url), **kwargs)
return cls(repo, name)
# add is an alias
- @ classmethod
- def add(cls, repo: 'Repo', name: str, url: str, **kwargs: Any) -> 'Remote':
+ @classmethod
+ def add(cls, repo: "Repo", name: str, url: str, **kwargs: Any) -> "Remote":
return cls.create(repo, name, url, **kwargs)
- @ classmethod
- def remove(cls, repo: 'Repo', name: str) -> str:
+ @classmethod
+ def remove(cls, repo: "Repo", name: str) -> str:
"""Remove the remote with the given name
:return: the passed remote name to remove
"""
@@ -698,9 +798,9 @@ class Remote(LazyMixin, IterableObj):
# alias
rm = remove
- def rename(self, new_name: str) -> 'Remote':
+ def rename(self, new_name: str) -> "Remote":
"""Rename self to the given new_name
- :return: self """
+ :return: self"""
if self.name == new_name:
return self
@@ -710,7 +810,7 @@ class Remote(LazyMixin, IterableObj):
return self
- def update(self, **kwargs: Any) -> 'Remote':
+ def update(self, **kwargs: Any) -> "Remote":
"""Fetch all changes for this remote, including new branches which will
be forced in ( in case your local remote branch is not part the new remote branches
ancestry anymore ).
@@ -718,21 +818,23 @@ class Remote(LazyMixin, IterableObj):
:param kwargs:
Additional arguments passed to git-remote update
- :return: self """
- scmd = 'update'
- kwargs['insert_kwargs_after'] = scmd
+ :return: self"""
+ scmd = "update"
+ kwargs["insert_kwargs_after"] = scmd
self.repo.git.remote(scmd, self.name, **kwargs)
return self
- def _get_fetch_info_from_stderr(self, proc: 'Git.AutoInterrupt',
- progress: Union[Callable[..., Any], RemoteProgress, None],
- kill_after_timeout: Union[None, float] = None,
- ) -> IterableList['FetchInfo']:
+ def _get_fetch_info_from_stderr(
+ self,
+ proc: "Git.AutoInterrupt",
+ progress: Union[Callable[..., Any], RemoteProgress, None],
+ kill_after_timeout: Union[None, float] = None,
+ ) -> IterableList["FetchInfo"]:
progress = to_progress_instance(progress)
# skip first line as it is some remote info we are not interested in
- output: IterableList['FetchInfo'] = IterableList('name')
+ output: IterableList["FetchInfo"] = IterableList("name")
# lines which are no progress are fetch info lines
# this also waits for the command to finish
@@ -743,10 +845,16 @@ class Remote(LazyMixin, IterableObj):
cmds = set(FetchInfo._flag_map.keys())
progress_handler = progress.new_message_handler()
- handle_process_output(proc, None, progress_handler, finalizer=None, decode_streams=False,
- kill_after_timeout=kill_after_timeout)
-
- stderr_text = progress.error_lines and '\n'.join(progress.error_lines) or ''
+ handle_process_output(
+ proc,
+ None,
+ progress_handler,
+ finalizer=None,
+ decode_streams=False,
+ kill_after_timeout=kill_after_timeout,
+ )
+
+ stderr_text = progress.error_lines and "\n".join(progress.error_lines) or ""
proc.wait(stderr=stderr_text)
if stderr_text:
log.warning("Error lines received while fetching: %s", stderr_text)
@@ -754,13 +862,13 @@ class Remote(LazyMixin, IterableObj):
for line in progress.other_lines:
line = force_text(line)
for cmd in cmds:
- if len(line) > 1 and line[0] == ' ' and line[1] == cmd:
+ if len(line) > 1 and line[0] == " " and line[1] == cmd:
fetch_info_lines.append(line)
continue
# read head information
fetch_head = SymbolicReference(self.repo, "FETCH_HEAD")
- with open(fetch_head.abspath, 'rb') as fp:
+ with open(fetch_head.abspath, "rb") as fp:
fetch_head_info = [line.decode(defenc) for line in fp.readlines()]
l_fil = len(fetch_info_lines)
@@ -788,9 +896,12 @@ class Remote(LazyMixin, IterableObj):
log.warning("Git informed while fetching: %s", err_line.strip())
return output
- def _get_push_info(self, proc: 'Git.AutoInterrupt',
- progress: Union[Callable[..., Any], RemoteProgress, None],
- kill_after_timeout: Union[None, float] = None) -> PushInfoList:
+ def _get_push_info(
+ self,
+ proc: "Git.AutoInterrupt",
+ progress: Union[Callable[..., Any], RemoteProgress, None],
+ kill_after_timeout: Union[None, float] = None,
+ ) -> PushInfoList:
progress = to_progress_instance(progress)
# read progress information from stderr
@@ -807,9 +918,15 @@ class Remote(LazyMixin, IterableObj):
# If an error happens, additional info is given which we parse below.
pass
- handle_process_output(proc, stdout_handler, progress_handler, finalizer=None, decode_streams=False,
- kill_after_timeout=kill_after_timeout)
- stderr_text = progress.error_lines and '\n'.join(progress.error_lines) or ''
+ handle_process_output(
+ proc,
+ stdout_handler,
+ progress_handler,
+ finalizer=None,
+ decode_streams=False,
+ kill_after_timeout=kill_after_timeout,
+ )
+ stderr_text = progress.error_lines and "\n".join(progress.error_lines) or ""
try:
proc.wait(stderr=stderr_text)
except Exception as e:
@@ -826,9 +943,9 @@ class Remote(LazyMixin, IterableObj):
def _assert_refspec(self) -> None:
"""Turns out we can't deal with remotes if the refspec is missing"""
config = self.config_reader
- unset = 'placeholder'
+ unset = "placeholder"
try:
- if config.get_value('fetch', default=unset) is unset:
+ if config.get_value("fetch", default=unset) is unset:
msg = "Remote '%s' has no refspec set.\n"
msg += "You can set it as follows:"
msg += " 'git config --add \"remote.%s.fetch +refs/heads/*:refs/heads/*\"'."
@@ -836,11 +953,14 @@ class Remote(LazyMixin, IterableObj):
finally:
config.release()
- def fetch(self, refspec: Union[str, List[str], None] = None,
- progress: Union[RemoteProgress, None, 'UpdateProgress'] = None,
- verbose: bool = True,
- kill_after_timeout: Union[None, float] = None,
- **kwargs: Any) -> IterableList[FetchInfo]:
+ def fetch(
+ self,
+ refspec: Union[str, List[str], None] = None,
+ progress: Union[RemoteProgress, None, "UpdateProgress"] = None,
+ verbose: bool = True,
+ kill_after_timeout: Union[None, float] = None,
+ **kwargs: Any
+ ) -> IterableList[FetchInfo]:
"""Fetch the latest changes for this remote
:param refspec:
@@ -881,18 +1001,29 @@ class Remote(LazyMixin, IterableObj):
else:
args = [refspec]
- proc = self.repo.git.fetch(self, *args, as_process=True, with_stdout=False,
- universal_newlines=True, v=verbose, **kwargs)
- res = self._get_fetch_info_from_stderr(proc, progress,
- kill_after_timeout=kill_after_timeout)
- if hasattr(self.repo.odb, 'update_cache'):
+ proc = self.repo.git.fetch(
+ self,
+ *args,
+ as_process=True,
+ with_stdout=False,
+ universal_newlines=True,
+ v=verbose,
+ **kwargs
+ )
+ res = self._get_fetch_info_from_stderr(
+ proc, progress, kill_after_timeout=kill_after_timeout
+ )
+ if hasattr(self.repo.odb, "update_cache"):
self.repo.odb.update_cache()
return res
- def pull(self, refspec: Union[str, List[str], None] = None,
- progress: Union[RemoteProgress, 'UpdateProgress', None] = None,
- kill_after_timeout: Union[None, float] = None,
- **kwargs: Any) -> IterableList[FetchInfo]:
+ def pull(
+ self,
+ refspec: Union[str, List[str], None] = None,
+ progress: Union[RemoteProgress, "UpdateProgress", None] = None,
+ kill_after_timeout: Union[None, float] = None,
+ **kwargs: Any
+ ) -> IterableList[FetchInfo]:
"""Pull changes from the given branch, being the same as a fetch followed
by a merge of branch with your local branch.
@@ -900,23 +1031,36 @@ class Remote(LazyMixin, IterableObj):
:param progress: see 'push' method
:param kill_after_timeout: see 'fetch' method
:param kwargs: Additional arguments to be passed to git-pull
- :return: Please see 'fetch' method """
+ :return: Please see 'fetch' method"""
if refspec is None:
# No argument refspec, then ensure the repo's config has a fetch refspec.
self._assert_refspec()
kwargs = add_progress(kwargs, self.repo.git, progress)
- proc = self.repo.git.pull(self, refspec, with_stdout=False, as_process=True,
- universal_newlines=True, v=True, **kwargs)
- res = self._get_fetch_info_from_stderr(proc, progress,
- kill_after_timeout=kill_after_timeout)
- if hasattr(self.repo.odb, 'update_cache'):
+ proc = self.repo.git.pull(
+ self,
+ refspec,
+ with_stdout=False,
+ as_process=True,
+ universal_newlines=True,
+ v=True,
+ **kwargs
+ )
+ res = self._get_fetch_info_from_stderr(
+ proc, progress, kill_after_timeout=kill_after_timeout
+ )
+ if hasattr(self.repo.odb, "update_cache"):
self.repo.odb.update_cache()
return res
- def push(self, refspec: Union[str, List[str], None] = None,
- progress: Union[RemoteProgress, 'UpdateProgress', Callable[..., RemoteProgress], None] = None,
- kill_after_timeout: Union[None, float] = None,
- **kwargs: Any) -> IterableList[PushInfo]:
+ def push(
+ self,
+ refspec: Union[str, List[str], None] = None,
+ progress: Union[
+ RemoteProgress, "UpdateProgress", Callable[..., RemoteProgress], None
+ ] = None,
+ kill_after_timeout: Union[None, float] = None,
+ **kwargs: Any
+ ) -> IterableList[PushInfo]:
"""Push changes from source branch in refspec to target branch in refspec.
:param refspec: see 'fetch' method
@@ -945,14 +1089,20 @@ class Remote(LazyMixin, IterableObj):
If the operation fails completely, the length of the returned IterableList will
be 0."""
kwargs = add_progress(kwargs, self.repo.git, progress)
- proc = self.repo.git.push(self, refspec, porcelain=True, as_process=True,
- universal_newlines=True,
- kill_after_timeout=kill_after_timeout,
- **kwargs)
- return self._get_push_info(proc, progress,
- kill_after_timeout=kill_after_timeout)
-
- @ property
+ proc = self.repo.git.push(
+ self,
+ refspec,
+ porcelain=True,
+ as_process=True,
+ universal_newlines=True,
+ kill_after_timeout=kill_after_timeout,
+ **kwargs
+ )
+ return self._get_push_info(
+ proc, progress, kill_after_timeout=kill_after_timeout
+ )
+
+ @property
def config_reader(self) -> SectionConstraint[GitConfigParser]:
"""
:return:
@@ -962,12 +1112,12 @@ class Remote(LazyMixin, IterableObj):
def _clear_cache(self) -> None:
try:
- del(self._config_reader)
+ del self._config_reader
except AttributeError:
pass
# END handle exception
- @ property
+ @property
def config_writer(self) -> SectionConstraint:
"""
:return: GitConfigParser compatible object able to write options for this remote.