"get_object_type_by_name",

"parse_date",

"parse_actor_and_date",

"ProcessStreamAdapter",

"Traversable",

"altz_to_utctz_str",

"utctz_to_altz",

"verify_utctz",

"Actor",

"tzoffset",

"utc",

"""Convert mode bits from an octal mode string to an integer mode for git.

if isinstance(modestr, bytes):

# A byte is an int. int(b"4"[0]) is 52, so b"100644" became 0o6767524.

modestr = modestr.decode("ascii")

mode = 0

for iteration, char in enumerate(reversed(modestr[-6:])):

mode += int(char) << iteration * 3

# END for each char

object_type_name: bytes,

) -> Union[Type["Commit"], Type["TagObject"], Type["Tree"], Type["Blob"]]:

"""Retrieve the Python class GitPython uses to represent a kind of Git object.

if object_type_name == b"commit":

from . import commit

return commit.Commit

elif object_type_name == b"tag":

from . import tag

return tag.TagObject

elif object_type_name == b"blob":

from . import blob

return blob.Blob

elif object_type_name == b"tree":

from . import tree

return tree.Tree

else:

"""Convert a git timezone offset into a timezone offset west of UTC in seconds

int_utctz = int(utctz)

seconds = (abs(int_utctz) // 100) * 3600 + (abs(int_utctz) % 100) * 60

"""Convert a timezone offset west of UTC in seconds into a Git timezone offset

hours = abs(altz) // 3600

minutes = (abs(altz) % 3600) // 60

sign = "-" if altz >= 60 else "+"

"""

fmt_exc = ValueError("Invalid timezone offset format: %s" % offset)

if len(offset) != 5:

raise fmt_exc

if offset[0] not in "+-":

raise fmt_exc

if offset[1] not in digits or offset[2] not in digits or offset[3] not in digits or offset[4] not in digits:

raise fmt_exc

# END for each char

def __init__(self, secs_west_of_utc: float, name: Union[None, str] = None) -> None:

self._offset = timedelta(seconds=-secs_west_of_utc)

self._name = name or "fixed"

def __reduce__(self) -> Tuple[Type["tzoffset"], Tuple[float, str]]:

return tzoffset, (-self._offset.total_seconds(), self._name)

def utcoffset(self, dt: Union[datetime, None]) -> timedelta:

return self._offset

def tzname(self, dt: Union[datetime, None]) -> str:

return self._name

def dst(self, dt: Union[datetime, None]) -> timedelta:

"""Convert a `timestamp` + `tz_offset` into an aware :class:`~datetime.datetime`

utc_dt = datetime.fromtimestamp(timestamp, utc)

try:

local_dt = utc_dt.astimezone(tzoffset(tz_offset))

return local_dt

except ValueError:

"""Parse the given date as one of the following:

if isinstance(string_date, datetime):

if string_date.tzinfo:

utcoffset = cast(timedelta, string_date.utcoffset()) # typeguard, if tzinfoand is not None

offset = -int(utcoffset.total_seconds())

return int(string_date.astimezone(utc).timestamp()), offset

else:

raise ValueError(f"string_date datetime object without tzinfo, {string_date}")

# Git time

try:

if string_date.count(" ") == 1 and string_date.rfind(":") == -1:

timestamp, offset_str = string_date.split()

if timestamp.startswith("@"):

timestamp = timestamp[1:]

timestamp_int = int(timestamp)

return timestamp_int, utctz_to_altz(verify_utctz(offset_str))

else:

offset_str = "+0000" # Local time by default.

if string_date[-5] in "-+":

offset_str = verify_utctz(string_date[-5:])

string_date = string_date[:-6] # skip space as well

# END split timezone info

offset = utctz_to_altz(offset_str)

# Now figure out the date and time portion - split time.

date_formats = []

splitter = -1

if "," in string_date:

date_formats.append("%a, %d %b %Y")

splitter = string_date.rfind(" ")

else:

# ISO plus additional

date_formats.append("%Y-%m-%d")

date_formats.append("%Y.%m.%d")

date_formats.append("%m/%d/%Y")

date_formats.append("%d.%m.%Y")

splitter = string_date.rfind("T")

if splitter == -1:

splitter = string_date.rfind(" ")

# END handle 'T' and ' '

# END handle RFC or ISO

assert splitter > -1

# Split date and time.

time_part = string_date[splitter + 1 :] # Skip space.

date_part = string_date[:splitter]

# Parse time.

tstruct = time.strptime(time_part, "%H:%M:%S")

for fmt in date_formats:

try:

dtstruct = time.strptime(date_part, fmt)

utctime = calendar.timegm(

(

dtstruct.tm_year,

dtstruct.tm_mon,

dtstruct.tm_mday,

tstruct.tm_hour,

tstruct.tm_min,

tstruct.tm_sec,

dtstruct.tm_wday,

dtstruct.tm_yday,

tstruct.tm_isdst,

)

)

return int(utctime), offset

except ValueError:

continue

# END exception handling

# END for each fmt

# Still here ? fail.

raise ValueError("no format matched")

# END handle format

except Exception as e:

raise ValueError(f"Unsupported date format or type: {string_date}, type={type(string_date)}") from e

"""Parse out the actor (author or committer) info from a line like::

