shaba/libvirt-python
1
0
Fork 0
mirror of https://gitlab.com/libvirt/libvirt-python.git synced 2025-07-17 00:59:36 +03:00
Code Issues Projects Releases Activity

override: Add manual PEP 484 type annotations

Browse Source 
Signed-off-by: Philipp Hahn <hahn@univention.de>
This commit is contained in:
Philipp Hahn
2018-11-20 08:59:35 +01:00
committed by Daniel Berrange
parent 67af8b910b
commit abbd47f4ea
9 changed files with 173 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

137
libvirt-override-virConnect.py
View File

@ -1,4 +1,4 @@
def __del__(self):
def __del__(self) -> None:
try:
for cb, opaque in self.domainEventCallbacks.items():
del self.domainEventCallbacks[cb]
@ -11,13 +11,13 @@
libvirtmod.virConnectClose(self._o)
self._o = None
def __enter__(self):
def __enter__(self) -> 'virConnect':
return self
def __exit__(self, exc_type_, exc_value_, traceback_):
def __exit__(self, exc_type_: Optional[Type[BaseException]], exc_value_: Optional[BaseException], traceback_: Optional[TracebackType]) -> None:
self.close()
def domainEventDeregister(self, cb):
def domainEventDeregister(self, cb: _DomainCB) -> None:
"""Removes a Domain Event Callback. De-registering for a
domain callback will disable delivery of this event type """
try:
@ -30,18 +30,18 @@
except AttributeError:
pass
def domainEventRegister(self, cb, opaque):
def domainEventRegister(self, cb: _DomainCB, opaque: _T) -> None:
"""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}
self.domainEventCallbacks = {cb: opaque} # type: Dict[_DomainCB, _T]
ret = libvirtmod.virConnectDomainEventRegister(self._o, self)
if ret == -1:
raise libvirtError('virConnectDomainEventRegister() failed')
def _dispatchDomainEventCallbacks(self, dom, event, detail):
def _dispatchDomainEventCallbacks(self, dom: 'virDomain', event: int, detail: int) -> None:
"""Dispatches events to python user domain event callbacks
"""
try:
@ -51,7 +51,7 @@
except AttributeError:
pass
def _dispatchDomainEventLifecycleCallback(self, dom, event, detail, cbData):
def _dispatchDomainEventLifecycleCallback(self, dom: 'virDomain', event: int, detail: int, cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user domain lifecycle event callbacks
"""
cb = cbData["cb"]
@ -60,7 +60,7 @@
cb(self, virDomain(self, _obj=dom), event, detail, opaque)
return 0
def _dispatchDomainEventGenericCallback(self, dom, cbData):
def _dispatchDomainEventGenericCallback(self, dom: 'virDomain', cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user domain generic event callbacks
"""
cb = cbData["cb"]
@ -69,7 +69,7 @@
cb(self, virDomain(self, _obj=dom), opaque)
return 0
def _dispatchDomainEventRTCChangeCallback(self, dom, offset, cbData):
def _dispatchDomainEventRTCChangeCallback(self, dom: 'virDomain', offset: int, cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user domain RTC change event callbacks
"""
cb = cbData["cb"]
@ -78,7 +78,7 @@
cb(self, virDomain(self, _obj=dom), offset, opaque)
return 0
def _dispatchDomainEventWatchdogCallback(self, dom, action, cbData):
def _dispatchDomainEventWatchdogCallback(self, dom: 'virDomain', action: int, cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user domain watchdog event callbacks
"""
cb = cbData["cb"]
@ -87,8 +87,7 @@
cb(self, virDomain(self, _obj=dom), action, opaque)
return 0
def _dispatchDomainEventIOErrorCallback(self, dom, srcPath, devAlias,
action, cbData):
def _dispatchDomainEventIOErrorCallback(self, dom: 'virDomain', srcPath: str, devAlias: str, action: int, cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user domain IO error event callbacks
"""
cb = cbData["cb"]
@ -97,9 +96,7 @@
cb(self, virDomain(self, _obj=dom), srcPath, devAlias, action, opaque)
return 0
def _dispatchDomainEventIOErrorReasonCallback(self, dom, srcPath,
devAlias, action, reason,
cbData):
def _dispatchDomainEventIOErrorReasonCallback(self, dom: 'virDomain', srcPath: str, devAlias: str, action: int, reason: int, cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user domain IO error event callbacks
"""
cb = cbData["cb"]
@ -109,9 +106,7 @@
reason, opaque)
return 0
def _dispatchDomainEventGraphicsCallback(self, dom, phase, localAddr,
remoteAddr, authScheme, subject,
cbData):
def _dispatchDomainEventGraphicsCallback(self, dom: 'virDomain', phase: int, localAddr: Any, remoteAddr: Any, authScheme: str, subject: Any, cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user domain graphics event callbacks
"""
cb = cbData["cb"]
@ -121,7 +116,7 @@
authScheme, subject, opaque)
return 0
def _dispatchDomainEventBlockJobCallback(self, dom, disk, type, status, cbData):
def _dispatchDomainEventBlockJobCallback(self, dom: 'virDomain', disk: str, type: int, status: int, cbData: Dict[str, Any]) -> None:
"""Dispatches events to python user domain blockJob/blockJob2 event callbacks
"""
try:
@ -133,7 +128,7 @@
except AttributeError:
pass
def _dispatchDomainEventDiskChangeCallback(self, dom, oldSrcPath, newSrcPath, devAlias, reason, cbData):
def _dispatchDomainEventDiskChangeCallback(self, dom: 'virDomain', oldSrcPath: str, newSrcPath: str, devAlias: str, reason: int, cbData: Dict[str, Any]) -> int:
"""Dispatches event to python user domain diskChange event callbacks
"""
cb = cbData["cb"]
@ -142,7 +137,7 @@
cb(self, virDomain(self, _obj=dom), oldSrcPath, newSrcPath, devAlias, reason, opaque)
return 0
def _dispatchDomainEventTrayChangeCallback(self, dom, devAlias, reason, cbData):
def _dispatchDomainEventTrayChangeCallback(self, dom: 'virDomain', devAlias: str, reason: int, cbData: Dict[str, Any]) -> int:
"""Dispatches event to python user domain trayChange event callbacks
"""
cb = cbData["cb"]
@ -151,7 +146,7 @@
cb(self, virDomain(self, _obj=dom), devAlias, reason, opaque)
return 0
def _dispatchDomainEventPMWakeupCallback(self, dom, reason, cbData):
def _dispatchDomainEventPMWakeupCallback(self, dom: 'virDomain', reason: int, cbData: Dict[str, Any]) -> int:
"""Dispatches event to python user domain pmwakeup event callbacks
"""
cb = cbData["cb"]
@ -160,7 +155,7 @@
cb(self, virDomain(self, _obj=dom), reason, opaque)
return 0
def _dispatchDomainEventPMSuspendCallback(self, dom, reason, cbData):
def _dispatchDomainEventPMSuspendCallback(self, dom: 'virDomain', reason: int, cbData: Dict[str, Any]) -> int:
"""Dispatches event to python user domain pmsuspend event callbacks
"""
cb = cbData["cb"]
@ -169,7 +164,7 @@
cb(self, virDomain(self, _obj=dom), reason, opaque)
return 0
def _dispatchDomainEventBalloonChangeCallback(self, dom, actual, cbData):
def _dispatchDomainEventBalloonChangeCallback(self, dom: 'virDomain', actual: int, cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user domain balloon change event callbacks
"""
cb = cbData["cb"]
@ -178,7 +173,7 @@
cb(self, virDomain(self, _obj=dom), actual, opaque)
return 0
def _dispatchDomainEventPMSuspendDiskCallback(self, dom, reason, cbData):
def _dispatchDomainEventPMSuspendDiskCallback(self, dom: 'virDomain', reason: int, cbData: Dict[str, Any]) -> int:
"""Dispatches event to python user domain pmsuspend-disk event callbacks
"""
cb = cbData["cb"]
@ -187,7 +182,7 @@
cb(self, virDomain(self, _obj=dom), reason, opaque)
return 0
def _dispatchDomainEventDeviceRemovedCallback(self, dom, devAlias, cbData):
def _dispatchDomainEventDeviceRemovedCallback(self, dom: 'virDomain', devAlias: str, cbData: Dict[str, Any]) -> int:
"""Dispatches event to python user domain device removed event callbacks
"""
cb = cbData["cb"]
@ -196,7 +191,7 @@
cb(self, virDomain(self, _obj=dom), devAlias, opaque)
return 0
def _dispatchDomainEventTunableCallback(self, dom, params, cbData):
def _dispatchDomainEventTunableCallback(self, dom: 'virDomain', params: Any, cbData: Dict[str, Any]) -> int:
"""Dispatches event to python user domain tunable event callbacks
"""
cb = cbData["cb"]
@ -205,7 +200,7 @@
cb(self, virDomain(self, _obj=dom), params, opaque)
return 0
def _dispatchDomainEventAgentLifecycleCallback(self, dom, state, reason, cbData):
def _dispatchDomainEventAgentLifecycleCallback(self, dom: 'virDomain', state: int, reason: int, cbData: Dict[str, Any]) -> int:
"""Dispatches event to python user domain agent lifecycle event callback
"""
@ -215,7 +210,7 @@
cb(self, virDomain(self, _obj=dom), state, reason, opaque)
return 0
def _dispatchDomainEventDeviceAddedCallback(self, dom, devAlias, cbData):
def _dispatchDomainEventDeviceAddedCallback(self, dom: 'virDomain', devAlias: str, cbData: Dict[str, Any]) -> int:
"""Dispatches event to python user domain device added event callbacks
"""
cb = cbData["cb"]
@ -224,7 +219,7 @@
cb(self, virDomain(self, _obj=dom), devAlias, opaque)
return 0
def _dispatchDomainEventMigrationIterationCallback(self, dom, iteration, cbData):
def _dispatchDomainEventMigrationIterationCallback(self, dom: 'virDomain', iteration: int, cbData: Dict[str, Any]) -> int:
"""Dispatches event to python user domain migration iteration event callbacks
"""
cb = cbData["cb"]
@ -233,7 +228,7 @@
cb(self, virDomain(self, _obj=dom), iteration, opaque)
return 0
def _dispatchDomainEventJobCompletedCallback(self, dom, params, cbData):
def _dispatchDomainEventJobCompletedCallback(self, dom: 'virDomain', params: Dict[str, Any], cbData: Dict[str, Any]) -> int:
"""Dispatches event to python user domain job completed callbacks
"""
cb = cbData["cb"]
@ -242,7 +237,7 @@
cb(self, virDomain(self, _obj=dom), params, opaque)
return 0
def _dispatchDomainEventDeviceRemovalFailedCallback(self, dom, devAlias, cbData):
def _dispatchDomainEventDeviceRemovalFailedCallback(self, dom: 'virDomain', devAlias: str, cbData: Dict[str, Any]) -> int:
"""Dispatches event to python user domain device removal failed event callbacks
"""
cb = cbData["cb"]
@ -251,7 +246,7 @@
cb(self, virDomain(self, _obj=dom), devAlias, opaque)
return 0
def _dispatchDomainEventMetadataChangeCallback(self, dom, mtype, nsuri, cbData):
def _dispatchDomainEventMetadataChangeCallback(self, dom: 'virDomain', mtype: int, nsuri: str, cbData: Dict[str, Any]) -> int:
"""Dispatches event to python user domain metadata change event callbacks
"""
cb = cbData["cb"]
@ -260,7 +255,7 @@
cb(self, virDomain(self, _obj=dom), mtype, nsuri, opaque)
return 0
def _dispatchDomainEventBlockThresholdCallback(self, dom, dev, path, threshold, excess, cbData):
def _dispatchDomainEventBlockThresholdCallback(self, dom: 'virDomain', dev: str, path: str, threshold: int, excess: int, cbData: Dict[str, Any]) -> int:
"""Dispatches event to python user domain block device threshold event callbacks
"""
cb = cbData["cb"]
@ -269,7 +264,7 @@
cb(self, virDomain(self, _obj=dom), dev, path, threshold, excess, opaque)
return 0
def domainEventDeregisterAny(self, callbackID):
def domainEventDeregisterAny(self, callbackID: int) -> None:
"""Removes a Domain Event Callback. De-registering for a
domain callback will disable delivery of this event type """
try:
@ -280,7 +275,7 @@
except AttributeError:
pass
def _dispatchNetworkEventLifecycleCallback(self, net, event, detail, cbData):
def _dispatchNetworkEventLifecycleCallback(self, net: 'virNetwork', event: int, detail: int, cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user network lifecycle event callbacks
"""
cb = cbData["cb"]
@ -289,7 +284,7 @@
cb(self, virNetwork(self, _obj=net), event, detail, opaque)
return 0
def networkEventDeregisterAny(self, callbackID):
def networkEventDeregisterAny(self, callbackID: int) -> None:
"""Removes a Network Event Callback. De-registering for a
network callback will disable delivery of this event type"""
try:
@ -300,11 +295,11 @@
except AttributeError:
pass
def networkEventRegisterAny(self, net, eventID, cb, opaque):
def networkEventRegisterAny(self, net: Optional['virNetwork'], eventID: int, cb: Callable, opaque: _T) -> int:
"""Adds a Network Event Callback. Registering for a network
callback will enable delivery of the events"""
if not hasattr(self, 'networkEventCallbackID'):
self.networkEventCallbackID = {}
self.networkEventCallbackID = {} # type: Dict[int, _T]
cbData = {"cb": cb, "conn": self, "opaque": opaque}
if net is None:
ret = libvirtmod.virConnectNetworkEventRegisterAny(self._o, None, eventID, cbData)
@ -315,11 +310,11 @@
self.networkEventCallbackID[ret] = opaque
return ret
def domainEventRegisterAny(self, dom, eventID, cb, opaque):
def domainEventRegisterAny(self, dom: Optional['virDomain'], eventID: int, cb: Callable, opaque: _T) -> int:
"""Adds a Domain Event Callback. Registering for a domain
callback will enable delivery of the events """
if not hasattr(self, 'domainEventCallbackID'):
self.domainEventCallbackID = {}
self.domainEventCallbackID = {} # type: Dict[int, _T]
cbData = {"cb": cb, "conn": self, "opaque": opaque}
if dom is None:
ret = libvirtmod.virConnectDomainEventRegisterAny(self._o, None, eventID, cbData)
@ -330,7 +325,7 @@
self.domainEventCallbackID[ret] = opaque
return ret
def _dispatchStoragePoolEventLifecycleCallback(self, pool, event, detail, cbData):
def _dispatchStoragePoolEventLifecycleCallback(self, pool: 'virStoragePool', event: int, detail: int, cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user storage pool
lifecycle event callbacks
"""
@ -340,7 +335,7 @@
cb(self, virStoragePool(self, _obj=pool), event, detail, opaque)
return 0
def _dispatchStoragePoolEventGenericCallback(self, pool, cbData):
def _dispatchStoragePoolEventGenericCallback(self, pool: 'virStoragePool', cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user storage pool
generic event callbacks
"""
@ -350,7 +345,7 @@
cb(self, virStoragePool(self, _obj=pool), opaque)
return 0
def storagePoolEventDeregisterAny(self, callbackID):
def storagePoolEventDeregisterAny(self, callbackID: int) -> None:
"""Removes a Storage Pool Event Callback. De-registering for a
storage pool callback will disable delivery of this event type"""
try:
@ -361,11 +356,11 @@
except AttributeError:
pass
def storagePoolEventRegisterAny(self, pool, eventID, cb, opaque):
def storagePoolEventRegisterAny(self, pool: Optional['virStoragePool'], eventID: int, cb: Callable, opaque: _T) -> int:
"""Adds a Storage Pool Event Callback. Registering for a storage pool
callback will enable delivery of the events"""
if not hasattr(self, 'storagePoolEventCallbackID'):
self.storagePoolEventCallbackID = {}
self.storagePoolEventCallbackID = {} # type: Dict[int, _T]
cbData = {"cb": cb, "conn": self, "opaque": opaque}
if pool is None:
ret = libvirtmod.virConnectStoragePoolEventRegisterAny(self._o, None, eventID, cbData)
@ -376,7 +371,7 @@
self.storagePoolEventCallbackID[ret] = opaque
return ret
def _dispatchNodeDeviceEventLifecycleCallback(self, dev, event, detail, cbData):
def _dispatchNodeDeviceEventLifecycleCallback(self, dev: 'virNodeDevice', event: int, detail: int, cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user node device
lifecycle event callbacks
"""
@ -386,7 +381,7 @@
cb(self, virNodeDevice(self, _obj=dev), event, detail, opaque)
return 0
def _dispatchNodeDeviceEventGenericCallback(self, dev, cbData):
def _dispatchNodeDeviceEventGenericCallback(self, dev: 'virNodeDevice', cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user node device
generic event callbacks
"""
@ -396,7 +391,7 @@
cb(self, virNodeDevice(self, _obj=dev), opaque)
return 0
def nodeDeviceEventDeregisterAny(self, callbackID):
def nodeDeviceEventDeregisterAny(self, callbackID: int) -> None:
"""Removes a Node Device Event Callback. De-registering for a
node device callback will disable delivery of this event type"""
try:
@ -407,11 +402,11 @@
except AttributeError:
pass
def nodeDeviceEventRegisterAny(self, dev, eventID, cb, opaque):
def nodeDeviceEventRegisterAny(self, dev: Optional['virNodeDevice'], eventID: int, cb: Callable, opaque: _T) -> int:
"""Adds a Node Device Event Callback. Registering for a node device
callback will enable delivery of the events"""
if not hasattr(self, 'nodeDeviceEventCallbackID'):
self.nodeDeviceEventCallbackID = {}
self.nodeDeviceEventCallbackID = {} # type: Dict[int, _T]
cbData = {"cb": cb, "conn": self, "opaque": opaque}
if dev is None:
ret = libvirtmod.virConnectNodeDeviceEventRegisterAny(self._o, None, eventID, cbData)
@ -422,7 +417,7 @@
self.nodeDeviceEventCallbackID[ret] = opaque
return ret
def _dispatchSecretEventLifecycleCallback(self, secret, event, detail, cbData):
def _dispatchSecretEventLifecycleCallback(self, secret: 'virSecret', event: int, detail: int, cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user secret lifecycle event callbacks
"""
cb = cbData["cb"]
@ -431,7 +426,7 @@
cb(self, virSecret(self, _obj=secret), event, detail, opaque)
return 0
def _dispatchSecretEventGenericCallback(self, secret, cbData):
def _dispatchSecretEventGenericCallback(self, secret: 'virSecret', cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user secret generic event callbacks
"""
cb = cbData["cb"]
@ -440,7 +435,7 @@
cb(self, virSecret(self, _obj=secret), opaque)
return 0
def secretEventDeregisterAny(self, callbackID):
def secretEventDeregisterAny(self, callbackID: int) -> None:
"""Removes a Secret Event Callback. De-registering for a
secret callback will disable delivery of this event type"""
try:
@ -451,11 +446,11 @@
except AttributeError:
pass
def secretEventRegisterAny(self, secret, eventID, cb, opaque):
def secretEventRegisterAny(self, secret: Optional['virSecret'], eventID: int, cb: Callable, opaque: _T) -> int:
"""Adds a Secret Event Callback. Registering for a secret
callback will enable delivery of the events"""
if not hasattr(self, 'secretEventCallbackID'):
self.secretEventCallbackID = {}
self.secretEventCallbackID = {} # type: Dict[int, _T]
cbData = {"cb": cb, "conn": self, "opaque": opaque}
if secret is None:
ret = libvirtmod.virConnectSecretEventRegisterAny(self._o, None, eventID, cbData)
@ -466,7 +461,7 @@
self.secretEventCallbackID[ret] = opaque
return ret
def listAllDomains(self, flags=0):
def listAllDomains(self, flags: int = 0) -> List['virDomain']:
"""List all domains and returns a list of domain objects"""
ret = libvirtmod.virConnectListAllDomains(self._o, flags)
if ret is None:
@ -478,7 +473,7 @@
return retlist
def listAllStoragePools(self, flags=0):
def listAllStoragePools(self, flags: int = 0) -> List['virStoragePool']:
"""Returns a list of storage pool objects"""
ret = libvirtmod.virConnectListAllStoragePools(self._o, flags)
if ret is None:
@ -490,7 +485,7 @@
return retlist
def listAllNetworks(self, flags=0):
def listAllNetworks(self, flags: int = 0) -> List['virNetwork']:
"""Returns a list of network objects"""
ret = libvirtmod.virConnectListAllNetworks(self._o, flags)
if ret is None:
@ -502,7 +497,7 @@
return retlist
def listAllInterfaces(self, flags=0):
def listAllInterfaces(self, flags: int = 0) -> List['virInterface']:
"""Returns a list of interface objects"""
ret = libvirtmod.virConnectListAllInterfaces(self._o, flags)
if ret is None:
@ -514,7 +509,7 @@
return retlist
def listAllDevices(self, flags=0):
def listAllDevices(self, flags: int = 0) -> List['virNodeDevice']:
"""Returns a list of host node device objects"""
ret = libvirtmod.virConnectListAllNodeDevices(self._o, flags)
if ret is None:
@ -526,7 +521,7 @@
return retlist
def listAllNWFilters(self, flags=0):
def listAllNWFilters(self, flags: int = 0) -> List['virNWFilter']:
"""Returns a list of network filter objects"""
ret = libvirtmod.virConnectListAllNWFilters(self._o, flags)
if ret is None:
@ -538,7 +533,7 @@
return retlist
def listAllNWFilterBindings(self, flags=0):
def listAllNWFilterBindings(self, flags: int = 0) -> List['virNWFilterBinding']:
"""Returns a list of network filter binding objects"""
ret = libvirtmod.virConnectListAllNWFilterBindings(self._o, flags)
if ret is None:
@ -550,7 +545,7 @@
return retlist
def listAllSecrets(self, flags=0):
def listAllSecrets(self, flags: int = 0) -> List['virSecret']:
"""Returns a list of secret objects"""
ret = libvirtmod.virConnectListAllSecrets(self._o, flags)
if ret is None:
@ -562,7 +557,7 @@
return retlist
def _dispatchCloseCallback(self, reason, cbData):
def _dispatchCloseCallback(self, reason: int, cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user close callback"""
cb = cbData["cb"]
opaque = cbData["opaque"]
@ -570,13 +565,13 @@
cb(self, reason, opaque)
return 0
def unregisterCloseCallback(self):
def unregisterCloseCallback(self) -> None:
"""Removes a close event callback"""
ret = libvirtmod.virConnectUnregisterCloseCallback(self._o)
if ret == -1:
raise libvirtError('virConnectUnregisterCloseCallback() failed')
def registerCloseCallback(self, cb, opaque):
def registerCloseCallback(self, cb: Callable, opaque: _T) -> int:
"""Adds a close event callback, providing a notification
when a connection fails / closes"""
cbData = {"cb": cb, "conn": self, "opaque": opaque}
@ -585,7 +580,7 @@
raise libvirtError('virConnectRegisterCloseCallback() failed')
return ret
def createXMLWithFiles(self, xmlDesc, files, flags=0):
def createXMLWithFiles(self, xmlDesc: str, files: List[int], flags: int = 0) -> 'virDomain':
"""Launch a new guest domain, based on an XML description similar
to the one returned by virDomainGetXMLDesc()
This function may require privileged access to the hypervisor.
@ -616,7 +611,7 @@
__tmp = virDomain(self, _obj=ret)
return __tmp
def getAllDomainStats(self, stats=0, flags=0):
def getAllDomainStats(self, stats: int = 0, flags: int = 0) -> List[Tuple['virDomain', Dict[str, Any]]]:
"""Query statistics for all domains on a given connection.
Report statistics of various parameters for a running VM according to @stats
@ -669,7 +664,7 @@
return retlist
def domainListGetStats(self, doms, stats=0, flags=0):
def domainListGetStats(self, doms: List['virDomain'], stats: int = 0, flags: int = 0) -> List[Tuple['virDomain', Dict[str, Any]]]:
""" Query statistics for given domains.
Report statistics of various parameters for a running VM according to @stats

14
libvirt-override-virDomain.py
View File

@ -1,4 +1,4 @@
def listAllSnapshots(self, flags=0):
def listAllSnapshots(self, flags: int = 0) -> List['virDomainSnapshot']:
"""List all snapshots and returns a list of snapshot objects"""
ret = libvirtmod.virDomainListAllSnapshots(self._o, flags)
if ret is None:
@ -10,7 +10,7 @@
return retlist
def listAllCheckpoints(self, flags=0):
def listAllCheckpoints(self, flags: int = 0) -> List['virDomainCheckpoint']:
"""List all checkpoints and returns a list of checkpoint objects"""
ret = libvirtmod.virDomainListAllCheckpoints(self._o, flags)
if ret is None:
@ -22,7 +22,7 @@
return retlist
def createWithFiles(self, files, flags=0):
def createWithFiles(self, files: List[int], flags: int = 0) -> 'virDomain':
"""Launch a defined domain. If the call succeeds the domain moves from the
defined to the running domains pools.
@ -60,28 +60,28 @@
raise libvirtError('virDomainCreateWithFiles() failed')
return ret
def fsFreeze(self, mountpoints=None, flags=0):
def fsFreeze(self, mountpoints: List[str] = None, flags: int = 0) -> int:
"""Freeze specified filesystems within the guest """
ret = libvirtmod.virDomainFSFreeze(self._o, mountpoints, flags)
if ret == -1:
raise libvirtError('virDomainFSFreeze() failed')
return ret
def fsThaw(self, mountpoints=None, flags=0):
def fsThaw(self, mountpoints: List[str] = None, flags: int = 0) -> int:
"""Thaw specified filesystems within the guest """
ret = libvirtmod.virDomainFSThaw(self._o, mountpoints, flags)
if ret == -1:
raise libvirtError('virDomainFSThaw() failed')
return ret
def getTime(self, flags=0):
def getTime(self, flags: int = 0) -> int:
"""Extract information about guest time """
ret = libvirtmod.virDomainGetTime(self._o, flags)
if ret == None:
raise libvirtError('virDomainGetTime() failed')
return ret
def setTime(self, time=None, flags=0):
def setTime(self, time: int = None, flags: int = 0) -> int:
"""Set guest time to the given value. @time is a dict containing
'seconds' field for seconds and 'nseconds' field for nanoseconds """
ret = libvirtmod.virDomainSetTime(self._o, time, flags)

6
libvirt-override-virDomainCheckpoint.py
View File

@ -1,12 +1,12 @@
def getConnect(self):
def getConnect(self) -> 'virConnect':
"""Get the connection that owns the domain that a checkpoint was created for"""
return self.connect()
def getDomain(self):
def getDomain(self) -> 'virDomain':
"""Get the domain that a checkpoint was created for"""
return self.domain()
def listAllChildren(self, flags=0):
def listAllChildren(self, flags: int = 0) -> List['virDomainCheckpoint']:
"""List all child checkpoints and returns a list of checkpoint objects"""
ret = libvirtmod.virDomainCheckpointListAllChildren(self._o, flags)
if ret is None:

6
libvirt-override-virDomainSnapshot.py
View File

@ -1,12 +1,12 @@
def getConnect(self):
def getConnect(self) -> 'virConnect':
"""Get the connection that owns the domain that a snapshot was created for"""
return self.connect()
def getDomain(self):
def getDomain(self) -> 'virDomain':
"""Get the domain that a snapshot was created for"""
return self.domain()
def listAllChildren(self, flags=0):
def listAllChildren(self, flags: int = 0) -> List['virDomainSnapshot']:
"""List all child snapshots and returns a list of snapshot objects"""
ret = libvirtmod.virDomainSnapshotListAllChildren(self._o, flags)
if ret is None:

2
libvirt-override-virNetwork.py
View File

@ -1,4 +1,4 @@
def listAllPorts(self, flags=0):
def listAllPorts(self, flags: int = 0) -> List['virNetworkPort']:
"""List all ports on the network and returns a list of network port objects"""
ret = libvirtmod.virNetworkListAllPorts(self._o, flags)
if ret is None:

2
libvirt-override-virStoragePool.py
View File

@ -1,4 +1,4 @@
def listAllVolumes(self, flags=0):
def listAllVolumes(self, flags: int = 0) -> List['virStorageVol']:
"""List all storage volumes and returns a list of storage volume objects"""
ret = libvirtmod.virStoragePoolListAllVolumes(self._o, flags)
if ret is None:

25
libvirt-override-virStream.py
View File

@ -1,4 +1,4 @@
def __del__(self):
def __del__(self) -> None:
try:
if self.cb:
libvirtmod.virStreamEventRemoveCallback(self._o)
@ -9,7 +9,7 @@
libvirtmod.virStreamFree(self._o)
self._o = None
def _dispatchStreamEventCallback(self, events, cbData):
def _dispatchStreamEventCallback(self, events: int, cbData: Dict[str, Any]) -> int:
"""
Dispatches events to python user's stream event callbacks
"""
@ -19,14 +19,14 @@
cb(self, events, opaque)
return 0
def eventAddCallback(self, events, cb, opaque):
def eventAddCallback(self, events: int, cb: Callable[['virStream', int, _T], None], opaque: _T) -> None:
self.cb = cb
cbData = {"stream": self, "cb": cb, "opaque": opaque}
ret = libvirtmod.virStreamEventAddCallback(self._o, events, cbData)
if ret == -1:
raise libvirtError('virStreamEventAddCallback() failed')
def recvAll(self, handler, opaque):
def recvAll(self, handler: Callable[['virStream', bytes, _T], int], opaque: _T) -> None:
"""Receive the entire data stream, sending the data to the
requested data sink. This is simply a convenient alternative
to virStreamRecv, for apps that do blocking-I/O.
@ -59,7 +59,7 @@
pass
raise e
def sendAll(self, handler, opaque):
def sendAll(self, handler: Callable[['virStream', int, _T], bytes], opaque: _T) -> None:
"""
Send the entire data stream, reading the data from the
requested data source. This is simply a convenient alternative
@ -92,7 +92,7 @@
raise libvirtError("cannot use sendAll with "
"nonblocking stream")
def recv(self, nbytes):
def recv(self, nbytes: int) -> bytes:
"""Reads a series of bytes from the stream. This method may
block the calling application for an arbitrary amount
of time.
@ -110,7 +110,7 @@
raise libvirtError('virStreamRecv() failed')
return ret
def send(self, data):
def send(self, data: bytes) -> int:
"""Write a series of bytes to the stream. This method may
block the calling application for an arbitrary amount
of time. Once an application has finished sending data
@ -129,7 +129,7 @@
raise libvirtError('virStreamSend() failed')
return ret
def recvHole(self, flags=0):
def recvHole(self, flags: int = 0) -> int:
"""This method is used to determine the length in bytes
of the empty space to be created in a stream's target
file when uploading or downloading sparsely populated
@ -140,7 +140,7 @@
raise libvirtError('virStreamRecvHole() failed')
return ret
def sendHole(self, length, flags=0):
def sendHole(self, length: int, flags: int = 0) -> int:
"""Rather than transmitting empty file space, this method
directs the stream target to create length bytes of empty
space. This method would be used when uploading or
@ -152,7 +152,7 @@
raise libvirtError('virStreamSendHole() failed')
return ret
def recvFlags(self, nbytes, flags=0):
def recvFlags(self, nbytes: int, flags: int = 0) -> Union[bytes, int]:
"""Reads a series of bytes from the stream. This method may
block the calling application for an arbitrary amount
of time. This is just like recv except it has flags
@ -171,7 +171,7 @@
raise libvirtError('virStreamRecvFlags() failed')
return ret
def sparseRecvAll(self, handler, holeHandler, opaque):
def sparseRecvAll(self, handler: Callable[['virStream', bytes, _T], Union[bytes, int]], holeHandler: Callable[['virStream', int, _T], Optional[int]], opaque: _T) -> None:
"""Receive the entire data stream, sending the data to
the requested data sink handler and calling the skip
holeHandler to generate holes for sparse stream targets.
@ -219,7 +219,7 @@
self.abort()
raise RuntimeError("sparseRecvAll handler returned %d" % ret)
def sparseSendAll(self, handler, holeHandler, skipHandler, opaque):
def sparseSendAll(self, handler: Callable[['virStream', int, _T], Union[bytes, int]], holeHandler: Callable[['virStream', _T], Tuple[bool, int]], skipHandler: Callable[['virStream', int, _T], int], opaque: _T) -> None:
"""Send the entire data stream, reading the data from the
requested data source. This is simply a convenient
alternative to virStreamSend, for apps that do
@ -269,6 +269,7 @@
if not got:
break
assert isinstance(got, bytes)
ret = self.send(got)
if ret == -2:
raise libvirtError("cannot use sparseSendAll with "

95
libvirt-override.py
View File

@ -4,22 +4,39 @@
# On cygwin, the DLL is called cygvirtmod.dll
try:
import libvirtmod
import libvirtmod # type: ignore
except ImportError as lib_e:
try:
import cygvirtmod as libvirtmod
import cygvirtmod as libvirtmod # type: ignore
except ImportError as cyg_e:
if "No module named" in str(cyg_e):
raise lib_e
from types import TracebackType
from typing import Any, Callable, Dict, List, Optional, overload, Tuple, Type, TypeVar, Union
_T = TypeVar('_T')
_EventCB = Callable[[int, int, int, _T], None]
_EventAddHandleFunc = Callable[[int, int, _EventCB, _T], int]
_EventUpdateHandleFunc = Callable[[int, int], None]
_EventRemoveHandleFunc = Callable[[int], int]
_TimerCB = Callable[[int, _T], None]
_EventAddTimeoutFunc = Callable[[int, _TimerCB, _T], int]
_EventUpdateTimeoutFunc = Callable[[int, int], None]
_EventRemoveTimeoutFunc = Callable[[int], int]
_DomainCB = Callable[['virConnect', 'virDomain', int, int, _T], Optional[int]]
_BlkioParameter = Dict[str, Any]
_MemoryParameter = Dict[str, Any]
_SchedParameter = Dict[str, Any]
_TypedParameter = Dict[str, Any]
# The root of all libvirt errors.
class libvirtError(Exception):
def __init__(self, defmsg):
def __init__(self, defmsg: str) -> None:
# Never call virConnGetLastError().
# virGetLastError() is now thread local
err = libvirtmod.virGetLastError()
err = libvirtmod.virGetLastError() # type: Optional[Tuple[int, int, str, int, str, Optional[str], Optional[str], int, int]]
if err is None:
msg = defmsg
else:
@ -29,47 +46,47 @@ class libvirtError(Exception):
self.err = err
def get_error_code(self):
def get_error_code(self) -> Optional[int]:
if self.err is None:
return None
return self.err[0]
def get_error_domain(self):
def get_error_domain(self) -> Optional[int]:
if self.err is None:
return None
return self.err[1]
def get_error_message(self):
def get_error_message(self) -> Optional[str]:
if self.err is None:
return None
return self.err[2]
def get_error_level(self):
def get_error_level(self) -> Optional[int]:
if self.err is None:
return None
return self.err[3]
def get_str1(self):
def get_str1(self) -> Optional[str]:
if self.err is None:
return None
return self.err[4]
def get_str2(self):
def get_str2(self) -> Optional[str]:
if self.err is None:
return None
return self.err[5]
def get_str3(self):
def get_str3(self) -> Optional[str]:
if self.err is None:
return None
return self.err[6]
def get_int1(self):
def get_int1(self) -> Optional[int]:
if self.err is None:
return None
return self.err[7]
def get_int2(self):
def get_int2(self) -> Optional[int]:
if self.err is None:
return None
return self.err[8]
@ -78,7 +95,7 @@ class libvirtError(Exception):
#
# register the libvirt global error handler
#
def registerErrorHandler(f, ctx):
def registerErrorHandler(f: Callable[[_T, List], None], ctx: _T) -> int:
"""Register a Python function for error reporting.
The function is called back as f(ctx, error), with error
being a list of information about the error being raised.
@ -86,7 +103,35 @@ def registerErrorHandler(f, ctx):
return libvirtmod.virRegisterErrorHandler(f, ctx)
def openAuth(uri, auth, flags=0):
def openAuth(uri: str, auth: List, flags: int = 0) -> 'virConnect':
# TODO: The C code rquires a List and there is not *Mutable*Tuple for a better description such as
# auth: Tuple[List[int], Callable[[List[MutableTuple[int, str, str, str, Any]], _T], int], _T]
"""
This function should be called first to get a connection to the
Hypervisor. If necessary, authentication will be performed fetching
credentials via the callback.
See :py:func:`open` for notes about environment variables which can
have an effect on opening drivers and freeing the connection resources.
:param str uri: (Optional) connection URI, see https://libvirt.org/uri.html
:param auth: a list that contains 3 items:
- a list of supported credential types
- a callable that takes 2 arguments (credentials, user-data) and returns 0 on succcess and -1 on errors.
The credentials argument is a list of credentials that libvirt (actually
the ESX driver) would like to request. An element of this list is itself a
list containing 5 items (4 inputs, 1 output):
- the credential type, e.g. :py:const:`libvirt.VIR_CRED_AUTHNAME`
- a prompt to be displayed to the user
- a challenge, the ESX driver sets this to the hostname to allow automatic
distinction between requests for ESX and vCenter credentials
- a default result for the request
- a place to store the actual result for the request
- user data that will be passed to the callable as second argument
:param int flags: bitwise-OR of virConnectFlags
:returns: a :py:class:`virConnect` instance on success.
:raises libvirtError: on errors.
"""
ret = libvirtmod.virConnectOpenAuth(uri, auth, flags)
if ret is None:
raise libvirtError('virConnectOpenAuth() failed')
@ -96,7 +141,7 @@ def openAuth(uri, auth, flags=0):
#
# Return library version.
#
def getVersion(name=None):
def getVersion(name: Optional[str] = None) -> int:
"""If no name parameter is passed (or name is None) then the
version of the libvirt library is returned as an integer.
@ -120,7 +165,11 @@ def getVersion(name=None):
#
# Invoke an EventHandle callback
#
def _eventInvokeHandleCallback(watch, fd, event, opaque, opaquecompat=None):
@overload
def _eventInvokeHandleCallback(watch: int, fd: int, event: int, opaque: Tuple[_EventCB, _T], opaquecompat: None = None) -> None: ... # noqa E704
@overload # noqa F811
def _eventInvokeHandleCallback(watch: int, fd: int, event: int, opaque: _EventCB, opaquecompat: _T = None) -> None: ... # noqa E704
def _eventInvokeHandleCallback(watch: int, fd: int, event: int, opaque: Union[Tuple[_EventCB, _T], _EventCB], opaquecompat: Optional[_T] = None) -> None: # noqa F811
"""
Invoke the Event Impl Handle Callback in C
"""
@ -141,7 +190,7 @@ def _eventInvokeHandleCallback(watch, fd, event, opaque, opaquecompat=None):
#
# Invoke an EventTimeout callback
#
def _eventInvokeTimeoutCallback(timer, opaque, opaquecompat=None):
def _eventInvokeTimeoutCallback(timer: int, opaque: Union[Tuple[_TimerCB, _T], _TimerCB], opaquecompat: Optional[_T] = None) -> None:
"""
Invoke the Event Impl Timeout Callback in C
"""
@ -159,7 +208,7 @@ def _eventInvokeTimeoutCallback(timer, opaque, opaquecompat=None):
libvirtmod.virEventInvokeTimeoutCallback(timer, callback, opaque)
def _dispatchEventHandleCallback(watch, fd, events, cbData):
def _dispatchEventHandleCallback(watch: int, fd: int, events: int, cbData: Dict[str, Any]) -> int:
cb = cbData["cb"]
opaque = cbData["opaque"]
@ -167,7 +216,7 @@ def _dispatchEventHandleCallback(watch, fd, events, cbData):
return 0
def _dispatchEventTimeoutCallback(timer, cbData):
def _dispatchEventTimeoutCallback(timer: int, cbData: Dict[str, Any]) -> int:
cb = cbData["cb"]
opaque = cbData["opaque"]
@ -175,7 +224,7 @@ def _dispatchEventTimeoutCallback(timer, cbData):
return 0
def virEventAddHandle(fd, events, cb, opaque):
def virEventAddHandle(fd: int, events: int, cb: _EventCB, opaque: _T) -> int:
"""
register a callback for monitoring file handle events
@ -197,7 +246,7 @@ def virEventAddHandle(fd, events, cb, opaque):
return ret
def virEventAddTimeout(timeout, cb, opaque):
def virEventAddTimeout(timeout: int, cb: _TimerCB, opaque: _T) -> int:
"""
register a callback for a timer event
@ -223,7 +272,7 @@ def virEventAddTimeout(timeout, cb, opaque):
# a caller for the ff callbacks for custom event loop implementations
#
def virEventInvokeFreeCallback(opaque):
def virEventInvokeFreeCallback(opaque: Any) -> None:
"""
Execute callback which frees the opaque buffer

13
libvirt-qemu-override.py
View File

@ -1,7 +1,8 @@
# Manually written part of python bindings for libvirt-qemu
from typing import Any, Callable, Dict
def _dispatchQemuMonitorEventCallback(conn, dom, event, seconds, micros, details, cbData):
def _dispatchQemuMonitorEventCallback(conn: libvirt.virConnect, dom: libvirt.virDomain, event: str, seconds: int, micros: int, details: str, cbData: Dict[str, Any]) -> int:
"""Dispatches events to python user qemu monitor event callbacks
"""
cb = cbData["cb"]
@ -11,23 +12,23 @@ def _dispatchQemuMonitorEventCallback(conn, dom, event, seconds, micros, details
return 0
def qemuMonitorEventDeregister(conn, callbackID):
def qemuMonitorEventDeregister(conn: libvirt.virConnect, callbackID: int) -> None:
"""Removes a qemu monitor event callback. De-registering for a callback
will disable delivery of this event type"""
try:
ret = libvirtmod_qemu.virConnectDomainQemuMonitorEventDeregister(conn._o, callbackID)
if ret == -1:
raise libvirt.libvirtError('virConnectDomainQemuMonitorEventDeregister() failed')
del conn.qemuMonitorEventCallbackID[callbackID]
del conn.qemuMonitorEventCallbackID[callbackID] # type: ignore
except AttributeError:
pass
def qemuMonitorEventRegister(conn, dom, event, cb, opaque, flags=0):
def qemuMonitorEventRegister(conn: libvirt.virConnect, dom: libvirt.virDomain, event: str, cb: Callable[[libvirt.virConnect, libvirt.virDomain, str, int, int, str, libvirt._T], None], opaque: libvirt._T, flags: int = 0) -> int:
"""Adds a qemu monitor event callback. Registering for a monitor
callback will enable delivery of the events"""
if not hasattr(conn, 'qemuMonitorEventCallbackID'):
conn.qemuMonitorEventCallbackID = {}
conn.qemuMonitorEventCallbackID = {} # type: ignore
cbData = {"cb": cb, "conn": conn, "opaque": opaque}
if dom is None:
ret = libvirtmod_qemu.virConnectDomainQemuMonitorEventRegister(conn._o, None, event, cbData, flags)
@ -35,5 +36,5 @@ def qemuMonitorEventRegister(conn, dom, event, cb, opaque, flags=0):
ret = libvirtmod_qemu.virConnectDomainQemuMonitorEventRegister(conn._o, dom._o, event, cbData, flags)
if ret == -1:
raise libvirt.libvirtError('virConnectDomainQemuMonitorEventRegister() failed')
conn.qemuMonitorEventCallbackID[ret] = opaque
conn.qemuMonitorEventCallbackID[ret] = opaque # type: ignore
return ret