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