summaryrefslogtreecommitdiff
path: root/git
diff options
context:
space:
mode:
Diffstat (limited to 'git')
-rw-r--r--git/cmd.py44
-rw-r--r--git/compat.py2
-rw-r--r--git/config.py20
-rw-r--r--git/db.py2
-rw-r--r--git/diff.py8
-rw-r--r--git/index/__init__.py2
-rw-r--r--git/index/base.py31
-rw-r--r--git/index/fun.py10
-rw-r--r--git/objects/__init__.py2
-rw-r--r--git/objects/base.py6
-rw-r--r--git/objects/blob.py4
-rw-r--r--git/objects/commit.py21
-rw-r--r--git/objects/fun.py2
-rw-r--r--git/objects/submodule/base.py7
-rw-r--r--git/objects/tag.py8
-rw-r--r--git/objects/tree.py14
-rw-r--r--git/objects/util.py82
-rw-r--r--git/refs/__init__.py1
-rw-r--r--git/refs/head.py37
-rw-r--r--git/refs/log.py140
-rw-r--r--git/refs/reference.py37
-rw-r--r--git/refs/remote.py31
-rw-r--r--git/refs/symbolic.py42
-rw-r--r--git/refs/tag.py46
-rw-r--r--git/remote.py69
-rw-r--r--git/repo/__init__.py1
-rw-r--r--git/repo/base.py27
-rw-r--r--git/repo/fun.py16
-rw-r--r--git/types.py5
-rw-r--r--git/util.py43
30 files changed, 459 insertions, 301 deletions
diff --git a/git/cmd.py b/git/cmd.py
index dd887a18..4404981e 100644
--- a/git/cmd.py
+++ b/git/cmd.py
@@ -15,7 +15,6 @@ from subprocess import (
PIPE
)
import subprocess
-import sys
import threading
from textwrap import dedent
@@ -539,7 +538,7 @@ class Git(LazyMixin):
return self
def __next__(self) -> bytes:
- return self.next()
+ return next(self)
def next(self) -> bytes:
line = self.readline()
@@ -566,11 +565,11 @@ class Git(LazyMixin):
.git directory in case of bare repositories."""
super(Git, self).__init__()
self._working_dir = expand_path(working_dir)
- self._git_options = () # type: Union[List[str], Tuple[str, ...]]
- self._persistent_git_options = [] # type: List[str]
+ self._git_options: Union[List[str], Tuple[str, ...]] = ()
+ self._persistent_git_options: List[str] = []
# Extra environment variables to pass to git commands
- self._environment = {} # type: Dict[str, str]
+ self._environment: Dict[str, str] = {}
# cached command slots
self.cat_file_header = None
@@ -604,19 +603,19 @@ class Git(LazyMixin):
process_version = self._call_process('version') # should be as default *args and **kwargs used
version_numbers = process_version.split(' ')[2]
- self._version_info = tuple(
- int(n) for n in version_numbers.split('.')[:4] if n.isdigit()
- ) # type: Tuple[int, int, int, int] # type: ignore
+ self._version_info = cast(Tuple[int, int, int, int],
+ tuple(int(n) for n in version_numbers.split('.')[:4] if n.isdigit())
+ )
else:
super(Git, self)._set_cache_(attr)
# END handle version info
- @property
+ @ property
def working_dir(self) -> Union[None, PathLike]:
""":return: Git directory we are working on"""
return self._working_dir
- @property
+ @ property
def version_info(self) -> Tuple[int, int, int, int]:
"""
:return: tuple(int, int, int, int) tuple with integers representing the major, minor
@@ -624,7 +623,7 @@ class Git(LazyMixin):
This value is generated on demand and is cached"""
return self._version_info
- @overload
+ @ overload
def execute(self,
command: Union[str, Sequence[Any]],
*,
@@ -632,7 +631,7 @@ class Git(LazyMixin):
) -> 'AutoInterrupt':
...
- @overload
+ @ overload
def execute(self,
command: Union[str, Sequence[Any]],
*,
@@ -641,7 +640,7 @@ class Git(LazyMixin):
) -> Union[str, Tuple[int, str, str]]:
...
- @overload
+ @ overload
def execute(self,
command: Union[str, Sequence[Any]],
*,
@@ -650,7 +649,7 @@ class Git(LazyMixin):
) -> Union[bytes, Tuple[int, bytes, str]]:
...
- @overload
+ @ overload
def execute(self,
command: Union[str, Sequence[Any]],
*,
@@ -660,7 +659,7 @@ class Git(LazyMixin):
) -> str:
...
- @overload
+ @ overload
def execute(self,
command: Union[str, Sequence[Any]],
*,
@@ -799,10 +798,7 @@ class Git(LazyMixin):
if kill_after_timeout:
raise GitCommandError(redacted_command, '"kill_after_timeout" feature is not supported on Windows.')
else:
- if sys.version_info[0] > 2:
- cmd_not_found_exception = FileNotFoundError # NOQA # exists, flake8 unknown @UndefinedVariable
- else:
- cmd_not_found_exception = OSError
+ cmd_not_found_exception = FileNotFoundError # NOQA # exists, flake8 unknown @UndefinedVariable
# end handle
stdout_sink = (PIPE
@@ -872,8 +868,8 @@ class Git(LazyMixin):
# Wait for the process to return
status = 0
- stdout_value = b'' # type: Union[str, bytes]
- stderr_value = b'' # type: Union[str, bytes]
+ stdout_value: Union[str, bytes] = b''
+ stderr_value: Union[str, bytes] = b''
newline = "\n" if universal_newlines else b"\n"
try:
if output_stream is None:
@@ -1070,8 +1066,8 @@ class Git(LazyMixin):
It contains key-values for the following:
- the :meth:`execute()` kwds, as listed in :var:`execute_kwargs`;
- "command options" to be converted by :meth:`transform_kwargs()`;
- - the `'insert_kwargs_after'` key which its value must match one of ``*args``,
- and any cmd-options will be appended after the matched arg.
+ - the `'insert_kwargs_after'` key which its value must match one of ``*args``
+ and any cmd-options will be appended after the matched arg.
Examples::
@@ -1149,7 +1145,7 @@ class Git(LazyMixin):
# required for command to separate refs on stdin, as bytes
if isinstance(ref, bytes):
# Assume 40 bytes hexsha - bin-to-ascii for some reason returns bytes, not text
- refstr = ref.decode('ascii') # type: str
+ refstr: str = ref.decode('ascii')
elif not isinstance(ref, str):
refstr = str(ref) # could be ref-object
else:
diff --git a/git/compat.py b/git/compat.py
index 187618a2..7a0a15d2 100644
--- a/git/compat.py
+++ b/git/compat.py
@@ -34,7 +34,7 @@ from git.types import TBD
# ---------------------------------------------------------------------------
-is_win = (os.name == 'nt') # type: bool
+is_win: bool = (os.name == 'nt')
is_posix = (os.name == 'posix')
is_darwin = (os.name == 'darwin')
defenc = sys.getfilesystemencoding()
diff --git a/git/config.py b/git/config.py
index 2c863f93..345cb40e 100644
--- a/git/config.py
+++ b/git/config.py
@@ -208,7 +208,7 @@ class _OMD(OrderedDict):
def getall(self, key: str) -> Any:
return super(_OMD, self).__getitem__(key)
- def items(self) -> List[Tuple[str, Any]]: # type: ignore ## mypy doesn't like overwriting supertype signitures
+ def items(self) -> List[Tuple[str, Any]]: # type: ignore[override]
"""List of (key, last value for key)."""
return [(k, self[k]) for k in self]
@@ -238,7 +238,7 @@ def get_config_path(config_level: Lit_config_levels) -> str:
assert_never(config_level, ValueError(f"Invalid configuration level: {config_level!r}"))
-class GitConfigParser(with_metaclass(MetaParserBuilder, cp.RawConfigParser, object)): # type: ignore ## mypy does not understand dynamic class creation # noqa: E501
+class GitConfigParser(with_metaclass(MetaParserBuilder, cp.RawConfigParser)): # type: ignore ## mypy does not understand dynamic class creation # noqa: E501
"""Implements specifics required to read git style configuration files.
@@ -322,7 +322,7 @@ class GitConfigParser(with_metaclass(MetaParserBuilder, cp.RawConfigParser, obje
self._is_initialized = False
self._merge_includes = merge_includes
self._repo = repo
- self._lock = None # type: Union['LockFile', None]
+ self._lock: Union['LockFile', None] = None
self._acquire_lock()
def _acquire_lock(self) -> None:
@@ -545,13 +545,15 @@ class GitConfigParser(with_metaclass(MetaParserBuilder, cp.RawConfigParser, obje
return None
self._is_initialized = True
- files_to_read = [""] # type: List[Union[PathLike, IO]] ## just for types until 3.5 dropped
- if isinstance(self._file_or_files, (str)): # replace with PathLike once 3.5 dropped
- files_to_read = [self._file_or_files] # for str, as str is a type of Sequence
+ files_to_read: List[Union[PathLike, IO]] = [""]
+ if isinstance(self._file_or_files, (str, os.PathLike)):
+ # for str or Path, as str is a type of Sequence
+ files_to_read = [self._file_or_files]
elif not isinstance(self._file_or_files, (tuple, list, Sequence)):
- files_to_read = [self._file_or_files] # for IO or Path
- else:
- files_to_read = list(self._file_or_files) # for lists or tuples
+ # could merge with above isinstance once runtime type known
+ files_to_read = [self._file_or_files]
+ else: # for lists or tuples
+ files_to_read = list(self._file_or_files)
# end assure we have a copy of the paths to handle
seen = set(files_to_read)
diff --git a/git/db.py b/git/db.py
index 47cccda8..3a7adc7d 100644
--- a/git/db.py
+++ b/git/db.py
@@ -17,7 +17,7 @@ from git.types import PathLike
if TYPE_CHECKING:
from git.cmd import Git
-
+
# --------------------------------------------------------
diff --git a/git/diff.py b/git/diff.py
index 51dac390..98a5cfd9 100644
--- a/git/diff.py
+++ b/git/diff.py
@@ -143,7 +143,7 @@ class Diffable(object):
paths = [paths]
if hasattr(self, 'Has_Repo'):
- self.repo: Repo = self.repo
+ self.repo: 'Repo' = self.repo
diff_cmd = self.repo.git.diff
if other is self.Index:
@@ -351,13 +351,13 @@ class Diff(object):
return hash(tuple(getattr(self, n) for n in self.__slots__))
def __str__(self) -> str:
- h = "%s" # type: str
+ h: str = "%s"
if self.a_blob:
h %= self.a_blob.path
elif self.b_blob:
h %= self.b_blob.path
- msg = '' # type: str
+ msg: str = ''
line = None # temp line
line_length = 0 # line length
for b, n in zip((self.a_blob, self.b_blob), ('lhs', 'rhs')):
@@ -449,7 +449,7 @@ class Diff(object):
:return: git.DiffIndex """
## FIXME: Here SLURPING raw, need to re-phrase header-regexes linewise.
- text_list = [] # type: List[bytes]
+ text_list: List[bytes] = []
handle_process_output(proc, text_list.append, None, finalize_process, decode_streams=False)
# for now, we have to bake the stream
diff --git a/git/index/__init__.py b/git/index/__init__.py
index 2516f01f..96b721f0 100644
--- a/git/index/__init__.py
+++ b/git/index/__init__.py
@@ -1,6 +1,4 @@
"""Initialize the index package"""
# flake8: noqa
-from __future__ import absolute_import
-
from .base import *
from .typ import *
diff --git a/git/index/base.py b/git/index/base.py
index 3aa06e38..6452419c 100644
--- a/git/index/base.py
+++ b/git/index/base.py
@@ -123,7 +123,7 @@ class IndexFile(LazyMixin, git_diff.Diffable, Serializable):
self.repo = repo
self.version = self._VERSION
self._extension_data = b''
- self._file_path = file_path or self._index_path() # type: PathLike
+ self._file_path: PathLike = file_path or self._index_path()
def _set_cache_(self, attr: str) -> None:
if attr == "entries":
@@ -136,7 +136,7 @@ class IndexFile(LazyMixin, git_diff.Diffable, Serializable):
ok = True
except OSError:
# in new repositories, there may be no index, which means we are empty
- self.entries = {} # type: Dict[Tuple[PathLike, StageType], IndexEntry]
+ self.entries: Dict[Tuple[PathLike, StageType], IndexEntry] = {}
return None
finally:
if not ok:
@@ -266,7 +266,7 @@ class IndexFile(LazyMixin, git_diff.Diffable, Serializable):
# -i : ignore working tree status
# --aggressive : handle more merge cases
# -m : do an actual merge
- args = ["--aggressive", "-i", "-m"] # type: List[Union[Treeish, str]]
+ args: List[Union[Treeish, str]] = ["--aggressive", "-i", "-m"]
if base is not None:
args.append(base)
args.append(rhs)
@@ -288,14 +288,14 @@ class IndexFile(LazyMixin, git_diff.Diffable, Serializable):
New IndexFile instance. Its path will be undefined.
If you intend to write such a merged Index, supply an alternate file_path
to its 'write' method."""
- tree_sha_bytes = [to_bin_sha(str(t)) for t in tree_sha] # List[bytes]
+ tree_sha_bytes: List[bytes] = [to_bin_sha(str(t)) for t in tree_sha]
base_entries = aggressive_tree_merge(repo.odb, tree_sha_bytes)
inst = cls(repo)
# convert to entries dict
- entries = dict(zip(
+ entries: Dict[Tuple[PathLike, int], IndexEntry] = dict(zip(
((e.path, e.stage) for e in base_entries),
- (IndexEntry.from_base(e) for e in base_entries))) # type: Dict[Tuple[PathLike, int], IndexEntry]
+ (IndexEntry.from_base(e) for e in base_entries)))
inst.entries = entries
return inst
@@ -338,7 +338,7 @@ class IndexFile(LazyMixin, git_diff.Diffable, Serializable):
if len(treeish) == 0 or len(treeish) > 3:
raise ValueError("Please specify between 1 and 3 treeish, got %i" % len(treeish))
- arg_list = [] # type: List[Union[Treeish, str]]
+ arg_list: List[Union[Treeish, str]] = []
# ignore that working tree and index possibly are out of date
if len(treeish) > 1:
# drop unmerged entries when reading our index and merging
@@ -445,7 +445,7 @@ class IndexFile(LazyMixin, git_diff.Diffable, Serializable):
we will close stdin to break the pipe."""
fprogress(filepath, False, item)
- rval = None # type: Union[None, str]
+ rval: Union[None, str] = None
if proc.stdin is not None:
try:
@@ -492,7 +492,7 @@ class IndexFile(LazyMixin, git_diff.Diffable, Serializable):
are at stage 3 will not have a stage 3 entry.
"""
is_unmerged_blob = lambda t: t[0] != 0
- path_map = {} # type: Dict[PathLike, List[Tuple[TBD, Blob]]]
+ path_map: Dict[PathLike, List[Tuple[TBD, Blob]]] = {}
for stage, blob in self.iter_blobs(is_unmerged_blob):
path_map.setdefault(blob.path, []).append((stage, blob))
# END for each unmerged blob
@@ -572,7 +572,7 @@ class IndexFile(LazyMixin, git_diff.Diffable, Serializable):
# note: additional deserialization could be saved if write_tree_from_cache
# would return sorted tree entries
root_tree = Tree(self.repo, binsha, path='')
- root_tree._cache = tree_items # type: ignore # should this be encoded to [bytes, int, str]?
+ root_tree._cache = tree_items
return root_tree
def _process_diff_args(self, # type: ignore[override]
@@ -586,8 +586,9 @@ class IndexFile(LazyMixin, git_diff.Diffable, Serializable):
return args
def _to_relative_path(self, path: PathLike) -> PathLike:
- """:return: Version of path relative to our git directory or raise ValueError
- if it is not within our git direcotory"""
+ """
+ :return: Version of path relative to our git directory or raise ValueError
+ if it is not within our git direcotory"""
if not osp.isabs(path):
return path
if self.repo.bare:
@@ -623,8 +624,8 @@ class IndexFile(LazyMixin, git_diff.Diffable, Serializable):
st = os.lstat(filepath) # handles non-symlinks as well
if S_ISLNK(st.st_mode):
# in PY3, readlink is string, but we need bytes. In PY2, it's just OS encoded bytes, we assume UTF-8
- open_stream = lambda: BytesIO(force_bytes(os.readlink(filepath),
- encoding=defenc)) # type: Callable[[], BinaryIO]
+ open_stream: Callable[[], BinaryIO] = lambda: BytesIO(force_bytes(os.readlink(filepath),
+ encoding=defenc))
else:
open_stream = lambda: open(filepath, 'rb')
with open_stream() as stream:
@@ -1159,7 +1160,7 @@ class IndexFile(LazyMixin, git_diff.Diffable, Serializable):
proc = self.repo.git.checkout_index(args, **kwargs)
# FIXME: Reading from GIL!
make_exc = lambda: GitCommandError(("git-checkout-index",) + tuple(args), 128, proc.stderr.read())
- checked_out_files = [] # type: List[PathLike]
+ checked_out_files: List[PathLike] = []
for path in paths:
co_path = to_native_path_linux(self._to_relative_path(path))
diff --git a/git/index/fun.py b/git/index/fun.py
index e5e566a0..e071e15c 100644
--- a/git/index/fun.py
+++ b/git/index/fun.py
@@ -99,8 +99,8 @@ def run_commit_hook(name: str, index: 'IndexFile', *args: str) -> None:
except Exception as ex:
raise HookExecutionError(hp, ex) from ex
else:
- stdout_list = [] # type: List[str]
- stderr_list = [] # type: List[str]
+ stdout_list: List[str] = []
+ stderr_list: List[str] = []
handle_process_output(cmd, stdout_list.append, stderr_list.append, finalize_process)
stdout = ''.join(stdout_list)
stderr = ''.join(stderr_list)
@@ -151,8 +151,8 @@ def write_cache(entries: Sequence[Union[BaseIndexEntry, 'IndexEntry']], stream:
beginoffset = tell()
write(entry[4]) # ctime
write(entry[5]) # mtime
- path_str = entry[3] # type: str
- path = force_bytes(path_str, encoding=defenc)
+ path_str: str = entry[3]
+ path: bytes = force_bytes(path_str, encoding=defenc)
plen = len(path) & CE_NAMEMASK # path length
assert plen == len(path), "Path %s too long to fit into index" % entry[3]
flags = plen | (entry[2] & CE_NAMEMASK_INV) # clear possible previous values
@@ -210,7 +210,7 @@ def read_cache(stream: IO[bytes]) -> Tuple[int, Dict[Tuple[PathLike, int], 'Inde
* content_sha is a 20 byte sha on all cache file contents"""
version, num_entries = read_header(stream)
count = 0
- entries = {} # type: Dict[Tuple[PathLike, int], 'IndexEntry']
+ entries: Dict[Tuple[PathLike, int], 'IndexEntry'] = {}
read = stream.read
tell = stream.tell
diff --git a/git/objects/__init__.py b/git/objects/__init__.py
index 897eb98f..1d0bb7a5 100644
--- a/git/objects/__init__.py
+++ b/git/objects/__init__.py
@@ -2,8 +2,6 @@
Import all submodules main classes into the package space
"""
# flake8: noqa
-from __future__ import absolute_import
-
import inspect
from .base import *
diff --git a/git/objects/base.py b/git/objects/base.py
index 4e2ed493..64f105ca 100644
--- a/git/objects/base.py
+++ b/git/objects/base.py
@@ -15,9 +15,9 @@ from .util import get_object_type_by_name
# typing ------------------------------------------------------------------
-from typing import Any, TYPE_CHECKING, Optional, Union
+from typing import Any, TYPE_CHECKING, Union
-from git.types import PathLike, Commit_ish
+from git.types import PathLike, Commit_ish, Lit_commit_ish
if TYPE_CHECKING:
from git.repo import Repo
@@ -44,7 +44,7 @@ class Object(LazyMixin):
TYPES = (dbtyp.str_blob_type, dbtyp.str_tree_type, dbtyp.str_commit_type, dbtyp.str_tag_type)
__slots__ = ("repo", "binsha", "size")
- type = None # type: Optional[str] # to be set by subclass
+ type: Union[Lit_commit_ish, None] = None
def __init__(self, repo: 'Repo', binsha: bytes):
"""Initialize an object by identifying it by its binary sha.
diff --git a/git/objects/blob.py b/git/objects/blob.py
index 017178f0..99b5c636 100644
--- a/git/objects/blob.py
+++ b/git/objects/blob.py
@@ -6,6 +6,8 @@
from mimetypes import guess_type
from . import base
+from git.types import Literal
+
__all__ = ('Blob', )
@@ -13,7 +15,7 @@ class Blob(base.IndexObject):
"""A Blob encapsulates a git blob object"""
DEFAULT_MIME_TYPE = "text/plain"
- type = "blob"
+ type: Literal['blob'] = "blob"
# valid blob modes
executable_mode = 0o100755
diff --git a/git/objects/commit.py b/git/objects/commit.py
index 65a87591..11cf52a5 100644
--- a/git/objects/commit.py
+++ b/git/objects/commit.py
@@ -41,10 +41,11 @@ import logging
from typing import Any, IO, Iterator, List, Sequence, Tuple, Union, TYPE_CHECKING
-from git.types import PathLike, TypeGuard
+from git.types import PathLike, TypeGuard, Literal
if TYPE_CHECKING:
from git.repo import Repo
+ from git.refs import SymbolicReference
# ------------------------------------------------------------------------
@@ -73,14 +74,14 @@ class Commit(base.Object, TraversableIterableObj, Diffable, Serializable):
default_encoding = "UTF-8"
# object configuration
- type = "commit"
+ type: Literal['commit'] = "commit"
__slots__ = ("tree",
"author", "authored_date", "author_tz_offset",
"committer", "committed_date", "committer_tz_offset",
"message", "parents", "encoding", "gpgsig")
_id_attribute_ = "hexsha"
- def __init__(self, repo: 'Repo', binsha: bytes, tree: Union['Tree', None] = None,
+ def __init__(self, repo: 'Repo', binsha: bytes, tree: Union[Tree, None] = None,
author: Union[Actor, None] = None,
authored_date: Union[int, None] = None,
author_tz_offset: Union[None, float] = None,
@@ -201,11 +202,11 @@ class Commit(base.Object, TraversableIterableObj, Diffable, Serializable):
# END handle attrs
@property
- def authored_datetime(self) -> 'datetime.datetime':
+ def authored_datetime(self) -> datetime.datetime:
return from_timestamp(self.authored_date, self.author_tz_offset)
@property
- def committed_datetime(self) -> 'datetime.datetime':
+ def committed_datetime(self) -> datetime.datetime:
return from_timestamp(self.committed_date, self.committer_tz_offset)
@property
@@ -242,7 +243,7 @@ class Commit(base.Object, TraversableIterableObj, Diffable, Serializable):
return self.repo.git.name_rev(self)
@classmethod
- def iter_items(cls, repo: 'Repo', rev: str, # type: ignore
+ def iter_items(cls, repo: 'Repo', rev: Union[str, 'Commit', 'SymbolicReference'], # type: ignore
paths: Union[PathLike, Sequence[PathLike]] = '', **kwargs: Any
) -> Iterator['Commit']:
"""Find all commits matching the given criteria.
@@ -354,7 +355,7 @@ class Commit(base.Object, TraversableIterableObj, Diffable, Serializable):
finalize_process(proc_or_stream)
@ classmethod
- def create_from_tree(cls, repo: 'Repo', tree: Union['Tree', str], message: str,
+ def create_from_tree(cls, repo: 'Repo', tree: Union[Tree, str], message: str,
parent_commits: Union[None, List['Commit']] = None, head: bool = False,
author: Union[None, Actor] = None, committer: Union[None, Actor] = None,
author_date: Union[None, str] = None, commit_date: Union[None, str] = None):
@@ -516,8 +517,10 @@ class Commit(base.Object, TraversableIterableObj, Diffable, Serializable):
return self
def _deserialize(self, stream: BytesIO) -> 'Commit':
- """:param from_rev_list: if true, the stream format is coming from the rev-list command
- Otherwise it is assumed to be a plain data stream from our object"""
+ """
+ :param from_rev_list: if true, the stream format is coming from the rev-list command
+ Otherwise it is assumed to be a plain data stream from our object
+ """
readline = stream.readline
self.tree = Tree(self.repo, hex_to_bin(readline().split()[1]), Tree.tree_id << 12, '')
diff --git a/git/objects/fun.py b/git/objects/fun.py
index fc2ea1e7..d6cdafe1 100644
--- a/git/objects/fun.py
+++ b/git/objects/fun.py
@@ -167,7 +167,7 @@ def traverse_trees_recursive(odb: 'GitCmdObjectDB', tree_shas: Sequence[Union[by
data: List[EntryTupOrNone] = []
else:
# make new list for typing as list invariant
- data = [x for x in tree_entries_from_data(odb.stream(tree_sha).read())]
+ data = list(tree_entries_from_data(odb.stream(tree_sha).read()))
# END handle muted trees
trees_data.append(data)
# END for each sha to get data for
diff --git a/git/objects/submodule/base.py b/git/objects/submodule/base.py
index b485dbf6..d5ba118f 100644
--- a/git/objects/submodule/base.py
+++ b/git/objects/submodule/base.py
@@ -52,7 +52,7 @@ from .util import (
from typing import Callable, Dict, Mapping, Sequence, TYPE_CHECKING, cast
from typing import Any, Iterator, Union
-from git.types import Commit_ish, PathLike, TBD
+from git.types import Commit_ish, Literal, PathLike, TBD
if TYPE_CHECKING:
from git.index import IndexFile
@@ -105,7 +105,7 @@ class Submodule(IndexObject, TraversableIterableObj):
k_default_mode = stat.S_IFDIR | stat.S_IFLNK # submodules are directories with link-status
# this is a bogus type for base class compatibility
- type = 'submodule'
+ type: Literal['submodule'] = 'submodule' # type: ignore
__slots__ = ('_parent_commit', '_url', '_branch_path', '_name', '__weakref__')
_cache_attrs = ('path', '_url', '_branch_path')
@@ -475,7 +475,8 @@ class Submodule(IndexObject, TraversableIterableObj):
sm._branch_path = br.path
# we deliberately assume that our head matches our index !
- sm.binsha = mrepo.head.commit.binsha # type: ignore
+ if mrepo:
+ sm.binsha = mrepo.head.commit.binsha
index.add([sm], write=True)
return sm
diff --git a/git/objects/tag.py b/git/objects/tag.py
index cb6efbe9..7048eb40 100644
--- a/git/objects/tag.py
+++ b/git/objects/tag.py
@@ -11,6 +11,8 @@ from ..compat import defenc
from typing import List, TYPE_CHECKING, Union
+from git.types import Literal
+
if TYPE_CHECKING:
from git.repo import Repo
from git.util import Actor
@@ -24,7 +26,7 @@ __all__ = ("TagObject", )
class TagObject(base.Object):
"""Non-Lightweight tag carrying additional information about an object we are pointing to."""
- type = "tag"
+ type: Literal['tag'] = "tag"
__slots__ = ("object", "tag", "tagger", "tagged_date", "tagger_tz_offset", "message")
def __init__(self, repo: 'Repo', binsha: bytes,
@@ -49,7 +51,7 @@ class TagObject(base.Object):
authored_date is in, in a format similar to time.altzone"""
super(TagObject, self).__init__(repo, binsha)
if object is not None:
- self.object = object # type: Union['Commit', 'Blob', 'Tree', 'TagObject']
+ self.object: Union['Commit', 'Blob', 'Tree', 'TagObject'] = object
if tag is not None:
self.tag = tag
if tagger is not None:
@@ -65,7 +67,7 @@ class TagObject(base.Object):
"""Cache all our attributes at once"""
if attr in TagObject.__slots__:
ostream = self.repo.odb.stream(self.binsha)
- lines = ostream.read().decode(defenc, 'replace').splitlines() # type: List[str]
+ lines: List[str] = ostream.read().decode(defenc, 'replace').splitlines()
_obj, hexsha = lines[0].split(" ")
_type_token, type_name = lines[1].split(" ")
diff --git a/git/objects/tree.py b/git/objects/tree.py
index a9656c1d..dd1fe783 100644
--- a/git/objects/tree.py
+++ b/git/objects/tree.py
@@ -24,7 +24,7 @@ from .fun import (
from typing import (Any, Callable, Dict, Iterable, Iterator, List,
Tuple, Type, Union, cast, TYPE_CHECKING)
-from git.types import PathLike, TypeGuard
+from git.types import PathLike, TypeGuard, Literal
if TYPE_CHECKING:
from git.repo import Repo
@@ -195,7 +195,7 @@ class Tree(IndexObject, git_diff.Diffable, util.Traversable, util.Serializable):
blob = tree[0]
"""
- type = "tree"
+ type: Literal['tree'] = "tree"
__slots__ = "_cache"
# actual integer ids for comparison
@@ -285,7 +285,7 @@ class Tree(IndexObject, git_diff.Diffable, util.Traversable, util.Serializable):
return [i for i in self if i.type == "tree"]
@ property
- def blobs(self) -> List['Blob']:
+ def blobs(self) -> List[Blob]:
""":return: list(Blob, ...) list of blobs directly below this tree"""
return [i for i in self if i.type == "blob"]
@@ -298,7 +298,7 @@ class Tree(IndexObject, git_diff.Diffable, util.Traversable, util.Serializable):
See the ``TreeModifier`` for more information on how to alter the cache"""
return TreeModifier(self._cache)
- def traverse(self, # type: ignore # overrides super()
+ def traverse(self, # type: ignore[override]
predicate: Callable[[Union[IndexObjUnion, TraversedTreeTup], int], bool] = lambda i, d: True,
prune: Callable[[Union[IndexObjUnion, TraversedTreeTup], int], bool] = lambda i, d: False,
depth: int = -1,
@@ -322,8 +322,8 @@ class Tree(IndexObject, git_diff.Diffable, util.Traversable, util.Serializable):
# assert is_tree_traversed(ret_tup), f"Type is {[type(x) for x in list(ret_tup[0])]}"
# return ret_tup[0]"""
return cast(Union[Iterator[IndexObjUnion], Iterator[TraversedTreeTup]],
- super(Tree, self).traverse(predicate, prune, depth, # type: ignore
- branch_first, visit_once, ignore_self))
+ super(Tree, self)._traverse(predicate, prune, depth, # type: ignore
+ branch_first, visit_once, ignore_self))
def list_traverse(self, *args: Any, **kwargs: Any) -> IterableList[IndexObjUnion]:
"""
@@ -331,7 +331,7 @@ class Tree(IndexObject, git_diff.Diffable, util.Traversable, util.Serializable):
traverse()
Tree -> IterableList[Union['Submodule', 'Tree', 'Blob']]
"""
- return super(Tree, self).list_traverse(* args, **kwargs)
+ return super(Tree, self)._list_traverse(* args, **kwargs)
# List protocol
diff --git a/git/objects/util.py b/git/objects/util.py
index fbe3d9de..ef1ae77b 100644
--- a/git/objects/util.py
+++ b/git/objects/util.py
@@ -5,6 +5,8 @@
# the BSD License: http://www.opensource.org/licenses/bsd-license.php
"""Module for general utility functions"""
+from abc import abstractmethod
+import warnings
from git.util import (
IterableList,
IterableObj,
@@ -23,7 +25,7 @@ from datetime import datetime, timedelta, tzinfo
from typing import (Any, Callable, Deque, Iterator, NamedTuple, overload, Sequence,
TYPE_CHECKING, Tuple, Type, TypeVar, Union, cast)
-from git.types import Has_id_attribute, Literal
+from git.types import Has_id_attribute, Literal, Protocol, runtime_checkable
if TYPE_CHECKING:
from io import BytesIO, StringIO
@@ -283,13 +285,14 @@ class ProcessStreamAdapter(object):
def __init__(self, process: 'Popen', stream_name: str) -> None:
self._proc = process
- self._stream = getattr(process, stream_name) # type: StringIO ## guess
+ self._stream: StringIO = getattr(process, stream_name) # guessed type
def __getattr__(self, attr: str) -> Any:
return getattr(self._stream, attr)
-class Traversable(object):
+@runtime_checkable
+class Traversable(Protocol):
"""Simple interface to perform depth-first or breadth-first traversals
into one direction.
@@ -301,6 +304,7 @@ class Traversable(object):
__slots__ = ()
@classmethod
+ @abstractmethod
def _get_intermediate_items(cls, item) -> Sequence['Traversable']:
"""
Returns:
@@ -313,7 +317,18 @@ class Traversable(object):
"""
raise NotImplementedError("To be implemented in subclass")
- def list_traverse(self, *args: Any, **kwargs: Any) -> IterableList[Union['Commit', 'Submodule', 'Tree', 'Blob']]:
+ @abstractmethod
+ def list_traverse(self, *args: Any, **kwargs: Any) -> Any:
+ """ """
+ warnings.warn("list_traverse() method should only be called from subclasses."
+ "Calling from Traversable abstract class will raise NotImplementedError in 3.1.20"
+ "Builtin sublclasses are 'Submodule', 'Tree' and 'Commit",
+ DeprecationWarning,
+ stacklevel=2)
+ return self._list_traverse(*args, **kwargs)
+
+ def _list_traverse(self, as_edge=False, *args: Any, **kwargs: Any
+ ) -> IterableList[Union['Commit', 'Submodule', 'Tree', 'Blob']]:
"""
:return: IterableList with the results of the traversal as produced by
traverse()
@@ -329,22 +344,34 @@ class Traversable(object):
id = "" # shouldn't reach here, unless Traversable subclass created with no _id_attribute_
# could add _id_attribute_ to Traversable, or make all Traversable also Iterable?
- out: IterableList[Union['Commit', 'Submodule', 'Tree', 'Blob']] = IterableList(id)
- # overloads in subclasses (mypy does't allow typing self: subclass)
- # Union[IterableList['Commit'], IterableList['Submodule'], IterableList[Union['Submodule', 'Tree', 'Blob']]]
-
- # NOTE: if is_edge=True, self.traverse returns a Tuple, so should be prevented or flattened?
- kwargs['as_edge'] = False
- out.extend(self.traverse(*args, **kwargs)) # type: ignore
- return out
-
- def traverse(self,
- predicate: Callable[[Union['Traversable', 'Blob', TraversedTup], int], bool] = lambda i, d: True,
- prune: Callable[[Union['Traversable', 'Blob', TraversedTup], int], bool] = lambda i, d: False,
- depth: int = -1, branch_first: bool = True, visit_once: bool = True,
- ignore_self: int = 1, as_edge: bool = False
- ) -> Union[Iterator[Union['Traversable', 'Blob']],
- Iterator[TraversedTup]]:
+ if not as_edge:
+ out: IterableList[Union['Commit', 'Submodule', 'Tree', 'Blob']] = IterableList(id)
+ out.extend(self.traverse(as_edge=as_edge, *args, **kwargs)) # type: ignore
+ return out
+ # overloads in subclasses (mypy does't allow typing self: subclass)
+ # Union[IterableList['Commit'], IterableList['Submodule'], IterableList[Union['Submodule', 'Tree', 'Blob']]]
+ else:
+ # Raise deprecationwarning, doesn't make sense to use this
+ out_list: IterableList = IterableList(self.traverse(*args, **kwargs))
+ return out_list
+
+ @ abstractmethod
+ def traverse(self, *args: Any, **kwargs: Any) -> Any:
+ """ """
+ warnings.warn("traverse() method should only be called from subclasses."
+ "Calling from Traversable abstract class will raise NotImplementedError in 3.1.20"
+ "Builtin sublclasses are 'Submodule', 'Tree' and 'Commit",
+ DeprecationWarning,
+ stacklevel=2)
+ return self._traverse(*args, **kwargs)
+
+ def _traverse(self,
+ predicate: Callable[[Union['Traversable', 'Blob', TraversedTup], int], bool] = lambda i, d: True,
+ prune: Callable[[Union['Traversable', 'Blob', TraversedTup], int], bool] = lambda i, d: False,
+ depth: int = -1, branch_first: bool = True, visit_once: bool = True,
+ ignore_self: int = 1, as_edge: bool = False
+ ) -> Union[Iterator[Union['Traversable', 'Blob']],
+ Iterator[TraversedTup]]:
""":return: iterator yielding of items found when traversing self
:param predicate: f(i,d) returns False if item i at depth d should not be included in the result
@@ -387,7 +414,7 @@ class Traversable(object):
ignore_self=False is_edge=False -> Iterator[Tuple[src, item]]"""
visited = set()
- stack = deque() # type: Deque[TraverseNT]
+ stack: Deque[TraverseNT] = deque()
stack.append(TraverseNT(0, self, None)) # self is always depth level 0
def addToStack(stack: Deque[TraverseNT],
@@ -435,11 +462,13 @@ class Traversable(object):
# END for each item on work stack
-class Serializable(object):
+@ runtime_checkable
+class Serializable(Protocol):
"""Defines methods to serialize and deserialize objects from and into a data stream"""
__slots__ = ()
+ # @abstractmethod
def _serialize(self, stream: 'BytesIO') -> 'Serializable':
"""Serialize the data of this object into the given data stream
:note: a serialized object would ``_deserialize`` into the same object
@@ -447,6 +476,7 @@ class Serializable(object):
:return: self"""
raise NotImplementedError("To be implemented in subclass")
+ # @abstractmethod
def _deserialize(self, stream: 'BytesIO') -> 'Serializable':
"""Deserialize all information regarding this object from the stream
:param stream: a file-like object
@@ -454,13 +484,13 @@ class Serializable(object):
raise NotImplementedError("To be implemented in subclass")
-class TraversableIterableObj(Traversable, IterableObj):
+class TraversableIterableObj(IterableObj, Traversable):
__slots__ = ()
TIobj_tuple = Tuple[Union[T_TIobj, None], T_TIobj]
- def list_traverse(self: T_TIobj, *args: Any, **kwargs: Any) -> IterableList[T_TIobj]: # type: ignore[override]
- return super(TraversableIterableObj, self).list_traverse(* args, **kwargs)
+ def list_traverse(self: T_TIobj, *args: Any, **kwargs: Any) -> IterableList[T_TIobj]:
+ return super(TraversableIterableObj, self)._list_traverse(* args, **kwargs)
@ overload # type: ignore
def traverse(self: T_TIobj,
@@ -522,6 +552,6 @@ class TraversableIterableObj(Traversable, IterableObj):
"""
return cast(Union[Iterator[T_TIobj],
Iterator[Tuple[Union[None, T_TIobj], T_TIobj]]],
- super(TraversableIterableObj, self).traverse(
+ super(TraversableIterableObj, self)._traverse(
predicate, prune, depth, branch_first, visit_once, ignore_self, as_edge # type: ignore
))
diff --git a/git/refs/__init__.py b/git/refs/__init__.py
index ded8b1f7..1486dffe 100644
--- a/git/refs/__init__.py
+++ b/git/refs/__init__.py
@@ -1,5 +1,4 @@
# flake8: noqa
-from __future__ import absolute_import
# import all modules in order, fix the names they require
from .symbolic import *
from .reference import *
diff --git a/git/refs/head.py b/git/refs/head.py
index 97c8e6a1..338efce9 100644
--- a/git/refs/head.py
+++ b/git/refs/head.py
@@ -5,12 +5,18 @@ from git.exc import GitCommandError
from .symbolic import SymbolicReference
from .reference import Reference
-from typing import Union, TYPE_CHECKING
+# typinng ---------------------------------------------------
-from git.types import Commit_ish
+from typing import Any, Sequence, Union, TYPE_CHECKING
+
+from git.types import PathLike, Commit_ish
if TYPE_CHECKING:
from git.repo import Repo
+ from git.objects import Commit
+ from git.refs import RemoteReference
+
+# -------------------------------------------------------------------
__all__ = ["HEAD", "Head"]
@@ -29,20 +35,21 @@ class HEAD(SymbolicReference):
_ORIG_HEAD_NAME = 'ORIG_HEAD'
__slots__ = ()
- def __init__(self, repo: 'Repo', path=_HEAD_NAME):
+ def __init__(self, repo: 'Repo', path: PathLike = _HEAD_NAME):
if path != self._HEAD_NAME:
raise ValueError("HEAD instance must point to %r, got %r" % (self._HEAD_NAME, path))
super(HEAD, self).__init__(repo, path)
- self.commit: 'Commit_ish'
+ self.commit: 'Commit'
- def orig_head(self) -> 'SymbolicReference':
+ def orig_head(self) -> SymbolicReference:
"""
:return: SymbolicReference pointing at the ORIG_HEAD, which is maintained
to contain the previous value of HEAD"""
return SymbolicReference(self.repo, self._ORIG_HEAD_NAME)
- def reset(self, commit: Union[Commit_ish, SymbolicReference, str] = 'HEAD', index=True, working_tree=False,
- paths=None, **kwargs):
+ def reset(self, commit: Union[Commit_ish, SymbolicReference, str] = 'HEAD',
+ index: bool = True, working_tree: bool = False,
+ paths: Union[PathLike, Sequence[PathLike], None] = None, **kwargs: Any) -> 'HEAD':
"""Reset our HEAD to the given commit optionally synchronizing
the index and working tree. The reference we refer to will be set to
commit as well.
@@ -122,7 +129,7 @@ class Head(Reference):
k_config_remote_ref = "merge" # branch to merge from remote
@classmethod
- def delete(cls, repo, *heads, **kwargs):
+ def delete(cls, repo: 'Repo', *heads: 'Head', **kwargs: Any):
"""Delete the given heads
:param force:
@@ -135,7 +142,7 @@ class Head(Reference):
flag = "-D"
repo.git.branch(flag, *heads)
- def set_tracking_branch(self, remote_reference):
+ def set_tracking_branch(self, remote_reference: 'RemoteReference') -> 'Head':
"""
Configure this branch to track the given remote reference. This will alter
this branch's configuration accordingly.
@@ -160,7 +167,7 @@ class Head(Reference):
return self
- def tracking_branch(self):
+ def tracking_branch(self) -> Union['RemoteReference', None]:
"""
:return: The remote_reference we are tracking, or None if we are
not a tracking branch"""
@@ -175,7 +182,7 @@ class Head(Reference):
# we are not a tracking branch
return None
- def rename(self, new_path, force=False):
+ def rename(self, new_path: PathLike, force: bool = False) -> 'Head':
"""Rename self to a new path
:param new_path:
@@ -196,7 +203,7 @@ class Head(Reference):
self.path = "%s/%s" % (self._common_path_default, new_path)
return self
- def checkout(self, force=False, **kwargs):
+ def checkout(self, force: bool = False, **kwargs: Any):
"""Checkout this head by setting the HEAD to this reference, by updating the index
to reflect the tree we point to and by updating the working tree to reflect
the latest index.
@@ -231,7 +238,7 @@ class Head(Reference):
return self.repo.active_branch
#{ Configuration
- def _config_parser(self, read_only):
+ def _config_parser(self, read_only: bool) -> SectionConstraint:
if read_only:
parser = self.repo.config_reader()
else:
@@ -240,13 +247,13 @@ class Head(Reference):
return SectionConstraint(parser, 'branch "%s"' % self.name)
- def config_reader(self):
+ def config_reader(self) -> SectionConstraint:
"""
:return: A configuration parser instance constrained to only read
this instance's values"""
return self._config_parser(read_only=True)
- def config_writer(self):
+ def config_writer(self) -> SectionConstraint:
"""
:return: A configuration writer instance with read-and write access
to options of this head"""
diff --git a/git/refs/log.py b/git/refs/log.py
index f850ba24..643b4114 100644
--- a/git/refs/log.py
+++ b/git/refs/log.py
@@ -1,5 +1,7 @@
+
+from mmap import mmap
import re
-import time
+import time as _time
from git.compat import defenc
from git.objects.util import (
@@ -20,20 +22,33 @@ from git.util import (
import os.path as osp
+# typing ------------------------------------------------------------------
+
+from typing import Iterator, List, Tuple, Union, TYPE_CHECKING
+
+from git.types import PathLike
+
+if TYPE_CHECKING:
+ from git.refs import SymbolicReference
+ from io import BytesIO
+ from git.config import GitConfigParser, SectionConstraint # NOQA
+
+# ------------------------------------------------------------------------------
+
__all__ = ["RefLog", "RefLogEntry"]
-class RefLogEntry(tuple):
+class RefLogEntry(Tuple[str, str, Actor, Tuple[int, int], str]):
"""Named tuple allowing easy access to the revlog data fields"""
_re_hexsha_only = re.compile('^[0-9A-Fa-f]{40}$')
__slots__ = ()
- def __repr__(self):
+ def __repr__(self) -> str:
"""Representation of ourselves in git reflog format"""
return self.format()
- def format(self):
+ def format(self) -> str:
""":return: a string suitable to be placed in a reflog file"""
act = self.actor
time = self.time
@@ -46,22 +61,22 @@ class RefLogEntry(tuple):
self.message)
@property
- def oldhexsha(self):
+ def oldhexsha(self) -> str:
"""The hexsha to the commit the ref pointed to before the change"""
return self[0]
@property
- def newhexsha(self):
+ def newhexsha(self) -> str:
"""The hexsha to the commit the ref now points to, after the change"""
return self[1]
@property
- def actor(self):
+ def actor(self) -> Actor:
"""Actor instance, providing access"""
return self[2]
@property
- def time(self):
+ def time(self) -> Tuple[int, int]:
"""time as tuple:
* [0] = int(time)
@@ -69,12 +84,13 @@ class RefLogEntry(tuple):
return self[3]
@property
- def message(self):
+ def message(self) -> str:
"""Message describing the operation that acted on the reference"""
return self[4]
@classmethod
- def new(cls, oldhexsha, newhexsha, actor, time, tz_offset, message): # skipcq: PYL-W0621
+ def new(cls, oldhexsha: str, newhexsha: str, actor: Actor, time: int, tz_offset: int, message: str
+ ) -> 'RefLogEntry': # skipcq: PYL-W0621
""":return: New instance of a RefLogEntry"""
if not isinstance(actor, Actor):
raise ValueError("Need actor instance, got %s" % actor)
@@ -111,14 +127,15 @@ class RefLogEntry(tuple):
# END handle missing end brace
actor = Actor._from_string(info[82:email_end + 1])
- time, tz_offset = parse_date(info[email_end + 2:]) # skipcq: PYL-W0621
+ time, tz_offset = parse_date(
+ info[email_end + 2:]) # skipcq: PYL-W0621
return RefLogEntry((oldhexsha, newhexsha, actor, (time, tz_offset), msg))
-class RefLog(list, Serializable):
+class RefLog(List[RefLogEntry], Serializable):
- """A reflog contains reflog entries, each of which defines a certain state
+ """A reflog contains RefLogEntrys, each of which defines a certain state
of the head in question. Custom query methods allow to retrieve log entries
by date or by other criteria.
@@ -127,11 +144,11 @@ class RefLog(list, Serializable):
__slots__ = ('_path', )
- def __new__(cls, filepath=None):
+ def __new__(cls, filepath: Union[PathLike, None] = None) -> 'RefLog':
inst = super(RefLog, cls).__new__(cls)
return inst
- def __init__(self, filepath=None):
+ def __init__(self, filepath: Union[PathLike, None] = None):
"""Initialize this instance with an optional filepath, from which we will
initialize our data. The path is also used to write changes back using
the write() method"""
@@ -142,7 +159,8 @@ class RefLog(list, Serializable):
def _read_from_file(self):
try:
- fmap = file_contents_ro_filepath(self._path, stream=True, allow_mmap=True)
+ fmap = file_contents_ro_filepath(
+ self._path, stream=True, allow_mmap=True)
except OSError:
# it is possible and allowed that the file doesn't exist !
return
@@ -154,10 +172,10 @@ class RefLog(list, Serializable):
fmap.close()
# END handle closing of handle
- #{ Interface
+ # { Interface
@classmethod
- def from_file(cls, filepath):
+ def from_file(cls, filepath: PathLike) -> 'RefLog':
"""
:return: a new RefLog instance containing all entries from the reflog
at the given filepath
@@ -166,7 +184,7 @@ class RefLog(list, Serializable):
return cls(filepath)
@classmethod
- def path(cls, ref):
+ def path(cls, ref: 'SymbolicReference') -> str:
"""
:return: string to absolute path at which the reflog of the given ref
instance would be found. The path is not guaranteed to point to a valid
@@ -175,28 +193,34 @@ class RefLog(list, Serializable):
return osp.join(ref.repo.git_dir, "logs", to_native_path(ref.path))
@classmethod
- def iter_entries(cls, stream):
+ def iter_entries(cls, stream: Union[str, 'BytesIO', mmap]) -> Iterator[RefLogEntry]:
"""
:return: Iterator yielding RefLogEntry instances, one for each line read
sfrom the given stream.
:param stream: file-like object containing the revlog in its native format
- or basestring instance pointing to a file to read"""
+ or string instance pointing to a file to read"""
new_entry = RefLogEntry.from_line
if isinstance(stream, str):
- stream = file_contents_ro_filepath(stream)
+ # default args return mmap on py>3
+ _stream = file_contents_ro_filepath(stream)
+ assert isinstance(_stream, mmap)
+ else:
+ _stream = stream
# END handle stream type
while True:
- line = stream.readline()
+ line = _stream.readline()
if not line:
return
yield new_entry(line.strip())
# END endless loop
- stream.close()
@classmethod
- def entry_at(cls, filepath, index):
- """:return: RefLogEntry at the given index
+ def entry_at(cls, filepath: PathLike, index: int) -> 'RefLogEntry':
+ """
+ :return: RefLogEntry at the given index
+
:param filepath: full path to the index file from which to read the entry
+
:param index: python list compatible index, i.e. it may be negative to
specify an entry counted from the end of the list
@@ -210,21 +234,19 @@ class RefLog(list, Serializable):
if index < 0:
return RefLogEntry.from_line(fp.readlines()[index].strip())
# read until index is reached
+
for i in range(index + 1):
line = fp.readline()
if not line:
- break
+ raise IndexError(
+ f"Index file ended at line {i+1}, before given index was reached")
# END abort on eof
# END handle runup
- if i != index or not line: # skipcq:PYL-W0631
- raise IndexError
- # END handle exception
-
return RefLogEntry.from_line(line.strip())
# END handle index
- def to_file(self, filepath):
+ def to_file(self, filepath: PathLike) -> None:
"""Write the contents of the reflog instance to a file at the given filepath.
:param filepath: path to file, parent directories are assumed to exist"""
lfd = LockedFD(filepath)
@@ -241,65 +263,75 @@ class RefLog(list, Serializable):
# END handle change
@classmethod
- def append_entry(cls, config_reader, filepath, oldbinsha, newbinsha, message):
+ def append_entry(cls, config_reader: Union[Actor, 'GitConfigParser', 'SectionConstraint', None],
+ filepath: PathLike, oldbinsha: bytes, newbinsha: bytes, message: str,
+ write: bool = True) -> 'RefLogEntry':
"""Append a new log entry to the revlog at filepath.
:param config_reader: configuration reader of the repository - used to obtain
- user information. May also be an Actor instance identifying the committer directly.
- May also be None
+ user information. May also be an Actor instance identifying the committer directly or None.
:param filepath: full path to the log file
:param oldbinsha: binary sha of the previous commit
:param newbinsha: binary sha of the current commit
:param message: message describing the change to the reference
:param write: If True, the changes will be written right away. Otherwise
the change will not be written
+
:return: RefLogEntry objects which was appended to the log
+
:note: As we are append-only, concurrent access is not a problem as we
do not interfere with readers."""
+
if len(oldbinsha) != 20 or len(newbinsha) != 20:
raise ValueError("Shas need to be given in binary format")
# END handle sha type
assure_directory_exists(filepath, is_file=True)
first_line = message.split('\n')[0]
- committer = isinstance(config_reader, Actor) and config_reader or Actor.committer(config_reader)
+ if isinstance(config_reader, Actor):
+ committer = config_reader # mypy thinks this is Actor | Gitconfigparser, but why?
+ else:
+ committer = Actor.committer(config_reader)
entry = RefLogEntry((
bin_to_hex(oldbinsha).decode('ascii'),
bin_to_hex(newbinsha).decode('ascii'),
- committer, (int(time.time()), time.altzone), first_line
+ committer, (int(_time.time()), _time.altzone), first_line
))
- lf = LockFile(filepath)
- lf._obtain_lock_or_raise()
- fd = open(filepath, 'ab')
- try:
- fd.write(entry.format().encode(defenc))
- finally:
- fd.close()
- lf._release_lock()
- # END handle write operation
-
+ if write:
+ lf = LockFile(filepath)
+ lf._obtain_lock_or_raise()
+ fd = open(filepath, 'ab')
+ try:
+ fd.write(entry.format().encode(defenc))
+ finally:
+ fd.close()
+ lf._release_lock()
+ # END handle write operation
return entry
- def write(self):
+ def write(self) -> 'RefLog':
"""Write this instance's data to the file we are originating from
:return: self"""
if self._path is None:
- raise ValueError("Instance was not initialized with a path, use to_file(...) instead")
+ raise ValueError(
+ "Instance was not initialized with a path, use to_file(...) instead")
# END assert path
self.to_file(self._path)
return self
- #} END interface
+ # } END interface
- #{ Serializable Interface
- def _serialize(self, stream):
+ # { Serializable Interface
+ def _serialize(self, stream: 'BytesIO') -> 'RefLog':
write = stream.write
# write all entries
for e in self:
write(e.format().encode(defenc))
# END for each entry
+ return self
- def _deserialize(self, stream):
+ def _deserialize(self, stream: 'BytesIO') -> 'RefLog':
self.extend(self.iter_entries(stream))
- #} END serializable interface
+ # } END serializable interface
+ return self
diff --git a/git/refs/reference.py b/git/refs/reference.py
index 8a9b0487..f584bb54 100644
--- a/git/refs/reference.py
+++ b/git/refs/reference.py
@@ -2,7 +2,18 @@ from git.util import (
LazyMixin,
IterableObj,
)
-from .symbolic import SymbolicReference
+from .symbolic import SymbolicReference, T_References
+
+
+# typing ------------------------------------------------------------------
+
+from typing import Any, Callable, Iterator, List, Match, Optional, Tuple, Type, TypeVar, Union, TYPE_CHECKING # NOQA
+from git.types import Commit_ish, PathLike, TBD, Literal, TypeGuard, _T # NOQA
+
+if TYPE_CHECKING:
+ from git.repo import Repo
+
+# ------------------------------------------------------------------------------
__all__ = ["Reference"]
@@ -10,10 +21,10 @@ __all__ = ["Reference"]
#{ Utilities
-def require_remote_ref_path(func):
+def require_remote_ref_path(func: Callable[..., _T]) -> Callable[..., _T]:
"""A decorator raising a TypeError if we are not a valid remote, based on the path"""
- def wrapper(self, *args):
+ def wrapper(self: T_References, *args: Any) -> _T:
if not self.is_remote():
raise ValueError("ref path does not point to a remote reference: %s" % self.path)
return func(self, *args)
@@ -32,7 +43,7 @@ class Reference(SymbolicReference, LazyMixin, IterableObj):
_resolve_ref_on_create = True
_common_path_default = "refs"
- def __init__(self, repo, path, check_path=True):
+ def __init__(self, repo: 'Repo', path: PathLike, check_path: bool = True) -> None:
"""Initialize this instance
:param repo: Our parent repository
@@ -41,16 +52,17 @@ class Reference(SymbolicReference, LazyMixin, IterableObj):
refs/heads/master
:param check_path: if False, you can provide any path. Otherwise the path must start with the
default path prefix of this type."""
- if check_path and not path.startswith(self._common_path_default + '/'):
- raise ValueError("Cannot instantiate %r from path %s" % (self.__class__.__name__, path))
+ if check_path and not str(path).startswith(self._common_path_default + '/'):
+ raise ValueError(f"Cannot instantiate {self.__class__.__name__!r} from path {path}")
+ self.path: str # SymbolicReference converts to string atm
super(Reference, self).__init__(repo, path)
- def __str__(self):
+ def __str__(self) -> str:
return self.name
#{ Interface
- def set_object(self, object, logmsg=None): # @ReservedAssignment
+ def set_object(self, object: Commit_ish, logmsg: Union[str, None] = None) -> 'Reference': # @ReservedAssignment
"""Special version which checks if the head-log needs an update as well
:return: self"""
oldbinsha = None
@@ -84,7 +96,7 @@ class Reference(SymbolicReference, LazyMixin, IterableObj):
# NOTE: Don't have to overwrite properties as the will only work without a the log
@property
- def name(self):
+ def name(self) -> str:
""":return: (shortest) Name of this reference - it may contain path components"""
# first two path tokens are can be removed as they are
# refs/heads or refs/tags or refs/remotes
@@ -94,7 +106,8 @@ class Reference(SymbolicReference, LazyMixin, IterableObj):
return '/'.join(tokens[2:])
@classmethod
- def iter_items(cls, repo, common_path=None):
+ def iter_items(cls: Type[T_References], repo: 'Repo', common_path: Union[PathLike, None] = None,
+ *args: Any, **kwargs: Any) -> Iterator[T_References]:
"""Equivalent to SymbolicReference.iter_items, but will return non-detached
references as well."""
return cls._iter_items(repo, common_path)
@@ -105,7 +118,7 @@ class Reference(SymbolicReference, LazyMixin, IterableObj):
@property # type: ignore ## mypy cannot deal with properties with an extra decorator (2021-04-21)
@require_remote_ref_path
- def remote_name(self):
+ def remote_name(self) -> str:
"""
:return:
Name of the remote we are a reference of, such as 'origin' for a reference
@@ -116,7 +129,7 @@ class Reference(SymbolicReference, LazyMixin, IterableObj):
@property # type: ignore ## mypy cannot deal with properties with an extra decorator (2021-04-21)
@require_remote_ref_path
- def remote_head(self):
+ def remote_head(self) -> str:
""":return: Name of the remote head itself, i.e. master.
:note: The returned name is usually not qualified enough to uniquely identify
a branch"""
diff --git a/git/refs/remote.py b/git/refs/remote.py
index 0164e110..8a680a4a 100644
--- a/git/refs/remote.py
+++ b/git/refs/remote.py
@@ -2,13 +2,23 @@ import os
from git.util import join_path
-import os.path as osp
-
from .head import Head
__all__ = ["RemoteReference"]
+# typing ------------------------------------------------------------------
+
+from typing import Any, NoReturn, Union, TYPE_CHECKING
+from git.types import PathLike
+
+
+if TYPE_CHECKING:
+ from git.repo import Repo
+ from git import Remote
+
+# ------------------------------------------------------------------------------
+
class RemoteReference(Head):
@@ -16,16 +26,19 @@ class RemoteReference(Head):
_common_path_default = Head._remote_common_path_default
@classmethod
- def iter_items(cls, repo, common_path=None, remote=None):
+ def iter_items(cls, repo: 'Repo', common_path: Union[PathLike, None] = None,
+ remote: Union['Remote', None] = None, *args: Any, **kwargs: Any
+ ) -> 'RemoteReference':
"""Iterate remote references, and if given, constrain them to the given remote"""
common_path = common_path or cls._common_path_default
if remote is not None:
common_path = join_path(common_path, str(remote))
# END handle remote constraint
+ # super is Reference
return super(RemoteReference, cls).iter_items(repo, common_path)
- @classmethod
- def delete(cls, repo, *refs, **kwargs):
+ @ classmethod
+ def delete(cls, repo: 'Repo', *refs: 'RemoteReference', **kwargs: Any) -> None:
"""Delete the given remote references
:note:
@@ -37,16 +50,16 @@ class RemoteReference(Head):
# and delete remainders manually
for ref in refs:
try:
- os.remove(osp.join(repo.common_dir, ref.path))
+ os.remove(os.path.join(repo.common_dir, ref.path))
except OSError:
pass
try:
- os.remove(osp.join(repo.git_dir, ref.path))
+ os.remove(os.path.join(repo.git_dir, ref.path))
except OSError:
pass
# END for each ref
- @classmethod
- def create(cls, *args, **kwargs):
+ @ classmethod
+ def create(cls, *args: Any, **kwargs: Any) -> NoReturn:
"""Used to disable this method"""
raise TypeError("Cannot explicitly create remote references")
diff --git a/git/refs/symbolic.py b/git/refs/symbolic.py
index f0bd9316..426d40d4 100644
--- a/git/refs/symbolic.py
+++ b/git/refs/symbolic.py
@@ -21,6 +21,19 @@ import os.path as osp
from .log import RefLog
+# typing ------------------------------------------------------------------
+
+from typing import Any, Iterator, List, Match, Optional, Tuple, Type, TypeVar, Union, TYPE_CHECKING # NOQA
+from git.types import Commit_ish, PathLike, TBD, Literal, TypeGuard # NOQA
+
+if TYPE_CHECKING:
+ from git.repo import Repo
+
+T_References = TypeVar('T_References', bound='SymbolicReference')
+
+# ------------------------------------------------------------------------------
+
+
__all__ = ["SymbolicReference"]
@@ -46,11 +59,11 @@ class SymbolicReference(object):
_remote_common_path_default = "refs/remotes"
_id_attribute_ = "name"
- def __init__(self, repo, path, check_path=None):
+ def __init__(self, repo: 'Repo', path: PathLike, check_path: bool = False):
self.repo = repo
- self.path = path
+ self.path = str(path)
- def __str__(self):
+ def __str__(self) -> str:
return self.path
def __repr__(self):
@@ -115,8 +128,8 @@ class SymbolicReference(object):
yield tuple(line.split(' ', 1))
# END for each line
- except (OSError, IOError):
- return
+ except OSError:
+ return None
# END no packed-refs file handling
# NOTE: Had try-finally block around here to close the fp,
# but some python version wouldn't allow yields within that.
@@ -149,7 +162,7 @@ class SymbolicReference(object):
# 60b64ef992065e2600bfef6187a97f92398a9144 branch 'master' of git-server:/path/to/repo
tokens = value.split()
assert(len(tokens) != 0)
- except (OSError, IOError):
+ except OSError:
# Probably we are just packed, find our entry in the packed refs file
# NOTE: We are not a symbolic ref if we are in a packed file, as these
# are excluded explicitly
@@ -205,7 +218,7 @@ class SymbolicReference(object):
# END handle type
return obj
- def set_commit(self, commit, logmsg=None):
+ def set_commit(self, commit: Union[Commit, 'SymbolicReference', str], logmsg=None):
"""As set_object, but restricts the type of object to be a Commit
:raise ValueError: If commit is not a Commit object or doesn't point to
@@ -344,7 +357,7 @@ class SymbolicReference(object):
# aliased reference
reference = property(_get_reference, set_reference, doc="Returns the Reference we point to")
- ref = reference
+ ref: Union[Commit_ish] = reference # type: ignore # Union[str, Commit_ish, SymbolicReference]
def is_valid(self):
"""
@@ -471,7 +484,7 @@ class SymbolicReference(object):
with open(pack_file_path, 'wb') as fd:
fd.writelines(line.encode(defenc) for line in new_lines)
- except (OSError, IOError):
+ except OSError:
pass # it didn't exist at all
# delete the reflog
@@ -514,8 +527,9 @@ class SymbolicReference(object):
return ref
@classmethod
- def create(cls, repo, path, reference='HEAD', force=False, logmsg=None, **kwargs):
- """Create a new symbolic reference, hence a reference pointing to another reference.
+ def create(cls, repo: 'Repo', path: PathLike, reference: Union[Commit_ish, str] = 'HEAD',
+ logmsg: Union[str, None] = None, force: bool = False, **kwargs: Any):
+ """Create a new symbolic reference, hence a reference pointing , to another reference.
:param repo:
Repository to create the reference in
@@ -591,7 +605,8 @@ class SymbolicReference(object):
return self
@classmethod
- def _iter_items(cls, repo, common_path=None):
+ def _iter_items(cls: Type[T_References], repo: 'Repo', common_path: Union[PathLike, None] = None
+ ) -> Iterator[T_References]:
if common_path is None:
common_path = cls._common_path_default
rela_paths = set()
@@ -629,7 +644,8 @@ class SymbolicReference(object):
# END for each sorted relative refpath
@classmethod
- def iter_items(cls, repo, common_path=None):
+ # type: ignore[override]
+ def iter_items(cls, repo: 'Repo', common_path: Union[PathLike, None] = None, *args, **kwargs):
"""Find all refs in the repository
:param repo: is the Repo
diff --git a/git/refs/tag.py b/git/refs/tag.py
index 4d84239e..281ce09a 100644
--- a/git/refs/tag.py
+++ b/git/refs/tag.py
@@ -2,6 +2,19 @@ from .reference import Reference
__all__ = ["TagReference", "Tag"]
+# typing ------------------------------------------------------------------
+
+from typing import Any, Union, TYPE_CHECKING
+from git.types import Commit_ish, PathLike
+
+if TYPE_CHECKING:
+ from git.repo import Repo
+ from git.objects import Commit
+ from git.objects import TagObject
+
+
+# ------------------------------------------------------------------------------
+
class TagReference(Reference):
@@ -22,9 +35,9 @@ class TagReference(Reference):
_common_path_default = Reference._common_path_default + "/" + _common_default
@property
- def commit(self):
+ def commit(self) -> 'Commit': # type: ignore[override] # LazyMixin has unrelated comit method
""":return: Commit object the tag ref points to
-
+
:raise ValueError: if the tag points to a tree or blob"""
obj = self.object
while obj.type != 'commit':
@@ -37,7 +50,7 @@ class TagReference(Reference):
return obj
@property
- def tag(self):
+ def tag(self) -> Union['TagObject', None]:
"""
:return: Tag object this tag ref points to or None in case
we are a light weight tag"""
@@ -48,10 +61,16 @@ class TagReference(Reference):
# make object read-only
# It should be reasonably hard to adjust an existing tag
- object = property(Reference._get_object)
+
+ # object = property(Reference._get_object)
+ @property
+ def object(self) -> Commit_ish: # type: ignore[override]
+ return Reference._get_object(self)
@classmethod
- def create(cls, repo, path, ref='HEAD', message=None, force=False, **kwargs):
+ def create(cls, repo: 'Repo', path: PathLike, reference: Union[Commit_ish, str] = 'HEAD',
+ logmsg: Union[str, None] = None,
+ force: bool = False, **kwargs: Any) -> 'TagReference':
"""Create a new tag reference.
:param path:
@@ -62,12 +81,16 @@ class TagReference(Reference):
A reference to the object you want to tag. It can be a commit, tree or
blob.
- :param message:
+ :param logmsg:
If not None, the message will be used in your tag object. This will also
create an additional tag object that allows to obtain that information, i.e.::
tagref.tag.message
+ :param message:
+ Synonym for :param logmsg:
+ Included for backwards compatability. :param logmsg is used in preference if both given.
+
:param force:
If True, to force creation of a tag even though that tag already exists.
@@ -75,9 +98,12 @@ class TagReference(Reference):
Additional keyword arguments to be passed to git-tag
:return: A new TagReference"""
- args = (path, ref)
- if message:
- kwargs['m'] = message
+ args = (path, reference)
+ if logmsg:
+ kwargs['m'] = logmsg
+ elif 'message' in kwargs and kwargs['message']:
+ kwargs['m'] = kwargs['message']
+
if force:
kwargs['f'] = True
@@ -85,7 +111,7 @@ class TagReference(Reference):
return TagReference(repo, "%s/%s" % (cls._common_path_default, path))
@classmethod
- def delete(cls, repo, *tags):
+ def delete(cls, repo: 'Repo', *tags: 'TagReference') -> None:
"""Delete the given existing tag or tags"""
repo.git.tag("-d", *tags)
diff --git a/git/remote.py b/git/remote.py
index f59b3245..d903552f 100644
--- a/git/remote.py
+++ b/git/remote.py
@@ -36,9 +36,10 @@ from .refs import (
# typing-------------------------------------------------------
-from typing import Any, Callable, Dict, Iterator, List, Optional, Sequence, TYPE_CHECKING, Union, overload
+from typing import (Any, Callable, Dict, Iterator, List, NoReturn, Optional, Sequence, # NOQA[TC002]
+ TYPE_CHECKING, Type, Union, overload)
-from git.types import PathLike, Literal, TBD, TypeGuard, Commit_ish
+from git.types import PathLike, Literal, TBD, TypeGuard, Commit_ish # NOQA[TC002]
if TYPE_CHECKING:
from git.repo.base import Repo
@@ -83,17 +84,17 @@ def add_progress(kwargs: Any, git: Git,
#} 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:
...
@@ -155,11 +156,11 @@ class PushInfo(IterableObj, object):
self._old_commit_sha = old_commit
self.summary = summary
- @property
- def old_commit(self) -> Union[str, SymbolicReference, 'Commit_ish', None]:
+ @ 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
- @property
+ @ property
def remote_ref(self) -> Union[RemoteReference, TagReference]:
"""
:return:
@@ -175,7 +176,7 @@ class PushInfo(IterableObj, object):
raise ValueError("Could not handle remote ref: %r" % self.remote_ref_string)
# END
- @classmethod
+ @ 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"""
@@ -192,7 +193,7 @@ class PushInfo(IterableObj, object):
# from_to handling
from_ref_string, to_ref_string = from_to.split(':')
if flags & cls.DELETED:
- from_ref = None # type: Union[SymbolicReference, None]
+ from_ref: Union[SymbolicReference, None] = None
else:
if from_ref_string == "(delete)":
from_ref = None
@@ -200,7 +201,7 @@ class PushInfo(IterableObj, object):
from_ref = Reference.from_path(remote.repo, from_ref_string)
# commit handling, could be message or commit info
- old_commit = None # type: Optional[str]
+ old_commit: Optional[str] = None
if summary.startswith('['):
if "[rejected]" in summary:
flags |= cls.REJECTED
@@ -228,6 +229,11 @@ class PushInfo(IterableObj, object):
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 FetchInfo(IterableObj, object):
@@ -253,16 +259,16 @@ class FetchInfo(IterableObj, object):
_re_fetch_result = re.compile(r'^\s*(.) (\[?[\w\s\.$@]+\]?)\s+(.+) -> ([^\s]+)( \(.*\)?$)?')
- _flag_map = {
+ _flag_map: Dict[flagKeyLiteral, int] = {
'!': ERROR,
'+': FORCED_UPDATE,
'*': 0,
'=': HEAD_UPTODATE,
' ': FAST_FORWARD,
'-': TAG_UPDATE,
- } # type: Dict[flagKeyLiteral, int]
+ }
- @classmethod
+ @ classmethod
def refresh(cls) -> Literal[True]:
"""This gets called by the refresh function (see the top level
__init__).
@@ -301,25 +307,25 @@ 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
+ @ 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.
- We can handle a line as follows
- "%c %-*s %-*s -> %s%s"
+ We can handle a line as follows:
+ "%c %-\\*s %-\\*s -> %s%s"
- Where c is either ' ', !, +, -, *, or =
+ Where c is either ' ', !, +, -, \\*, or =
! means error
+ means success forcing update
- means a tag was updated
@@ -334,6 +340,7 @@ class FetchInfo(IterableObj, object):
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()
assert is_flagKeyLiteral(control_character), f"{control_character}"
@@ -352,7 +359,7 @@ class FetchInfo(IterableObj, object):
# END control char exception handling
# parse operation string for more info - makes no sense for symbolic refs, but we parse it anyway
- old_commit = None # type: Union[Commit_ish, None]
+ old_commit: Union[Commit_ish, None] = None
is_tag_operation = False
if 'rejected' in operation:
flags |= cls.REJECTED
@@ -375,7 +382,7 @@ class FetchInfo(IterableObj, object):
# 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 = None
+ 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:
@@ -404,14 +411,15 @@ class FetchInfo(IterableObj, object):
# 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 = None # type: Optional[PathLike]
+ 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
# 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 + "/"):
+ 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:
@@ -430,6 +438,11 @@ class FetchInfo(IterableObj, object):
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):
@@ -507,7 +520,7 @@ class Remote(LazyMixin, IterableObj):
return False
# end
- @classmethod
+ @ 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():
@@ -833,7 +846,7 @@ class Remote(LazyMixin, IterableObj):
kwargs = add_progress(kwargs, self.repo.git, progress)
if isinstance(refspec, list):
- args = refspec # type: Sequence[Optional[str]] # should need this - check logic for passing None through
+ args: Sequence[Optional[str]] = refspec
else:
args = [refspec]
@@ -897,7 +910,7 @@ class Remote(LazyMixin, IterableObj):
universal_newlines=True, **kwargs)
return self._get_push_info(proc, progress)
- @property
+ @ property
def config_reader(self) -> SectionConstraint:
"""
:return:
@@ -912,7 +925,7 @@ class Remote(LazyMixin, IterableObj):
pass
# END handle exception
- @property
+ @ property
def config_writer(self) -> SectionConstraint:
"""
:return: GitConfigParser compatible object able to write options for this remote.
diff --git a/git/repo/__init__.py b/git/repo/__init__.py
index 5619aa69..712df60d 100644
--- a/git/repo/__init__.py
+++ b/git/repo/__init__.py
@@ -1,4 +1,3 @@
"""Initialize the Repo package"""
# flake8: noqa
-from __future__ import absolute_import
from .base import *
diff --git a/git/repo/base.py b/git/repo/base.py
index 3214b528..64f32bd3 100644
--- a/git/repo/base.py
+++ b/git/repo/base.py
@@ -83,10 +83,10 @@ class Repo(object):
DAEMON_EXPORT_FILE = 'git-daemon-export-ok'
git = cast('Git', None) # Must exist, or __del__ will fail in case we raise on `__init__()`
- working_dir = None # type: Optional[PathLike]
- _working_tree_dir = None # type: Optional[PathLike]
- git_dir = "" # type: PathLike
- _common_dir = "" # type: PathLike
+ working_dir: Optional[PathLike] = None
+ _working_tree_dir: Optional[PathLike] = None
+ git_dir: PathLike = ""
+ _common_dir: PathLike = ""
# precompiled regex
re_whitespace = re.compile(r'\s+')
@@ -221,7 +221,7 @@ class Repo(object):
self._working_tree_dir = None
# END working dir handling
- self.working_dir = self._working_tree_dir or self.common_dir # type: Optional[PathLike]
+ self.working_dir: Optional[PathLike] = self._working_tree_dir or self.common_dir
self.git = self.GitCommandWrapperType(self.working_dir)
# special handling, in special times
@@ -426,7 +426,7 @@ class Repo(object):
For more documentation, please see the Head.create method.
:return: newly created Head Reference"""
- return Head.create(self, path, commit, force, logmsg)
+ return Head.create(self, path, commit, logmsg, force)
def delete_head(self, *heads: 'SymbolicReference', **kwargs: Any) -> None:
"""Delete the given heads
@@ -518,7 +518,7 @@ class Repo(object):
repository = configuration file for this repository only"""
return GitConfigParser(self._get_config_path(config_level), read_only=False, repo=self)
- def commit(self, rev: Optional[str] = None
+ def commit(self, rev: Union[str, Commit_ish, None] = None
) -> Commit:
"""The Commit object for the specified revision
@@ -551,7 +551,8 @@ class Repo(object):
return self.head.commit.tree
return self.rev_parse(str(rev) + "^{tree}")
- def iter_commits(self, rev: Optional[TBD] = None, paths: Union[PathLike, Sequence[PathLike]] = '',
+ def iter_commits(self, rev: Union[str, Commit, 'SymbolicReference', None] = None,
+ paths: Union[PathLike, Sequence[PathLike]] = '',
**kwargs: Any) -> Iterator[Commit]:
"""A list of Commit objects representing the history of a given ref/commit
@@ -590,7 +591,7 @@ class Repo(object):
raise ValueError("Please specify at least two revs, got only %i" % len(rev))
# end handle input
- res = [] # type: List[Union[Commit_ish, None]]
+ res: List[Union[Commit_ish, None]] = []
try:
lines = self.git.merge_base(*rev, **kwargs).splitlines() # List[str]
except GitCommandError as err:
@@ -812,7 +813,7 @@ class Repo(object):
line = next(stream) # when exhausted, causes a StopIteration, terminating this function
except StopIteration:
return
- split_line = line.split() # type: Tuple[str, str, str, str]
+ split_line: Tuple[str, str, str, str] = line.split()
hexsha, orig_lineno_str, lineno_str, num_lines_str = split_line
lineno = int(lineno_str)
num_lines = int(num_lines_str)
@@ -878,10 +879,10 @@ class Repo(object):
return self.blame_incremental(rev, file, **kwargs)
data = self.git.blame(rev, '--', file, p=True, stdout_as_string=False, **kwargs)
- commits = {} # type: Dict[str, Any]
- blames = [] # type: List[List[Union[Optional['Commit'], List[str]]]]
+ commits: Dict[str, TBD] = {}
+ blames: List[List[Union[Optional['Commit'], List[str]]]] = []
- info = {} # type: Dict[str, Any] # use Any until TypedDict available
+ info: Dict[str, TBD] = {} # use Any until TypedDict available
keepends = True
for line in data.splitlines(keepends):
diff --git a/git/repo/fun.py b/git/repo/fun.py
index e96b62e0..7d5c7823 100644
--- a/git/repo/fun.py
+++ b/git/repo/fun.py
@@ -1,5 +1,4 @@
"""Package with general repository related functions"""
-from git.refs.tag import Tag
import os
import stat
from string import digits
@@ -19,11 +18,14 @@ from git.cmd import Git
# Typing ----------------------------------------------------------------------
from typing import Union, Optional, cast, TYPE_CHECKING
-from git.types import PathLike
+
+
if TYPE_CHECKING:
+ from git.types import PathLike
from .base import Repo
from git.db import GitCmdObjectDB
from git.objects import Commit, TagObject, Blob, Tree
+ from git.refs.tag import Tag
# ----------------------------------------------------------------------------
@@ -37,7 +39,7 @@ def touch(filename: str) -> str:
return filename
-def is_git_dir(d: PathLike) -> bool:
+def is_git_dir(d: 'PathLike') -> bool:
""" This is taken from the git setup.c:is_git_directory
function.
@@ -59,7 +61,7 @@ def is_git_dir(d: PathLike) -> bool:
return False
-def find_worktree_git_dir(dotgit: PathLike) -> Optional[str]:
+def find_worktree_git_dir(dotgit: 'PathLike') -> Optional[str]:
"""Search for a gitdir for this worktree."""
try:
statbuf = os.stat(dotgit)
@@ -78,7 +80,7 @@ def find_worktree_git_dir(dotgit: PathLike) -> Optional[str]:
return None
-def find_submodule_git_dir(d: PathLike) -> Optional[PathLike]:
+def find_submodule_git_dir(d: 'PathLike') -> Optional['PathLike']:
"""Search for a submodule repo."""
if is_git_dir(d):
return d
@@ -122,7 +124,7 @@ def name_to_object(repo: 'Repo', name: str, return_ref: bool = False
:param return_ref: if name specifies a reference, we will return the reference
instead of the object. Otherwise it will raise BadObject or BadName
"""
- hexsha = None # type: Union[None, str, bytes]
+ hexsha: Union[None, str, bytes] = None
# is it a hexsha ? Try the most common ones, which is 7 to 40
if repo.re_hexsha_shortened.match(name):
@@ -162,7 +164,7 @@ def name_to_object(repo: 'Repo', name: str, return_ref: bool = False
return Object.new_from_sha(repo, hex_to_bin(hexsha))
-def deref_tag(tag: Tag) -> 'TagObject':
+def deref_tag(tag: 'Tag') -> 'TagObject':
"""Recursively dereference a tag and return the resulting object"""
while True:
try:
diff --git a/git/types.py b/git/types.py
index 9181e040..53f0f1e4 100644
--- a/git/types.py
+++ b/git/types.py
@@ -7,9 +7,6 @@ import sys
from typing import (Callable, Dict, NoReturn, Sequence, Tuple, Union, Any, Iterator, # noqa: F401
NamedTuple, TYPE_CHECKING, TypeVar) # noqa: F401
-if TYPE_CHECKING:
- from git.repo import Repo
-
if sys.version_info[:2] >= (3, 8):
from typing import Final, Literal, SupportsIndex, TypedDict, Protocol, runtime_checkable # noqa: F401
else:
@@ -28,6 +25,7 @@ elif sys.version_info[:2] >= (3, 9):
PathLike = Union[str, 'os.PathLike[str]'] # forward ref as pylance complains unless editing with py3.9+
if TYPE_CHECKING:
+ from git.repo import Repo
from git.objects import Commit, Tree, TagObject, Blob
# from git.refs import SymbolicReference
@@ -36,6 +34,7 @@ _T = TypeVar('_T')
Tree_ish = Union['Commit', 'Tree']
Commit_ish = Union['Commit', 'TagObject', 'Blob', 'Tree']
+Lit_commit_ish = Literal['commit', 'tag', 'blob', 'tree']
# Config_levels ---------------------------------------------------------
diff --git a/git/util.py b/git/util.py
index 571e261e..c0c0ecb7 100644
--- a/git/util.py
+++ b/git/util.py
@@ -4,6 +4,7 @@
# This module is part of GitPython and is released under
# the BSD License: http://www.opensource.org/licenses/bsd-license.php
+from abc import abstractmethod
from .exc import InvalidGitRepositoryError
import os.path as osp
from .compat import is_win
@@ -28,7 +29,8 @@ import warnings
# typing ---------------------------------------------------------
from typing import (Any, AnyStr, BinaryIO, Callable, Dict, Generator, IO, Iterator, List,
- Optional, Pattern, Sequence, Tuple, TypeVar, Union, cast, TYPE_CHECKING, overload)
+ Optional, Pattern, Sequence, Tuple, TypeVar, Union, cast,
+ TYPE_CHECKING, overload, )
import pathlib
@@ -39,8 +41,8 @@ if TYPE_CHECKING:
# from git.objects.base import IndexObject
-from .types import (Literal, SupportsIndex, # because behind py version guards
- PathLike, HSH_TD, Total_TD, Files_TD, # aliases
+from .types import (Literal, SupportsIndex, Protocol, runtime_checkable, # because behind py version guards
+ PathLike, HSH_TD, Total_TD, Files_TD, # aliases
Has_id_attribute)
T_IterableObj = TypeVar('T_IterableObj', bound=Union['IterableObj', 'Has_id_attribute'], covariant=True)
@@ -265,7 +267,7 @@ def _cygexpath(drive: Optional[str], path: str) -> str:
return p_str.replace('\\', '/')
-_cygpath_parsers = (
+_cygpath_parsers: Tuple[Tuple[Pattern[str], Callable, bool], ...] = (
# See: https://msdn.microsoft.com/en-us/library/windows/desktop/aa365247(v=vs.85).aspx
# and: https://www.cygwin.com/cygwin-ug-net/using.html#unc-paths
(re.compile(r"\\\\\?\\UNC\\([^\\]+)\\([^\\]+)(?:\\(.*))?"),
@@ -292,7 +294,7 @@ _cygpath_parsers = (
(lambda url: url),
False
),
-) # type: Tuple[Tuple[Pattern[str], Callable, bool], ...]
+)
def cygpath(path: str) -> str:
@@ -328,7 +330,7 @@ def decygpath(path: PathLike) -> str:
#: Store boolean flags denoting if a specific Git executable
#: is from a Cygwin installation (since `cache_lru()` unsupported on PY2).
-_is_cygwin_cache = {} # type: Dict[str, Optional[bool]]
+_is_cygwin_cache: Dict[str, Optional[bool]] = {}
@overload
@@ -460,10 +462,10 @@ class RemoteProgress(object):
re_op_relative = re.compile(r"(remote: )?([\w\s]+):\s+(\d+)% \((\d+)/(\d+)\)(.*)")
def __init__(self) -> None:
- self._seen_ops = [] # type: List[int]
- self._cur_line = None # type: Optional[str]
- self.error_lines = [] # type: List[str]
- self.other_lines = [] # type: List[str]
+ self._seen_ops: List[int] = []
+ self._cur_line: Optional[str] = None
+ self.error_lines: List[str] = []
+ self.other_lines: List[str] = []
def _parse_progress_line(self, line: AnyStr) -> None:
"""Parse progress information from the given line as retrieved by git-push
@@ -471,7 +473,7 @@ class RemoteProgress(object):
- Lines that do not contain progress info are stored in :attr:`other_lines`.
- Lines that seem to contain an error (i.e. start with error: or fatal:) are stored
- in :attr:`error_lines`."""
+ in :attr:`error_lines`."""
# handle
# Counting objects: 4, done.
# Compressing objects: 50% (1/2)
@@ -993,7 +995,7 @@ class IterableList(List[T_IterableObj]):
# END for each item
return list.__getattribute__(self, attr)
- def __getitem__(self, index: Union[SupportsIndex, int, slice, str]) -> 'T_IterableObj': # type: ignore
+ def __getitem__(self, index: Union[SupportsIndex, int, slice, str]) -> T_IterableObj: # type: ignore
assert isinstance(index, (int, str, slice)), "Index of IterableList should be an int or str"
@@ -1030,23 +1032,24 @@ class IterableList(List[T_IterableObj]):
class IterableClassWatcher(type):
+ """ Metaclass that watches """
def __init__(cls, name, bases, clsdict):
for base in bases:
if type(base) == IterableClassWatcher:
warnings.warn(f"GitPython Iterable subclassed by {name}. "
- "Iterable is deprecated due to naming clash, "
+ "Iterable is deprecated due to naming clash since v3.1.18"
+ " and will be removed in 3.1.20, "
"Use IterableObj instead \n",
DeprecationWarning,
stacklevel=2)
-class Iterable(object):
+class Iterable(metaclass=IterableClassWatcher):
"""Defines an interface for iterable items which is to assure a uniform
way to retrieve and iterate items within the git repository"""
__slots__ = ()
_id_attribute_ = "attribute that most suitably identifies your instance"
- __metaclass__ = IterableClassWatcher
@classmethod
def list_items(cls, repo, *args, **kwargs):
@@ -1064,14 +1067,15 @@ class Iterable(object):
return out_list
@classmethod
- def iter_items(cls, repo: 'Repo', *args: Any, **kwargs: Any):
+ def iter_items(cls, repo: 'Repo', *args: Any, **kwargs: Any) -> Any:
# return typed to be compatible with subtypes e.g. Remote
"""For more information about the arguments, see list_items
:return: iterator yielding Items"""
raise NotImplementedError("To be implemented by Subclass")
-class IterableObj():
+@runtime_checkable
+class IterableObj(Protocol):
"""Defines an interface for iterable items which is to assure a uniform
way to retrieve and iterate items within the git repository
@@ -1095,11 +1099,12 @@ class IterableObj():
return out_list
@classmethod
+ @abstractmethod
def iter_items(cls, repo: 'Repo', *args: Any, **kwargs: Any
- ) -> Iterator[T_IterableObj]:
+ ) -> Iterator[T_IterableObj]: # Iterator[T_IterableObj]:
# return typed to be compatible with subtypes e.g. Remote
"""For more information about the arguments, see list_items
- :return: iterator yielding Items"""
+ :return: iterator yielding Items"""
raise NotImplementedError("To be implemented by Subclass")
# } END classes