This repository was archived by the owner on Nov 4, 2020. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 104
Expand file tree
/
Copy pathbase.py
More file actions
172 lines (138 loc) · 6.25 KB
/
base.py
File metadata and controls
172 lines (138 loc) · 6.25 KB
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
# -*- coding: utf-8 -*-
import requests
class Base(object):
"""
Base class
On init we setup the token used for all the api calls and all the urls
:param host: host of gitlab
:param token: token
:param verify_ssl: Weather or not to verify the SSL cert
:param auth: Authentication
:param timeout: Timeout
:param suppress_http_error: Use :obj:`False` to unsuppress
:class:`requests.exceptions.HTTPError` exceptions on failure
:return: None
"""
def __init__(self, host, token=None, oauth_token=None, verify_ssl=True, auth=None, timeout=None,
suppress_http_error=True):
self.suppress_http_error = suppress_http_error
if token:
self.token = token
self.headers = {'PRIVATE-TOKEN': self.token}
if oauth_token:
self.oauth_token = oauth_token
self.headers = {'Authorization': 'Bearer {}'.format(
self.oauth_token)}
if not host:
raise ValueError('host argument may not be empty')
self.host = host.rstrip('/')
if self.host.startswith('http://') or self.host.startswith('https://'):
pass
else:
self.host = 'https://' + self.host
self.auth = auth
self.api_url = self.host + '/api/v3'
self.projects_url = self.api_url + '/projects'
self.users_url = self.api_url + '/users'
self.keys_url = self.api_url + '/user/keys'
self.groups_url = self.api_url + '/groups'
self.search_url = self.api_url + '/projects/search'
self.hook_url = self.api_url + '/hooks'
self.namespaces_url = self.api_url + '/namespaces'
self.verify_ssl = verify_ssl
self.timeout = timeout
def get(self, uri, default_response=None, **kwargs):
"""
Call GET on the Gitlab server
>>> gitlab = Gitlab(host='http://localhost:10080', verify_ssl=False)
>>> gitlab.login(user='root', password='5iveL!fe')
>>> gitlab.get('/users/5')
:param uri: String with the URI for the endpoint to GET from
:param default_response: Return value if JSONDecodeError
:param kwargs: Key word arguments to use as GET arguments
:return: Dictionary containing response data
:raise: HttpError: If invalid response returned
"""
url = self.api_url + uri
response = requests.get(url, params=kwargs, headers=self.headers,
verify=self.verify_ssl, auth=self.auth,
timeout=self.timeout)
return self.success_or_raise(response, default_response=default_response)
def post(self, uri, default_response=None, **kwargs):
"""
Call POST on the Gitlab server
>>> gitlab = Gitlab(host='http://localhost:10080', verify_ssl=False)
>>> gitlab.login(user='root', password='5iveL!fe')
>>> password = 'MyTestPassword1'
>>> email = 'example@example.com'
>>> data = {'name': 'test', 'username': 'test1', 'password': password, 'email': email}
>>> gitlab.post('/users/5', **data)
:param uri: String with the URI for the endpoint to POST to
:param default_response: Return value if JSONDecodeError
:param kwargs: Key word arguments representing the data to use in the POST
:return: Dictionary containing response data
:raise: HttpError: If invalid response returned
"""
url = self.api_url + uri
response = requests.post(
url, headers=self.headers, data=kwargs,
verify=self.verify_ssl, auth=self.auth, timeout=self.timeout)
return self.success_or_raise(response, default_response=default_response)
def delete(self, uri, default_response=None):
"""
Call DELETE on the Gitlab server
>>> gitlab = Gitlab(host='http://localhost:10080', verify_ssl=False)
>>> gitlab.login(user='root', password='5iveL!fe')
>>> gitlab.delete('/users/5')
:param uri: String with the URI you wish to delete
:param default_response: Return value if JSONDecodeError
:return: Dictionary containing response data
:raise: HttpError: If invalid response returned
"""
url = self.api_url + uri
response = requests.delete(
url, headers=self.headers, verify=self.verify_ssl,
auth=self.auth, timeout=self.timeout)
return self.success_or_raise(response, default_response=default_response)
def success_or_raise(self, response, default_response=None):
"""
Check if request was successful or raises an HttpError
:param response: Response Object to check
:param default_response: Return value if JSONDecodeError
:returns dict: Dictionary containing response data
:returns bool: :obj:`False` on failure when exceptions are suppressed
:raises requests.exceptions.HTTPError: If invalid response returned
"""
if self.suppress_http_error and not response.ok:
return False
response_json = default_response
if response_json is None:
response_json = {}
response.raise_for_status()
try:
response_json = response.json()
except ValueError:
pass
return response_json
@staticmethod
def getall(fn, page=None, *args, **kwargs):
"""
Auto-iterate over the paginated results of various methods of the API.
Pass the GitLabAPI method as the first argument, followed by the
other parameters as normal. Include `page` to determine first page to poll.
Remaining kwargs are passed on to the called method, including `per_page`.
:param fn: Actual method to call
:param page: Optional, page number to start at, defaults to 1
:param args: Positional arguments to actual method
:param kwargs: Keyword arguments to actual method
:return: Yields each item in the result until exhausted, and then implicit StopIteration; or no elements if error
"""
if not page:
page = 1
while True:
results = fn(*args, page=page, **kwargs)
if not results:
break
for x in results:
yield x
page += 1