/
rest_client.py
144 lines (110 loc) · 5.77 KB
/
rest_client.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
from conans import CHECKSUM_DEPLOY, REVISIONS, ONLY_V2, OAUTH_TOKEN
from conans.client.rest.rest_client_v1 import RestV1Methods
from conans.client.rest.rest_client_v2 import RestV2Methods
from conans.errors import OnlyV2Available, AuthenticationException
from conans.search.search import filter_packages
from conans.util.log import logger
class RestApiClient(object):
"""
Rest Api Client for handle remote.
"""
def __init__(self, output, requester, revisions_enabled, put_headers=None):
# Set to instance
self.token = None
self.refresh_token = None
self.remote_url = None
self.custom_headers = {} # Can set custom headers to each request
self._output = output
self.requester = requester
# Remote manager will set it to True or False dynamically depending on the remote
self.verify_ssl = True
self._put_headers = put_headers
self._revisions_enabled = revisions_enabled
self._cached_capabilities = {}
def _capable(self, capability):
capabilities = self._cached_capabilities.get(self.remote_url)
if capabilities is None:
tmp = RestV1Methods(self.remote_url, self.token, self.custom_headers, self._output,
self.requester, self.verify_ssl, self._put_headers)
capabilities = tmp.server_capabilities()
self._cached_capabilities[self.remote_url] = capabilities
logger.debug("REST: Cached capabilities for the remote: %s" % capabilities)
if not self._revisions_enabled and ONLY_V2 in capabilities:
raise OnlyV2Available(self.remote_url)
return capability in capabilities
def _get_api(self):
revisions = self._capable(REVISIONS)
if self._revisions_enabled and revisions:
checksum_deploy = self._capable(CHECKSUM_DEPLOY)
return RestV2Methods(self.remote_url, self.token, self.custom_headers, self._output,
self.requester, self.verify_ssl, self._put_headers,
checksum_deploy)
else:
return RestV1Methods(self.remote_url, self.token, self.custom_headers, self._output,
self.requester, self.verify_ssl, self._put_headers)
def get_recipe_manifest(self, ref):
return self._get_api().get_recipe_manifest(ref)
def get_package_manifest(self, pref):
return self._get_api().get_package_manifest(pref)
def get_package_info(self, pref):
return self._get_api().get_package_info(pref)
def get_recipe(self, ref, dest_folder):
return self._get_api().get_recipe(ref, dest_folder)
def get_recipe_snapshot(self, ref):
return self._get_api().get_recipe_snapshot(ref)
def get_recipe_sources(self, ref, dest_folder):
return self._get_api().get_recipe_sources(ref, dest_folder)
def get_package(self, pref, dest_folder):
return self._get_api().get_package(pref, dest_folder)
def get_package_snapshot(self, ref):
return self._get_api().get_package_snapshot(ref)
def get_recipe_path(self, ref, path):
return self._get_api().get_recipe_path(ref, path)
def get_package_path(self, pref, path):
return self._get_api().get_package_path(pref, path)
def upload_recipe(self, ref, files_to_upload, deleted, retry, retry_wait):
return self._get_api().upload_recipe(ref, files_to_upload, deleted, retry,
retry_wait)
def upload_package(self, pref, files_to_upload, deleted, retry, retry_wait):
return self._get_api().upload_package(pref, files_to_upload, deleted, retry, retry_wait)
def authenticate(self, user, password):
api_v1 = RestV1Methods(self.remote_url, self.token, self.custom_headers, self._output,
self.requester, self.verify_ssl, self._put_headers)
if self.refresh_token and self.token:
token, refresh_token = api_v1.refresh_token(self.token, self.refresh_token)
else:
try:
# Check capabilities can raise also 401 until the new Artifactory is released
oauth_capable = self._capable(OAUTH_TOKEN)
except AuthenticationException:
oauth_capable = False
if oauth_capable:
# Artifactory >= 6.13.X
token, refresh_token = api_v1.authenticate_oauth(user, password)
else:
token = api_v1.authenticate(user, password)
refresh_token = None
return token, refresh_token
def check_credentials(self):
return self._get_api().check_credentials()
def search(self, pattern=None, ignorecase=True):
return self._get_api().search(pattern, ignorecase)
def search_packages(self, reference, query):
# Do not send the query to the server, as it will fail
# https://github.com/conan-io/conan/issues/4951
package_infos = self._get_api().search_packages(reference, query=None)
return filter_packages(query, package_infos)
def remove_conanfile(self, ref):
return self._get_api().remove_conanfile(ref)
def remove_packages(self, ref, package_ids=None):
return self._get_api().remove_packages(ref, package_ids)
def server_capabilities(self):
return self._get_api().server_capabilities()
def get_recipe_revisions(self, ref):
return self._get_api().get_recipe_revisions(ref)
def get_package_revisions(self, pref):
return self._get_api().get_package_revisions(pref)
def get_latest_recipe_revision(self, ref):
return self._get_api().get_latest_recipe_revision(ref)
def get_latest_package_revision(self, pref):
return self._get_api().get_latest_package_revision(pref)