override: Add manual PEP 484 type annotations

Signed-off-by: Philipp Hahn <hahn@univention.de>

9 files changed, 173 insertions, 127 deletions

diff --git a/libvirt-override-virConnect.py b/libvirt-override-virConnect.py
index ed5f45b..255522c 100644
--- a/libvirt-override-virConnect.py
+++ b/libvirt-override-virConnect.py
@@ -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
diff --git a/libvirt-override-virDomain.py b/libvirt-override-virDomain.py
index 4e42053..aa86a00 100644
--- a/libvirt-override-virDomain.py
+++ b/libvirt-override-virDomain.py
@@ -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)
diff --git a/libvirt-override-virDomainCheckpoint.py b/libvirt-override-virDomainCheckpoint.py
index d4ccd88..8b9d212 100644
--- a/libvirt-override-virDomainCheckpoint.py
+++ b/libvirt-override-virDomainCheckpoint.py
@@ -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:
diff --git a/libvirt-override-virDomainSnapshot.py b/libvirt-override-virDomainSnapshot.py
index fd4b311..64bd319 100644
--- a/libvirt-override-virDomainSnapshot.py
+++ b/libvirt-override-virDomainSnapshot.py
@@ -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:
diff --git a/libvirt-override-virNetwork.py b/libvirt-override-virNetwork.py
index df38914..892f8e4 100644
--- a/libvirt-override-virNetwork.py
+++ b/libvirt-override-virNetwork.py
@@ -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:
diff --git a/libvirt-override-virStoragePool.py b/libvirt-override-virStoragePool.py
index 8bd5745..669df3e 100644
--- a/libvirt-override-virStoragePool.py
+++ b/libvirt-override-virStoragePool.py
@@ -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:
diff --git a/libvirt-override-virStream.py b/libvirt-override-virStream.py
index c9d70db..4aa6f54 100644
--- a/libvirt-override-virStream.py
+++ b/libvirt-override-virStream.py
@@ -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 "
diff --git a/libvirt-override.py b/libvirt-override.py
index aeeee52..e342884 100644
--- a/libvirt-override.py
+++ b/libvirt-override.py
@@ -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
 
diff --git a/libvirt-qemu-override.py b/libvirt-qemu-override.py
index cba96a1..fae1db7 100644
--- a/libvirt-qemu-override.py
+++ b/libvirt-qemu-override.py
@@ -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