1    def __del__(self) -> None:
2        try:
3            self.domainEventCallbacks.clear()
4            del self.domainEventCallbacks
5            libvirtmod.virConnectDomainEventDeregister(self._o, self)
6        except AttributeError:
7            pass
8
9        if self._o is not None:
10            libvirtmod.virConnectClose(self._o)
11        self._o = None
12
13    def __enter__(self) -> 'virConnect':
14        return self
15
16    def __exit__(self, exc_type_: Optional[Type[BaseException]], exc_value_: Optional[BaseException], traceback_: Optional[TracebackType]) -> None:
17        self.close()
18
19    def domainEventDeregister(self, cb: _DomainCB) -> None:
20        """Removes a Domain Event Callback. De-registering for a
21           domain callback will disable delivery of this event type """
22        try:
23            del self.domainEventCallbacks[cb]
24            if len(self.domainEventCallbacks) == 0:
25                del self.domainEventCallbacks
26                ret = libvirtmod.virConnectDomainEventDeregister(self._o, self)
27                if ret == -1:
28                    raise libvirtError('virConnectDomainEventDeregister() failed')
29        except AttributeError:
30            pass
31
32    def domainEventRegister(self, cb: _DomainCB, opaque: _T) -> None:
33        """Adds a Domain Event Callback. Registering for a domain
34           callback will enable delivery of the events """
35        try:
36            self.domainEventCallbacks[cb] = opaque
37        except AttributeError:
38            self.domainEventCallbacks = {cb: opaque}  # type: Dict[_DomainCB, _T]
39            ret = libvirtmod.virConnectDomainEventRegister(self._o, self)
40            if ret == -1:
41                raise libvirtError('virConnectDomainEventRegister() failed')
42
43    def _dispatchDomainEventCallbacks(self, dom: 'virDomain', event: int, detail: int) -> None:
44        """Dispatches events to python user domain event callbacks
45        """
46        try:
47            for cb, opaque in self.domainEventCallbacks.items():
48                cb(self, virDomain(self, _obj=dom), event, detail, opaque)
49        except AttributeError:
50            pass
51
52    def _dispatchDomainEventLifecycleCallback(self, dom: 'virDomain', event: int, detail: int, cbData: Dict[str, Any]) -> int:
53        """Dispatches events to python user domain lifecycle event callbacks
54        """
55        cb = cbData["cb"]
56        opaque = cbData["opaque"]
57
58        cb(self, virDomain(self, _obj=dom), event, detail, opaque)
59        return 0
60
61    def _dispatchDomainEventGenericCallback(self, dom: 'virDomain', cbData: Dict[str, Any]) -> int:
62        """Dispatches events to python user domain generic event callbacks
63        """
64        cb = cbData["cb"]
65        opaque = cbData["opaque"]
66
67        cb(self, virDomain(self, _obj=dom), opaque)
68        return 0
69
70    def _dispatchDomainEventRTCChangeCallback(self, dom: 'virDomain', offset: int, cbData: Dict[str, Any]) -> int:
71        """Dispatches events to python user domain RTC change event callbacks
72        """
73        cb = cbData["cb"]
74        opaque = cbData["opaque"]
75
76        cb(self, virDomain(self, _obj=dom), offset, opaque)
77        return 0
78
79    def _dispatchDomainEventWatchdogCallback(self, dom: 'virDomain', action: int, cbData: Dict[str, Any]) -> int:
80        """Dispatches events to python user domain watchdog event callbacks
81        """
82        cb = cbData["cb"]
83        opaque = cbData["opaque"]
84
85        cb(self, virDomain(self, _obj=dom), action, opaque)
86        return 0
87
88    def _dispatchDomainEventIOErrorCallback(self, dom: 'virDomain', srcPath: str, devAlias: str, action: int, cbData: Dict[str, Any]) -> int:
89        """Dispatches events to python user domain IO error event callbacks
90        """
91        cb = cbData["cb"]
92        opaque = cbData["opaque"]
93
94        cb(self, virDomain(self, _obj=dom), srcPath, devAlias, action, opaque)
95        return 0
96
97    def _dispatchDomainEventIOErrorReasonCallback(self, dom: 'virDomain', srcPath: str, devAlias: str, action: int, reason: int, cbData: Dict[str, Any]) -> int:
98        """Dispatches events to python user domain IO error event callbacks
99        """
100        cb = cbData["cb"]
101        opaque = cbData["opaque"]
102
103        cb(self, virDomain(self, _obj=dom), srcPath, devAlias, action,
104           reason, opaque)
105        return 0
106
107    def _dispatchDomainEventGraphicsCallback(self, dom: 'virDomain', phase: int, localAddr: Any, remoteAddr: Any, authScheme: str, subject: Any, cbData: Dict[str, Any]) -> int:
108        """Dispatches events to python user domain graphics event callbacks
109        """
110        cb = cbData["cb"]
111        opaque = cbData["opaque"]
112
113        cb(self, virDomain(self, _obj=dom), phase, localAddr, remoteAddr,
114           authScheme, subject, opaque)
115        return 0
116
117    def _dispatchDomainEventBlockJobCallback(self, dom: 'virDomain', disk: str, type: int, status: int, cbData: Dict[str, Any]) -> None:
118        """Dispatches events to python user domain blockJob/blockJob2 event callbacks
119        """
120        try:
121            cb = cbData["cb"]
122            opaque = cbData["opaque"]
123
124            cb(self, virDomain(self, _obj=dom), disk, type, status, opaque)
125        except AttributeError:
126            pass
127
128    def _dispatchDomainEventDiskChangeCallback(self, dom: 'virDomain', oldSrcPath: str, newSrcPath: str, devAlias: str, reason: int, cbData: Dict[str, Any]) -> int:
129        """Dispatches event to python user domain diskChange event callbacks
130        """
131        cb = cbData["cb"]
132        opaque = cbData["opaque"]
133
134        cb(self, virDomain(self, _obj=dom), oldSrcPath, newSrcPath, devAlias, reason, opaque)
135        return 0
136
137    def _dispatchDomainEventTrayChangeCallback(self, dom: 'virDomain', devAlias: str, reason: int, cbData: Dict[str, Any]) -> int:
138        """Dispatches event to python user domain trayChange event callbacks
139        """
140        cb = cbData["cb"]
141        opaque = cbData["opaque"]
142
143        cb(self, virDomain(self, _obj=dom), devAlias, reason, opaque)
144        return 0
145
146    def _dispatchDomainEventPMWakeupCallback(self, dom: 'virDomain', reason: int, cbData: Dict[str, Any]) -> int:
147        """Dispatches event to python user domain pmwakeup event callbacks
148        """
149        cb = cbData["cb"]
150        opaque = cbData["opaque"]
151
152        cb(self, virDomain(self, _obj=dom), reason, opaque)
153        return 0
154
155    def _dispatchDomainEventPMSuspendCallback(self, dom: 'virDomain', reason: int, cbData: Dict[str, Any]) -> int:
156        """Dispatches event to python user domain pmsuspend event callbacks
157        """
158        cb = cbData["cb"]
159        opaque = cbData["opaque"]
160
161        cb(self, virDomain(self, _obj=dom), reason, opaque)
162        return 0
163
164    def _dispatchDomainEventBalloonChangeCallback(self, dom: 'virDomain', actual: int, cbData: Dict[str, Any]) -> int:
165        """Dispatches events to python user domain balloon change event callbacks
166        """
167        cb = cbData["cb"]
168        opaque = cbData["opaque"]
169
170        cb(self, virDomain(self, _obj=dom), actual, opaque)
171        return 0
172
173    def _dispatchDomainEventPMSuspendDiskCallback(self, dom: 'virDomain', reason: int, cbData: Dict[str, Any]) -> int:
174        """Dispatches event to python user domain pmsuspend-disk event callbacks
175        """
176        cb = cbData["cb"]
177        opaque = cbData["opaque"]
178
179        cb(self, virDomain(self, _obj=dom), reason, opaque)
180        return 0
181
182    def _dispatchDomainEventDeviceRemovedCallback(self, dom: 'virDomain', devAlias: str, cbData: Dict[str, Any]) -> int:
183        """Dispatches event to python user domain device removed event callbacks
184        """
185        cb = cbData["cb"]
186        opaque = cbData["opaque"]
187
188        cb(self, virDomain(self, _obj=dom), devAlias, opaque)
189        return 0
190
191    def _dispatchDomainEventTunableCallback(self, dom: 'virDomain', params: Any, cbData: Dict[str, Any]) -> int:
192        """Dispatches event to python user domain tunable event callbacks
193        """
194        cb = cbData["cb"]
195        opaque = cbData["opaque"]
196
197        cb(self, virDomain(self, _obj=dom), params, opaque)
198        return 0
199
200    def _dispatchDomainEventAgentLifecycleCallback(self, dom: 'virDomain', state: int, reason: int, cbData: Dict[str, Any]) -> int:
201        """Dispatches event to python user domain agent lifecycle event callback
202        """
203
204        cb = cbData["cb"]
205        opaque = cbData["opaque"]
206
207        cb(self, virDomain(self, _obj=dom), state, reason, opaque)
208        return 0
209
210    def _dispatchDomainEventDeviceAddedCallback(self, dom: 'virDomain', devAlias: str, cbData: Dict[str, Any]) -> int:
211        """Dispatches event to python user domain device added event callbacks
212        """
213        cb = cbData["cb"]
214        opaque = cbData["opaque"]
215
216        cb(self, virDomain(self, _obj=dom), devAlias, opaque)
217        return 0
218
219    def _dispatchDomainEventMigrationIterationCallback(self, dom: 'virDomain', iteration: int, cbData: Dict[str, Any]) -> int:
220        """Dispatches event to python user domain migration iteration event callbacks
221        """
222        cb = cbData["cb"]
223        opaque = cbData["opaque"]
224
225        cb(self, virDomain(self, _obj=dom), iteration, opaque)
226        return 0
227
228    def _dispatchDomainEventJobCompletedCallback(self, dom: 'virDomain', params: Dict[str, Any], cbData: Dict[str, Any]) -> int:
229        """Dispatches event to python user domain job completed callbacks
230        """
231        cb = cbData["cb"]
232        opaque = cbData["opaque"]
233
234        cb(self, virDomain(self, _obj=dom), params, opaque)
235        return 0
236
237    def _dispatchDomainEventDeviceRemovalFailedCallback(self, dom: 'virDomain', devAlias: str, cbData: Dict[str, Any]) -> int:
238        """Dispatches event to python user domain device removal failed event callbacks
239        """
240        cb = cbData["cb"]
241        opaque = cbData["opaque"]
242
243        cb(self, virDomain(self, _obj=dom), devAlias, opaque)
244        return 0
245
246    def _dispatchDomainEventMetadataChangeCallback(self, dom: 'virDomain', mtype: int, nsuri: str, cbData: Dict[str, Any]) -> int:
247        """Dispatches event to python user domain metadata change event callbacks
248        """
249        cb = cbData["cb"]
250        opaque = cbData["opaque"]
251
252        cb(self, virDomain(self, _obj=dom), mtype, nsuri, opaque)
253        return 0
254
255    def _dispatchDomainEventBlockThresholdCallback(self, dom: 'virDomain', dev: str, path: str, threshold: int, excess: int, cbData: Dict[str, Any]) -> int:
256        """Dispatches event to python user domain block device threshold event callbacks
257        """
258        cb = cbData["cb"]
259        opaque = cbData["opaque"]
260
261        cb(self, virDomain(self, _obj=dom), dev, path, threshold, excess, opaque)
262        return 0
263
264    def _dispatchDomainEventMemoryFailureCallback(self, dom: 'virDomain', recipient: int, action: int, flags: int, cbData: Dict[str, Any]) -> int:
265        """Dispatches event to python user domain memory failure event callbacks
266        """
267        cb = cbData["cb"]
268        opaque = cbData["opaque"]
269
270        cb(self, virDomain(self, _obj=dom), recipient, action, flags, opaque)
271        return 0
272
273    def _dispatchDomainEventMemoryDeviceSizeChangeCallback(self, dom: 'virDomain', alias: str, size: int, cbData: Dict[str, Any]) -> int:
274        """Dispatches event to python user domain memory device size change event callbacks
275        """
276        cb = cbData["cb"]
277        opaque = cbData["opaque"]
278
279        cb(self, virDomain(self, _obj=dom), alias, size, opaque)
280        return 0
281
282    def domainEventDeregisterAny(self, callbackID: int) -> None:
283        """Removes a Domain Event Callback. De-registering for a
284           domain callback will disable delivery of this event type """
285        try:
286            ret = libvirtmod.virConnectDomainEventDeregisterAny(self._o, callbackID)
287            if ret == -1:
288                raise libvirtError('virConnectDomainEventDeregisterAny() failed')
289            del self.domainEventCallbackID[callbackID]
290        except AttributeError:
291            pass
292
293    def _dispatchNetworkEventLifecycleCallback(self, net: 'virNetwork', event: int, detail: int, cbData: Dict[str, Any]) -> int:
294        """Dispatches events to python user network lifecycle event callbacks
295        """
296        cb = cbData["cb"]
297        opaque = cbData["opaque"]
298
299        cb(self, virNetwork(self, _obj=net), event, detail, opaque)
300        return 0
301
302    def networkEventDeregisterAny(self, callbackID: int) -> None:
303        """Removes a Network Event Callback. De-registering for a
304           network callback will disable delivery of this event type"""
305        try:
306            ret = libvirtmod.virConnectNetworkEventDeregisterAny(self._o, callbackID)
307            if ret == -1:
308                raise libvirtError('virConnectNetworkEventDeregisterAny() failed')
309            del self.networkEventCallbackID[callbackID]
310        except AttributeError:
311            pass
312
313    def networkEventRegisterAny(self, net: Optional['virNetwork'], eventID: int, cb: Callable, opaque: _T) -> int:
314        """Adds a Network Event Callback. Registering for a network
315           callback will enable delivery of the events"""
316        if not hasattr(self, 'networkEventCallbackID'):
317            self.networkEventCallbackID = {}  # type: Dict[int, _T]
318        cbData = {"cb": cb, "conn": self, "opaque": opaque}
319        if net is None:
320            ret = libvirtmod.virConnectNetworkEventRegisterAny(self._o, None, eventID, cbData)
321        else:
322            ret = libvirtmod.virConnectNetworkEventRegisterAny(self._o, net._o, eventID, cbData)
323        if ret == -1:
324            raise libvirtError('virConnectNetworkEventRegisterAny() failed')
325        self.networkEventCallbackID[ret] = opaque
326        return ret
327
328    def domainEventRegisterAny(self, dom: Optional['virDomain'], eventID: int, cb: Callable, opaque: _T) -> int:
329        """Adds a Domain Event Callback. Registering for a domain
330           callback will enable delivery of the events """
331        if not hasattr(self, 'domainEventCallbackID'):
332            self.domainEventCallbackID = {}  # type: Dict[int, _T]
333        cbData = {"cb": cb, "conn": self, "opaque": opaque}
334        if dom is None:
335            ret = libvirtmod.virConnectDomainEventRegisterAny(self._o, None, eventID, cbData)
336        else:
337            ret = libvirtmod.virConnectDomainEventRegisterAny(self._o, dom._o, eventID, cbData)
338        if ret == -1:
339            raise libvirtError('virConnectDomainEventRegisterAny() failed')
340        self.domainEventCallbackID[ret] = opaque
341        return ret
342
343    def _dispatchStoragePoolEventLifecycleCallback(self, pool: 'virStoragePool', event: int, detail: int, cbData: Dict[str, Any]) -> int:
344        """Dispatches events to python user storage pool
345           lifecycle event callbacks
346        """
347        cb = cbData["cb"]
348        opaque = cbData["opaque"]
349
350        cb(self, virStoragePool(self, _obj=pool), event, detail, opaque)
351        return 0
352
353    def _dispatchStoragePoolEventGenericCallback(self, pool: 'virStoragePool', cbData: Dict[str, Any]) -> int:
354        """Dispatches events to python user storage pool
355           generic event callbacks
356        """
357        cb = cbData["cb"]
358        opaque = cbData["opaque"]
359
360        cb(self, virStoragePool(self, _obj=pool), opaque)
361        return 0
362
363    def storagePoolEventDeregisterAny(self, callbackID: int) -> None:
364        """Removes a Storage Pool Event Callback. De-registering for a
365           storage pool callback will disable delivery of this event type"""
366        try:
367            ret = libvirtmod.virConnectStoragePoolEventDeregisterAny(self._o, callbackID)
368            if ret == -1:
369                raise libvirtError('virConnectStoragePoolEventDeregisterAny() failed')
370            del self.storagePoolEventCallbackID[callbackID]
371        except AttributeError:
372            pass
373
374    def storagePoolEventRegisterAny(self, pool: Optional['virStoragePool'], eventID: int, cb: Callable, opaque: _T) -> int:
375        """Adds a Storage Pool Event Callback. Registering for a storage pool
376           callback will enable delivery of the events"""
377        if not hasattr(self, 'storagePoolEventCallbackID'):
378            self.storagePoolEventCallbackID = {}  # type: Dict[int, _T]
379        cbData = {"cb": cb, "conn": self, "opaque": opaque}
380        if pool is None:
381            ret = libvirtmod.virConnectStoragePoolEventRegisterAny(self._o, None, eventID, cbData)
382        else:
383            ret = libvirtmod.virConnectStoragePoolEventRegisterAny(self._o, pool._o, eventID, cbData)
384        if ret == -1:
385            raise libvirtError('virConnectStoragePoolEventRegisterAny() failed')
386        self.storagePoolEventCallbackID[ret] = opaque
387        return ret
388
389    def _dispatchNodeDeviceEventLifecycleCallback(self, dev: 'virNodeDevice', event: int, detail: int, cbData: Dict[str, Any]) -> int:
390        """Dispatches events to python user node device
391           lifecycle event callbacks
392        """
393        cb = cbData["cb"]
394        opaque = cbData["opaque"]
395
396        cb(self, virNodeDevice(self, _obj=dev), event, detail, opaque)
397        return 0
398
399    def _dispatchNodeDeviceEventGenericCallback(self, dev: 'virNodeDevice', cbData: Dict[str, Any]) -> int:
400        """Dispatches events to python user node device
401           generic event callbacks
402        """
403        cb = cbData["cb"]
404        opaque = cbData["opaque"]
405
406        cb(self, virNodeDevice(self, _obj=dev), opaque)
407        return 0
408
409    def nodeDeviceEventDeregisterAny(self, callbackID: int) -> None:
410        """Removes a Node Device Event Callback. De-registering for a
411           node device callback will disable delivery of this event type"""
412        try:
413            ret = libvirtmod.virConnectNodeDeviceEventDeregisterAny(self._o, callbackID)
414            if ret == -1:
415                raise libvirtError('virConnectNodeDeviceEventDeregisterAny() failed')
416            del self.nodeDeviceEventCallbackID[callbackID]
417        except AttributeError:
418            pass
419
420    def nodeDeviceEventRegisterAny(self, dev: Optional['virNodeDevice'], eventID: int, cb: Callable, opaque: _T) -> int:
421        """Adds a Node Device Event Callback. Registering for a node device
422           callback will enable delivery of the events"""
423        if not hasattr(self, 'nodeDeviceEventCallbackID'):
424            self.nodeDeviceEventCallbackID = {}  # type: Dict[int, _T]
425        cbData = {"cb": cb, "conn": self, "opaque": opaque}
426        if dev is None:
427            ret = libvirtmod.virConnectNodeDeviceEventRegisterAny(self._o, None, eventID, cbData)
428        else:
429            ret = libvirtmod.virConnectNodeDeviceEventRegisterAny(self._o, dev._o, eventID, cbData)
430        if ret == -1:
431            raise libvirtError('virConnectNodeDeviceEventRegisterAny() failed')
432        self.nodeDeviceEventCallbackID[ret] = opaque
433        return ret
434
435    def _dispatchSecretEventLifecycleCallback(self, secret: 'virSecret', event: int, detail: int, cbData: Dict[str, Any]) -> int:
436        """Dispatches events to python user secret lifecycle event callbacks
437        """
438        cb = cbData["cb"]
439        opaque = cbData["opaque"]
440
441        cb(self, virSecret(self, _obj=secret), event, detail, opaque)
442        return 0
443
444    def _dispatchSecretEventGenericCallback(self, secret: 'virSecret', cbData: Dict[str, Any]) -> int:
445        """Dispatches events to python user secret generic event callbacks
446        """
447        cb = cbData["cb"]
448        opaque = cbData["opaque"]
449
450        cb(self, virSecret(self, _obj=secret), opaque)
451        return 0
452
453    def secretEventDeregisterAny(self, callbackID: int) -> None:
454        """Removes a Secret Event Callback. De-registering for a
455           secret callback will disable delivery of this event type"""
456        try:
457            ret = libvirtmod.virConnectSecretEventDeregisterAny(self._o, callbackID)
458            if ret == -1:
459                raise libvirtError('virConnectSecretEventDeregisterAny() failed')
460            del self.secretEventCallbackID[callbackID]
461        except AttributeError:
462            pass
463
464    def secretEventRegisterAny(self, secret: Optional['virSecret'], eventID: int, cb: Callable, opaque: _T) -> int:
465        """Adds a Secret Event Callback. Registering for a secret
466           callback will enable delivery of the events"""
467        if not hasattr(self, 'secretEventCallbackID'):
468            self.secretEventCallbackID = {}  # type: Dict[int, _T]
469        cbData = {"cb": cb, "conn": self, "opaque": opaque}
470        if secret is None:
471            ret = libvirtmod.virConnectSecretEventRegisterAny(self._o, None, eventID, cbData)
472        else:
473            ret = libvirtmod.virConnectSecretEventRegisterAny(self._o, secret._o, eventID, cbData)
474        if ret == -1:
475            raise libvirtError('virConnectSecretEventRegisterAny() failed')
476        self.secretEventCallbackID[ret] = opaque
477        return ret
478
479    def listAllDomains(self, flags: int = 0) -> List['virDomain']:
480        """List all domains and returns a list of domain objects"""
481        ret = libvirtmod.virConnectListAllDomains(self._o, flags)
482        if ret is None:
483            raise libvirtError("virConnectListAllDomains() failed")
484
485        return [virDomain(self, _obj=domptr) for domptr in ret]
486
487    def listAllStoragePools(self, flags: int = 0) -> List['virStoragePool']:
488        """Returns a list of storage pool objects"""
489        ret = libvirtmod.virConnectListAllStoragePools(self._o, flags)
490        if ret is None:
491            raise libvirtError("virConnectListAllStoragePools() failed")
492
493        return [virStoragePool(self, _obj=poolptr) for poolptr in ret]
494
495    def listAllNetworks(self, flags: int = 0) -> List['virNetwork']:
496        """Returns a list of network objects"""
497        ret = libvirtmod.virConnectListAllNetworks(self._o, flags)
498        if ret is None:
499            raise libvirtError("virConnectListAllNetworks() failed")
500
501        return [virNetwork(self, _obj=netptr) for netptr in ret]
502
503    def listAllInterfaces(self, flags: int = 0) -> List['virInterface']:
504        """Returns a list of interface objects"""
505        ret = libvirtmod.virConnectListAllInterfaces(self._o, flags)
506        if ret is None:
507            raise libvirtError("virConnectListAllInterfaces() failed")
508
509        return [virInterface(self, _obj=ifaceptr) for ifaceptr in ret]
510
511    def listAllDevices(self, flags: int = 0) -> List['virNodeDevice']:
512        """Returns a list of host node device objects"""
513        ret = libvirtmod.virConnectListAllNodeDevices(self._o, flags)
514        if ret is None:
515            raise libvirtError("virConnectListAllNodeDevices() failed")
516
517        return [virNodeDevice(self, _obj=devptr) for devptr in ret]
518
519    def listAllNWFilters(self, flags: int = 0) -> List['virNWFilter']:
520        """Returns a list of network filter objects"""
521        ret = libvirtmod.virConnectListAllNWFilters(self._o, flags)
522        if ret is None:
523            raise libvirtError("virConnectListAllNWFilters() failed")
524
525        return [virNWFilter(self, _obj=filter_ptr) for filter_ptr in ret]
526
527    def listAllNWFilterBindings(self, flags: int = 0) -> List['virNWFilterBinding']:
528        """Returns a list of network filter binding objects"""
529        ret = libvirtmod.virConnectListAllNWFilterBindings(self._o, flags)
530        if ret is None:
531            raise libvirtError("virConnectListAllNWFilterBindings() failed")
532
533        return [virNWFilterBinding(self, _obj=filter_ptr) for filter_ptr in ret]
534
535    def listAllSecrets(self, flags: int = 0) -> List['virSecret']:
536        """Returns a list of secret objects"""
537        ret = libvirtmod.virConnectListAllSecrets(self._o, flags)
538        if ret is None:
539            raise libvirtError("virConnectListAllSecrets() failed")
540
541        return [virSecret(self, _obj=secret_ptr) for secret_ptr in ret]
542
543    def _dispatchCloseCallback(self, reason: int, cbData: Dict[str, Any]) -> int:
544        """Dispatches events to python user close callback"""
545        cb = cbData["cb"]
546        opaque = cbData["opaque"]
547
548        cb(self, reason, opaque)
549        return 0
550
551    def unregisterCloseCallback(self) -> None:
552        """Removes a close event callback"""
553        ret = libvirtmod.virConnectUnregisterCloseCallback(self._o)
554        if ret == -1:
555            raise libvirtError('virConnectUnregisterCloseCallback() failed')
556
557    def registerCloseCallback(self, cb: Callable, opaque: _T) -> int:
558        """Adds a close event callback, providing a notification
559         when a connection fails / closes"""
560        cbData = {"cb": cb, "conn": self, "opaque": opaque}
561        ret = libvirtmod.virConnectRegisterCloseCallback(self._o, cbData)
562        if ret == -1:
563            raise libvirtError('virConnectRegisterCloseCallback() failed')
564        return ret
565
566    def createXMLWithFiles(self, xmlDesc: str, files: List[int], flags: int = 0) -> 'virDomain':
567        """Launch a new guest domain, based on an XML description similar
568        to the one returned by virDomainGetXMLDesc()
569        This function may require privileged access to the hypervisor.
570        The domain is not persistent, so its definition will disappear when it
571        is destroyed, or if the host is restarted (see virDomainDefineXML() to
572        define persistent domains).
573
574        @files provides an array of file descriptors which will be
575        made available to the 'init' process of the guest. The file
576        handles exposed to the guest will be renumbered to start
577        from 3 (ie immediately following stderr). This is only
578        supported for guests which use container based virtualization
579        technology.
580
581        If the VIR_DOMAIN_START_PAUSED flag is set, the guest domain
582        will be started, but its CPUs will remain paused. The CPUs
583        can later be manually started using virDomainResume.
584
585        If the VIR_DOMAIN_START_AUTODESTROY flag is set, the guest
586        domain will be automatically destroyed when the virConnectPtr
587        object is finally released. This will also happen if the
588        client application crashes / loses its connection to the
589        libvirtd daemon. Any domains marked for auto destroy will
590        block attempts at migration, save-to-file, or snapshots. """
591        ret = libvirtmod.virDomainCreateXMLWithFiles(self._o, xmlDesc, files, flags)
592        if ret is None:
593            raise libvirtError('virDomainCreateXMLWithFiles() failed')
594        __tmp = virDomain(self, _obj=ret)
595        return __tmp
596
597    def getAllDomainStats(self, stats: int = 0, flags: int = 0) -> List[Tuple['virDomain', Dict[str, Any]]]:
598        """Query statistics for all domains on a given connection.
599
600        Report statistics of various parameters for a running VM according to @stats
601        field. The statistics are returned as an array of structures for each queried
602        domain. The structure contains an array of typed parameters containing the
603        individual statistics. The typed parameter name for each statistic field
604        consists of a dot-separated string containing name of the requested group
605        followed by a group specific description of the statistic value.
606
607        The statistic groups are enabled using the @stats parameter which is a
608        binary-OR of enum virDomainStatsTypes. The following groups are available
609        (although not necessarily implemented for each hypervisor):
610
611        VIR_DOMAIN_STATS_STATE: Return domain state and reason for entering that
612        state. The typed parameter keys are in this format:
613        "state.state" - state of the VM, returned as int from virDomainState enum
614        "state.reason" - reason for entering given state, returned as int from
615                         virDomain*Reason enum corresponding to given state.
616
617        Using 0 for @stats returns all stats groups supported by the given
618        hypervisor.
619
620        Specifying VIR_CONNECT_GET_ALL_DOMAINS_STATS_ENFORCE_STATS as @flags makes
621        the function return error in case some of the stat types in @stats were
622        not recognized by the daemon.
623
624        Similarly to virConnectListAllDomains, @flags can contain various flags to
625        filter the list of domains to provide stats for.
626
627        VIR_CONNECT_GET_ALL_DOMAINS_STATS_ACTIVE selects online domains while
628        VIR_CONNECT_GET_ALL_DOMAINS_STATS_INACTIVE selects offline ones.
629
630        VIR_CONNECT_GET_ALL_DOMAINS_STATS_PERSISTENT and
631        VIR_CONNECT_GET_ALL_DOMAINS_STATS_TRANSIENT allow to filter the list
632        according to their persistence.
633
634        To filter the list of VMs by domain state @flags can contain
635        VIR_CONNECT_GET_ALL_DOMAINS_STATS_RUNNING,
636        VIR_CONNECT_GET_ALL_DOMAINS_STATS_PAUSED,
637        VIR_CONNECT_GET_ALL_DOMAINS_STATS_SHUTOFF and/or
638        VIR_CONNECT_GET_ALL_DOMAINS_STATS_OTHER for all other states. """
639        ret = libvirtmod.virConnectGetAllDomainStats(self._o, stats, flags)
640        if ret is None:
641            raise libvirtError("virConnectGetAllDomainStats() failed")
642
643        return [(virDomain(self, _obj=elem[0]), elem[1]) for elem in ret]
644
645    def domainListGetStats(self, doms: List['virDomain'], stats: int = 0, flags: int = 0) -> List[Tuple['virDomain', Dict[str, Any]]]:
646        """ Query statistics for given domains.
647
648        Report statistics of various parameters for a running VM according to @stats
649        field. The statistics are returned as an array of structures for each queried
650        domain. The structure contains an array of typed parameters containing the
651        individual statistics. The typed parameter name for each statistic field
652        consists of a dot-separated string containing name of the requested group
653        followed by a group specific description of the statistic value.
654
655        The statistic groups are enabled using the @stats parameter which is a
656        binary-OR of enum virDomainStatsTypes. The following groups are available
657        (although not necessarily implemented for each hypervisor):
658
659        VIR_DOMAIN_STATS_STATE: Return domain state and reason for entering that
660        state. The typed parameter keys are in this format:
661        "state.state" - state of the VM, returned as int from virDomainState enum
662        "state.reason" - reason for entering given state, returned as int from
663                         virDomain*Reason enum corresponding to given state.
664
665        Using 0 for @stats returns all stats groups supported by the given
666        hypervisor.
667
668        Specifying VIR_CONNECT_GET_ALL_DOMAINS_STATS_ENFORCE_STATS as @flags makes
669        the function return error in case some of the stat types in @stats were
670        not recognized by the daemon.
671
672        Get statistics about domains provided as a list in @doms. @stats is
673        a bit field selecting requested statistics types."""
674        domlist = list()
675        for dom in doms:
676            if not isinstance(dom, virDomain):
677                raise libvirtError("domain list contains non-domain elements")
678
679            domlist.append(dom._o)
680
681        ret = libvirtmod.virDomainListGetStats(self._o, domlist, stats, flags)
682        if ret is None:
683            raise libvirtError("virDomainListGetStats() failed")
684
685        return [(virDomain(self, _obj=elem[0]), elem[1]) for elem in ret]
686