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
|
# pylint: skip-file
# flake8: noqa
# pylint: disable=too-many-instance-attributes
class PersistentVolumeClaimConfig(object):
''' Handle pvc options '''
# pylint: disable=too-many-arguments
def __init__(self,
sname,
namespace,
kubeconfig,
access_modes=None,
vol_capacity='1G',
selector=None,
storage_class_name=None):
''' constructor for handling pvc options '''
self.kubeconfig = kubeconfig
self.name = sname
self.namespace = namespace
self.access_modes = access_modes
self.vol_capacity = vol_capacity
self.data = {}
self.selector = selector
self.storage_class_name = storage_class_name
self.create_dict()
def create_dict(self):
''' return a service as a dict '''
# version
self.data['apiVersion'] = 'v1'
# kind
self.data['kind'] = 'PersistentVolumeClaim'
# metadata
self.data['metadata'] = {}
self.data['metadata']['name'] = self.name
# spec
self.data['spec'] = {}
self.data['spec']['accessModes'] = ['ReadWriteOnce']
if self.access_modes:
self.data['spec']['accessModes'] = self.access_modes
if self.selector:
self.data['spec']['selector'] = {'matchLabels': self.selector}
# storage capacity
self.data['spec']['resources'] = {}
self.data['spec']['resources']['requests'] = {}
self.data['spec']['resources']['requests']['storage'] = self.vol_capacity
if self.storage_class_name:
self.data['spec']['storageClassName'] = self.storage_class_name
# pylint: disable=too-many-instance-attributes,too-many-public-methods
class PersistentVolumeClaim(Yedit):
''' Class to wrap the oc command line tools '''
access_modes_path = "spec.accessModes"
volume_capacity_path = "spec.requests.storage"
volume_name_path = "spec.volumeName"
bound_path = "status.phase"
kind = 'PersistentVolumeClaim'
selector_path = "spec.selector.matchLabels"
storage_class_name_path = "spec.storageClassName"
def __init__(self, content):
'''PersistentVolumeClaim constructor'''
super(PersistentVolumeClaim, self).__init__(content=content)
self._access_modes = None
self._volume_capacity = None
self._volume_name = None
self._selector = None
self._storage_class_name = None
@property
def storage_class_name(self):
''' storage_class_name property '''
if self._storage_class_name is None:
self._storage_class_name = self.get_storage_class_name()
return self._storage_class_name
@storage_class_name.setter
def storage_class_name(self, data):
''' storage_class_name property setter'''
self._storage_class_name = data
@property
def volume_name(self):
''' volume_name property '''
if self._volume_name is None:
self._volume_name = self.get_volume_name()
return self._volume_name
@volume_name.setter
def volume_name(self, data):
''' volume_name property setter'''
self._volume_name = data
@property
def selector(self):
''' selector property '''
if self._selector is None:
self._selector = self.get_selector()
if not isinstance(self._selector, dict):
self._selector = dict(self._selector)
return self._selector
@selector.setter
def selector(self, data):
''' selector property setter'''
if not isinstance(data, dict):
data = dict(data)
self._selector = data
@property
def access_modes(self):
''' access_modes property '''
if self._access_modes is None:
self._access_modes = self.get_access_modes()
if not isinstance(self._access_modes, list):
self._access_modes = list(self._access_modes)
return self._access_modes
@access_modes.setter
def access_modes(self, data):
''' access_modes property setter'''
if not isinstance(data, list):
data = list(data)
self._access_modes = data
@property
def volume_capacity(self):
''' volume_capacity property '''
if self._volume_capacity is None:
self._volume_capacity = self.get_volume_capacity()
return self._volume_capacity
@volume_capacity.setter
def volume_capacity(self, data):
''' volume_capacity property setter'''
self._volume_capacity = data
def get_storage_class_name(self):
'''get storage_class_name'''
return self.get(PersistentVolumeClaim.storage_class_name_path) or []
def get_selector(self):
'''get selector'''
return self.get(PersistentVolumeClaim.selector_path) or []
def get_access_modes(self):
'''get access_modes'''
return self.get(PersistentVolumeClaim.access_modes_path) or []
def get_volume_capacity(self):
'''get volume_capacity'''
return self.get(PersistentVolumeClaim.volume_capacity_path) or []
def get_volume_name(self):
'''get volume_name'''
return self.get(PersistentVolumeClaim.volume_name_path) or []
def is_bound(self):
'''return whether volume is bound'''
return self.get(PersistentVolumeClaim.bound_path) or []
#### ADD #####
def add_access_mode(self, inc_mode):
''' add an access_mode'''
if self.access_modes:
self.access_modes.append(inc_mode)
else:
self.put(PersistentVolumeClaim.access_modes_path, [inc_mode])
return True
#### /ADD #####
#### Remove #####
def remove_access_mode(self, inc_mode):
''' remove an access_mode'''
try:
self.access_modes.remove(inc_mode)
except ValueError as _:
return False
return True
#### /REMOVE #####
#### UPDATE #####
def update_access_mode(self, inc_mode):
''' update an access_mode'''
try:
index = self.access_modes.index(inc_mode)
except ValueError as _:
return self.add_access_mode(inc_mode)
self.access_modes[index] = inc_mode
return True
#### /UPDATE #####
#### FIND ####
def find_access_mode(self, inc_mode):
''' find a user '''
index = None
try:
index = self.access_modes.index(inc_mode)
except ValueError as _:
return index
return index
|