-
Notifications
You must be signed in to change notification settings - Fork 86
/
github.py
216 lines (187 loc) · 8.98 KB
/
github.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
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
#!/usr/bin/env python
import logging
log = logging.getLogger('git_repo.github')
from ..service import register_target, RepositoryService, os
from ...exceptions import ResourceError, ResourceExistsError, ResourceNotFoundError
import github3
from git.exc import GitCommandError
@register_target('hub', 'github')
class GithubService(RepositoryService):
fqdn = 'github.com'
def __init__(self, *args, **kwarg):
self.gh = github3.GitHub()
super(GithubService, self).__init__(*args, **kwarg)
def connect(self):
try:
self.gh.login(token=self._privatekey)
self.username = self.gh.user().login
except github3.models.GitHubError as err:
if err.code is 401:
if not self._privatekey:
raise ConnectionError('Could not connect to Github. '
'Please configure .gitconfig '
'with your github private key.') from err
else:
raise ConnectionError('Could not connect to Github. '
'Check your configuration and try again.') from err
def create(self, user, repo, add=False):
if user != self.username:
raise NotImplementedError("Project creation supported for authentified user only!")
try:
self.gh.create_repo(repo)
except github3.models.GitHubError as err:
if err.code == 422 or err.message == 'name already exists on this account':
raise ResourceExistsError("Project already exists.") from err
else: # pragma: no cover
raise ResourceError("Unhandled error.") from err
if add:
self.add(user=self.username, repo=repo, tracking=self.name)
def fork(self, user, repo):
try:
return self.gh.repository(user, repo).create_fork().full_name
except github3.models.GitHubError as err:
if err.message == 'name already exists on this account':
raise ResourceExistsError("Project already exists.") from err
else: # pragma: no cover
raise ResourceError("Unhandled error: {}".format(err)) from err
def delete(self, repo, user=None):
if not user:
user = self.username
try:
repository = self.gh.repository(user, repo)
if repository:
result = repository.delete()
if not repository or not result:
raise ResourceNotFoundError('Cannot delete: repository {}/{} does not exists.'.format(user, repo))
except github3.models.GitHubError as err: # pragma: no cover
if err.code == 403:
raise ResourcePermissionError('You don\'t have enough permissions for deleting the repository. '
'Check the namespace or the private token\'s privileges') from err
raise ResourceError('Unhandled exception: {}'.format(err)) from err
def get_repository(self, user, repo):
repository = self.gh.repository(user, repo)
if not repository:
raise ResourceNotFoundError('Cannot delete: repository {}/{} does not exists.'.format(user, repo))
return repository
def _format_gist(self, gist):
return gist.split('https://gist.github.com/')[-1].split('.git')[0]
def gist_list(self, gist=None):
if not gist:
for gist in self.gh.iter_gists(self.gh.user().login):
yield (gist.html_url, gist.description)
else:
gist = self.gh.gist(self._format_gist(gist))
if gist is None:
raise ResourceNotFoundError('Gist does not exists.')
for gist_file in gist.iter_files():
yield (gist_file.language if gist_file.language else 'Raw text',
gist_file.size,
gist_file.filename)
def gist_fetch(self, gist, fname=None):
try:
gist = self.gh.gist(self._format_gist(gist))
except Exception as err:
raise ResourceNotFoundError('Could not find gist') from err
if gist.files == 1 and not fname:
gist_file = list(gist.iter_files())[0]
else:
for gist_file in gist.iter_files():
if gist_file.filename == fname:
break
else:
raise ResourceNotFoundError('Could not find file within gist.')
return gist_file.content
def gist_clone(self, gist):
try:
gist = self.gh.gist(gist.split('https://gist.github.com/')[-1])
except Exception as err:
raise ResourceNotFoundError('Could not find gist') from err
remote = self.repository.create_remote('gist', gist.git_push_url)
self.pull(remote, 'master')
def gist_create(self, gist_pathes, description, secret=False):
def load_file(fname, path='.'):
with open(os.path.join(path, fname), 'r') as f:
return {'content': f.read()}
gist_files = dict()
for gist_path in gist_pathes:
if not os.path.isdir(gist_path):
gist_files[os.path.basename(gist_path)] = load_file(gist_path)
else:
for gist_file in os.listdir(gist_path):
if not os.path.isdir(os.path.join(gist_path, gist_file)) and not gist_file.startswith('.'):
gist_files[gist_file] = load_file(gist_file, gist_path)
gist = self.gh.create_gist(
description=description,
files=gist_files,
public=not secret # isn't it obvious? ☺
)
return gist.html_url
def gist_delete(self, gist_id):
gist = self.gh.gist(self._format_gist(gist_id))
if not gist:
raise ResourceNotFoundError('Could not find gist')
gist.delete()
def request_create(self, user, repo, local_branch, remote_branch, title, description=None):
repository = self.gh.repository(user, repo)
if not repository:
raise ResourceNotFoundError('Could not find repository `{}/{}`!'.format(user, repo))
if not remote_branch:
remote_branch = self.repository.active_branch.name
if not local_branch:
local_branch = repository.master_branch or 'master'
try:
request = repository.create_pull(title,
base=local_branch,
head=':'.join([user, remote_branch]),
body=description)
except github3.models.GitHubError as err:
if err.code == 422:
if err.message == 'Validation Failed':
for error in err.errors:
if 'message' in error:
raise ResourceError(error['message'])
raise ResourceError("Unhandled formatting error: {}".format(err.errors))
raise ResourceError(err.message)
return {'local': local_branch, 'remote': remote_branch, 'ref': request.number}
def request_list(self, user, repo):
repository = self.gh.repository(user, repo)
for pull in repository.iter_pulls():
yield ( str(pull.number), pull.title, pull.links['issue'] )
def request_fetch(self, user, repo, request, pull=False):
if pull:
raise NotImplementedError('Pull operation on requests for merge are not yet supported')
try:
for remote in self.repository.remotes:
if remote.name == self.name:
local_branch_name = 'request/{}'.format(request)
self.fetch(
remote,
'pull/{}/head'.format(request),
local_branch_name
)
return local_branch_name
else:
raise ResourceNotFoundError('Could not find remote {}'.format(self.name))
except GitCommandError as err:
if 'Error when fetching: fatal: Couldn\'t find remote ref' in err.command[0]:
raise ResourceNotFoundError('Could not find opened request #{}'.format(request)) from err
raise err
@classmethod
def get_auth_token(cls, login, password, prompt=None):
import platform
gh = github3.GitHub()
gh.login(login, password, two_factor_callback=lambda: prompt('2FA code> '))
try:
auth = gh.authorize(login, password,
scopes=[ 'repo', 'delete_repo', 'gist' ],
note='git-repo2 token used on {}'.format(platform.node()),
note_url='https://github.com/guyzmo/git-repo')
return auth.token
except github3.models.GitHubError as err:
if len(err.args) > 0 and 422 == err.args[0].status_code:
raise ResourceExistsError("A token already exist for this machine on your github account.")
else:
raise err
@property
def user(self):
return self.gh.user().login