From cc63210d122ac7a113990e27b48e1bdbd07ceb4c Mon Sep 17 00:00:00 2001 From: Yobmod Date: Mon, 19 Jul 2021 16:52:00 +0100 Subject: Add types to refs/tag.py --- git/refs/tag.py | 46 ++++++++++++++++++++++++++++++++++++---------- 1 file changed, 36 insertions(+), 10 deletions(-) (limited to 'git/refs') diff --git a/git/refs/tag.py b/git/refs/tag.py index 4d84239e..aa3b82a2 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 """: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) -- cgit v1.2.1 From 8dd3d0d6308f97249d69d43b6636d13fd3813d44 Mon Sep 17 00:00:00 2001 From: Yobmod Date: Mon, 19 Jul 2021 16:52:27 +0100 Subject: Add types to refs/symbolic.py --- git/refs/symbolic.py | 674 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 674 insertions(+) (limited to 'git/refs') diff --git a/git/refs/symbolic.py b/git/refs/symbolic.py index f0bd9316..48b94cfa 100644 --- a/git/refs/symbolic.py +++ b/git/refs/symbolic.py @@ -21,6 +21,680 @@ 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"] + + +def _git_dir(repo, path): + """ Find the git dir that's appropriate for the path""" + name = "%s" % (path,) + if name in ['HEAD', 'ORIG_HEAD', 'FETCH_HEAD', 'index', 'logs']: + return repo.git_dir + return repo.common_dir + + +class SymbolicReference(object): + + """Represents a special case of a reference such that this reference is symbolic. + It does not point to a specific commit, but to another Head, which itself + 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 + _common_path_default = "" + _remote_common_path_default = "refs/remotes" + _id_attribute_ = "name" + + def __init__(self, repo: 'Repo', path: PathLike, check_path: bool = False): + self.repo = repo + self.path = str(path) + + def __str__(self) -> str: + return self.path + + def __repr__(self): + return '' % (self.__class__.__name__, self.path) + + def __eq__(self, other): + if hasattr(other, 'path'): + return self.path == other.path + return False + + def __ne__(self, other): + return not (self == other) + + def __hash__(self): + return hash(self.path) + + @property + def name(self): + """ + :return: + In case of symbolic references, the shortest assumable name + is the path itself.""" + return self.path + + @property + def abspath(self): + return join_path_native(_git_dir(self.repo, self.path), self.path) + + @classmethod + def _get_packed_refs_path(cls, repo): + return osp.join(repo.common_dir, 'packed-refs') + + @classmethod + def _iter_packed_refs(cls, repo): + """Returns an iterator yielding pairs of sha1/path pairs (as bytes) 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: + for line in fp: + line = line.strip() + if not line: + continue + if line.startswith('#'): + # "# pack-refs with: peeled fully-peeled sorted" + # the git source code shows "peeled", + # "fully-peeled" and "sorted" as the keywords + # that can go on this line, as per comments in git file + # refs/packed-backend.c + # 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) + # 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] == '^': + continue + + yield tuple(line.split(' ', 1)) + # END for each line + 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. + # I believe files are closing themselves on destruction, so it is + # alright. + + @classmethod + def dereference_recursive(cls, repo, ref_path): + """ + :return: hexsha stored in the reference at the given ref_path, recursively dereferencing all + intermediate references as required + :param repo: the repository containing the reference at ref_path""" + while True: + hexsha, ref_path = cls._get_ref_info(repo, ref_path) + if hexsha is not None: + return hexsha + # END recursive dereferencing + + @classmethod + def _get_ref_info_helper(cls, repo, ref_path): + """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 = None + repodir = _git_dir(repo, ref_path) + try: + with open(osp.join(repodir, 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) + 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 + for sha, path in cls._iter_packed_refs(repo): + if path != ref_path: + continue + # sha will be used + tokens = sha, path + break + # END for each packed ref + # END handle packed refs + if tokens is None: + raise ValueError("Reference at %r does not exist" % ref_path) + + # is it a reference ? + if tokens[0] == 'ref:': + return (None, tokens[1]) + + # its a commit + if repo.re_hexsha_only.match(tokens[0]): + return (tokens[0], None) + + raise ValueError("Failed to parse reference information from %r" % ref_path) + + @classmethod + def _get_ref_info(cls, repo, ref_path): + """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""" + return cls._get_ref_info_helper(repo, ref_path) + + def _get_object(self): + """ + :return: + The object our ref currently refers to. Refs can be cached, they will + 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))) + + def _get_commit(self): + """ + :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': + obj = obj.object + # END dereference tag + + if obj.type != Commit.type: + 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=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 + a commit + :return: self""" + # check the type - assume the best if it is a base-string + invalid_type = False + if isinstance(commit, Object): + invalid_type = commit.type != Commit.type + elif isinstance(commit, SymbolicReference): + invalid_type = commit.object.type != Commit.type + else: + try: + invalid_type = self.repo.rev_parse(commit).type != Commit.type + except (BadObject, BadName) as e: + raise ValueError("Invalid object: %s" % commit) from e + # END handle exception + # END verify type + + if invalid_type: + raise ValueError("Need commit, got %r" % commit) + # END handle raise + + # we leave strings to the rev-parse method below + self.set_object(commit, logmsg) + + return self + + def set_object(self, object, logmsg=None): # @ReservedAssignment + """Set the object we point to, possibly dereference our symbolic reference first. + If the reference does not exist, it will be created + + :param object: a refspec, a SymbolicReference or an Object instance. SymbolicReferences + will be dereferenced beforehand to obtain the object they point to + :param logmsg: If not None, the message will be used in the reflog entry to be + written. Otherwise the reflog is not altered + :note: plain SymbolicReferences may not actually point to objects by convention + :return: self""" + if isinstance(object, SymbolicReference): + object = object.object # @ReservedAssignment + # END resolve references + + is_detached = True + try: + is_detached = self.is_detached + except ValueError: + pass + # END handle non-existing ones + + if is_detached: + return self.set_reference(object, logmsg) + + # 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") + object = property(_get_object, set_object, doc="Return the object our ref currently refers to") + + def _get_reference(self): + """: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)) + return self.from_path(self.repo, target_ref_path) + + def set_reference(self, ref, logmsg=None): + """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 refererence if it was a purely + symbolic one. + + :param ref: SymbolicReference instance, Object instance or refspec string + Only if the ref is a SymbolicRef instance, we will point to it. Everything + else is dereferenced to obtain the actual object. + :param logmsg: If set to a string, the message will be used in the reflog. + Otherwise, a reflog entry is not written for the changed reference. + The previous commit of the entry will be the commit we point to now. + + See also: log_append() + + :return: self + :note: This symbolic reference will not be dereferenced. For that, see + ``set_object(...)``""" + write_value = None + obj = None + if isinstance(ref, SymbolicReference): + write_value = "ref: %s" % ref.path + elif isinstance(ref, Object): + obj = ref + write_value = ref.hexsha + elif isinstance(ref, str): + try: + 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 + # END end try string + else: + raise ValueError("Unrecognized Value: %r" % ref) + # END try commit attribute + + # typecheck + if obj is not None and self._points_to_commits_only and obj.type != Commit.type: + raise TypeError("Require commit, got %r" % obj) + # END verify type + + oldbinsha = None + if logmsg is not None: + try: + oldbinsha = self.commit.binsha + except ValueError: + oldbinsha = Commit.NULL_BIN_SHA + # END handle non-existing + # END retrieve old hexsha + + fpath = self.abspath + assure_directory_exists(fpath, is_file=True) + + lfd = LockedFD(fpath) + fd = lfd.open(write=True, stream=True) + ok = True + try: + fd.write(write_value.encode('ascii') + b'\n') + lfd.commit() + ok = True + finally: + if not ok: + lfd.rollback() + # Adjust the reflog + if logmsg is not None: + self.log_append(oldbinsha, logmsg) + + return self + + # aliased reference + reference = property(_get_reference, set_reference, doc="Returns the Reference we point to") + ref: Union[Commit_ish] = reference # type: ignore # Union[str, Commit_ish, SymbolicReference] + + def is_valid(self): + """ + :return: + True if the reference is valid, hence it can be read and points to + a valid object or reference.""" + try: + self.object + except (OSError, ValueError): + return False + else: + return True + + @property + def is_detached(self): + """ + :return: + True if we are a detached reference, hence we point to a specific commit + instead to another reference""" + try: + self.ref + return False + except TypeError: + return True + + def log(self): + """ + :return: RefLog for this reference. Its last entry reflects the latest change + applied to this reference + + .. note:: As the log is parsed every time, its recommended to cache it for use + instead of calling this method repeatedly. It should be considered read-only.""" + return RefLog.from_file(RefLog.path(self)) + + def log_append(self, oldbinsha, message, newbinsha=None): + """Append a logentry to the logfile of this ref + + :param oldbinsha: binary sha this ref used to point to + :param message: A message describing the change + :param newbinsha: The sha the ref points to now. If None, our current commit sha + will be used + :return: added RefLogEntry instance""" + # NOTE: we use the committer of the currently active commit - this should be + # correct to allow overriding the committer on a per-commit level. + # See https://github.com/gitpython-developers/GitPython/pull/146 + try: + committer_or_reader = self.commit.committer + except ValueError: + committer_or_reader = self.repo.config_reader() + # end handle newly cloned repositories + return RefLog.append_entry(committer_or_reader, RefLog.path(self), oldbinsha, + (newbinsha is None and self.commit.binsha) or newbinsha, + message) + + def log_entry(self, index): + """:return: RefLogEntry at the given index + :param index: python list compatible positive or negative index + + .. note:: This method must read part of the reflog during execution, hence + it should be used sparringly, or only if you need just one index. + In that case, it will be faster than the ``log()`` method""" + return RefLog.entry_at(RefLog.path(self), index) + + @classmethod + def to_full_path(cls, path) -> 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``""" + if isinstance(path, SymbolicReference): + path = path.path + full_ref_path = path + if not cls._common_path_default: + return full_ref_path + if not path.startswith(cls._common_path_default + "/"): + full_ref_path = '%s/%s' % (cls._common_path_default, path) + return full_ref_path + + @classmethod + def delete(cls, repo, path): + """Delete the reference at the given path + + :param repo: + Repository to delete the reference from + + :param path: + Short or full path pointing to the reference, i.e. refs/myreference + or just "myreference", hence 'refs/' is implied. + Alternatively the symbolic reference to be deleted""" + full_ref_path = cls.to_full_path(path) + abs_path = osp.join(repo.common_dir, full_ref_path) + if osp.exists(abs_path): + os.remove(abs_path) + else: + # check packed refs + pack_file_path = cls._get_packed_refs_path(repo) + try: + with open(pack_file_path, 'rb') as reader: + new_lines = [] + made_change = False + dropped_last_line = False + for line in reader: + line = line.decode(defenc) + _, _, 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('^')): + new_lines.append(line) + dropped_last_line = False + continue + # END skip comments and lines without our path + + # drop this line + made_change = True + dropped_last_line = True + + # write the new lines + 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: + fd.writelines(line.encode(defenc) for line in new_lines) + + except OSError: + pass # it didn't exist at all + + # delete the reflog + reflog_path = RefLog.path(cls(repo, full_ref_path)) + if osp.isfile(reflog_path): + os.remove(reflog_path) + # END remove reflog + + @classmethod + def _create(cls, repo, path, resolve, reference, force, logmsg=None): + """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 + corresponding object and a detached symbolic reference will be created + instead""" + git_dir = _git_dir(repo, path) + full_ref_path = cls.to_full_path(path) + abs_ref_path = osp.join(git_dir, full_ref_path) + + # figure out target data + target = reference + if resolve: + target = repo.rev_parse(str(reference)) + + if not force and osp.isfile(abs_ref_path): + target_data = str(target) + if isinstance(target, SymbolicReference): + target_data = target.path + if not resolve: + target_data = "ref: " + target_data + 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)) + # END no force handling + + ref = cls(repo, full_ref_path) + ref.set_reference(target, logmsg) + return ref + + @classmethod + 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 + + :param path: + full path at which the new symbolic reference is supposed to be + created at, i.e. "NEW_HEAD" or "symrefs/my_new_symref" + + :param reference: + The reference to which the new symbolic reference should point to. + If it is a commit'ish, the symbolic ref will be detached. + + :param force: + if True, force creation even if a symbolic reference with that name already exists. + Raise OSError otherwise + + :param logmsg: + If not None, the message to append to the reflog. Otherwise no reflog + entry is written. + + :return: Newly created symbolic Reference + + :raise OSError: + If a (Symbolic)Reference with the same name but different contents + 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) + + def rename(self, new_path, force=False): + """Rename self to a new path + + :param new_path: + Either a simple name or a full path, i.e. new_name or features/new_name. + The prefix refs/ is implied for references and will be set as needed. + In case this is a symbolic ref, there is no implied prefix + + :param force: + If True, the rename will succeed even if a head with the target name + 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 """ + new_path = self.to_full_path(new_path) + if self.path == new_path: + return self + + new_abs_path = osp.join(_git_dir(self.repo, new_path), new_path) + cur_abs_path = osp.join(_git_dir(self.repo, self.path), self.path) + if osp.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: + f1 = fd1.read().strip() + 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) + # else: we could remove ourselves and use the otherone, but + # but clarity we just continue as usual + # END not force handling + os.remove(new_abs_path) + # END handle existing target file + + dname = osp.dirname(new_abs_path) + if not osp.isdir(dname): + os.makedirs(dname) + # END create directory + + os.rename(cur_abs_path, new_abs_path) + self.path = new_path + + return self + + @classmethod + 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'] + if refs_id: + dirs[0:] = ['refs'] + # END prune non-refs folders + + for f in files: + 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) + '/', "")) + # END for each file in root directory + # END for each directory to walk + + # read packed refs + for _sha, rela_path in cls._iter_packed_refs(repo): + if rela_path.startswith(common_path): + rela_paths.add(rela_path) + # END relative path matches common path + # END packed refs reading + + # return paths in sorted order + for path in sorted(rela_paths): + try: + yield cls.from_path(repo, path) + except ValueError: + continue + # END for each sorted relative refpath + + @classmethod + # 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 + + :param common_path: + Optional keyword argument to the path which is to be shared by all + returned Ref objects. + Defaults to class specific portion if None assuring that only + refs suitable for the actual class are returned. + + :return: + git.SymbolicReference[], each of them is guaranteed to be a symbolic + ref which is not detached and pointing to a valid ref + + 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) + + @classmethod + def from_path(cls, repo, path): + """ + :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 + :return: + Instance of type Reference, Head, or Tag + depending on the given path""" + if not path: + raise ValueError("Cannot create Reference from %r" % path) + + # 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): + try: + instance = ref_type(repo, path) + if instance.__class__ == SymbolicReference and instance.is_detached: + raise ValueError("SymbolRef was detached, we drop it") + return instance + except ValueError: + pass + # END exception handling + # END for each type to try + raise ValueError("Could not find reference type suitable to handle path %r" % path) + + def is_remote(self): + """:return: True if this symbolic reference points to a remote branch""" + return self.path.startswith(self._remote_common_path_default + "/") + + __all__ = ["SymbolicReference"] -- cgit v1.2.1 From 4b9ca921e8722f4e7359bea174b2c823059c5542 Mon Sep 17 00:00:00 2001 From: Yobmod Date: Mon, 19 Jul 2021 16:53:14 +0100 Subject: Add types to refs/remote.py --- git/refs/remote.py | 31 ++++++++++++++++++++++--------- 1 file changed, 22 insertions(+), 9 deletions(-) (limited to 'git/refs') 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") -- cgit v1.2.1 From 1dd4596294d2302cc091a337ff6f89761795efe7 Mon Sep 17 00:00:00 2001 From: Yobmod Date: Mon, 19 Jul 2021 16:53:28 +0100 Subject: Add types to refs/reference.py --- git/refs/reference.py | 37 +++++++++++++++++++++++++------------ 1 file changed, 25 insertions(+), 12 deletions(-) (limited to 'git/refs') 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""" -- cgit v1.2.1 From ae13c6dfd6124604c30de1841dfd669195568608 Mon Sep 17 00:00:00 2001 From: Yobmod Date: Mon, 19 Jul 2021 16:53:47 +0100 Subject: Add types to refs/log.py --- git/refs/log.py | 140 ++++++++++++++++++++++++++++++++++---------------------- 1 file changed, 86 insertions(+), 54 deletions(-) (limited to 'git/refs') 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 -- cgit v1.2.1 From 8fc25c63d9282ddc6b3162c2d92679a89e934ec5 Mon Sep 17 00:00:00 2001 From: Yobmod Date: Mon, 19 Jul 2021 16:54:05 +0100 Subject: Add types to refs/head.py --- git/refs/head.py | 37 ++++++++++++++++++++++--------------- 1 file changed, 22 insertions(+), 15 deletions(-) (limited to 'git/refs') 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""" -- cgit v1.2.1 From 795d2b1f75c6c210ebabd81f33c0c9ac8b57729e Mon Sep 17 00:00:00 2001 From: Yobmod Date: Mon, 19 Jul 2021 17:03:50 +0100 Subject: rmv redundant IOerror except --- git/refs/symbolic.py | 658 --------------------------------------------------- 1 file changed, 658 deletions(-) (limited to 'git/refs') diff --git a/git/refs/symbolic.py b/git/refs/symbolic.py index 48b94cfa..426d40d4 100644 --- a/git/refs/symbolic.py +++ b/git/refs/symbolic.py @@ -693,661 +693,3 @@ class SymbolicReference(object): def is_remote(self): """:return: True if this symbolic reference points to a remote branch""" return self.path.startswith(self._remote_common_path_default + "/") - - -__all__ = ["SymbolicReference"] - - -def _git_dir(repo, path): - """ Find the git dir that's appropriate for the path""" - name = "%s" % (path,) - if name in ['HEAD', 'ORIG_HEAD', 'FETCH_HEAD', 'index', 'logs']: - return repo.git_dir - return repo.common_dir - - -class SymbolicReference(object): - - """Represents a special case of a reference such that this reference is symbolic. - It does not point to a specific commit, but to another Head, which itself - 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 - _common_path_default = "" - _remote_common_path_default = "refs/remotes" - _id_attribute_ = "name" - - def __init__(self, repo, path, check_path=None): - self.repo = repo - self.path = path - - def __str__(self): - return self.path - - def __repr__(self): - return '' % (self.__class__.__name__, self.path) - - def __eq__(self, other): - if hasattr(other, 'path'): - return self.path == other.path - return False - - def __ne__(self, other): - return not (self == other) - - def __hash__(self): - return hash(self.path) - - @property - def name(self): - """ - :return: - In case of symbolic references, the shortest assumable name - is the path itself.""" - return self.path - - @property - def abspath(self): - return join_path_native(_git_dir(self.repo, self.path), self.path) - - @classmethod - def _get_packed_refs_path(cls, repo): - return osp.join(repo.common_dir, 'packed-refs') - - @classmethod - def _iter_packed_refs(cls, repo): - """Returns an iterator yielding pairs of sha1/path pairs (as bytes) 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: - for line in fp: - line = line.strip() - if not line: - continue - if line.startswith('#'): - # "# pack-refs with: peeled fully-peeled sorted" - # the git source code shows "peeled", - # "fully-peeled" and "sorted" as the keywords - # that can go on this line, as per comments in git file - # refs/packed-backend.c - # 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) - # 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] == '^': - continue - - yield tuple(line.split(' ', 1)) - # END for each line - except (OSError, IOError): - return - # 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. - # I believe files are closing themselves on destruction, so it is - # alright. - - @classmethod - def dereference_recursive(cls, repo, ref_path): - """ - :return: hexsha stored in the reference at the given ref_path, recursively dereferencing all - intermediate references as required - :param repo: the repository containing the reference at ref_path""" - while True: - hexsha, ref_path = cls._get_ref_info(repo, ref_path) - if hexsha is not None: - return hexsha - # END recursive dereferencing - - @classmethod - def _get_ref_info_helper(cls, repo, ref_path): - """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 = None - repodir = _git_dir(repo, ref_path) - try: - with open(osp.join(repodir, 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) - except (OSError, IOError): - # 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 - for sha, path in cls._iter_packed_refs(repo): - if path != ref_path: - continue - # sha will be used - tokens = sha, path - break - # END for each packed ref - # END handle packed refs - if tokens is None: - raise ValueError("Reference at %r does not exist" % ref_path) - - # is it a reference ? - if tokens[0] == 'ref:': - return (None, tokens[1]) - - # its a commit - if repo.re_hexsha_only.match(tokens[0]): - return (tokens[0], None) - - raise ValueError("Failed to parse reference information from %r" % ref_path) - - @classmethod - def _get_ref_info(cls, repo, ref_path): - """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""" - return cls._get_ref_info_helper(repo, ref_path) - - def _get_object(self): - """ - :return: - The object our ref currently refers to. Refs can be cached, they will - 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))) - - def _get_commit(self): - """ - :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': - obj = obj.object - # END dereference tag - - if obj.type != Commit.type: - raise TypeError("Symbolic Reference pointed to object %r, commit was required" % obj) - # END handle type - return obj - - def set_commit(self, commit, 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 - a commit - :return: self""" - # check the type - assume the best if it is a base-string - invalid_type = False - if isinstance(commit, Object): - invalid_type = commit.type != Commit.type - elif isinstance(commit, SymbolicReference): - invalid_type = commit.object.type != Commit.type - else: - try: - invalid_type = self.repo.rev_parse(commit).type != Commit.type - except (BadObject, BadName) as e: - raise ValueError("Invalid object: %s" % commit) from e - # END handle exception - # END verify type - - if invalid_type: - raise ValueError("Need commit, got %r" % commit) - # END handle raise - - # we leave strings to the rev-parse method below - self.set_object(commit, logmsg) - - return self - - def set_object(self, object, logmsg=None): # @ReservedAssignment - """Set the object we point to, possibly dereference our symbolic reference first. - If the reference does not exist, it will be created - - :param object: a refspec, a SymbolicReference or an Object instance. SymbolicReferences - will be dereferenced beforehand to obtain the object they point to - :param logmsg: If not None, the message will be used in the reflog entry to be - written. Otherwise the reflog is not altered - :note: plain SymbolicReferences may not actually point to objects by convention - :return: self""" - if isinstance(object, SymbolicReference): - object = object.object # @ReservedAssignment - # END resolve references - - is_detached = True - try: - is_detached = self.is_detached - except ValueError: - pass - # END handle non-existing ones - - if is_detached: - return self.set_reference(object, logmsg) - - # 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") - object = property(_get_object, set_object, doc="Return the object our ref currently refers to") - - def _get_reference(self): - """: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)) - return self.from_path(self.repo, target_ref_path) - - def set_reference(self, ref, logmsg=None): - """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 refererence if it was a purely - symbolic one. - - :param ref: SymbolicReference instance, Object instance or refspec string - Only if the ref is a SymbolicRef instance, we will point to it. Everything - else is dereferenced to obtain the actual object. - :param logmsg: If set to a string, the message will be used in the reflog. - Otherwise, a reflog entry is not written for the changed reference. - The previous commit of the entry will be the commit we point to now. - - See also: log_append() - - :return: self - :note: This symbolic reference will not be dereferenced. For that, see - ``set_object(...)``""" - write_value = None - obj = None - if isinstance(ref, SymbolicReference): - write_value = "ref: %s" % ref.path - elif isinstance(ref, Object): - obj = ref - write_value = ref.hexsha - elif isinstance(ref, str): - try: - 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 - # END end try string - else: - raise ValueError("Unrecognized Value: %r" % ref) - # END try commit attribute - - # typecheck - if obj is not None and self._points_to_commits_only and obj.type != Commit.type: - raise TypeError("Require commit, got %r" % obj) - # END verify type - - oldbinsha = None - if logmsg is not None: - try: - oldbinsha = self.commit.binsha - except ValueError: - oldbinsha = Commit.NULL_BIN_SHA - # END handle non-existing - # END retrieve old hexsha - - fpath = self.abspath - assure_directory_exists(fpath, is_file=True) - - lfd = LockedFD(fpath) - fd = lfd.open(write=True, stream=True) - ok = True - try: - fd.write(write_value.encode('ascii') + b'\n') - lfd.commit() - ok = True - finally: - if not ok: - lfd.rollback() - # Adjust the reflog - if logmsg is not None: - self.log_append(oldbinsha, logmsg) - - return self - - # aliased reference - reference = property(_get_reference, set_reference, doc="Returns the Reference we point to") - ref = reference - - def is_valid(self): - """ - :return: - True if the reference is valid, hence it can be read and points to - a valid object or reference.""" - try: - self.object - except (OSError, ValueError): - return False - else: - return True - - @property - def is_detached(self): - """ - :return: - True if we are a detached reference, hence we point to a specific commit - instead to another reference""" - try: - self.ref - return False - except TypeError: - return True - - def log(self): - """ - :return: RefLog for this reference. Its last entry reflects the latest change - applied to this reference - - .. note:: As the log is parsed every time, its recommended to cache it for use - instead of calling this method repeatedly. It should be considered read-only.""" - return RefLog.from_file(RefLog.path(self)) - - def log_append(self, oldbinsha, message, newbinsha=None): - """Append a logentry to the logfile of this ref - - :param oldbinsha: binary sha this ref used to point to - :param message: A message describing the change - :param newbinsha: The sha the ref points to now. If None, our current commit sha - will be used - :return: added RefLogEntry instance""" - # NOTE: we use the committer of the currently active commit - this should be - # correct to allow overriding the committer on a per-commit level. - # See https://github.com/gitpython-developers/GitPython/pull/146 - try: - committer_or_reader = self.commit.committer - except ValueError: - committer_or_reader = self.repo.config_reader() - # end handle newly cloned repositories - return RefLog.append_entry(committer_or_reader, RefLog.path(self), oldbinsha, - (newbinsha is None and self.commit.binsha) or newbinsha, - message) - - def log_entry(self, index): - """:return: RefLogEntry at the given index - :param index: python list compatible positive or negative index - - .. note:: This method must read part of the reflog during execution, hence - it should be used sparringly, or only if you need just one index. - In that case, it will be faster than the ``log()`` method""" - return RefLog.entry_at(RefLog.path(self), index) - - @classmethod - def to_full_path(cls, path) -> 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``""" - if isinstance(path, SymbolicReference): - path = path.path - full_ref_path = path - if not cls._common_path_default: - return full_ref_path - if not path.startswith(cls._common_path_default + "/"): - full_ref_path = '%s/%s' % (cls._common_path_default, path) - return full_ref_path - - @classmethod - def delete(cls, repo, path): - """Delete the reference at the given path - - :param repo: - Repository to delete the reference from - - :param path: - Short or full path pointing to the reference, i.e. refs/myreference - or just "myreference", hence 'refs/' is implied. - Alternatively the symbolic reference to be deleted""" - full_ref_path = cls.to_full_path(path) - abs_path = osp.join(repo.common_dir, full_ref_path) - if osp.exists(abs_path): - os.remove(abs_path) - else: - # check packed refs - pack_file_path = cls._get_packed_refs_path(repo) - try: - with open(pack_file_path, 'rb') as reader: - new_lines = [] - made_change = False - dropped_last_line = False - for line in reader: - line = line.decode(defenc) - _, _, 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('^')): - new_lines.append(line) - dropped_last_line = False - continue - # END skip comments and lines without our path - - # drop this line - made_change = True - dropped_last_line = True - - # write the new lines - 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: - fd.writelines(line.encode(defenc) for line in new_lines) - - except (OSError, IOError): - pass # it didn't exist at all - - # delete the reflog - reflog_path = RefLog.path(cls(repo, full_ref_path)) - if osp.isfile(reflog_path): - os.remove(reflog_path) - # END remove reflog - - @classmethod - def _create(cls, repo, path, resolve, reference, force, logmsg=None): - """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 - corresponding object and a detached symbolic reference will be created - instead""" - git_dir = _git_dir(repo, path) - full_ref_path = cls.to_full_path(path) - abs_ref_path = osp.join(git_dir, full_ref_path) - - # figure out target data - target = reference - if resolve: - target = repo.rev_parse(str(reference)) - - if not force and osp.isfile(abs_ref_path): - target_data = str(target) - if isinstance(target, SymbolicReference): - target_data = target.path - if not resolve: - target_data = "ref: " + target_data - 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)) - # END no force handling - - ref = cls(repo, full_ref_path) - ref.set_reference(target, logmsg) - 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. - - :param repo: - Repository to create the reference in - - :param path: - full path at which the new symbolic reference is supposed to be - created at, i.e. "NEW_HEAD" or "symrefs/my_new_symref" - - :param reference: - The reference to which the new symbolic reference should point to. - If it is a commit'ish, the symbolic ref will be detached. - - :param force: - if True, force creation even if a symbolic reference with that name already exists. - Raise OSError otherwise - - :param logmsg: - If not None, the message to append to the reflog. Otherwise no reflog - entry is written. - - :return: Newly created symbolic Reference - - :raise OSError: - If a (Symbolic)Reference with the same name but different contents - 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) - - def rename(self, new_path, force=False): - """Rename self to a new path - - :param new_path: - Either a simple name or a full path, i.e. new_name or features/new_name. - The prefix refs/ is implied for references and will be set as needed. - In case this is a symbolic ref, there is no implied prefix - - :param force: - If True, the rename will succeed even if a head with the target name - 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 """ - new_path = self.to_full_path(new_path) - if self.path == new_path: - return self - - new_abs_path = osp.join(_git_dir(self.repo, new_path), new_path) - cur_abs_path = osp.join(_git_dir(self.repo, self.path), self.path) - if osp.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: - f1 = fd1.read().strip() - 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) - # else: we could remove ourselves and use the otherone, but - # but clarity we just continue as usual - # END not force handling - os.remove(new_abs_path) - # END handle existing target file - - dname = osp.dirname(new_abs_path) - if not osp.isdir(dname): - os.makedirs(dname) - # END create directory - - os.rename(cur_abs_path, new_abs_path) - self.path = new_path - - return self - - @classmethod - def _iter_items(cls, repo, common_path=None): - 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'] - if refs_id: - dirs[0:] = ['refs'] - # END prune non-refs folders - - for f in files: - 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) + '/', "")) - # END for each file in root directory - # END for each directory to walk - - # read packed refs - for _sha, rela_path in cls._iter_packed_refs(repo): - if rela_path.startswith(common_path): - rela_paths.add(rela_path) - # END relative path matches common path - # END packed refs reading - - # return paths in sorted order - for path in sorted(rela_paths): - try: - yield cls.from_path(repo, path) - except ValueError: - continue - # END for each sorted relative refpath - - @classmethod - def iter_items(cls, repo, common_path=None): - """Find all refs in the repository - - :param repo: is the Repo - - :param common_path: - Optional keyword argument to the path which is to be shared by all - returned Ref objects. - Defaults to class specific portion if None assuring that only - refs suitable for the actual class are returned. - - :return: - git.SymbolicReference[], each of them is guaranteed to be a symbolic - ref which is not detached and pointing to a valid ref - - 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) - - @classmethod - def from_path(cls, repo, path): - """ - :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 - :return: - Instance of type Reference, Head, or Tag - depending on the given path""" - if not path: - raise ValueError("Cannot create Reference from %r" % path) - - # 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): - try: - instance = ref_type(repo, path) - if instance.__class__ == SymbolicReference and instance.is_detached: - raise ValueError("SymbolRef was detached, we drop it") - return instance - except ValueError: - pass - # END exception handling - # END for each type to try - raise ValueError("Could not find reference type suitable to handle path %r" % path) - - def is_remote(self): - """:return: True if this symbolic reference points to a remote branch""" - return self.path.startswith(self._remote_common_path_default + "/") -- cgit v1.2.1 From 9e5e969479ec6018e1ba06b95bcdefca5b0082a4 Mon Sep 17 00:00:00 2001 From: Yobmod Date: Mon, 19 Jul 2021 19:10:45 +0100 Subject: Change remaining type comments to py3.6+ types --- git/refs/tag.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'git/refs') diff --git a/git/refs/tag.py b/git/refs/tag.py index aa3b82a2..281ce09a 100644 --- a/git/refs/tag.py +++ b/git/refs/tag.py @@ -35,7 +35,7 @@ class TagReference(Reference): _common_path_default = Reference._common_path_default + "/" + _common_default @property - def commit(self) -> 'Commit': # type: ignore[override] # LazyMixin has unrelated + 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""" -- cgit v1.2.1 From 600df043e76924d43a4f9f88f4e3241740f34c77 Mon Sep 17 00:00:00 2001 From: Yobmod Date: Mon, 19 Jul 2021 20:26:53 +0100 Subject: Rmv old py2.7 __future__ imports --- git/refs/__init__.py | 1 - 1 file changed, 1 deletion(-) (limited to 'git/refs') 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 * -- cgit v1.2.1