diff options
Diffstat (limited to 'git/refs')
| -rw-r--r-- | git/refs/head.py | 87 | ||||
| -rw-r--r-- | git/refs/log.py | 117 | ||||
| -rw-r--r-- | git/refs/reference.py | 58 | ||||
| -rw-r--r-- | git/refs/remote.py | 21 | ||||
| -rw-r--r-- | git/refs/symbolic.py | 266 | ||||
| -rw-r--r-- | git/refs/tag.py | 42 |
6 files changed, 385 insertions, 206 deletions
diff --git a/git/refs/head.py b/git/refs/head.py index d1d72c7b..befdc135 100644 --- a/git/refs/head.py +++ b/git/refs/head.py @@ -31,15 +31,18 @@ class HEAD(SymbolicReference): """Special case of a Symbolic Reference as it represents the repository's HEAD reference.""" - _HEAD_NAME = 'HEAD' - _ORIG_HEAD_NAME = 'ORIG_HEAD' + + _HEAD_NAME = "HEAD" + _ORIG_HEAD_NAME = "ORIG_HEAD" __slots__ = () - def __init__(self, repo: 'Repo', path: PathLike = _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)) + raise ValueError( + "HEAD instance must point to %r, got %r" % (self._HEAD_NAME, path) + ) super(HEAD, self).__init__(repo, path) - self.commit: 'Commit' + self.commit: "Commit" def orig_head(self) -> SymbolicReference: """ @@ -47,9 +50,14 @@ class HEAD(SymbolicReference): 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: bool = True, working_tree: bool = False, - paths: Union[PathLike, Sequence[PathLike], None] = None, **kwargs: Any) -> 'HEAD': + 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. @@ -90,12 +98,14 @@ class HEAD(SymbolicReference): if working_tree: mode = "--hard" if not index: - raise ValueError("Cannot reset the working tree if the index is not reset as well") + raise ValueError( + "Cannot reset the working tree if the index is not reset as well" + ) # END working tree handling try: - self.repo.git.reset(mode, commit, '--', paths, **kwargs) + self.repo.git.reset(mode, commit, "--", paths, **kwargs) except GitCommandError as e: # git nowadays may use 1 as status to indicate there are still unstaged # modifications after the reset @@ -124,12 +134,19 @@ class Head(Reference): >>> head.commit.hexsha '1c09f116cbc2cb4100fb6935bb162daa4723f455'""" + _common_path_default = "refs/heads" k_config_remote = "remote" - k_config_remote_ref = "merge" # branch to merge from remote + k_config_remote_ref = "merge" # branch to merge from remote @classmethod - def delete(cls, repo: 'Repo', *heads: 'Union[Head, str]', force: bool = False, **kwargs: Any) -> None: + def delete( + cls, + repo: "Repo", + *heads: "Union[Head, str]", + force: bool = False, + **kwargs: Any + ) -> None: """Delete the given heads :param force: @@ -141,7 +158,9 @@ class Head(Reference): flag = "-D" repo.git.branch(flag, *heads) - def set_tracking_branch(self, remote_reference: Union['RemoteReference', None]) -> 'Head': + def set_tracking_branch( + self, remote_reference: Union["RemoteReference", None] + ) -> "Head": """ Configure this branch to track the given remote reference. This will alter this branch's configuration accordingly. @@ -150,7 +169,10 @@ class Head(Reference): any references :return: self""" from .remote import RemoteReference - if remote_reference is not None and not isinstance(remote_reference, RemoteReference): + + if remote_reference is not None and not isinstance( + remote_reference, RemoteReference + ): raise ValueError("Incorrect parameter type: %r" % remote_reference) # END handle type @@ -162,26 +184,39 @@ class Head(Reference): writer.remove_section() else: writer.set_value(self.k_config_remote, remote_reference.remote_name) - writer.set_value(self.k_config_remote_ref, Head.to_full_path(remote_reference.remote_head)) + writer.set_value( + self.k_config_remote_ref, + Head.to_full_path(remote_reference.remote_head), + ) return self - def tracking_branch(self) -> Union['RemoteReference', None]: + def tracking_branch(self) -> Union["RemoteReference", None]: """ :return: The remote_reference we are tracking, or None if we are not a tracking branch""" from .remote import RemoteReference + reader = self.config_reader() - if reader.has_option(self.k_config_remote) and reader.has_option(self.k_config_remote_ref): - ref = Head(self.repo, Head.to_full_path(strip_quotes(reader.get_value(self.k_config_remote_ref)))) - remote_refpath = RemoteReference.to_full_path(join_path(reader.get_value(self.k_config_remote), ref.name)) + if reader.has_option(self.k_config_remote) and reader.has_option( + self.k_config_remote_ref + ): + ref = Head( + self.repo, + Head.to_full_path( + strip_quotes(reader.get_value(self.k_config_remote_ref)) + ), + ) + remote_refpath = RemoteReference.to_full_path( + join_path(reader.get_value(self.k_config_remote), ref.name) + ) return RemoteReference(self.repo, remote_refpath) # END handle have tracking branch # we are not a tracking branch return None - def rename(self, new_path: PathLike, force: bool = False) -> 'Head': + def rename(self, new_path: PathLike, force: bool = False) -> "Head": """Rename self to a new path :param new_path: @@ -202,7 +237,7 @@ class Head(Reference): self.path = "%s/%s" % (self._common_path_default, new_path) return self - def checkout(self, force: bool = False, **kwargs: Any) -> Union['HEAD', 'Head']: + def checkout(self, force: bool = False, **kwargs: Any) -> Union["HEAD", "Head"]: """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. @@ -227,9 +262,9 @@ class Head(Reference): By default it is only allowed to checkout heads - everything else will leave the HEAD detached which is allowed and possible, but remains a special state that some tools might not be able to handle.""" - kwargs['f'] = force - if kwargs['f'] is False: - kwargs.pop('f') + kwargs["f"] = force + if kwargs["f"] is False: + kwargs.pop("f") self.repo.git.checkout(self, **kwargs) if self.repo.head.is_detached: @@ -237,7 +272,7 @@ class Head(Reference): else: return self.repo.active_branch - #{ Configuration + # { Configuration def _config_parser(self, read_only: bool) -> SectionConstraint[GitConfigParser]: if read_only: parser = self.repo.config_reader() @@ -259,4 +294,4 @@ class Head(Reference): to options of this head""" return self._config_parser(read_only=False) - #} END configuration + # } END configuration diff --git a/git/refs/log.py b/git/refs/log.py index ddd78bc7..908f93d1 100644 --- a/git/refs/log.py +++ b/git/refs/log.py @@ -1,4 +1,3 @@ - from mmap import mmap import re import time as _time @@ -16,7 +15,7 @@ from git.util import ( assure_directory_exists, to_native_path, bin_to_hex, - file_contents_ro_filepath + file_contents_ro_filepath, ) import os.path as osp @@ -41,7 +40,8 @@ __all__ = ["RefLog", "RefLogEntry"] 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}$') + + _re_hexsha_only = re.compile("^[0-9A-Fa-f]{40}$") __slots__ = () def __repr__(self) -> str: @@ -52,13 +52,15 @@ class RefLogEntry(Tuple[str, str, Actor, Tuple[int, int], str]): """:return: a string suitable to be placed in a reflog file""" act = self.actor time = self.time - return "{} {} {} <{}> {!s} {}\t{}\n".format(self.oldhexsha, - self.newhexsha, - act.name, - act.email, - time[0], - altz_to_utctz_str(time[1]), - self.message) + return "{} {} {} <{}> {!s} {}\t{}\n".format( + self.oldhexsha, + self.newhexsha, + act.name, + act.email, + time[0], + altz_to_utctz_str(time[1]), + self.message, + ) @property def oldhexsha(self) -> str: @@ -80,7 +82,7 @@ class RefLogEntry(Tuple[str, str, Actor, Tuple[int, int], str]): """time as tuple: * [0] = int(time) - * [1] = int(timezone_offset) in time.altzone format """ + * [1] = int(timezone_offset) in time.altzone format""" return self[3] @property @@ -89,8 +91,15 @@ class RefLogEntry(Tuple[str, str, Actor, Tuple[int, int], str]): return self[4] @classmethod - def new(cls, oldhexsha: str, newhexsha: str, actor: Actor, time: int, tz_offset: int, message: str - ) -> 'RefLogEntry': # 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) @@ -98,19 +107,21 @@ class RefLogEntry(Tuple[str, str, Actor, Tuple[int, int], str]): return RefLogEntry((oldhexsha, newhexsha, actor, (time, tz_offset), message)) @classmethod - def from_line(cls, line: bytes) -> 'RefLogEntry': + def from_line(cls, line: bytes) -> "RefLogEntry": """:return: New RefLogEntry instance from the given revlog line. :param line: line bytes without trailing newline :raise ValueError: If line could not be parsed""" line_str = line.decode(defenc) - fields = line_str.split('\t', 1) + fields = line_str.split("\t", 1) if len(fields) == 1: info, msg = fields[0], None elif len(fields) == 2: info, msg = fields else: - raise ValueError("Line must have up to two TAB-separated fields." - " Got %s" % repr(line_str)) + raise ValueError( + "Line must have up to two TAB-separated fields." + " Got %s" % repr(line_str) + ) # END handle first split oldhexsha = info[:40] @@ -121,14 +132,13 @@ class RefLogEntry(Tuple[str, str, Actor, Tuple[int, int], str]): # END if hexsha re doesn't match # END for each hexsha - email_end = info.find('>', 82) + email_end = info.find(">", 82) if email_end == -1: raise ValueError("Missing token: >") # 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 + actor = Actor._from_string(info[82 : email_end + 1]) + time, tz_offset = parse_date(info[email_end + 2 :]) # skipcq: PYL-W0621 return RefLogEntry((oldhexsha, newhexsha, actor, (time, tz_offset), msg)) @@ -142,9 +152,9 @@ class RefLog(List[RefLogEntry], Serializable): Reflog entries are ordered, the first added entry is first in the list, the last entry, i.e. the last change of the head or reference, is last in the list.""" - __slots__ = ('_path', ) + __slots__ = ("_path",) - def __new__(cls, filepath: Union[PathLike, None] = None) -> 'RefLog': + def __new__(cls, filepath: Union[PathLike, None] = None) -> "RefLog": inst = super(RefLog, cls).__new__(cls) return inst @@ -159,8 +169,7 @@ class RefLog(List[RefLogEntry], Serializable): def _read_from_file(self) -> None: 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 @@ -175,7 +184,7 @@ class RefLog(List[RefLogEntry], Serializable): # { Interface @classmethod - def from_file(cls, filepath: PathLike) -> 'RefLog': + def from_file(cls, filepath: PathLike) -> "RefLog": """ :return: a new RefLog instance containing all entries from the reflog at the given filepath @@ -184,7 +193,7 @@ class RefLog(List[RefLogEntry], Serializable): return cls(filepath) @classmethod - def path(cls, ref: 'SymbolicReference') -> str: + 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 @@ -193,7 +202,7 @@ class RefLog(List[RefLogEntry], Serializable): return osp.join(ref.repo.git_dir, "logs", to_native_path(ref.path)) @classmethod - def iter_entries(cls, stream: Union[str, 'BytesIO', mmap]) -> Iterator[RefLogEntry]: + 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. @@ -215,7 +224,7 @@ class RefLog(List[RefLogEntry], Serializable): # END endless loop @classmethod - def entry_at(cls, filepath: PathLike, index: int) -> 'RefLogEntry': + def entry_at(cls, filepath: PathLike, index: int) -> "RefLogEntry": """ :return: RefLogEntry at the given index @@ -230,7 +239,7 @@ class RefLog(List[RefLogEntry], Serializable): all other lines. Nonetheless, the whole file has to be read if the index is negative """ - with open(filepath, 'rb') as fp: + with open(filepath, "rb") as fp: if index < 0: return RefLogEntry.from_line(fp.readlines()[index].strip()) # read until index is reached @@ -239,7 +248,8 @@ class RefLog(List[RefLogEntry], Serializable): line = fp.readline() if not line: raise IndexError( - f"Index file ended at line {i+1}, before given index was reached") + f"Index file ended at line {i+1}, before given index was reached" + ) # END abort on eof # END handle runup @@ -263,9 +273,15 @@ class RefLog(List[RefLogEntry], Serializable): # END handle change @classmethod - def append_entry(cls, config_reader: Union[Actor, 'GitConfigParser', 'SectionConstraint', None], - filepath: PathLike, oldbinsha: bytes, newbinsha: bytes, message: str, - write: bool = True) -> 'RefLogEntry': + 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 @@ -286,21 +302,27 @@ class RefLog(List[RefLogEntry], Serializable): 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] + first_line = message.split("\n")[0] if isinstance(config_reader, Actor): - committer = config_reader # mypy thinks this is Actor | Gitconfigparser, but why? + 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 - )) + entry = RefLogEntry( + ( + bin_to_hex(oldbinsha).decode("ascii"), + bin_to_hex(newbinsha).decode("ascii"), + committer, + (int(_time.time()), _time.altzone), + first_line, + ) + ) if write: lf = LockFile(filepath) lf._obtain_lock_or_raise() - fd = open(filepath, 'ab') + fd = open(filepath, "ab") try: fd.write(entry.format().encode(defenc)) finally: @@ -309,12 +331,13 @@ class RefLog(List[RefLogEntry], Serializable): # END handle write operation return entry - def write(self) -> 'RefLog': + 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") + "Instance was not initialized with a path, use to_file(...) instead" + ) # END assert path self.to_file(self._path) return self @@ -322,7 +345,7 @@ class RefLog(List[RefLogEntry], Serializable): # } END interface # { Serializable Interface - def _serialize(self, stream: 'BytesIO') -> 'RefLog': + def _serialize(self, stream: "BytesIO") -> "RefLog": write = stream.write # write all entries @@ -331,7 +354,7 @@ class RefLog(List[RefLogEntry], Serializable): # END for each entry return self - def _deserialize(self, stream: 'BytesIO') -> 'RefLog': + 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 2a33fbff..9b946ec4 100644 --- a/git/refs/reference.py +++ b/git/refs/reference.py @@ -8,7 +8,7 @@ from .symbolic import SymbolicReference, T_References # typing ------------------------------------------------------------------ from typing import Any, Callable, Iterator, Type, Union, TYPE_CHECKING # NOQA -from git.types import Commit_ish, PathLike, _T # NOQA +from git.types import Commit_ish, PathLike, _T # NOQA if TYPE_CHECKING: from git.repo import Repo @@ -18,7 +18,7 @@ if TYPE_CHECKING: __all__ = ["Reference"] -#{ Utilities +# { Utilities def require_remote_ref_path(func: Callable[..., _T]) -> Callable[..., _T]: @@ -26,24 +26,30 @@ def require_remote_ref_path(func: Callable[..., _T]) -> Callable[..., _T]: 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) + raise ValueError( + "ref path does not point to a remote reference: %s" % self.path + ) return func(self, *args) + # END wrapper wrapper.__name__ = func.__name__ return wrapper -#}END utilities + + +# }END utilities class Reference(SymbolicReference, LazyMixin, IterableObj): """Represents a named reference to any object. Subclasses may apply restrictions though, i.e. Heads can only point to commits.""" + __slots__ = () _points_to_commits_only = False _resolve_ref_on_create = True _common_path_default = "refs" - def __init__(self, repo: 'Repo', path: PathLike, check_path: bool = True) -> None: + def __init__(self, repo: "Repo", path: PathLike, check_path: bool = True) -> None: """Initialize this instance :param repo: Our parent repository @@ -52,19 +58,24 @@ 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 str(path).startswith(self._common_path_default + '/'): - raise ValueError(f"Cannot instantiate {self.__class__.__name__!r} from path {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) -> str: return self.name - #{ Interface + # { Interface # @ReservedAssignment - def set_object(self, object: Union[Commit_ish, 'SymbolicReference', str], logmsg: Union[str, None] = None - ) -> 'Reference': + def set_object( + self, + object: Union[Commit_ish, "SymbolicReference", str], + logmsg: Union[str, None] = None, + ) -> "Reference": """Special version which checks if the head-log needs an update as well :return: self""" oldbinsha = None @@ -102,21 +113,26 @@ class Reference(SymbolicReference, LazyMixin, IterableObj): """: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 - tokens = self.path.split('/') + tokens = self.path.split("/") if len(tokens) < 3: - return self.path # could be refs/HEAD - return '/'.join(tokens[2:]) + return self.path # could be refs/HEAD + return "/".join(tokens[2:]) @classmethod - def iter_items(cls: Type[T_References], repo: 'Repo', common_path: Union[PathLike, None] = None, - *args: Any, **kwargs: Any) -> Iterator[T_References]: + 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) - #}END interface + # }END interface - #{ Remote Interface + # { Remote Interface @property # type: ignore ## mypy cannot deal with properties with an extra decorator (2021-04-21) @require_remote_ref_path @@ -125,7 +141,7 @@ class Reference(SymbolicReference, LazyMixin, IterableObj): :return: Name of the remote we are a reference of, such as 'origin' for a reference named 'origin/master'""" - tokens = self.path.split('/') + tokens = self.path.split("/") # /refs/remotes/<remote name>/<branch_name> return tokens[2] @@ -135,7 +151,7 @@ class Reference(SymbolicReference, LazyMixin, IterableObj): """:return: Name of the remote head itself, i.e. master. :note: The returned name is usually not qualified enough to uniquely identify a branch""" - tokens = self.path.split('/') - return '/'.join(tokens[3:]) + tokens = self.path.split("/") + return "/".join(tokens[3:]) - #} END remote interface + # } END remote interface diff --git a/git/refs/remote.py b/git/refs/remote.py index 1b416bd0..8ac6bcd2 100644 --- a/git/refs/remote.py +++ b/git/refs/remote.py @@ -23,12 +23,18 @@ if TYPE_CHECKING: class RemoteReference(Head): """Represents a reference pointing to a remote head.""" + _common_path_default = Head._remote_common_path_default @classmethod - def iter_items(cls, repo: 'Repo', common_path: Union[PathLike, None] = None, - remote: Union['Remote', None] = None, *args: Any, **kwargs: Any - ) -> Iterator['RemoteReference']: + def iter_items( + cls, + repo: "Repo", + common_path: Union[PathLike, None] = None, + remote: Union["Remote", None] = None, + *args: Any, + **kwargs: Any + ) -> Iterator["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: @@ -41,9 +47,10 @@ class RemoteReference(Head): # implementation does not. mypy doesn't have a way of representing # tightening the types of arguments in subclasses and recommends Any or # "type: ignore". (See https://github.com/python/typing/issues/241) - @ classmethod - def delete(cls, repo: 'Repo', *refs: 'RemoteReference', # type: ignore - **kwargs: Any) -> None: + @classmethod + def delete( + cls, repo: "Repo", *refs: "RemoteReference", **kwargs: Any # type: ignore + ) -> None: """Delete the given remote references :note: @@ -64,7 +71,7 @@ class RemoteReference(Head): pass # END for each ref - @ classmethod + @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 8d869173..6d9ebb96 100644 --- a/git/refs/symbolic.py +++ b/git/refs/symbolic.py @@ -10,19 +10,26 @@ from git.util import ( to_native_path_linux, assure_directory_exists, hex_to_bin, - LockedFD -) -from gitdb.exc import ( - BadObject, - BadName + LockedFD, ) +from gitdb.exc import BadObject, BadName from .log import RefLog # typing ------------------------------------------------------------------ -from typing import Any, Iterator, List, Tuple, Type, TypeVar, Union, TYPE_CHECKING, cast # NOQA -from git.types import Commit_ish, PathLike # NOQA +from typing import ( + Any, + Iterator, + List, + Tuple, + Type, + TypeVar, + Union, + TYPE_CHECKING, + cast, +) # NOQA +from git.types import Commit_ish, PathLike # NOQA if TYPE_CHECKING: from git.repo import Repo @@ -32,7 +39,7 @@ if TYPE_CHECKING: from git.objects.commit import Actor -T_References = TypeVar('T_References', bound='SymbolicReference') +T_References = TypeVar("T_References", bound="SymbolicReference") # ------------------------------------------------------------------------------ @@ -40,10 +47,10 @@ T_References = TypeVar('T_References', bound='SymbolicReference') __all__ = ["SymbolicReference"] -def _git_dir(repo: 'Repo', path: Union[PathLike, None]) -> PathLike: - """ Find the git dir that's appropriate for the path""" +def _git_dir(repo: "Repo", path: Union[PathLike, None]) -> PathLike: + """Find the git dir that's appropriate for the path""" name = f"{path}" - if name in ['HEAD', 'ORIG_HEAD', 'FETCH_HEAD', 'index', 'logs']: + if name in ["HEAD", "ORIG_HEAD", "FETCH_HEAD", "index", "logs"]: return repo.git_dir return repo.common_dir @@ -55,6 +62,7 @@ class SymbolicReference(object): specifies a commit. A typical example for a symbolic reference is HEAD.""" + __slots__ = ("repo", "path") _resolve_ref_on_create = False _points_to_commits_only = True @@ -62,7 +70,7 @@ class SymbolicReference(object): _remote_common_path_default = "refs/remotes" _id_attribute_ = "name" - def __init__(self, repo: 'Repo', path: PathLike, check_path: bool = False): + def __init__(self, repo: "Repo", path: PathLike, check_path: bool = False): self.repo = repo self.path = path @@ -73,7 +81,7 @@ class SymbolicReference(object): return '<git.%s "%s">' % (self.__class__.__name__, self.path) def __eq__(self, other: object) -> bool: - if hasattr(other, 'path'): + if hasattr(other, "path"): other = cast(SymbolicReference, other) return self.path == other.path return False @@ -97,20 +105,20 @@ class SymbolicReference(object): return join_path_native(_git_dir(self.repo, self.path), self.path) @classmethod - def _get_packed_refs_path(cls, repo: 'Repo') -> str: - return os.path.join(repo.common_dir, 'packed-refs') + def _get_packed_refs_path(cls, repo: "Repo") -> str: + return os.path.join(repo.common_dir, "packed-refs") @classmethod - def _iter_packed_refs(cls, repo: 'Repo') -> Iterator[Tuple[str, str]]: + def _iter_packed_refs(cls, repo: "Repo") -> Iterator[Tuple[str, str]]: """Returns an iterator yielding pairs of sha1/path pairs (as strings) for the corresponding refs. :note: The packed refs file will be kept open as long as we iterate""" try: - with open(cls._get_packed_refs_path(repo), 'rt', encoding='UTF-8') as fp: + with open(cls._get_packed_refs_path(repo), "rt", encoding="UTF-8") as fp: for line in fp: line = line.strip() if not line: continue - if line.startswith('#'): + if line.startswith("#"): # "# pack-refs with: peeled fully-peeled sorted" # the git source code shows "peeled", # "fully-peeled" and "sorted" as the keywords @@ -119,18 +127,23 @@ class SymbolicReference(object): # I looked at master on 2017-10-11, # commit 111ef79afe, after tag v2.15.0-rc1 # from repo https://github.com/git/git.git - if line.startswith('# pack-refs with:') and 'peeled' not in line: - raise TypeError("PackingType of packed-Refs not understood: %r" % line) + if ( + line.startswith("# pack-refs with:") + and "peeled" not in line + ): + raise TypeError( + "PackingType of packed-Refs not understood: %r" % line + ) # END abort if we do not understand the packing scheme continue # END parse comment # skip dereferenced tag object entries - previous line was actual # tag reference for it - if line[0] == '^': + if line[0] == "^": continue - yield cast(Tuple[str, str], tuple(line.split(' ', 1))) + yield cast(Tuple[str, str], tuple(line.split(" ", 1))) # END for each line except OSError: return None @@ -141,7 +154,9 @@ class SymbolicReference(object): # alright. @classmethod - def dereference_recursive(cls, repo: 'Repo', ref_path: Union[PathLike, None]) -> str: + def dereference_recursive( + cls, repo: "Repo", ref_path: Union[PathLike, None] + ) -> str: """ :return: hexsha stored in the reference at the given ref_path, recursively dereferencing all intermediate references as required @@ -154,20 +169,23 @@ class SymbolicReference(object): # END recursive dereferencing @classmethod - def _get_ref_info_helper(cls, repo: 'Repo', ref_path: Union[PathLike, None] - ) -> Union[Tuple[str, None], Tuple[None, str]]: + def _get_ref_info_helper( + cls, repo: "Repo", ref_path: Union[PathLike, None] + ) -> Union[Tuple[str, None], Tuple[None, str]]: """Return: (str(sha), str(target_ref_path)) if available, the sha the file at rela_path points to, or None. target_ref_path is the reference we point to, or None""" tokens: Union[None, List[str], Tuple[str, str]] = None repodir = _git_dir(repo, ref_path) try: - with open(os.path.join(repodir, str(ref_path)), 'rt', encoding='UTF-8') as fp: + with open( + os.path.join(repodir, str(ref_path)), "rt", encoding="UTF-8" + ) as fp: value = fp.read().rstrip() # Don't only split on spaces, but on whitespace, which allows to parse lines like # 60b64ef992065e2600bfef6187a97f92398a9144 branch 'master' of git-server:/path/to/repo tokens = value.split() - assert(len(tokens) != 0) + assert len(tokens) != 0 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 @@ -184,7 +202,7 @@ class SymbolicReference(object): raise ValueError("Reference at %r does not exist" % ref_path) # is it a reference ? - if tokens[0] == 'ref:': + if tokens[0] == "ref:": return (None, tokens[1]) # its a commit @@ -194,7 +212,9 @@ class SymbolicReference(object): raise ValueError("Failed to parse reference information from %r" % ref_path) @classmethod - def _get_ref_info(cls, repo: 'Repo', ref_path: Union[PathLike, None]) -> Union[Tuple[str, None], Tuple[None, str]]: + def _get_ref_info( + cls, repo: "Repo", ref_path: Union[PathLike, None] + ) -> Union[Tuple[str, None], Tuple[None, str]]: """Return: (str(sha), str(target_ref_path)) if available, the sha the file at rela_path points to, or None. target_ref_path is the reference we point to, or None""" @@ -207,25 +227,32 @@ class SymbolicReference(object): always point to the actual object as it gets re-created on each query""" # have to be dynamic here as we may be a tag which can point to anything # Our path will be resolved to the hexsha which will be used accordingly - return Object.new_from_sha(self.repo, hex_to_bin(self.dereference_recursive(self.repo, self.path))) + return Object.new_from_sha( + self.repo, hex_to_bin(self.dereference_recursive(self.repo, self.path)) + ) - def _get_commit(self) -> 'Commit': + def _get_commit(self) -> "Commit": """ :return: Commit object we point to, works for detached and non-detached SymbolicReferences. The symbolic reference will be dereferenced recursively.""" obj = self._get_object() - if obj.type == 'tag': + if obj.type == "tag": obj = obj.object # END dereference tag if obj.type != Commit.type: - raise TypeError("Symbolic Reference pointed to object %r, commit was required" % obj) + raise TypeError( + "Symbolic Reference pointed to object %r, commit was required" % obj + ) # END handle type return obj - def set_commit(self, commit: Union[Commit, 'SymbolicReference', str], logmsg: Union[str, None] = None - ) -> 'SymbolicReference': + def set_commit( + self, + commit: Union[Commit, "SymbolicReference", str], + logmsg: Union[str, None] = None, + ) -> "SymbolicReference": """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 @@ -254,8 +281,11 @@ class SymbolicReference(object): return self - def set_object(self, object: Union[Commit_ish, 'SymbolicReference', str], logmsg: Union[str, None] = None - ) -> 'SymbolicReference': + def set_object( + self, + object: Union[Commit_ish, "SymbolicReference", str], + logmsg: Union[str, None] = None, + ) -> "SymbolicReference": """Set the object we point to, possibly dereference our symbolic reference first. If the reference does not exist, it will be created @@ -282,20 +312,25 @@ class SymbolicReference(object): # set the commit on our reference return self._get_reference().set_object(object, logmsg) - commit = property(_get_commit, set_commit, doc="Query or set commits directly") # type: ignore + commit = property(_get_commit, set_commit, doc="Query or set commits directly") # type: ignore object = property(_get_object, set_object, doc="Return the object our ref currently refers to") # type: ignore - def _get_reference(self) -> 'SymbolicReference': + def _get_reference(self) -> "SymbolicReference": """:return: Reference Object we point to :raise TypeError: If this symbolic reference is detached, hence it doesn't point to a reference, but to a commit""" sha, target_ref_path = self._get_ref_info(self.repo, self.path) if target_ref_path is None: - raise TypeError("%s is a detached symbolic reference as it points to %r" % (self, sha)) + raise TypeError( + "%s is a detached symbolic reference as it points to %r" % (self, sha) + ) return self.from_path(self.repo, target_ref_path) - def set_reference(self, ref: Union[Commit_ish, 'SymbolicReference', str], - logmsg: Union[str, None] = None) -> 'SymbolicReference': + def set_reference( + self, + ref: Union[Commit_ish, "SymbolicReference", str], + logmsg: Union[str, None] = None, + ) -> "SymbolicReference": """Set ourselves to the given ref. It will stay a symbol if the ref is a Reference. Otherwise an Object, given as Object instance or refspec, is assumed and if valid, will be set which effectively detaches the reference if it was a purely @@ -322,7 +357,7 @@ class SymbolicReference(object): write_value = ref.hexsha elif isinstance(ref, str): try: - obj = self.repo.rev_parse(ref + "^{}") # optionally deref tags + obj = self.repo.rev_parse(ref + "^{}") # optionally deref tags write_value = obj.hexsha except (BadObject, BadName) as e: raise ValueError("Could not extract object from %s" % ref) from e @@ -336,7 +371,7 @@ class SymbolicReference(object): raise TypeError("Require commit, got %r" % obj) # END verify type - oldbinsha: bytes = b'' + oldbinsha: bytes = b"" if logmsg is not None: try: oldbinsha = self.commit.binsha @@ -352,7 +387,7 @@ class SymbolicReference(object): fd = lfd.open(write=True, stream=True) ok = True try: - fd.write(write_value.encode('utf-8') + b'\n') + fd.write(write_value.encode("utf-8") + b"\n") lfd.commit() ok = True finally: @@ -365,7 +400,7 @@ class SymbolicReference(object): return self # aliased reference - reference: Union['Head', 'TagReference', 'RemoteReference', 'Reference'] + reference: Union["Head", "TagReference", "RemoteReference", "Reference"] reference = property(_get_reference, set_reference, doc="Returns the Reference we point to") # type: ignore ref = reference @@ -393,7 +428,7 @@ class SymbolicReference(object): except TypeError: return True - def log(self) -> 'RefLog': + def log(self) -> "RefLog": """ :return: RefLog for this reference. Its last entry reflects the latest change applied to this reference @@ -402,8 +437,12 @@ class SymbolicReference(object): instead of calling this method repeatedly. It should be considered read-only.""" return RefLog.from_file(RefLog.path(self)) - def log_append(self, oldbinsha: bytes, message: Union[str, None], - newbinsha: Union[bytes, None] = None) -> 'RefLogEntry': + def log_append( + self, + oldbinsha: bytes, + message: Union[str, None], + newbinsha: Union[bytes, None] = None, + ) -> "RefLogEntry": """Append a logentry to the logfile of this ref :param oldbinsha: binary sha this ref used to point to @@ -415,7 +454,9 @@ class SymbolicReference(object): # correct to allow overriding the committer on a per-commit level. # See https://github.com/gitpython-developers/GitPython/pull/146 try: - committer_or_reader: Union['Actor', 'GitConfigParser'] = self.commit.committer + committer_or_reader: Union[ + "Actor", "GitConfigParser" + ] = self.commit.committer except ValueError: committer_or_reader = self.repo.config_reader() # end handle newly cloned repositories @@ -423,11 +464,13 @@ class SymbolicReference(object): newbinsha = self.commit.binsha if message is None: - message = '' + message = "" - return RefLog.append_entry(committer_or_reader, RefLog.path(self), oldbinsha, newbinsha, message) + return RefLog.append_entry( + committer_or_reader, RefLog.path(self), oldbinsha, newbinsha, message + ) - def log_entry(self, index: int) -> 'RefLogEntry': + def log_entry(self, index: int) -> "RefLogEntry": """:return: RefLogEntry at the given index :param index: python list compatible positive or negative index @@ -437,7 +480,7 @@ class SymbolicReference(object): return RefLog.entry_at(RefLog.path(self), index) @classmethod - def to_full_path(cls, path: Union[PathLike, 'SymbolicReference']) -> PathLike: + def to_full_path(cls, path: Union[PathLike, "SymbolicReference"]) -> PathLike: """ :return: string with a full repository-relative path which can be used to initialize a Reference instance, for instance by using ``Reference.from_path``""" @@ -447,11 +490,11 @@ class SymbolicReference(object): if not cls._common_path_default: return full_ref_path if not str(path).startswith(cls._common_path_default + "/"): - full_ref_path = '%s/%s' % (cls._common_path_default, path) + full_ref_path = "%s/%s" % (cls._common_path_default, path) return full_ref_path @classmethod - def delete(cls, repo: 'Repo', path: PathLike) -> None: + def delete(cls, repo: "Repo", path: PathLike) -> None: """Delete the reference at the given path :param repo: @@ -469,20 +512,23 @@ class SymbolicReference(object): # check packed refs pack_file_path = cls._get_packed_refs_path(repo) try: - with open(pack_file_path, 'rb') as reader: + with open(pack_file_path, "rb") as reader: new_lines = [] made_change = False dropped_last_line = False for line_bytes in reader: line = line_bytes.decode(defenc) - _, _, line_ref = line.partition(' ') + _, _, line_ref = line.partition(" ") line_ref = line_ref.strip() # keep line if it is a comment or if the ref to delete is not # in the line # If we deleted the last line and this one is a tag-reference object, # we drop it as well - if (line.startswith('#') or full_ref_path != line_ref) and \ - (not dropped_last_line or dropped_last_line and not line.startswith('^')): + if (line.startswith("#") or full_ref_path != line_ref) and ( + not dropped_last_line + or dropped_last_line + and not line.startswith("^") + ): new_lines.append(line) dropped_last_line = False continue @@ -496,7 +542,7 @@ class SymbolicReference(object): if made_change: # write-binary is required, otherwise windows will # open the file in text mode and change LF to CRLF ! - with open(pack_file_path, 'wb') as fd: + with open(pack_file_path, "wb") as fd: fd.writelines(line.encode(defenc) for line in new_lines) except OSError: @@ -509,9 +555,15 @@ class SymbolicReference(object): # END remove reflog @classmethod - def _create(cls: Type[T_References], repo: 'Repo', path: PathLike, resolve: bool, - reference: Union['SymbolicReference', str], force: bool, - logmsg: Union[str, None] = None) -> T_References: + def _create( + cls: Type[T_References], + repo: "Repo", + path: PathLike, + resolve: bool, + reference: Union["SymbolicReference", str], + force: bool, + logmsg: Union[str, None] = None, + ) -> T_References: """internal method used to create a new symbolic reference. If resolve is False, the reference will be taken as is, creating a proper symbolic reference. Otherwise it will be resolved to the @@ -532,11 +584,13 @@ class SymbolicReference(object): target_data = str(target.path) if not resolve: target_data = "ref: " + target_data - with open(abs_ref_path, 'rb') as fd: + with open(abs_ref_path, "rb") as fd: existing_data = fd.read().decode(defenc).strip() if existing_data != target_data: - raise OSError("Reference at %r does already exist, pointing to %r, requested was %r" % - (full_ref_path, existing_data, target_data)) + raise OSError( + "Reference at %r does already exist, pointing to %r, requested was %r" + % (full_ref_path, existing_data, target_data) + ) # END no force handling ref = cls(repo, full_ref_path) @@ -544,9 +598,15 @@ class SymbolicReference(object): return ref @classmethod - def create(cls: Type[T_References], repo: 'Repo', path: PathLike, - reference: Union['SymbolicReference', str] = 'HEAD', - logmsg: Union[str, None] = None, force: bool = False, **kwargs: Any) -> T_References: + def create( + cls: Type[T_References], + repo: "Repo", + path: PathLike, + reference: Union["SymbolicReference", str] = "HEAD", + logmsg: Union[str, None] = None, + force: bool = False, + **kwargs: Any, + ) -> T_References: """Create a new symbolic reference, hence a reference pointing , to another reference. :param repo: @@ -575,9 +635,11 @@ class SymbolicReference(object): already exists. :note: This does not alter the current HEAD, index or Working Tree""" - return cls._create(repo, path, cls._resolve_ref_on_create, reference, force, logmsg) + return cls._create( + repo, path, cls._resolve_ref_on_create, reference, force, logmsg + ) - def rename(self, new_path: PathLike, force: bool = False) -> 'SymbolicReference': + def rename(self, new_path: PathLike, force: bool = False) -> "SymbolicReference": """Rename self to a new path :param new_path: @@ -590,7 +652,7 @@ class SymbolicReference(object): already exists. It will be overwritten in that case :return: self - :raise OSError: In case a file at path but a different contents already exists """ + :raise OSError: In case a file at path but a different contents already exists""" new_path = self.to_full_path(new_path) if self.path == new_path: return self @@ -600,9 +662,9 @@ class SymbolicReference(object): if os.path.isfile(new_abs_path): if not force: # if they point to the same file, its not an error - with open(new_abs_path, 'rb') as fd1: + with open(new_abs_path, "rb") as fd1: f1 = fd1.read().strip() - with open(cur_abs_path, 'rb') as fd2: + with open(cur_abs_path, "rb") as fd2: f2 = fd2.read().strip() if f1 != f2: raise OSError("File at path %r already exists" % new_abs_path) @@ -623,26 +685,31 @@ class SymbolicReference(object): return self @classmethod - def _iter_items(cls: Type[T_References], repo: 'Repo', common_path: Union[PathLike, None] = None - ) -> Iterator[T_References]: + 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() # walk loose refs # Currently we do not follow links - for root, dirs, files in os.walk(join_path_native(repo.common_dir, common_path)): - if 'refs' not in root.split(os.sep): # skip non-refs subfolders - refs_id = [d for d in dirs if d == 'refs'] + for root, dirs, files in os.walk( + join_path_native(repo.common_dir, common_path) + ): + if "refs" not in root.split(os.sep): # skip non-refs subfolders + refs_id = [d for d in dirs if d == "refs"] if refs_id: - dirs[0:] = ['refs'] + dirs[0:] = ["refs"] # END prune non-refs folders for f in files: - if f == 'packed-refs': + if f == "packed-refs": continue abs_path = to_native_path_linux(join_path(root, f)) - rela_paths.add(abs_path.replace(to_native_path_linux(repo.common_dir) + '/', "")) + rela_paths.add( + abs_path.replace(to_native_path_linux(repo.common_dir) + "/", "") + ) # END for each file in root directory # END for each directory to walk @@ -662,8 +729,13 @@ class SymbolicReference(object): # END for each sorted relative refpath @classmethod - def iter_items(cls: Type[T_References], repo: 'Repo', common_path: Union[PathLike, None] = None, - *args: Any, **kwargs: Any) -> Iterator[T_References]: + def iter_items( + cls: Type[T_References], + repo: "Repo", + common_path: Union[PathLike, None] = None, + *args: Any, + **kwargs: Any, + ) -> Iterator[T_References]: """Find all refs in the repository :param repo: is the Repo @@ -680,10 +752,16 @@ class SymbolicReference(object): List is lexicographically sorted The returned objects represent actual subclasses, such as Head or TagReference""" - return (r for r in cls._iter_items(repo, common_path) if r.__class__ == SymbolicReference or not r.is_detached) + return ( + r + for r in cls._iter_items(repo, common_path) + if r.__class__ == SymbolicReference or not r.is_detached + ) @classmethod - def from_path(cls: Type[T_References], repo: 'Repo', path: PathLike) -> T_References: + def from_path( + cls: Type[T_References], repo: "Repo", path: PathLike + ) -> T_References: """ :param path: full .git-directory-relative path name to the Reference to instantiate :note: use to_full_path() if you only have a partial path of a known Reference Type @@ -696,7 +774,15 @@ class SymbolicReference(object): # Names like HEAD are inserted after the refs module is imported - we have an import dependency # cycle and don't want to import these names in-function from . import HEAD, Head, RemoteReference, TagReference, Reference - for ref_type in (HEAD, Head, RemoteReference, TagReference, Reference, SymbolicReference): + + for ref_type in ( + HEAD, + Head, + RemoteReference, + TagReference, + Reference, + SymbolicReference, + ): try: instance: T_References instance = ref_type(repo, path) @@ -709,7 +795,9 @@ class SymbolicReference(object): pass # END exception handling # END for each type to try - raise ValueError("Could not find reference type suitable to handle path %r" % path) + raise ValueError( + "Could not find reference type suitable to handle path %r" % path + ) def is_remote(self) -> bool: """:return: True if this symbolic reference points to a remote branch""" diff --git a/git/refs/tag.py b/git/refs/tag.py index 8cc79edd..96494148 100644 --- a/git/refs/tag.py +++ b/git/refs/tag.py @@ -36,22 +36,27 @@ class TagReference(Reference): _common_path_default = Reference._common_path_default + "/" + _common_default @property - def commit(self) -> 'Commit': # type: ignore[override] # LazyMixin has unrelated commit method + def commit(self) -> "Commit": # type: ignore[override] # LazyMixin has unrelated commit 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': + while obj.type != "commit": if obj.type == "tag": # it is a tag object which carries the commit as an object - we can point to anything obj = obj.object else: - raise ValueError(("Cannot resolve commit as tag %s points to a %s object - " + - "use the `.object` property instead to access it") % (self, obj.type)) + raise ValueError( + ( + "Cannot resolve commit as tag %s points to a %s object - " + + "use the `.object` property instead to access it" + ) + % (self, obj.type) + ) return obj @property - def tag(self) -> Union['TagObject', None]: + def tag(self) -> Union["TagObject", None]: """ :return: Tag object this tag ref points to or None in case we are a light weight tag""" @@ -69,10 +74,15 @@ class TagReference(Reference): return Reference._get_object(self) @classmethod - def create(cls: Type['TagReference'], repo: 'Repo', path: PathLike, - reference: Union[str, 'SymbolicReference'] = 'HEAD', - logmsg: Union[str, None] = None, - force: bool = False, **kwargs: Any) -> 'TagReference': + def create( + cls: Type["TagReference"], + repo: "Repo", + path: PathLike, + reference: Union[str, "SymbolicReference"] = "HEAD", + logmsg: Union[str, None] = None, + force: bool = False, + **kwargs: Any + ) -> "TagReference": """Create a new tag reference. :param path: @@ -100,16 +110,16 @@ class TagReference(Reference): Additional keyword arguments to be passed to git-tag :return: A new TagReference""" - if 'ref' in kwargs and kwargs['ref']: - reference = kwargs['ref'] + if "ref" in kwargs and kwargs["ref"]: + reference = kwargs["ref"] if logmsg: - kwargs['m'] = logmsg - elif 'message' in kwargs and kwargs['message']: - kwargs['m'] = kwargs['message'] + kwargs["m"] = logmsg + elif "message" in kwargs and kwargs["message"]: + kwargs["m"] = kwargs["message"] if force: - kwargs['f'] = True + kwargs["f"] = True args = (path, reference) @@ -117,7 +127,7 @@ class TagReference(Reference): return TagReference(repo, "%s/%s" % (cls._common_path_default, path)) @classmethod - def delete(cls, repo: 'Repo', *tags: 'TagReference') -> None: # type: ignore[override] + def delete(cls, repo: "Repo", *tags: "TagReference") -> None: # type: ignore[override] """Delete the given existing tag or tags""" repo.git.tag("-d", *tags) |