Refactoring server/inprocserver.py, server/localserver.py and `cl…

…ient/_events.py`. (#815)

* Replace `ctypes.byref` with `byref` in `client/_events.py`.

* Replace `import comtypes.typeinfo` with specific imports in `client/_events.py`.

* Replace `comtypes.COMObject` and `comtypes.IUnknown` with specific imports in `server/localserver.py`.

* Replace `import ctypes` with specific imports in `server/inprocserver.py`.

* Remove assignment to unused variable `res` from `PumpEvents`.

* Convert to `not in` from `not ... in` in `inproc_find_class`.

* Add type assertions.

comtypes/client/_events.py

@@ -2,7 +2,7 @@
 import logging
 import traceback
 from _ctypes import COMError
-from ctypes import HRESULT, POINTER, WINFUNCTYPE, OleDLL, Structure, WinDLL
+from ctypes import HRESULT, POINTER, WINFUNCTYPE, OleDLL, Structure, WinDLL, byref
 from ctypes.wintypes import (
     BOOL,
     DWORD,
@@ -17,12 +17,12 @@
 from typing import Union as _UnionT
 
 import comtypes
-import comtypes.typeinfo
 from comtypes import COMObject, IUnknown
 from comtypes._comobject import _MethodFinder
 from comtypes.automation import DISPATCH_METHOD, IDispatch
 from comtypes.client._generate import GetModule
 from comtypes.connectionpoints import IConnectionPoint, IConnectionPointContainer
+from comtypes.typeinfo import IProvideClassInfo2
 
 logger = logging.getLogger(__name__)
 
@@ -82,7 +82,7 @@ def _connect(
         self, source: IUnknown, interface: Type[IUnknown], receiver: _ReceiverType
     ) -> None:
         cpc = source.QueryInterface(IConnectionPointContainer)
-        self.cp = cpc.FindConnectionPoint(ctypes.byref(interface._iid_))
+        self.cp = cpc.FindConnectionPoint(byref(interface._iid_))
         logger.debug("Start advise %s", interface)
         # Since `POINTER(IUnknown).from_param`(`_compointer_base.from_param`)
         # can accept a `COMObject` instance, `IConnectionPoint.Advise` can
@@ -114,7 +114,7 @@ def FindOutgoingInterface(source: IUnknown) -> Type[IUnknown]:
     # If the COM object implements IProvideClassInfo2, it is easy to
     # find the default outgoing interface.
     try:
-        pci = source.QueryInterface(comtypes.typeinfo.IProvideClassInfo2)
+        pci = source.QueryInterface(IProvideClassInfo2)
         guid = pci.GetGUID(1)
     except COMError:
         pass
@@ -359,12 +359,12 @@ def HandlerRoutine(dwCtrlType):
 
     try:
         try:
-            res = _CoWaitForMultipleHandles(
+            _CoWaitForMultipleHandles(
                 0,
                 int(timeout * 1000),
                 len(handles),
                 handles,
-                ctypes.byref(ctypes.c_ulong()),
+                byref(ctypes.c_ulong()),
             )
         except WindowsError as details:
             if details.winerror != RPC_S_CALLPENDING:  # timeout expired

comtypes/server/inprocserver.py

@@ -1,7 +1,7 @@
-import ctypes
 import logging
 import sys
 import winreg
+from ctypes import c_void_p, pointer
 from typing import TYPE_CHECKING, Any, Literal, Optional, Type
 
 from comtypes import GUID, COMObject, IUnknown, hresult
@@ -29,14 +29,15 @@ def IClassFactory_CreateInstance(
         this: Any,
         punkOuter: Optional[Type["_Pointer[IUnknown]"]],
         riid: "_Pointer[GUID]",
-        ppv: ctypes.c_void_p,
+        ppv: c_void_p,
     ) -> int:
         _debug("ClassFactory.CreateInstance(%s)", riid[0])
         result = self._cls().IUnknown_QueryInterface(None, riid, ppv)
         _debug("CreateInstance() -> %s", result)
         return result
 
     def IClassFactory_LockServer(self, this: Any, fLock: bool) -> Literal[0]:
+        assert COMObject.__server__ is not None, "The inprocserver is not running yet"
         if fLock:
             COMObject.__server__.Lock()
         else:
@@ -58,7 +59,7 @@ def inproc_find_class(clsid: GUID) -> Type[COMObject]:
     except:
         _debug("NO path to insert")
     else:
-        if not pathdir in sys.path:
+        if pathdir not in sys.path:
             sys.path.insert(0, str(pathdir))
             _debug("insert path %r", pathdir)
         else:
@@ -138,7 +139,7 @@ def DllGetClassObject(rclsid: int, riid: int, ppv: int) -> int:
             return hresult.CLASS_E_CLASSNOTAVAILABLE
 
         result = ClassFactory(cls).IUnknown_QueryInterface(
-            None, ctypes.pointer(iid), ctypes.c_void_p(ppv)
+            None, pointer(iid), c_void_p(ppv)
         )
         _debug("DllGetClassObject() -> %s", result)
         return result

comtypes/server/localserver.py

@@ -5,7 +5,7 @@
 from typing import TYPE_CHECKING, Any, Literal, Optional, Sequence, Type
 
 import comtypes
-from comtypes import GUID, hresult
+from comtypes import GUID, COMObject, IUnknown, hresult
 from comtypes.server import IClassFactory
 
 if TYPE_CHECKING:
@@ -33,18 +33,18 @@
 _CoRevokeClassObject.restype = HRESULT
 
 
-def run(classes: Sequence[Type[comtypes.COMObject]]) -> None:
+def run(classes: Sequence[Type[COMObject]]) -> None:
     classobjects = [ClassFactory(cls) for cls in classes]
-    comtypes.COMObject.__run_localserver__(classobjects)
+    COMObject.__run_localserver__(classobjects)
 
 
-class ClassFactory(comtypes.COMObject):
+class ClassFactory(COMObject):
     _com_interfaces_ = [IClassFactory]
     _locks: int = 0
     _queue: Optional[queue.Queue] = None
     regcls: int = REGCLS_MULTIPLEUSE
 
-    def __init__(self, cls: Type[comtypes.COMObject], *args, **kw) -> None:
+    def __init__(self, cls: Type[COMObject], *args, **kw) -> None:
         super(ClassFactory, self).__init__()
         self._cls = cls
         self._register_class()
@@ -60,11 +60,11 @@ def IUnknown_Release(self, this: Any) -> int:
     def _register_class(self) -> None:
         regcls = getattr(self._cls, "_regcls_", self.regcls)
         cookie = c_ulong()
-        ptr = self._com_pointers_[comtypes.IUnknown._iid_]
+        ptr = self._com_pointers_[IUnknown._iid_]
         clsctx = self._cls._reg_clsctx_
         clsctx &= ~comtypes.CLSCTX_INPROC  # reset the inproc flags
         _CoRegisterClassObject(
-            byref(comtypes.GUID(self._cls._reg_clsid_)),
+            byref(GUID(self._cls._reg_clsid_)),
             ptr,
             clsctx,
             regcls,
@@ -78,8 +78,8 @@ def _revoke_class(self) -> None:
     def CreateInstance(
         self,
         this: Any,
-        punkOuter: Optional[Type["_Pointer[comtypes.IUnknown]"]],
-        riid: "_Pointer[comtypes.GUID]",
+        punkOuter: Optional[Type["_Pointer[IUnknown]"]],
+        riid: "_Pointer[GUID]",
         ppv: c_void_p,
     ) -> int:
         _debug("ClassFactory.CreateInstance(%s)", riid[0])
@@ -89,8 +89,9 @@ def CreateInstance(
         return result
 
     def LockServer(self, this: Any, fLock: bool) -> Literal[0]:
+        assert COMObject.__server__ is not None, "The localserver is not running yet"
         if fLock:
-            comtypes.COMObject.__server__.Lock()
+            COMObject.__server__.Lock()
         else:
-            comtypes.COMObject.__server__.Unlock()
+            COMObject.__server__.Unlock()
         return hresult.S_OK

pyproject.toml

@@ -16,10 +16,9 @@ ignore = ["E402"]
 "comtypes/util.py" = ["F403", "F405"]
 "comtypes/viewobject.py" = ["F403", "F405"]
 "comtypes/client/_constants.py" = ["F401"]
-"comtypes/client/_events.py" = ["F841"]
 "comtypes/server/automation.py" = ["F403", "F405"]
 "comtypes/server/connectionpoints.py" = ["F401", "F403", "F405"]
-"comtypes/server/inprocserver.py" = ["E713", "E722", "F841"]
+"comtypes/server/inprocserver.py" = ["E722", "F841"]
 "comtypes/server/register.py" = ["E713"]
 "comtypes/tools/codegenerator/packing.py" = ["F821", "F841"]
 "comtypes/tools/typedesc.py" = ["F403", "F405"]