diff options
Diffstat (limited to 'git/util.py')
| -rw-r--r-- | git/util.py | 235 |
1 files changed, 171 insertions, 64 deletions
diff --git a/git/util.py b/git/util.py index af499028..4f82219e 100644 --- a/git/util.py +++ b/git/util.py @@ -4,6 +4,10 @@ # 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 import contextlib from functools import wraps import getpass @@ -18,15 +22,32 @@ from sys import maxsize import time from unittest import SkipTest from urllib.parse import urlsplit, urlunsplit +import warnings + +# from git.objects.util import Traversable # typing --------------------------------------------------------- -from typing import (Any, AnyStr, BinaryIO, Callable, Dict, Generator, IO, List, - NoReturn, Optional, Pattern, Sequence, Tuple, Union, cast, TYPE_CHECKING) +from typing import (Any, AnyStr, BinaryIO, Callable, Dict, Generator, IO, Iterator, List, + Optional, Pattern, Sequence, Tuple, TypeVar, Union, cast, + TYPE_CHECKING, overload, ) + +import pathlib + if TYPE_CHECKING: from git.remote import Remote from git.repo.base import Repo -from .types import PathLike, TBD + from git.config import GitConfigParser, SectionConstraint + from git import Git + # from git.objects.base import IndexObject + + +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) +# So IterableList[Head] is subtype of IterableList[IterableObj] # --------------------------------------------------------------------- @@ -43,18 +64,13 @@ from gitdb.util import ( # NOQA @IgnorePep8 hex_to_bin, # @UnusedImport ) -from .compat import is_win -import os.path as osp - -from .exc import InvalidGitRepositoryError - # NOTE: Some of the unused imports might be used/imported by others. # Handle once test-cases are back up and running. # Most of these are unused here, but are for use by git-python modules so these # don't see gitdb all the time. Flake of course doesn't like it. __all__ = ["stream_copy", "join_path", "to_native_path_linux", - "join_path_native", "Stats", "IndexFileSHA1Writer", "Iterable", "IterableList", + "join_path_native", "Stats", "IndexFileSHA1Writer", "IterableObj", "IterableList", "BlockingLockFile", "LockFile", 'Actor', 'get_user_id', 'assure_directory_exists', 'RemoteProgress', 'CallableRemoteProgress', 'rmtree', 'unbare_repo', 'HIDE_WINDOWS_KNOWN_ERRORS'] @@ -70,15 +86,17 @@ log = logging.getLogger(__name__) HIDE_WINDOWS_KNOWN_ERRORS = is_win and os.environ.get('HIDE_WINDOWS_KNOWN_ERRORS', True) HIDE_WINDOWS_FREEZE_ERRORS = is_win and os.environ.get('HIDE_WINDOWS_FREEZE_ERRORS', True) -#{ Utility Methods +# { Utility Methods +T = TypeVar('T') -def unbare_repo(func: Callable) -> Callable: + +def unbare_repo(func: Callable[..., T]) -> Callable[..., T]: """Methods with this decorator raise InvalidGitRepositoryError if they encounter a bare repository""" @wraps(func) - def wrapper(self: 'Remote', *args: Any, **kwargs: Any) -> TBD: + def wrapper(self: 'Remote', *args: Any, **kwargs: Any) -> T: if self.repo.bare: raise InvalidGitRepositoryError("Method '%s' cannot operate on bare repositories" % func.__name__) # END bare method @@ -104,7 +122,7 @@ def rmtree(path: PathLike) -> None: :note: we use shutil rmtree but adjust its behaviour to see whether files that couldn't be deleted are read-only. Windows will not remove them in that case""" - def onerror(func: Callable, path: PathLike, exc_info: TBD) -> None: + def onerror(func: Callable, path: PathLike, exc_info: str) -> None: # Is the error an access error ? os.chmod(path, stat.S_IWUSR) @@ -165,7 +183,7 @@ if is_win: path = str(path) return path.replace('/', '\\') - def to_native_path_linux(path: PathLike) -> PathLike: + def to_native_path_linux(path: PathLike) -> str: path = str(path) return path.replace('\\', '/') @@ -173,8 +191,9 @@ if is_win: to_native_path = to_native_path_windows else: # no need for any work on linux - def to_native_path_linux(path: PathLike) -> PathLike: - return path + def to_native_path_linux(path: PathLike) -> str: + return str(path) + to_native_path = to_native_path_linux @@ -230,9 +249,9 @@ def py_where(program: str, path: Optional[PathLike] = None) -> List[str]: return progs -def _cygexpath(drive: Optional[str], path: PathLike) -> str: +def _cygexpath(drive: Optional[str], path: str) -> str: if osp.isabs(path) and not drive: - ## Invoked from `cygpath()` directly with `D:Apps\123`? + # Invoked from `cygpath()` directly with `D:Apps\123`? # It's an error, leave it alone just slashes) p = path # convert to str if AnyPath given else: @@ -249,9 +268,9 @@ def _cygexpath(drive: Optional[str], path: PathLike) -> str: return p_str.replace('\\', '/') -_cygpath_parsers = ( - ## 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 +_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\\([^\\]+)\\([^\\]+)(?:\\(.*))?"), (lambda server, share, rest_path: '//%s/%s/%s' % (server, share, rest_path.replace('\\', '/'))), False @@ -276,12 +295,13 @@ _cygpath_parsers = ( (lambda url: url), False ), -) # type: Tuple[Tuple[Pattern[str], Callable, bool], ...] +) -def cygpath(path: PathLike) -> PathLike: +def cygpath(path: str) -> str: """Use :meth:`git.cmd.Git.polish_url()` instead, that works on any environment.""" - path = str(path) # ensure is str and not AnyPath + path = str(path) # ensure is str and not AnyPath. + # Fix to use Paths when 3.5 dropped. or to be just str if only for urls? if not path.startswith(('/cygdrive', '//')): for regex, parser, recurse in _cygpath_parsers: match = regex.match(path) @@ -311,14 +331,26 @@ 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 +def is_cygwin_git(git_executable: None) -> Literal[False]: + ... + +@overload def is_cygwin_git(git_executable: PathLike) -> bool: + ... + + +def is_cygwin_git(git_executable: Union[None, PathLike]) -> bool: if not is_win: return False - #from subprocess import check_output + if git_executable is None: + return False + git_executable = str(git_executable) is_cygwin = _is_cygwin_cache.get(git_executable) # type: Optional[bool] if is_cygwin is None: @@ -329,7 +361,7 @@ def is_cygwin_git(git_executable: PathLike) -> bool: res = py_where(git_executable) git_dir = osp.dirname(res[0]) if res else "" - ## Just a name given, not a real path. + # Just a name given, not a real path. uname_cmd = osp.join(git_dir, 'uname') process = subprocess.Popen([uname_cmd], stdout=subprocess.PIPE, universal_newlines=True) @@ -348,23 +380,36 @@ def get_user_id() -> str: return "%s@%s" % (getpass.getuser(), platform.node()) -def finalize_process(proc: TBD, **kwargs: Any) -> None: +def finalize_process(proc: Union[subprocess.Popen, 'Git.AutoInterrupt'], **kwargs: Any) -> None: """Wait for the process (clone, fetch, pull or push) and handle its errors accordingly""" - ## TODO: No close proc-streams?? + # TODO: No close proc-streams?? proc.wait(**kwargs) -def expand_path(p: PathLike, expand_vars: bool = True) -> Optional[PathLike]: +@overload +def expand_path(p: None, expand_vars: bool = ...) -> None: + ... + + +@overload +def expand_path(p: PathLike, expand_vars: bool = ...) -> str: + # improve these overloads when 3.5 dropped + ... + + +def expand_path(p: Union[None, PathLike], expand_vars: bool = True) -> Optional[PathLike]: + if isinstance(p, pathlib.Path): + return p.resolve() try: - p = osp.expanduser(p) + p = osp.expanduser(p) # type: ignore if expand_vars: - p = osp.expandvars(p) - return osp.normpath(osp.abspath(p)) + p = osp.expandvars(p) # type: ignore + return osp.normpath(osp.abspath(p)) # type: ignore except Exception: return None -def remove_password_if_present(cmdline): +def remove_password_if_present(cmdline: Sequence[str]) -> List[str]: """ Parse any command line argument and if on of the element is an URL with a password, replace it by stars (in-place). @@ -391,9 +436,9 @@ def remove_password_if_present(cmdline): return new_cmdline -#} END utilities +# } END utilities -#{ Classes +# { Classes class RemoteProgress(object): @@ -401,7 +446,7 @@ class RemoteProgress(object): Handler providing an interface to parse progress information emitted by git-push and git-fetch and to dispatch callbacks allowing subclasses to react to the progress. """ - _num_op_codes = 9 + _num_op_codes: int = 9 BEGIN, END, COUNTING, COMPRESSING, WRITING, RECEIVING, RESOLVING, FINDING_SOURCES, CHECKING_OUT = \ [1 << x for x in range(_num_op_codes)] STAGE_MASK = BEGIN | END @@ -418,10 +463,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[TBD] - 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 @@ -429,7 +474,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) @@ -441,7 +486,7 @@ class RemoteProgress(object): line_str = line self._cur_line = line_str - if self.error_lines or self._cur_line.startswith(('error:', 'fatal:')): + if self._cur_line.startswith(('error:', 'fatal:')): self.error_lines.append(self._cur_line) return @@ -639,7 +684,8 @@ class Actor(object): # END handle name/email matching @classmethod - def _main_actor(cls, env_name: str, env_email: str, config_reader: Optional[TBD] = None) -> 'Actor': + def _main_actor(cls, env_name: str, env_email: str, + config_reader: Union[None, 'GitConfigParser', 'SectionConstraint'] = None) -> 'Actor': actor = Actor('', '') user_id = None # We use this to avoid multiple calls to getpass.getuser() @@ -668,7 +714,7 @@ class Actor(object): return actor @classmethod - def committer(cls, config_reader: Optional[TBD] = None) -> 'Actor': + def committer(cls, config_reader: Union[None, 'GitConfigParser', 'SectionConstraint'] = None) -> 'Actor': """ :return: Actor instance corresponding to the configured committer. It behaves similar to the git implementation, such that the environment will override @@ -679,7 +725,7 @@ class Actor(object): return cls._main_actor(cls.env_committer_name, cls.env_committer_email, config_reader) @classmethod - def author(cls, config_reader: Optional[TBD] = None) -> 'Actor': + def author(cls, config_reader: Union[None, 'GitConfigParser', 'SectionConstraint'] = None) -> 'Actor': """Same as committer(), but defines the main author. It may be specified in the environment, but defaults to the committer""" return cls._main_actor(cls.env_author_name, cls.env_author_email, config_reader) @@ -713,7 +759,7 @@ class Stats(object): files = number of changed files as int""" __slots__ = ("total", "files") - def __init__(self, total: Dict[str, Dict[str, int]], files: Dict[str, Dict[str, int]]): + def __init__(self, total: Total_TD, files: Dict[PathLike, Files_TD]): self.total = total self.files = files @@ -722,9 +768,13 @@ class Stats(object): """Create a Stat object from output retrieved by git-diff. :return: git.Stat""" - hsh = {'total': {'insertions': 0, 'deletions': 0, 'lines': 0, 'files': 0}, - 'files': {} - } # type: Dict[str, Dict[str, TBD]] ## need typeddict or refactor for mypy + + hsh: HSH_TD = {'total': {'insertions': 0, + 'deletions': 0, + 'lines': 0, + 'files': 0}, + 'files': {} + } for line in text.splitlines(): (raw_insertions, raw_deletions, filename) = line.split("\t") insertions = raw_insertions != '-' and int(raw_insertions) or 0 @@ -733,9 +783,10 @@ class Stats(object): hsh['total']['deletions'] += deletions hsh['total']['lines'] += insertions + deletions hsh['total']['files'] += 1 - hsh['files'][filename.strip()] = {'insertions': insertions, - 'deletions': deletions, - 'lines': insertions + deletions} + files_dict: Files_TD = {'insertions': insertions, + 'deletions': deletions, + 'lines': insertions + deletions} + hsh['files'][filename.strip()] = files_dict return Stats(hsh['total'], hsh['files']) @@ -891,7 +942,7 @@ class BlockingLockFile(LockFile): # END endless loop -class IterableList(list): +class IterableList(List[T_IterableObj]): """ List of iterable objects allowing to query an object by id or by named index:: @@ -901,6 +952,9 @@ class IterableList(list): heads['master'] heads[0] + Iterable parent objects = [Commit, SubModule, Reference, FetchInfo, PushInfo] + Iterable via inheritance = [Head, TagReference, RemoteReference] + ] It requires an id_attribute name to be set which will be queried from its contained items to have a means for comparison. @@ -909,7 +963,7 @@ class IterableList(list): can be left out.""" __slots__ = ('_id_attr', '_prefix') - def __new__(cls, id_attr: str, prefix: str = '') -> 'IterableList': + def __new__(cls, id_attr: str, prefix: str = '') -> 'IterableList[IterableObj]': return super(IterableList, cls).__new__(cls) def __init__(self, id_attr: str, prefix: str = '') -> None: @@ -934,7 +988,7 @@ class IterableList(list): return False # END handle membership - def __getattr__(self, attr: str) -> Any: + def __getattr__(self, attr: str) -> T_IterableObj: attr = self._prefix + attr for item in self: if getattr(item, self._id_attr) == attr: @@ -942,7 +996,10 @@ class IterableList(list): # END for each item return list.__getattribute__(self, attr) - def __getitem__(self, index: Union[int, slice, str]) -> Any: + 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" + if isinstance(index, int): return list.__getitem__(self, index) elif isinstance(index, slice): @@ -954,12 +1011,13 @@ class IterableList(list): raise IndexError("No item found with id %r" % (self._prefix + index)) from e # END handle getattr - def __delitem__(self, index: Union[int, str, slice]) -> None: + def __delitem__(self, index: Union[SupportsIndex, int, slice, str]) -> None: + + assert isinstance(index, (int, str)), "Index of IterableList should be an int or str" delindex = cast(int, index) if not isinstance(index, int): delindex = -1 - assert not isinstance(index, slice) name = self._prefix + index for i, item in enumerate(self): if getattr(item, self._id_attr) == name: @@ -974,7 +1032,20 @@ class IterableList(list): list.__delitem__(self, delindex) -class Iterable(object): +class IterableClassWatcher(type): + """ Metaclass that watches """ + def __init__(cls, name: str, bases: Tuple, clsdict: Dict) -> None: + for base in bases: + if type(base) == IterableClassWatcher: + warnings.warn(f"GitPython Iterable subclassed by {name}. " + "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(metaclass=IterableClassWatcher): """Defines an interface for iterable items which is to assure a uniform way to retrieve and iterate items within the git repository""" @@ -982,8 +1053,9 @@ class Iterable(object): _id_attribute_ = "attribute that most suitably identifies your instance" @classmethod - def list_items(cls, repo: 'Repo', *args: Any, **kwargs: Any) -> 'IterableList': + def list_items(cls, repo: 'Repo', *args: Any, **kwargs: Any) -> Any: """ + Deprecated, use IterableObj instead. Find all items of this type - subclasses can specify args and kwargs differently. If no args are given, subclasses are obliged to return all items if no additional arguments arg given. @@ -991,17 +1063,52 @@ class Iterable(object): :note: Favor the iter_items method as it will :return:list(Item,...) list of item instances""" - out_list = IterableList(cls._id_attribute_) + out_list: Any = IterableList(cls._id_attribute_) out_list.extend(cls.iter_items(repo, *args, **kwargs)) return out_list @classmethod - def iter_items(cls, repo: 'Repo', *args: Any, **kwargs: Any) -> NoReturn: + 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") -#} END classes + +@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 + + Subclasses = [Submodule, Commit, Reference, PushInfo, FetchInfo, Remote]""" + + __slots__ = () + _id_attribute_: str + + @classmethod + def list_items(cls, repo: 'Repo', *args: Any, **kwargs: Any) -> IterableList[T_IterableObj]: + """ + Find all items of this type - subclasses can specify args and kwargs differently. + If no args are given, subclasses are obliged to return all items if no additional + arguments arg given. + + :note: Favor the iter_items method as it will + + :return:list(Item,...) list of item instances""" + out_list: IterableList = IterableList(cls._id_attribute_) + out_list.extend(cls.iter_items(repo, *args, **kwargs)) + return out_list + + @classmethod + @abstractmethod + def iter_items(cls, repo: 'Repo', *args: Any, **kwargs: Any + ) -> 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""" + raise NotImplementedError("To be implemented by Subclass") + +# } END classes class NullHandler(logging.Handler): |