-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathhoster_bitbucket.py
234 lines (185 loc) · 8.23 KB
/
hoster_bitbucket.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
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
# -*- coding: utf-8 -*-
#
# Copyright (C) 2018 Sandro Lutz <[email protected]>
#
# This software is licensed under GPLv3, see LICENSE for details.
from hoster_base import BaseHoster
from remote_repo import RemoteRepository
import validators
import requests
import json
from urllib.parse import urlencode
from slugify import slugify
class BitbucketHoster(BaseHoster):
def __init__(self, name, user, access_token, api_version, team=None, ignored_repositories=None):
'''
Initialize a GitLab hoster instance
:param str name: Hoster name
:param str user: Username for the git service
:param str access_token: Personal access token for the user
:param int api_version: API version
:param str team: Team name used for syncing
:param str ignored_repositories: List of repository names to be ignored
:raises ValueError: If the given parameters are invalid
'''
super().__init__(name, user, access_token, team, ignored_repositories)
if api_version not in [2]:
raise ValueError('Bitbucket API v' + str(api_version) + ' is not supported')
self.api_version = api_version
def getRepositoryList(self, visibility):
self._checkVisibility(visibility)
param = dict()
if visibility in ['internal' 'private']:
param['q'] = 'is_private = true'
if self.api_version == 2:
if self.organization != None:
url = self._getAPIUrl('/2.0/repositories/%(team)s', {'team': self.organization})
else:
url = self._getAPIUrl('/2.0/repositories/%(user)s', {'user': self.user})
else:
raise NotImplementedError('Operation not supported with API v' + str(self.api_version))
response = requests.get(url, params = param, auth = self._getBasicAuthentication())
repo_list = self._parseRepositoryListResponse(response)
next_link = response.json()['next']
while (next_link):
response = requests.get(next_link, auth = self._getBasicAuthentication())
repo_list += self._parseRepositoryListResponse(response)
next_link = response.json()['next']
return repo_list
def getRepository(self, name):
if name in self.ignored_repositories:
raise LookupError('Repository \'' + name + '\' not found')
if self.api_version == 2:
param = {'q': 'name="%(name)s"' % {'name': name}}
if self.organization != None:
url = self._getAPIUrl('/2.0/repositories/%(team)s', {'team': self.organization})
else:
url = self._getAPIUrl('/2.0/repositories/%(user)s', {'user': self.user})
else:
raise NotImplementedError('Operation not supported with API v' + str(self.api_version))
response = requests.get(url, params = param, auth = self._getBasicAuthentication())
if response.status_code in [200, 201, 202]:
for repo in response.json()['values']:
if repo['name'] == name:
return self._parseProjectResponse(repo)
raise LookupError('Repository \'' + name + '\' not found')
elif response.status_code in [401, 403]:
self._raisePermissionError(response)
else:
self._raiseConnectionError(response)
def createRepository(self, name, visibility, description=None, website=None):
if name in self.ignored_repositories:
raise PermissionError('Repository name \'' + name + '\' is in ignored-list of this hoster')
if self.api_version == 2:
if visibility != 'public':
is_private = True
else:
is_private = False
if self.organization != None:
url = self._getAPIUrl('/2.0/repositories/%(team)s/%(slug)s', {'team': self.organization, 'slug': slugify(name)})
else:
url = self._getAPIUrl('/2.0/repositories/%(user)s/%(slug)s', {'user': self.user, 'slug': slugify(name)})
values = {
'name': name,
'is_private': is_private,
'description': description,
'scm': 'git',
'has_wiki': False,
'has_issues': False,
'fork_policy': 'allow_forks'
}
else:
raise NotImplementedError('Operation not supported with API v' + str(self.api_version))
response = requests.post(url, json = values, auth = self._getBasicAuthentication(), headers = {'Content-Type': 'application/json'})
if response.status_code in [200, 201, 202]:
return self._parseProjectResponse(response.json())
elif response.status_code in [401, 403]:
self._raisePermissionError(response)
else:
self._raiseConnectionError(response)
def updateRepository(self, repo):
if self.api_version == 2:
if repo.visibility != 'public':
is_private = True
else:
is_private = False
if self.organization != None:
url = self._getAPIUrl('/2.0/repositories/%(team)s/%(id)s', {'team': self.organization, 'id': repo.id})
else:
url = self._getAPIUrl('/2.0/repositories/%(user)s/%(id)s', {'user': self.user, 'id': repo.id})
values = {
'name': repo.name,
'is_private': is_private,
'description': repo.description,
'website': repo.website,
'scm': 'git',
'has_wiki': False,
'has_issues': False,
'fork_policy': 'allow_forks'
}
else:
raise NotImplementedError('Operation not supported with API v' + str(self.api_version))
response = requests.put(url, json = values, auth = self._getBasicAuthentication(), headers = {'Content-Type': 'application/json'})
if response.status_code in [401, 403]:
self._raisePermissionError(response)
elif response.status_code not in [200, 201, 202]:
self._raiseConnectionError(response)
def deleteRepository(self, repo):
if self.api_version == 2:
if self.organization != None:
url = self._getAPIUrl('/2.0/repositories/%(team)s/%(slug)s', {'team': self.organization, 'id': repp.id})
else:
url = self._getAPIUrl('/2.0/repositories/%(user)s/%(slug)s', {'user': self.user, 'id': repo.id})
else:
raise NotImplementedError('Operation not supported with API v' + str(self.api_version))
response = requests.delete(url, auth = self._getBasicAuthentication())
if response.status_code in [401, 403]:
self._raisePermissionError(response)
elif response.status_code not in [200, 201, 202, 203, 204]:
self._raiseConnectionError(response)
def _checkVisibility(self, visibility):
if visibility not in ['all', 'public', 'internal', 'private']:
raise ValueError('Type \'' + visibility +'\' is unknown')
def _parseProjectResponse(self, response):
if response['is_private']:
visibility = 'private'
else:
visibility = 'public'
for link in response['links']['clone']:
if link['name'] == 'https':
http_url_to_repo = 'https://' + link['href'].split('@', 2)[1]
return RemoteRepository(response['uuid'], response['name'], visibility, response['description'], response['website'],
http_url_to_repo, response['links']['self']['href'], self.name, self.user, self.password)
def _getAPIUrl(self, path, parameters=dict()):
'''
Generate the complete URL for the given parameters.
:param str path: Path within the API (can contain placeholders e.g. %(user_id)s)
:param dict parameters: Values to be filled into the placeholders of the given path
:return: Complete API URL
:rtype: str
'''
url = 'https://api.bitbucket.org'
url += path % parameters
return url
def _getBasicAuthentication(self):
return (self.user, self.password)
def _parseRepositoryListResponse(self, response):
'''
Parse the API response and return all (non-ignored) repositories as a list.
:param object response: request response object
:return: returns a list of RemoteRepository objects
:rtype: list
:raises PermissionError: if the access is denied by the server
:raises ConnectionError: if another HTTP error has occurred
'''
if response.status_code in [200, 201, 202]:
repo_list = list()
values = response.json()['values']
for repo in values:
if (repo['name'] not in self.ignored_repositories):
repo_list.append(self._parseProjectResponse(repo))
return repo_list
elif response.status_code in [401, 403]:
self._raisePermissionError(response)
else:
self._raiseConnectionError(response)