actor, epoch, offset = "", "0", "0"

m = _re_actor_epoch.search(line)

if m:

actor, epoch, offset = m.groups()

else:

m = _re_only_actor.search(line)

actor = m.group(1) if m else line or ""

"""Class wiring all calls to the contained Process instance.

__slots__ = ("_proc", "_stream")

def __init__(self, process: "Popen", stream_name: str) -> None:

self._proc = process

self._stream: StringIO = getattr(process, stream_name) # guessed type

def __getattr__(self, attr: str) -> Any:

"""Simple interface to perform depth-first or breadth-first traversals in one

__slots__ = ()

@classmethod

@abstractmethod

def _get_intermediate_items(cls, item: Any) -> Sequence["Traversable"]:

"""

raise NotImplementedError("To be implemented in subclass")

@abstractmethod

def list_traverse(self, *args: Any, **kwargs: Any) -> Any:

"""Traverse self and collect all items found.

warnings.warn(

"list_traverse() method should only be called from subclasses."

" Calling from Traversable abstract class will raise NotImplementedError in 4.0.0."

" The concrete subclasses in GitPython itself are 'Commit', 'RootModule', 'Submodule', and 'Tree'.",

DeprecationWarning,

stacklevel=2,

)

return self._list_traverse(*args, **kwargs)

def _list_traverse(

self, as_edge: bool = False, *args: Any, **kwargs: Any

) -> IterableList[Union["Commit", "Submodule", "Tree", "Blob"]]:

"""Traverse self and collect all items found.

# Commit and Submodule have id.__attribute__ as IterableObj.

# Tree has id.__attribute__ inherited from IndexObject.

if isinstance(self, Has_id_attribute):

id = self._id_attribute_

else:

# Shouldn't reach here, unless Traversable subclass created with no

# _id_attribute_.

id = ""

# Could add _id_attribute_ to Traversable, or make all Traversable also

# Iterable?

if not as_edge:

out: IterableList[Union["Commit", "Submodule", "Tree", "Blob"]] = IterableList(id)

out.extend(self.traverse(as_edge=as_edge, *args, **kwargs)) # noqa: B026

return out

# Overloads in subclasses (mypy doesn't allow typing self: subclass).

# Union[IterableList['Commit'], IterableList['Submodule'], IterableList[Union['Submodule', 'Tree', 'Blob']]]

else:

# Raise DeprecationWarning, it doesn't make sense to use this.

out_list: IterableList = IterableList(self.traverse(*args, **kwargs))

return out_list

@abstractmethod

def traverse(self, *args: Any, **kwargs: Any) -> Any:

"""Iterator yielding items found when traversing self.

warnings.warn(

"traverse() method should only be called from subclasses."

" Calling from Traversable abstract class will raise NotImplementedError in 4.0.0."

" The concrete subclasses in GitPython itself are 'Commit', 'RootModule', 'Submodule', and 'Tree'.",

DeprecationWarning,

stacklevel=2,

)

return self._traverse(*args, **kwargs)

def _traverse(

self,

predicate: Callable[[Union["Traversable", "Blob", TraversedTup], int], bool] = lambda i, d: True,

prune: Callable[[Union["Traversable", "Blob", TraversedTup], int], bool] = lambda i, d: False,

depth: int = -1,

branch_first: bool = True,

visit_once: bool = True,

ignore_self: int = 1,

as_edge: bool = False,

) -> Union[Iterator[Union["Traversable", "Blob"]], Iterator[TraversedTup]]:

