forked from objectrocket/python-client
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathbases.py
More file actions
358 lines (283 loc) · 11 KB
/
Copy pathbases.py
File metadata and controls
358 lines (283 loc) · 11 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
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
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
"""Base classes used throughout the library."""
import abc
import logging
import requests
import six
from objectrocket import errors
from objectrocket import util
from stevedore.extension import ExtensionManager
log = logging.getLogger(__name__)
@six.add_metaclass(abc.ABCMeta)
class BaseOperationsLayer(object):
"""A base class for operations layer classes."""
def __init__(self, base_client):
self.__client = base_client
######################
# Private interface. #
######################
@property
def _client(self):
"""An instance of the objectrocket.client.Client."""
return self.__client
@abc.abstractproperty
def _default_request_kwargs(self):
"""The default request keyword arguments to be passed to the requests library."""
default_kwargs = {
'headers': {
'Content-Type': 'application/json'
},
'hooks': {
'response': self._verify_auth
}
}
return default_kwargs
def _get_response_data(self, response):
"""Return the data from a ``requests.Response`` object.
:param requests.Response response: The ``Response`` object from which to get the data.
"""
try:
_json = response.json()
data = _json.get('data')
return data
except ValueError as ex:
log.exception(ex)
return None
@abc.abstractproperty
def _url(self):
"""The URL this operations layer is to interface with."""
pass
def _verify_auth(self, resp, *args, **kwargs):
"""A callback handler to verify that the given response object did not receive a 401."""
if resp.status_code == 401:
raise errors.AuthFailure(
'Received response code 401 from {} {}.'
.format(resp.request.method, resp.request.path_url)
)
class BaseAuthLayer(BaseOperationsLayer):
"""A base class for authentication layer classes."""
#####################
# Public interface. #
#####################
@abc.abstractmethod
def authenticate(self):
"""An implementation of this layer's authentication protocol."""
pass
######################
# Private interface. #
######################
@property
def _default_request_kwargs(self):
"""The default request keyword arguments to be passed to the requests library."""
default_kwargs = {
'headers': {
'Content-Type': 'application/json'
},
'hooks': {}
}
return default_kwargs
@abc.abstractmethod
def _refresh(self):
"""An implementation of this layer's authentication refresh protocol."""
pass
@abc.abstractproperty
def _url(self):
"""The URL this operations layer is to interface with."""
pass
@six.add_metaclass(abc.ABCMeta)
class BaseInstance(object):
"""The base class for ObjectRocket service instances.
:param dict instance_document: A dictionary representing the instance object, most likey coming
from the ObjectRocket API.
:param objectrocket.instances.Instances instances: An instance of
:py:class:`objectrocket.instances.Instances`.
"""
def __init__(self, instance_document, instances):
self.__client = instances._client
self.__instances = instances
self.__instance_document = instance_document
if 'connect_string' in instance_document:
self._connect_string = instance_document['connect_string']
elif 'connection_strings' in instance_document:
self._connect_string = instance_document['connection_strings']
else:
raise errors.InstancesException('No connection string found.')
# Bind required pseudo private attributes from API response document.
self._created = instance_document['created']
self._id = instance_document['id']
self._name = instance_document['name']
self._plan = instance_document['plan']
self._service = instance_document['service']
self._type = instance_document['type']
self._version = instance_document['version']
self._settings = instance_document.get('settings', [])
def __repr__(self):
"""Represent this object as a string."""
_id = hex(id(self))
rep = (
'<{!s} name={!s} id={!s} at {!s}>'
.format(self.__class__.__name__, self.name, self.id, _id)
)
return rep
def acl_sync(self, aws_sync=None, rackspace_sync=None):
"""Adjust Amazon Web Services and/or Rackspace Acl Sync feature for this instance.
:param bool aws_sync: True/False whether to enable AWS acl sync for this instance.
:param bool rackspace_sync: True/False whether to enable Rackspace acl sync for this
instance.
"""
url = self._url + 'acl_sync'
data = {"aws_acl_sync_enabled": False, "rackspace_acl_sync_enabled": False}
# Let's get current status of acl sync for this intance to set proper defaults.
response = requests.get(url, **self._instances._default_request_kwargs)
if response.status_code == 200:
resp_json = response.json()
current_status = resp_json.get('data', {})
current_aws_sync_status = current_status.get("aws_acl_sync_enabled", False)
current_rax_sync_status = current_status.get("rackspace_acl_sync_enabled", False)
data.update({
"aws_acl_sync_enabled": current_aws_sync_status,
"rackspace_acl_sync_enabled": current_rax_sync_status
})
if aws_sync is not None:
data.update({"aws_acl_sync_enabled": aws_sync})
if rackspace_sync is not None:
data.update({"rackspace_acl_sync_enabled": rackspace_sync})
response = requests.put(url, json=data, **self._instances._default_request_kwargs)
return response.json()
else:
raise errors.ObjectRocketException(
"Couldn't get current status of instance, failing. Error: {}".format(response.text)
)
def run_acl_sync(self, aws_sync=False, rackspace_sync=False):
"""Run Acl sync for this instance.
:param bool aws_sync: True/False whether to run AWS acl sync for this instance immediately.
:param bool rackspace_sync: True/False whether to run Rackspace acl sync for this instance
immediately.
"""
url = self._url + 'acl_sync'
data = {"aws_acl_sync_enabled": False, "rackspace_acl_sync_enabled": False}
if aws_sync:
data.update({"aws_acl_sync_enabled": True})
if rackspace_sync:
data.update({"rackspace_acl_sync_enabled": True})
response = requests.post(url, json=data, **self._instances._default_request_kwargs)
return response.json()
@property
def connect_string(self):
"""This instance's connection string."""
return self._connect_string
@property
def created(self):
"""The date this instance was created."""
return self._created
@abc.abstractmethod
def get_connection(self):
"""Get a live connection to this instance."""
pass
@property
def id(self):
"""This instance's ID."""
return self._id
@property
def name(self):
"""This instance's name."""
return self._name
@property
def plan(self):
"""The base plan size of this instance."""
return self._plan
@property
def service(self):
"""The service this instance provides."""
return self._service
@property
def type(self):
"""The type of service this instance provides."""
return self._type
@property
def version(self):
"""The version of this instance's service."""
return self._version
@property
def settings(self):
"""The settings on this instance's service."""
return self._settings
def to_dict(self):
"""Render this object as a dictionary."""
return self._instance_document
######################
# Private interface. #
######################
@property
def _client(self):
"""An instance of the objectrocket.client.Client."""
return self.__client
@property
def _instances(self):
"""An instance of the objectrocket.instances.Instances."""
return self.__instances
@property
def _instance_document(self):
"""The document used to construct this Instance object."""
return self.__instance_document
@property
def _url(self):
"""The URL of this instance object."""
return self._instances._url + '{}/'.format(self.name)
@property
def _service_url(self):
"""The service specific URL of this instance object."""
return self._client._url + '{}/{}/'.format(self.service, self.name)
###########
# Mixins. #
###########
class Extensible(object):
"""A mixin to implement support for class extensibility."""
def _register_extensions(self, namespace):
"""Register any extensions under the given namespace."""
# Register any extension classes for this class.
extmanager = ExtensionManager(
'extensions.classes.{}'.format(namespace),
propagate_map_exceptions=True
)
if extmanager.extensions:
extmanager.map(util.register_extension_class, base=self)
# Register any extension methods for this class.
extmanager = ExtensionManager(
'extensions.methods.{}'.format(namespace),
propagate_map_exceptions=True
)
if extmanager.extensions:
extmanager.map(util.register_extension_method, base=self)
class InstanceAclsInterface(object):
"""A mixin implementing support for the instance bound ACLs interface.
Should only be mixed in with an Instance class.
"""
_acls = None
@property
def acls(self):
"""The instance bound ACLs operations layer."""
if self._acls is None:
self._acls = InstanceAcls(instance=self)
return self._acls
class InstanceAcls(object):
"""An object implementing the ACLs interface bound to a specific instance."""
def __init__(self, instance):
self._instance = instance
def all(self):
"""Get all ACLs for this instance."""
return self._instance._client.acls.all(self._instance.name)
def create(self, cidr_mask, description, **kwargs):
"""Create an ACL for this instance.
See :py:meth:`Acls.create` for call signature.
"""
return self._instance._client.acls.create(
self._instance.name,
cidr_mask,
description,
**kwargs
)
def get(self, acl):
"""Get the ACL specified by ID belonging to this instance.
See :py:meth:`Acls.get` for call signature.
"""
return self._instance._client.acls.get(self._instance.name, acl)