Update

Author: junkmd

comtypes/client/_events.py

@@ -12,12 +12,14 @@
     LPVOID,
     ULONG,
 )
+from typing import Any, Callable, Optional, Protocol, Type, TypeVar
 
 import comtypes
 import comtypes.automation
 import comtypes.connectionpoints
 import comtypes.hresult
 import comtypes.typeinfo
+from comtypes import COMObject, IUnknown
 from comtypes.client._generate import GetModule
 
 logger = logging.getLogger(__name__)
@@ -58,29 +60,46 @@ class SECURITY_ATTRIBUTES(Structure):
 _CloseHandle.restype = BOOL
 
 
+_T_IUnknown = TypeVar("_T_IUnknown", bound=IUnknown)
+
+
+class _SupportsQueryInterface(Protocol):
+    def QueryInterface(self, interface: Type[_T_IUnknown]) -> _T_IUnknown: ...
+
+
 class _AdviseConnection(object):
-    def __init__(self, source, interface, receiver):
+    def __init__(
+        self,
+        source: IUnknown,
+        interface: Type[IUnknown],
+        receiver: _SupportsQueryInterface,
+    ) -> None:
         self.cp = None
         self.cookie = None
         self.receiver = None
         self._connect(source, interface, receiver)
 
-    def _connect(self, source, interface, receiver):
+    def _connect(
+        self,
+        source: IUnknown,
+        interface: Type[IUnknown],
+        receiver: _SupportsQueryInterface,
+    ) -> None:
         cpc = source.QueryInterface(comtypes.connectionpoints.IConnectionPointContainer)
         self.cp = cpc.FindConnectionPoint(ctypes.byref(interface._iid_))
         logger.debug("Start advise %s", interface)
         self.cookie = self.cp.Advise(receiver)
         self.receiver = receiver
 
-    def disconnect(self):
+    def disconnect(self) -> None:
         if self.cookie:
             self.cp.Unadvise(self.cookie)
             logger.debug("Unadvised %s", self.cp)
             self.cp = None
             self.cookie = None
             del self.receiver
 
-    def __del__(self):
+    def __del__(self) -> None:
         try:
             if self.cookie is not None:
                 self.cp.Unadvise(self.cookie)
@@ -89,7 +108,7 @@ def __del__(self):
             pass
 
 
-def FindOutgoingInterface(source):
+def FindOutgoingInterface(source: IUnknown) -> Type[IUnknown]:
     """XXX Describe the strategy that is used..."""
     # If the COM object implements IProvideClassInfo2, it is easy to
     # find the default outgoing interface.
@@ -129,7 +148,7 @@ def FindOutgoingInterface(source):
     raise TypeError("cannot determine source interface")
 
 
-def find_single_connection_interface(source):
+def find_single_connection_interface(source: IUnknown) -> Optional[Type[IUnknown]]:
     # Enumerate the connection interfaces.  If we find a single one,
     # return it, if there are more, we give up since we cannot
     # determine which one to use.
@@ -187,11 +206,11 @@ class _SinkMethodFinder(_MethodFinder):
     event handlers.
     """
 
-    def __init__(self, inst, sink):
+    def __init__(self, inst: COMObject, sink: Any) -> None:
         super(_SinkMethodFinder, self).__init__(inst)
         self.sink = sink
 
-    def find_method(self, fq_name, mthname):
+    def find_method(self, fq_name: str, mthname: str) -> Callable[..., Any]:
         impl = self._find_method(fq_name, mthname)
         # Caller of this method catches AttributeError,
         # so we need to be careful in the following code
@@ -206,7 +225,7 @@ def find_method(self, fq_name, mthname):
         except AttributeError as details:
             raise RuntimeError(details)
 
-    def _find_method(self, fq_name, mthname):
+    def _find_method(self, fq_name: str, mthname: str) -> Callable[..., Any]:
         try:
             return super(_SinkMethodFinder, self).find_method(fq_name, mthname)
         except AttributeError:
@@ -216,7 +235,7 @@ def _find_method(self, fq_name, mthname):
                 return getattr(self.sink, mthname)
 
 
-def CreateEventReceiver(interface, handler):
+def CreateEventReceiver(interface: Type[IUnknown], handler: Any) -> COMObject:
     class Sink(comtypes.COMObject):
         _com_interfaces_ = [interface]