"""Iterator yielding items found when traversing `self`.

visited = set()

stack: Deque[TraverseNT] = deque()

stack.append(TraverseNT(0, self, None)) # self is always depth level 0.

def addToStack(

stack: Deque[TraverseNT],

src_item: "Traversable",

branch_first: bool,

depth: int,

) -> None:

lst = self._get_intermediate_items(item)

if not lst: # Empty list

return

if branch_first:

stack.extendleft(TraverseNT(depth, i, src_item) for i in lst)

else:

reviter = (TraverseNT(depth, lst[i], src_item) for i in range(len(lst) - 1, -1, -1))

stack.extend(reviter)

# END addToStack local method

while stack:

d, item, src = stack.pop() # Depth of item, item, item_source

if visit_once and item in visited:

continue

if visit_once:

visited.add(item)

rval: Union[TraversedTup, "Traversable", "Blob"]

if as_edge:

# If as_edge return (src, item) unless rrc is None

# (e.g. for first item).

rval = (src, item)

else:

rval = item

if prune(rval, d):

continue

skipStartItem = ignore_self and (item is self)

if not skipStartItem and predicate(rval, d):

yield rval

# Only continue to next level if this is appropriate!

next_d = d + 1

if depth > -1 and next_d > depth:

continue

addToStack(stack, item, branch_first, next_d)

"""Defines methods to serialize and deserialize objects from and into a data

__slots__ = ()

# @abstractmethod

def _serialize(self, stream: "BytesIO") -> "Serializable":

"""Serialize the data of this object into the given data stream.

raise NotImplementedError("To be implemented in subclass")

# @abstractmethod

def _deserialize(self, stream: "BytesIO") -> "Serializable":

"""Deserialize all information regarding this object from the stream.

__slots__ = ()

TIobj_tuple = Tuple[Union[T_TIobj, None], T_TIobj]

def list_traverse(self: T_TIobj, *args: Any, **kwargs: Any) -> IterableList[T_TIobj]:

return super()._list_traverse(*args, **kwargs)

@overload

def traverse(self: T_TIobj) -> Iterator[T_TIobj]: ...

@overload

def traverse(

self: T_TIobj,

predicate: Callable[[Union[T_TIobj, Tuple[Union[T_TIobj, None], T_TIobj]], int], bool],

prune: Callable[[Union[T_TIobj, Tuple[Union[T_TIobj, None], T_TIobj]], int], bool],

depth: int,

branch_first: bool,

visit_once: bool,

ignore_self: Literal[True],

as_edge: Literal[False],

) -> Iterator[T_TIobj]: ...

@overload

def traverse(

self: T_TIobj,

predicate: Callable[[Union[T_TIobj, Tuple[Union[T_TIobj, None], T_TIobj]], int], bool],

prune: Callable[[Union[T_TIobj, Tuple[Union[T_TIobj, None], T_TIobj]], int], bool],

depth: int,

branch_first: bool,

visit_once: bool,

ignore_self: Literal[False],

as_edge: Literal[True],

) -> Iterator[Tuple[Union[T_TIobj, None], T_TIobj]]: ...

@overload

def traverse(

self: T_TIobj,

predicate: Callable[[Union[T_TIobj, TIobj_tuple], int], bool],

prune: Callable[[Union[T_TIobj, TIobj_tuple], int], bool],

depth: int,

branch_first: bool,

visit_once: bool,

ignore_self: Literal[True],

as_edge: Literal[True],

) -> Iterator[Tuple[T_TIobj, T_TIobj]]: ...

def traverse(

self: T_TIobj,

predicate: Callable[[Union[T_TIobj, TIobj_tuple], int], bool] = lambda i, d: True,

prune: Callable[[Union[T_TIobj, TIobj_tuple], int], bool] = lambda i, d: False,

depth: int = -1,

branch_first: bool = True,

visit_once: bool = True,

ignore_self: int = 1,

as_edge: bool = False,

) -> Union[Iterator[T_TIobj], Iterator[Tuple[T_TIobj, T_TIobj]], Iterator[TIobj_tuple]]:

"""For documentation, see :meth:`Traversable._traverse`."""

## To typecheck instead of using cast:

#

# import itertools

# from git.types import TypeGuard

# def is_commit_traversed(inp: Tuple) -> TypeGuard[Tuple[Iterator[Tuple['Commit', 'Commit']]]]:

# for x in inp[1]:

# if not isinstance(x, tuple) and len(x) != 2:

# if all(isinstance(inner, Commit) for inner in x):

# continue

# return True

#

# ret = super(Commit, self).traverse(predicate, prune, depth, branch_first, visit_once, ignore_self, as_edge)

# ret_tup = itertools.tee(ret, 2)

# assert is_commit_traversed(ret_tup), f"{[type(x) for x in list(ret_tup[0])]}"

# return ret_tup[0]

return cast(

Union[Iterator[T_TIobj], Iterator[Tuple[Union[None, T_TIobj], T_TIobj]]],

super()._traverse(

predicate, # type: ignore[arg-type]

prune, # type: ignore[arg-type]

depth,

branch_first,

visit_once,

ignore_self,

as_edge,

),