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
|
# vim: set fileencoding=utf-8 :
#
# (C) 2014 Guido Gunther <agx@sigxcpu.org>
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
import logging
import os
import posixpath
import requests
log = logging.getLogger(__name__)
class ArtifactoryError(Exception):
def __init__(self, msg, url, status_code=0):
self.url = url
self.status_code = status_code
Exception.__init__(self, msg)
class ArtifactoryObjectNotFound(ArtifactoryError):
"""Object not found aka 404"""
pass
class ArtifactoryNoPermission(ArtifactoryError):
"""Object not found aka 401"""
pass
class Artifactory(object):
"""
Extremely simple JFrog Artifactory API Wrapper
:arg str base: The base URL of to connect to, e.g. 'http://example.com/artifactory'
:arg str user: The user to connect as
:arg str password: The user's password
If not noted otherwise all methods return the deserialized JSON data.
"""
def __init__(self, base, user=None, password=None):
self.user = user
self.password = password
self.base = base
def search_gavc(self, groupid=None, artifactid=None, version=None, classifier=None):
"""
Query the artifactory via Groupid, ArtifacId, Version and classifier
:arg str groupid: The group id to search for
:arg str artifacid: The artifact id to search for
:arg str version: The version to search for
:arg str classifier: The classifier to search for (e.g. rpm)
:returns: The URLs of the found items
:rtype: list of str
"""
path='api/search/gavc'
param_map = {
'groupid': 'g',
'artifactid': 'a',
'version': 'v',
'classifier': 'c',
}
params = {}
for k,v in locals().items():
if k in param_map and v is not None:
params[param_map[k]] = v
url = posixpath.join(self.base, path)
log.debug("Searching on %s with %s" % (url, params))
return self.get(url, params)['results']
def calc_yum_metadata(self, repository):
"""
Recalculate YUM metadata
:arg str repository: The repository to calculate the metadata for
"""
url = posixpath.join(self.base, 'api/yum', repository)
params = { 'async': 0 }
log.debug("Refreshing on %s" % url)
return self.post(url, params)
def get_repositories(self):
"""
Get available repositories
"""
url = posixpath.join(self.base, 'api/repositories')
return self.get(url)
def _copy_or_move(self, action, source, target):
try:
if len(source) != 2:
raise ValueError("Source tuple not (repo, path)")
except TypeError:
raise TypeError("Source must be a tuple")
try:
if len(target) != 2:
raise ValueError("Target tuple not (repo, path)")
except TypeError:
raise TypeError("Source must be a tuple")
url = posixpath.join(self.base,
'api', action,
source[0], source[1])
dst = posixpath.join(target[0], target[1])
log.debug("%s %s", action.capitalize(), url)
# Parameter order matters for artifactory
return self.post(url, {'to': dst})
def copy(self, source, target):
"""
Copy artifacts in the repository from soure to target
:arg tuple source: The source in the form (repo, path) to copy
:arg tuple target: The target in the form (repo, path) to copy
"""
return self._copy_or_move('copy', source, target)
def move(self, source, target):
"""
Move artifacts in the repository from soure to target
:arg tuple source: The source in the form (repo, path) to copy
:arg tuple target: The target in the form (repo, path) to copy
"""
return self._copy_or_move('move', source, target)
def _transform_url(self, url):
"""
Helper to rewrite URLs. All artifactory URLs are passed through this method
in order to cope with broken installations where e.g. artifactory sits behind
a reverse proxy but doesn't know about it.
Rewrite this in derived classes as needed (but hopefully not).
"""
return url
@staticmethod
def _parse_error(data):
try:
if data.has_key('errors'):
return data['errors'][0]['message']
return None
except:
return None
def _request(self, http_method, url, params=None):
"""
Perform a HTTP request on the artifactory
"""
url = self._transform_url(url)
auth = None
if self.user and self.password:
auth = requests.auth.HTTPBasicAuth(self.user, self.password)
method = getattr(requests, http_method.lower())
r = method(url, params=params, auth=auth)
if r.status_code / 100 != 2:
err = self._parse_error(r.json())
if r.status_code == 404:
raise ArtifactoryObjectNotFound(err or "Not found" % url,
url,
r.status_code)
elif r.status_code == 401:
raise ArtifactoryNoPermission(err or "Unauthorized" % url,
url,
r.status_code)
raise ArtifactoryError(err or "Did not get a 2xx",
url,
r.status_code)
if r.headers['content-type'].endswith('json'):
return r.json()
else:
return r.text
def get(self, url, params=None):
"""
Perform a GET request on the artifactory
:arg str url: The URL to perform the get request on
:arg dict params: Additional URL params
:returns: The retrieved data deserialized from JSON if the
request returns JSON otherwise as plain string.
"""
return self._request('GET', url, params)
def post(self, url, params):
"""
Perform a POST request on the artifactory
:arg str url: The URL to perform the get request on
:arg dict params: Additional URL params
:returns: The retrieved data deserialized from JSON if the
request returns JSON otherwise as plain string.
"""
return self._request('POST', url, params)
|
