Make generic named objects collection instead of separate for each type

There will be more:
 - labels
 - storage pools
 - storage volumes
This commit is contained in:
Marek Marczykowski-Górecki 2017-03-12 20:23:07 +01:00
parent 0e775209fd
commit 64d2f13212
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
3 changed files with 62 additions and 48 deletions

View File

@ -113,7 +113,8 @@ class QubesBase(qubesmgmt.base.PropertyHolder):
def __init__(self):
super(QubesBase, self).__init__(self, 'mgmt.property.', 'dom0')
self.domains = VMCollection(self)
self.labels = qubesmgmt.label.LabelsCollection(self)
self.labels = qubesmgmt.base.WrapperObjectsCollection(
self, 'mgmt.label.List', qubesmgmt.label.Label)
class QubesLocal(QubesBase):

View File

@ -224,3 +224,63 @@ class PropertyHolder(object):
)
except qubesmgmt.exc.QubesDaemonNoResponseError:
raise qubesmgmt.exc.QubesPropertyAccessError(name)
class WrapperObjectsCollection(object):
'''Collection of simple named objects'''
def __init__(self, app, list_method, object_class):
'''
Construct manager of named wrapper objects.
:param app: Qubes() object
:param list_method: name of API method used to list objects,
must return simple "one name per line" list
:param object_class: object class (callable) for wrapper objects,
will be called with just two arguments: app and a name
'''
self.app = app
self._list_method = list_method
self._object_class = object_class
#: names cache
self._names_list = None
#: returned objects cache
self._objects = {}
def clear_cache(self):
'''Clear cached list of names'''
self._names_list = None
def refresh_cache(self, force=False):
'''Refresh cached list of names'''
if not force and self._names_list is not None:
return
list_data = self.app.qubesd_call('dom0', self._list_method)
list_data = list_data.decode('ascii')
assert list_data[-1] == '\n'
self._names_list = list_data[:-1].splitlines()
for name, obj in list(self._objects.items()):
if obj.name not in self._names_list:
# Object no longer exists
del self._objects[name]
def __getitem__(self, item):
if item not in self:
raise KeyError(item)
if item not in self._objects:
self._objects[item] = self._object_class(self.app, item)
return self._objects[item]
def __contains__(self, item):
self.refresh_cache()
return item in self._names_list
def __iter__(self):
self.refresh_cache()
for obj in self._names_list:
yield self[obj]
def keys(self):
'''Get list of names.'''
self.refresh_cache()
return self._names_list.copy()

View File

@ -55,50 +55,3 @@ class Label(object):
def __str__(self):
return self._name
class LabelsCollection(object):
'''Collection of VMs objects'''
def __init__(self, app):
self.app = app
self._label_list = None
self._label_objects = {}
def clear_cache(self):
'''Clear cached list of labels'''
self._label_list = None
def refresh_cache(self, force=False):
'''Refresh cached list of VMs'''
if not force and self._label_list is not None:
return
label_list_data = self.app.qubesd_call('dom0', 'mgmt.label.List')
label_list_data = label_list_data.decode('ascii')
assert label_list_data[-1] == '\n'
self._label_list = label_list_data[:-1].splitlines()
for name, label in list(self._label_objects.items()):
if label.name not in self._label_list:
# Label no longer exists
del self._label_objects[name]
def __getitem__(self, item):
if item not in self:
raise KeyError(item)
if item not in self._label_objects:
self._label_objects[item] = Label(self.app, item)
return self._label_objects[item]
def __contains__(self, item):
self.refresh_cache()
return item in self._label_list
def __iter__(self):
self.refresh_cache()
for vm in self._label_list:
yield self[vm]
def keys(self):
'''Get list of label names.'''
self.refresh_cache()
return self._label_list.keys()