diff options
Diffstat (limited to 'git/objects/commit.py')
| -rw-r--r-- | git/objects/commit.py | 142 |
1 files changed, 97 insertions, 45 deletions
diff --git a/git/objects/commit.py b/git/objects/commit.py index 45e6d772..b689167f 100644 --- a/git/objects/commit.py +++ b/git/objects/commit.py @@ -3,12 +3,12 @@ # # This module is part of GitPython and is released under # the BSD License: http://www.opensource.org/licenses/bsd-license.php - +import datetime +from subprocess import Popen from gitdb import IStream from git.util import ( hex_to_bin, Actor, - Iterable, Stats, finalize_process ) @@ -17,8 +17,8 @@ from git.diff import Diffable from .tree import Tree from . import base from .util import ( - Traversable, Serializable, + TraversableIterableObj, parse_date, altz_to_utctz_str, parse_actor_and_date, @@ -36,13 +36,26 @@ import os from io import BytesIO import logging + +# typing ------------------------------------------------------------------ + +from typing import Any, IO, Iterator, List, Sequence, Tuple, Union, TYPE_CHECKING, cast + +from git.types import PathLike, Literal + +if TYPE_CHECKING: + from git.repo import Repo + from git.refs import SymbolicReference + +# ------------------------------------------------------------------------ + log = logging.getLogger('git.objects.commit') log.addHandler(logging.NullHandler()) __all__ = ('Commit', ) -class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): +class Commit(base.Object, TraversableIterableObj, Diffable, Serializable): """Wraps a git Commit object. @@ -61,16 +74,24 @@ class Commit(base.Object, Iterable, Diffable, Traversable, 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, binsha, tree=None, author=None, authored_date=None, author_tz_offset=None, - committer=None, committed_date=None, committer_tz_offset=None, - message=None, parents=None, encoding=None, gpgsig=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, + committer: Union[Actor, None] = None, + committed_date: Union[int, None] = None, + committer_tz_offset: Union[None, float] = None, + message: Union[str, bytes, None] = None, + parents: Union[Sequence['Commit'], None] = None, + encoding: Union[str, None] = None, + gpgsig: Union[str, None] = None) -> None: """Instantiate a new Commit. All keyword arguments taking None as default will be implicitly set on first query. @@ -107,6 +128,7 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): as what time.altzone returns. The sign is inverted compared to git's UTC timezone.""" super(Commit, self).__init__(repo, binsha) + self.binsha = binsha if tree is not None: assert isinstance(tree, Tree), "Tree needs to be a Tree instance, was %s" % type(tree) if tree is not None: @@ -133,11 +155,11 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): self.gpgsig = gpgsig @classmethod - def _get_intermediate_items(cls, commit): - return commit.parents + def _get_intermediate_items(cls, commit: 'Commit') -> Tuple['Commit', ...]: + return tuple(commit.parents) @classmethod - def _calculate_sha_(cls, repo, commit): + def _calculate_sha_(cls, repo: 'Repo', commit: 'Commit') -> bytes: '''Calculate the sha of a commit. :param repo: Repo object the commit should be part of @@ -152,7 +174,7 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): istream = repo.odb.store(IStream(cls.type, streamlen, stream)) return istream.binsha - def replace(self, **kwargs): + def replace(self, **kwargs: Any) -> 'Commit': '''Create new commit object from existing commit object. Any values provided as keyword arguments will replace the @@ -171,7 +193,7 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): return new_commit - def _set_cache_(self, attr): + def _set_cache_(self, attr: str) -> None: if attr in Commit.__slots__: # read the data in a chunk, its faster - then provide a file wrapper _binsha, _typename, self.size, stream = self.repo.odb.stream(self.binsha) @@ -181,19 +203,22 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): # END handle attrs @property - def authored_datetime(self): + def authored_datetime(self) -> datetime.datetime: return from_timestamp(self.authored_date, self.author_tz_offset) @property - def committed_datetime(self): + def committed_datetime(self) -> datetime.datetime: return from_timestamp(self.committed_date, self.committer_tz_offset) @property - def summary(self): + def summary(self) -> Union[str, bytes]: """:return: First line of the commit message""" - return self.message.split('\n', 1)[0] + if isinstance(self.message, str): + return self.message.split('\n', 1)[0] + else: + return self.message.split(b'\n', 1)[0] - def count(self, paths='', **kwargs): + def count(self, paths: Union[PathLike, Sequence[PathLike]] = '', **kwargs: Any) -> int: """Count the number of commits reachable from this commit :param paths: @@ -211,7 +236,7 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): return len(self.repo.git.rev_list(self.hexsha, **kwargs).splitlines()) @property - def name_rev(self): + def name_rev(self) -> str: """ :return: String describing the commits hex sha based on the closest Reference. @@ -219,7 +244,9 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): return self.repo.git.name_rev(self) @classmethod - def iter_items(cls, repo, rev, paths='', **kwargs): + 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. :param repo: is the Repo @@ -239,15 +266,23 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): # use -- in any case, to prevent possibility of ambiguous arguments # see https://github.com/gitpython-developers/GitPython/issues/264 - args = ['--'] + + args_list: List[PathLike] = ['--'] + if paths: - args.extend((paths, )) + paths_tup: Tuple[PathLike, ...] + if isinstance(paths, (str, os.PathLike)): + paths_tup = (paths, ) + else: + paths_tup = tuple(paths) + + args_list.extend(paths_tup) # END if paths - proc = repo.git.rev_list(rev, args, as_process=True, **kwargs) + proc = repo.git.rev_list(rev, args_list, as_process=True, **kwargs) return cls._iter_from_process_or_stream(repo, proc) - def iter_parents(self, paths='', **kwargs): + def iter_parents(self, paths: Union[PathLike, Sequence[PathLike]] = '', **kwargs: Any) -> Iterator['Commit']: """Iterate _all_ parents of this commit. :param paths: @@ -263,8 +298,8 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): return self.iter_items(self.repo, self, paths, **kwargs) - @property - def stats(self): + @ property + def stats(self) -> Stats: """Create a git stat from changes between this commit and its first parent or from all changes done if this is the very first commit. @@ -280,17 +315,28 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): text = self.repo.git.diff(self.parents[0].hexsha, self.hexsha, '--', numstat=True) return Stats._list_from_string(self.repo, text) - @classmethod - def _iter_from_process_or_stream(cls, repo, proc_or_stream): + @ classmethod + def _iter_from_process_or_stream(cls, repo: 'Repo', proc_or_stream: Union[Popen, IO]) -> Iterator['Commit']: """Parse out commit information into a list of Commit objects We expect one-line per commit, and parse the actual commit information directly from our lighting fast object database :param proc: git-rev-list process instance - one sha per line :return: iterator returning Commit objects""" - stream = proc_or_stream - if not hasattr(stream, 'readline'): - stream = proc_or_stream.stdout + + # def is_proc(inp) -> TypeGuard[Popen]: + # return hasattr(proc_or_stream, 'wait') and not hasattr(proc_or_stream, 'readline') + + # def is_stream(inp) -> TypeGuard[IO]: + # return hasattr(proc_or_stream, 'readline') + + if hasattr(proc_or_stream, 'wait'): + proc_or_stream = cast(Popen, proc_or_stream) + if proc_or_stream.stdout is not None: + stream = proc_or_stream.stdout + elif hasattr(proc_or_stream, 'readline'): + proc_or_stream = cast(IO, proc_or_stream) + stream = proc_or_stream readline = stream.readline while True: @@ -309,18 +355,21 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): # TODO: Review this - it seems process handling got a bit out of control # due to many developers trying to fix the open file handles issue if hasattr(proc_or_stream, 'wait'): + proc_or_stream = cast(Popen, proc_or_stream) finalize_process(proc_or_stream) - @classmethod - def create_from_tree(cls, repo, tree, message, parent_commits=None, head=False, author=None, committer=None, - author_date=None, commit_date=None): + @ classmethod + 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) -> 'Commit': """Commit the given tree, creating a commit object. :param repo: Repo object the commit should be part of :param tree: Tree object or hex or bin sha the tree of the new commit :param message: Commit message. It may be an empty string if no message is provided. - It will be converted to a string in any case. + It will be converted to a string , in any case. :param parent_commits: Optional Commit objects to use as parents for the new commit. If empty list, the commit will have no parents at all and become @@ -354,7 +403,7 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): else: for p in parent_commits: if not isinstance(p, cls): - raise ValueError("Parent commit '%r' must be of type %s" % (p, cls)) + raise ValueError(f"Parent commit '{p!r}' must be of type {cls}") # end check parent commit types # END if parent commits are unset @@ -430,7 +479,7 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): #{ Serializable Implementation - def _serialize(self, stream): + def _serialize(self, stream: BytesIO) -> 'Commit': write = stream.write write(("tree %s\n" % self.tree).encode('ascii')) for p in self.parents: @@ -454,7 +503,7 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): write(("encoding %s\n" % self.encoding).encode('ascii')) try: - if self.__getattribute__('gpgsig') is not None: + if self.__getattribute__('gpgsig'): write(b"gpgsig") for sigline in self.gpgsig.rstrip("\n").split("\n"): write((" " + sigline + "\n").encode('ascii')) @@ -471,9 +520,11 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): # END handle encoding return self - def _deserialize(self, stream): - """: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""" + 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 + """ readline = stream.readline self.tree = Tree(self.repo, hex_to_bin(readline().split()[1]), Tree.tree_id << 12, '') @@ -504,14 +555,14 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): # now we can have the encoding line, or an empty line followed by the optional # message. self.encoding = self.default_encoding - self.gpgsig = None + self.gpgsig = "" # read headers enc = next_line buf = enc.strip() while buf: if buf[0:10] == b"encoding ": - self.encoding = buf[buf.find(' ') + 1:].decode( + self.encoding = buf[buf.find(b' ') + 1:].decode( self.encoding, 'ignore') elif buf[0:7] == b"gpgsig ": sig = buf[buf.find(b' ') + 1:] + b"\n" @@ -533,7 +584,7 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): # decode the authors name try: - self.author, self.authored_date, self.author_tz_offset = \ + (self.author, self.authored_date, self.author_tz_offset) = \ parse_actor_and_date(author_line.decode(self.encoding, 'replace')) except UnicodeDecodeError: log.error("Failed to decode author line '%s' using encoding %s", author_line, self.encoding, @@ -553,7 +604,8 @@ class Commit(base.Object, Iterable, Diffable, Traversable, Serializable): try: self.message = self.message.decode(self.encoding, 'replace') except UnicodeDecodeError: - log.error("Failed to decode message '%s' using encoding %s", self.message, self.encoding, exc_info=True) + log.error("Failed to decode message '%s' using encoding %s", + self.message, self.encoding, exc_info=True) # END exception handling return self |