-
Notifications
You must be signed in to change notification settings - Fork 844
/
Copy pathgit_wrapper.py
169 lines (137 loc) · 4.8 KB
/
git_wrapper.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
# -*- coding: utf-8 -*-
"""
Wrap python git interface for compatibility with older/newer version
"""
try:
from itertools import zip_longest
except ImportError:
from six.moves import zip_longest
import logging
import os
from time import mktime
from datetime import datetime
from pelican.utils import set_date_tzinfo
from git import Git, Repo
DEV_LOGGER = logging.getLogger(__name__)
def grouper(iterable, n, fillvalue=None):
'''
Collect data into fixed-length chunks or blocks
'''
# grouper('ABCDEFG', 3, 'x') --> ABC DEF Gxx
args = [iter(iterable)] * n
return zip_longest(fillvalue=fillvalue, *args)
class _GitWrapperCommon(object):
'''
Wrap git module to provide a more stable interface across versions
'''
def __init__(self, repo_path):
self.git = Git()
self.git.update_environment(
GIT_CONFIG_NOSYSTEM='true',
HOME=os.getcwd(),
XDG_CONFIG_HOME=os.getcwd()
)
self.repo = Repo(os.path.abspath("."), search_parent_directories=True)
def is_file_managed_by_git(self, path):
'''
:param path: Path to check
:returns: True if path is managed by git
'''
status, _stdout, _stderr = self.git.execute(
['git', 'ls-files', path, '--error-unmatch'],
with_extended_output=True,
with_exceptions=False)
return status == 0
def is_file_modified(self, path):
'''
Does a file have local changes not yet committed
:returns: True if file has local changes
'''
status, _stdout, _stderr = self.git.execute(
['git', 'diff', '--quiet', 'HEAD', path],
with_extended_output=True,
with_exceptions=False)
return status != 0
def get_commits_following(self, path):
'''
Get all commits including path following the file through
renames
:param path: Path which we will find commits for
:returns: Sequence of commit objects. Newest to oldest
'''
return [
commit for commit, _ in self.get_commits_and_names_iter(
path)]
def get_commits_and_names_iter(self, path):
'''
Get all commits including a given path following renames
'''
log_result = self.git.log(
'--pretty=%H',
'--follow',
'--name-only',
'--',
path).splitlines()
for commit_sha, _, filename in grouper(log_result, 3):
yield self.repo.commit(commit_sha), filename
def get_commits(self, path, follow=False):
'''
Get all commits including path
:param path: Path which we will find commits for
:param bool follow: If True we will follow path through renames
:returns: Sequence of commit objects. Newest to oldest
'''
if follow:
return self.get_commits_following(path)
else:
return self._get_commits(path)
class _GitWrapperLegacy(_GitWrapperCommon):
def _get_commits(self, path):
'''
Get all commits including path without following renames
:param path: Path which we will find commits for
:returns: Sequence of commit objects. Newest to oldest
'''
return self.repo.commits(path=path)
@staticmethod
def get_commit_date(commit, tz_name):
'''
Get datetime of commit comitted_date
'''
return set_date_tzinfo(
datetime.fromtimestamp(mktime(commit.committed_date)),
tz_name=tz_name)
class _GitWrapper(_GitWrapperCommon):
def _get_commits(self, path):
'''
Get all commits including path without following renames
:param path: Path which we will find commits for
:returns: Sequence of commit objects. Newest to oldest
.. NOTE ::
If this fails it could be that your gitpython version is out of
sync with the git binary on your distro.
Make sure you use the correct gitpython version.
Alternatively enabling GIT_FILETIME_FOLLOW may also make your
problem go away.
'''
return list(self.repo.iter_commits(paths=path))
@staticmethod
def get_commit_date(commit, tz_name):
'''
Get datetime of commit comitted_date
'''
return set_date_tzinfo(
datetime.fromtimestamp(commit.committed_date),
tz_name=tz_name)
_wrapper_cache = {}
def git_wrapper(path):
'''
Get appropriate wrapper factory and cache instance for path
'''
path = os.path.abspath(path)
if path not in _wrapper_cache:
if hasattr(Repo, 'commits'):
_wrapper_cache[path] = _GitWrapperLegacy(path)
else:
_wrapper_cache[path] = _GitWrapper(path)
return _wrapper_cache[path]