diff options
Diffstat (limited to 'git/objects/fun.py')
| -rw-r--r-- | git/objects/fun.py | 61 |
1 files changed, 43 insertions, 18 deletions
diff --git a/git/objects/fun.py b/git/objects/fun.py index 339a53b8..fc2ea1e7 100644 --- a/git/objects/fun.py +++ b/git/objects/fun.py @@ -1,6 +1,7 @@ """Module with functions which are supposed to be as fast as possible""" from stat import S_ISDIR + from git.compat import ( safe_decode, defenc @@ -8,8 +9,14 @@ from git.compat import ( # typing ---------------------------------------------- -from typing import List, Tuple +from typing import Callable, List, MutableSequence, Sequence, Tuple, TYPE_CHECKING, Union, overload + +if TYPE_CHECKING: + from _typeshed import ReadableBuffer + from git import GitCmdObjectDB +EntryTup = Tuple[bytes, int, str] # same as TreeCacheTup in tree.py +EntryTupOrNone = Union[EntryTup, None] # --------------------------------------------------- @@ -18,7 +25,7 @@ __all__ = ('tree_to_stream', 'tree_entries_from_data', 'traverse_trees_recursive 'traverse_tree_recursive') -def tree_to_stream(entries, write): +def tree_to_stream(entries: Sequence[EntryTup], write: Callable[['ReadableBuffer'], Union[int, None]]) -> None: """Write the give list of entries into a stream using its write method :param entries: **sorted** list of tuples with (binsha, mode, name) :param write: write method which takes a data string""" @@ -42,12 +49,14 @@ def tree_to_stream(entries, write): # According to my tests, this is exactly what git does, that is it just # takes the input literally, which appears to be utf8 on linux. if isinstance(name, str): - name = name.encode(defenc) - write(b''.join((mode_str, b' ', name, b'\0', binsha))) + name_bytes = name.encode(defenc) + else: + name_bytes = name + write(b''.join((mode_str, b' ', name_bytes, b'\0', binsha))) # END for each item -def tree_entries_from_data(data: bytes) -> List[Tuple[bytes, int, str]]: +def tree_entries_from_data(data: bytes) -> List[EntryTup]: """Reads the binary representation of a tree and returns tuples of Tree items :param data: data block with tree data (as bytes) :return: list(tuple(binsha, mode, tree_relative_path), ...)""" @@ -93,11 +102,13 @@ def tree_entries_from_data(data: bytes) -> List[Tuple[bytes, int, str]]: return out -def _find_by_name(tree_data, name, is_dir, start_at): +def _find_by_name(tree_data: MutableSequence[EntryTupOrNone], name: str, is_dir: bool, start_at: int + ) -> EntryTupOrNone: """return data entry matching the given name and tree mode or None. Before the item is returned, the respective data item is set None in the tree_data list to mark it done""" + try: item = tree_data[start_at] if item and item[2] == name and S_ISDIR(item[1]) == is_dir: @@ -115,16 +126,27 @@ def _find_by_name(tree_data, name, is_dir, start_at): return None -def _to_full_path(item, path_prefix): +@ overload +def _to_full_path(item: None, path_prefix: str) -> None: + ... + + +@ overload +def _to_full_path(item: EntryTup, path_prefix: str) -> EntryTup: + ... + + +def _to_full_path(item: EntryTupOrNone, path_prefix: str) -> EntryTupOrNone: """Rebuild entry with given path prefix""" if not item: return item return (item[0], item[1], path_prefix + item[2]) -def traverse_trees_recursive(odb, tree_shas, path_prefix): +def traverse_trees_recursive(odb: 'GitCmdObjectDB', tree_shas: Sequence[Union[bytes, None]], + path_prefix: str) -> List[Tuple[EntryTupOrNone, ...]]: """ - :return: list with entries according to the given binary tree-shas. + :return: list of list with entries according to the given binary tree-shas. The result is encoded in a list of n tuple|None per blob/commit, (n == len(tree_shas)), where * [0] == 20 byte sha @@ -137,28 +159,31 @@ def traverse_trees_recursive(odb, tree_shas, path_prefix): :param path_prefix: a prefix to be added to the returned paths on this level, set it '' for the first iteration :note: The ordering of the returned items will be partially lost""" - trees_data = [] + trees_data: List[List[EntryTupOrNone]] = [] + nt = len(tree_shas) for tree_sha in tree_shas: if tree_sha is None: - data = [] + data: List[EntryTupOrNone] = [] else: - data = tree_entries_from_data(odb.stream(tree_sha).read()) + # make new list for typing as list invariant + data = [x for x in tree_entries_from_data(odb.stream(tree_sha).read())] # END handle muted trees trees_data.append(data) # END for each sha to get data for - out = [] - out_append = out.append + out: List[Tuple[EntryTupOrNone, ...]] = [] # find all matching entries and recursively process them together if the match # is a tree. If the match is a non-tree item, put it into the result. # Processed items will be set None for ti, tree_data in enumerate(trees_data): + for ii, item in enumerate(tree_data): if not item: continue # END skip already done items + entries: List[EntryTupOrNone] entries = [None for _ in range(nt)] entries[ti] = item _sha, mode, name = item @@ -170,16 +195,16 @@ def traverse_trees_recursive(odb, tree_shas, path_prefix): for tio in range(ti + 1, ti + nt): tio = tio % nt entries[tio] = _find_by_name(trees_data[tio], name, is_dir, ii) - # END for each other item data + # END for each other item data # if we are a directory, enter recursion if is_dir: out.extend(traverse_trees_recursive( odb, [((ei and ei[0]) or None) for ei in entries], path_prefix + name + '/')) else: - out_append(tuple(_to_full_path(e, path_prefix) for e in entries)) - # END handle recursion + out.append(tuple(_to_full_path(e, path_prefix) for e in entries)) + # END handle recursion # finally mark it done tree_data[ii] = None # END for each item @@ -190,7 +215,7 @@ def traverse_trees_recursive(odb, tree_shas, path_prefix): return out -def traverse_tree_recursive(odb, tree_sha, path_prefix): +def traverse_tree_recursive(odb: 'GitCmdObjectDB', tree_sha: bytes, path_prefix: str) -> List[EntryTup]: """ :return: list of entries of the tree pointed to by the binary tree_sha. An entry has the following format: |