Coverage for src / ezxl / utils / _com_utils.py: 0.00%

1# /////////////////////////////////////////////////////////////// 

2# _com_utils - Internal COM utility functions 

3# Project: EzXl 

4# /////////////////////////////////////////////////////////////// 

5 

6""" 

7Internal COM utility functions. 

8 

9Low-level helpers shared across the COM proxy layer. Not part of the 

10public API — do not import from outside the ezxl package. 

11""" 

12 

13from __future__ import annotations 

14 

15# /////////////////////////////////////////////////////////////// 

16# IMPORTS 

17# /////////////////////////////////////////////////////////////// 

18# Standard library imports 

19import threading 

20import time 

21from collections.abc import Callable 

22from functools import wraps 

23from typing import Any, cast 

24 

25# Third-party imports 

26from ezplog.lib_mode import get_logger 

27 

28# Local imports 

29from ..exceptions import ( 

30 COMOperationError, 

31 ExcelSessionLostError, 

32 ExcelThreadViolationError, 

33) 

34from ._pywintypes_compat import COM_EXCEPTION_TYPE 

35 

36# /////////////////////////////////////////////////////////////// 

37# CONSTANTS 

38# /////////////////////////////////////////////////////////////// 

39 

40logger = get_logger(__name__) 

41 

42# COM HRESULT codes that indicate a dead/disconnected server. 

43# RPC_E_DISCONNECTED (0x80010108) and RPC_E_SERVER_DIED (0x80010007). 

44_DISCONNECTED_HRESULTS: frozenset[int] = frozenset( 

45 [ 

46 0x80010108, # RPC_E_DISCONNECTED 

47 0x80010007, # RPC_E_SERVER_DIED 

48 0x800706BA, # RPC_S_SERVER_UNAVAILABLE 

49 ] 

50) 

51 

52# /////////////////////////////////////////////////////////////// 

53# FUNCTIONS 

54# /////////////////////////////////////////////////////////////// 

55 

56 

57def wait_until_ready(xl_app: Any, timeout: float = 30.0) -> None: 

58 """Poll ``xl_app.Ready`` until Excel reports it is ready or the timeout expires. 

59 

60 Excel's ``Application.Ready`` property returns ``False`` while Excel is 

61 performing an operation (e.g., opening a file, calculating). This helper 

62 blocks the calling thread with short sleeps until Excel is available again. 

63 

64 Args: 

65 xl_app: A live COM ``Application`` object (``win32com.client`` dispatch). 

66 timeout: Maximum number of seconds to wait before raising. Defaults to 30. 

67 

68 Raises: 

69 COMOperationError: If ``timeout`` is exceeded before Excel becomes ready. 

70 

71 Example: 

72 >>> wait_until_ready(xl_app, timeout=60.0) 

73 """ 

74 deadline = time.monotonic() + timeout 

75 poll_interval = 0.25 # seconds between each Ready check 

76 

77 while time.monotonic() < deadline: 

78 try: 

79 if xl_app.Ready: 

80 return 

81 except Exception as exc: 

82 # Ready property itself can throw when Excel is in a transient state. 

83 # Log at DEBUG and retry until the deadline. 

84 logger.debug("wait_until_ready: Ready check raised (will retry): %s", exc) 

85 time.sleep(poll_interval) 

86 

87 raise COMOperationError(f"Excel did not become ready within {timeout:.1f} seconds.") 

88 

89 

90def wrap_com_error[**FuncP, FuncR]( 

91 func: Callable[FuncP, FuncR], 

92) -> Callable[FuncP, FuncR]: 

93 """Decorator that intercepts ``pywintypes.com_error`` and re-raises as EzXl exceptions. 

94 

95 Wraps any COM boundary function so that raw pywin32 errors never escape 

96 the ``ezxl`` package. Disconnection HRESULTs are mapped to 

97 ``ExcelSessionLostError``; all other COM errors become ``COMOperationError``. 

98 

99 Args: 

100 func: The callable to wrap. 

101 

102 Returns: 

103 The wrapped callable with identical signature. 

104 

105 Example: 

106 >>> @wrap_com_error 

107 ... def open_workbook(xl_app, path): 

108 ... return xl_app.Workbooks.Open(path) 

109 """ 

110 

111 @wraps(func) 

112 def _wrapper(*args: FuncP.args, **kwargs: FuncP.kwargs) -> FuncR: 

113 try: 

114 return func(*args, **kwargs) 

115 except Exception as exc: 

116 if isinstance(exc, COM_EXCEPTION_TYPE): 

117 first_arg: object = exc.args[0] if exc.args else 0 

118 # pywin32 delivers the HRESULT as a *signed* 32-bit int 

119 # (RPC_E_DISCONNECTED arrives as -2147417848), while the 

120 # constants below are unsigned. Mask before comparing, or the 

121 # lookup never matches and the hex formatting prints garbage. 

122 hresult: int = ( 

123 (first_arg & 0xFFFFFFFF) if isinstance(first_arg, int) else 0 

124 ) 

125 if hresult in _DISCONNECTED_HRESULTS: 

126 raise ExcelSessionLostError( 

127 f"Excel COM session lost (HRESULT 0x{hresult:08X}): {exc}", 

128 cause=exc, 

129 ) from exc 

130 raise COMOperationError( 

131 f"COM operation failed (HRESULT 0x{hresult:08X}): {exc}", 

132 cause=exc, 

133 ) from exc 

134 # Re-raise non-COM exceptions untouched. 

135 raise 

136 

137 return cast(Callable[FuncP, FuncR], _wrapper) 

138 

139 

140def assert_main_thread(thread_id: int) -> None: 

141 """Assert that the current thread matches the expected COM apartment thread. 

142 

143 COM STA (Single-Threaded Apartment) requires that all calls on a COM object 

144 are made from the thread that created it. This function raises proactively 

145 before reaching the COM dispatcher, giving callers a clear diagnostic. 

146 

147 Args: 

148 thread_id: The thread identifier recorded at ``ExcelApp`` construction time. 

149 

150 Raises: 

151 ExcelThreadViolationError: If the calling thread differs from ``thread_id``. 

152 

153 Example: 

154 >>> assert_main_thread(self._thread_id) 

155 """ 

156 current = threading.get_ident() 

157 if current != thread_id: 

158 raise ExcelThreadViolationError( 

159 f"COM call attempted from thread {current}, but ExcelApp was created " 

160 f"on thread {thread_id}. Excel COM uses STA — all calls must originate " 

161 f"from the creating thread." 

162 )