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
|
def __del__(self):
try:
for cb,opaque in self.domainEventCallbacks.items():
del self.domainEventCallbacks[cb]
del self.domainEventCallbacks
libvirtmod.virConnectDomainEventDeregister(self._o, self)
except AttributeError:
pass
if self._o != None:
libvirtmod.virConnectClose(self._o)
self._o = None
def domainEventDeregister(self, cb):
"""Removes a Domain Event Callback. De-registering for a
domain callback will disable delivery of this event type """
try:
del self.domainEventCallbacks[cb]
if len(self.domainEventCallbacks) == 0:
del self.domainEventCallbacks
ret = libvirtmod.virConnectDomainEventDeregister(self._o, self)
if ret == -1: raise libvirtError ('virConnectDomainEventDeregister() failed', conn=self)
except AttributeError:
pass
def domainEventRegister(self, cb, opaque):
"""Adds a Domain Event Callback. Registering for a domain
callback will enable delivery of the events """
try:
self.domainEventCallbacks[cb] = opaque
except AttributeError:
self.domainEventCallbacks = {cb:opaque}
ret = libvirtmod.virConnectDomainEventRegister(self._o, self)
if ret == -1: raise libvirtError ('virConnectDomainEventRegister() failed', conn=self)
def _dispatchDomainEventCallbacks(self, dom, event, detail):
"""Dispatches events to python user domain event callbacks
"""
try:
for cb,opaque in self.domainEventCallbacks.items():
cb(self,dom,event,detail,opaque)
return 0
except AttributeError:
pass
def _dispatchDomainEventLifecycleCallback(self, dom, event, detail, cbData):
"""Dispatches events to python user domain lifecycle event callbacks
"""
cb = cbData["cb"]
opaque = cbData["opaque"]
cb(self, virDomain(self, _obj=dom), event, detail, opaque)
return 0
def _dispatchDomainEventGenericCallback(self, dom, cbData):
"""Dispatches events to python user domain generic event callbacks
"""
cb = cbData["cb"]
opaque = cbData["opaque"]
cb(self, virDomain(self, _obj=dom), opaque)
return 0
def _dispatchDomainEventRTCChangeCallback(self, dom, offset, cbData):
"""Dispatches events to python user domain RTC change event callbacks
"""
cb = cbData["cb"]
opaque = cbData["opaque"]
cb(self, virDomain(self, _obj=dom), offset ,opaque)
return 0
def _dispatchDomainEventWatchdogCallback(self, dom, action, cbData):
"""Dispatches events to python user domain watchdog event callbacks
"""
cb = cbData["cb"]
opaque = cbData["opaque"]
cb(self, virDomain(self, _obj=dom), action, opaque)
return 0
def _dispatchDomainEventIOErrorCallback(self, dom, srcPath, devAlias,
action, cbData):
"""Dispatches events to python user domain IO error event callbacks
"""
cb = cbData["cb"]
opaque = cbData["opaque"]
cb(self, virDomain(self, _obj=dom), srcPath, devAlias, action, opaque)
return 0
def _dispatchDomainEventIOErrorReasonCallback(self, dom, srcPath,
devAlias, action, reason,
cbData):
"""Dispatches events to python user domain IO error event callbacks
"""
cb = cbData["cb"]
opaque = cbData["opaque"]
cb(self, virDomain(self, _obj=dom), srcPath, devAlias, action,
reason, opaque)
return 0
def _dispatchDomainEventGraphicsCallback(self, dom, phase, localAddr,
remoteAddr, authScheme, subject,
cbData):
"""Dispatches events to python user domain graphics event callbacks
"""
cb = cbData["cb"]
opaque = cbData["opaque"]
cb(self, virDomain(self, _obj=dom), phase, localAddr, remoteAddr,
authScheme, subject, opaque)
return 0
def dispatchDomainEventBlockPullCallback(self, dom, path, type, status, cbData):
"""Dispatches events to python user domain blockJob event callbacks
"""
try:
cb = cbData["cb"]
opaque = cbData["opaque"]
cb(self, virDomain(self, _obj=dom), path, type, status, opaque)
return 0
except AttributeError:
pass
def _dispatchDomainEventDiskChangeCallback(self, dom, oldSrcPath, newSrcPath, devAlias, reason, cbData):
"""Dispatches event to python user domain diskChange event callbacks
"""
cb = cbData["cb"]
opaque = cbData["opaque"]
cb(self, virDomain(self, _obj=dom), oldSrcPath, newSrcPath, devAlias, reason, opaque)
return 0;
def _dispatchDomainEventTrayChangeCallback(self, dom, devAlias, reason, cbData):
"""Dispatches event to python user domain trayChange event callbacks
"""
cb = cbData["cb"]
opaque = cbData["opaque"]
cb(self, virDomain(self, _obj=dom), devAlias, reason, opaque)
return 0;
def _dispatchDomainEventPMWakeupCallback(self, dom, reason, cbData):
"""Dispatches event to python user domain pmwakeup event callbacks
"""
cb = cbData["cb"]
opaque = cbData["opaque"]
cb(self, virDomain(self, _obj=dom), reason, opaque)
return 0;
def _dispatchDomainEventPMSuspendCallback(self, dom, reason, cbData):
"""Dispatches event to python user domain pmsuspend event callbacks
"""
cb = cbData["cb"]
opaque = cbData["opaque"]
cb(self, virDomain(self, _obj=dom), reason, opaque)
return 0;
def _dispatchDomainEventBalloonChangeCallback(self, dom, actual, cbData):
"""Dispatches events to python user domain balloon change event callbacks
"""
cb = cbData["cb"]
opaque = cbData["opaque"]
cb(self, virDomain(self, _obj=dom), actual, opaque)
return 0
def domainEventDeregisterAny(self, callbackID):
"""Removes a Domain Event Callback. De-registering for a
domain callback will disable delivery of this event type """
try:
ret = libvirtmod.virConnectDomainEventDeregisterAny(self._o, callbackID)
if ret == -1: raise libvirtError ('virConnectDomainEventDeregisterAny() failed', conn=self)
del self.domainEventCallbackID[callbackID]
except AttributeError:
pass
def domainEventRegisterAny(self, dom, eventID, cb, opaque):
"""Adds a Domain Event Callback. Registering for a domain
callback will enable delivery of the events """
if not hasattr(self, 'domainEventCallbackID'):
self.domainEventCallbackID = {}
cbData = { "cb": cb, "conn": self, "opaque": opaque }
if dom is None:
ret = libvirtmod.virConnectDomainEventRegisterAny(self._o, None, eventID, cbData)
else:
ret = libvirtmod.virConnectDomainEventRegisterAny(self._o, dom._o, eventID, cbData)
if ret == -1:
raise libvirtError ('virConnectDomainEventRegisterAny() failed', conn=self)
self.domainEventCallbackID[ret] = opaque
return ret
def listAllDomains(self, flags):
"""List all domains and returns a list of domain objects"""
ret = libvirtmod.virConnectListAllDomains(self._o, flags)
if ret is None:
raise libvirtError("virConnectListAllDomains() failed", conn=self)
retlist = list()
for domptr in ret:
retlist.append(virDomain(self, _obj=domptr))
return retlist
def listAllStoragePools(self, flags):
"""Returns a list of storage pool objects"""
ret = libvirtmod.virConnectListAllStoragePools(self._o, flags)
if ret is None:
raise libvirtError("virConnectListAllStoragePools() failed", conn=self)
retlist = list()
for poolptr in ret:
retlist.append(virStoragePool(self, _obj=poolptr))
return retlist
|
