Coverage for Users / vladimirpavlov / PycharmProjects / parameterizable / src / mixinforge / guarded_init_metaclass.py: 98%

1"""Metaclass for strict initialization control and lifecycle hooks. 

2 

3This module provides GuardedInitMeta, a metaclass that enforces a strict 

4initialization contract where _init_finished is set to True only after 

5complete initialization. It wraps __setstate__ to ensure proper state 

6restoration during unpickling and provides hooks for post-initialization 

7and post-unpickling tasks. 

8""" 

9import functools 

10from abc import ABCMeta 

11from dataclasses import is_dataclass 

12from typing import Any, Type, TypeVar 

13 

14T = TypeVar('T') 

15 

16 

17def _validate_pickle_state_integrity(state: Any, cls_name: str) -> None: 

18 """Ensure pickled state does not claim initialization is finished. 

19 

20 Args: 

21 state: The pickle state to validate. 

22 cls_name: Class name for error reporting. 

23 

24 Raises: 

25 RuntimeError: If _init_finished is True in the pickled state. 

26 """ 

27 candidate_dict, _ = _parse_pickle_state(state, cls_name) 

28 

29 if candidate_dict is not None and candidate_dict.get("_init_finished") is True: 

30 raise RuntimeError( 

31 f"{cls_name} must not be pickled with _init_finished=True") 

32 

33 

34def _parse_pickle_state(state: Any, cls_name: str) -> tuple[dict | None, dict | None]: 

35 """Extract __dict__ and __slots__ state from pickle data. 

36 

37 Args: 

38 state: The state object passed to __setstate__. 

39 cls_name: Class name for error reporting. 

40 

41 Returns: 

42 A tuple (dict_state, slots_state) where each element is a dictionary or None. 

43 

44 Raises: 

45 RuntimeError: If state format is unsupported. 

46 """ 

47 if state is None: 

48 return None, None 

49 elif isinstance(state, dict): 

50 return state, None 

51 elif (isinstance(state, tuple) and len(state) == 2 

52 and (state[0] is None or isinstance(state[0], dict)) 

53 and (state[1] is None or isinstance(state[1], dict))): 

54 return state 

55 else: 

56 raise RuntimeError( 

57 f"Unsupported pickle state for {cls_name}: {state!r}") 

58 

59 

60def _restore_dict_state(instance: Any, state_dict: dict, cls_name: str) -> None: 

61 """Update instance __dict__ with restored state. 

62 

63 Args: 

64 instance: The object instance being restored. 

65 state_dict: Dictionary of attribute values to restore. 

66 cls_name: Class name for error reporting. 

67 

68 Raises: 

69 RuntimeError: If instance has no __dict__ attribute. 

70 """ 

71 if hasattr(instance, "__dict__"): 

72 instance.__dict__.update(state_dict) 

73 else: 

74 raise RuntimeError( 

75 f"Cannot restore pickle state for {cls_name}: " 

76 f"instance has no __dict__ but state contains a dictionary.") 

77 

78 

79def _restore_slots_state(instance: Any, state_slots: dict[str,Any]) -> None: 

80 """Restore slot values using setattr. 

81 

82 Args: 

83 instance: The object instance being restored. 

84 state_slots: Dictionary mapping slot names to values. 

85 """ 

86 for key, value in state_slots.items(): 

87 setattr(instance, key, value) 

88 

89 

90def _invoke_post_setstate_hook(instance: Any) -> None: 

91 """Execute __post_setstate__ hook if defined. 

92 

93 Args: 

94 instance: The object instance to invoke the hook on. 

95 

96 Raises: 

97 TypeError: If __post_setstate__ is not callable. 

98 """ 

99 post_setstate = getattr(instance, "__post_setstate__", None) 

100 if post_setstate: 

101 if not callable(post_setstate): 

102 raise TypeError(f"__post_setstate__ must be callable, " 

103 f"got {instance.__post_setstate__!r}") 

104 try: 

105 post_setstate() 

106 except Exception as e: 

107 _re_raise_with_context("__post_setstate__", e) 

108 

109 

110class GuardedInitMeta(ABCMeta): 

111 """Metaclass for strict initialization control and lifecycle hooks. 

112 

113 Enforces a contract where _init_finished is False during initialization 

114 and only becomes True after all initialization code completes. This ensures 

115 that properties and methods can reliably check initialization state. 

116 

117 The metaclass automatically wraps __setstate__ to maintain the same 

118 contract during unpickling, and invokes __post_init__ and __post_setstate__ 

119 hooks when defined. 

120 

121 Contract: 

122 - __init__ must set self._init_finished = False immediately. 

123 - The metaclass sets self._init_finished = True after __init__ returns 

124 (but before __post_init__, if defined). 

125 - __setstate__ is wrapped to ensure _init_finished becomes True after 

126 full state restoration (but before __post_setstate__, if defined). 

127 """ 

128 

129 def __init__(cls, name, bases, dct): 

130 """Initialize the class and inject lifecycle enforcement. 

131 

132 Wraps __setstate__ to ensure proper initialization state after unpickling 

133 and validates that the class is compatible with the GuardedInitMeta contract. 

134 

135 Args: 

136 name: The class name. 

137 bases: Base classes. 

138 dct: Class dictionary. 

139 

140 Raises: 

141 TypeError: If class is a dataclass or has multiple GuardedInitMeta bases. 

142 """ 

143 super().__init__(name, bases, dct) 

144 _raise_if_dataclass(cls) 

145 

146 n_guarded_bases = sum(1 for base in bases if isinstance(base, GuardedInitMeta)) 

147 if n_guarded_bases > 1: 

148 raise TypeError(f"Class {name} has {n_guarded_bases} GuardedInitMeta bases, " 

149 "but only 1 is allowed.") 

150 

151 if '__setstate__' in dct: 

152 original_setstate = dct['__setstate__'] 

153 elif getattr(cls, '__setstate__', None) is not None: 

154 inherited = getattr(cls, '__setstate__') 

155 if getattr(inherited, "__guarded_init_meta_wrapped__", False): 

156 return 

157 original_setstate = inherited 

158 else: 

159 original_setstate = None 

160 

161 def setstate_wrapper(self, state): 

162 """Restore state, finalize initialization, and invoke hook.""" 

163 _validate_pickle_state_integrity(state, type(self).__name__) 

164 

165 if original_setstate is not None: 

166 original_setstate(self, state) 

167 else: 

168 state_dict, state_slots = _parse_pickle_state(state, type(self).__name__) 

169 

170 if state_dict is not None: 

171 _restore_dict_state(self, state_dict, type(self).__name__) 

172 

173 if state_slots is not None: 

174 _restore_slots_state(self, state_slots) 

175 

176 if isinstance(self, cls): 

177 self._init_finished = True 

178 _invoke_post_setstate_hook(self) 

179 

180 if original_setstate: 

181 setstate_wrapper = functools.wraps(original_setstate)(setstate_wrapper) 

182 

183 setstate_wrapper.__guarded_init_meta_wrapped__ = True 

184 setstate_wrapper.__name__ = '__setstate__' 

185 setattr(cls, '__setstate__', setstate_wrapper) 

186 

187 def __call__(cls: Type[T], *args: Any, **kwargs: Any) -> T: 

188 """Create instance, enforce initialization contract, and invoke hook. 

189 

190 Ensures _init_finished is False during __init__, sets it to True afterward, 

191 and invokes __post_init__ if defined. 

192 

193 Args: 

194 *args: Positional arguments for __init__. 

195 **kwargs: Keyword arguments for __init__. 

196 

197 Returns: 

198 The initialized instance. 

199 

200 Raises: 

201 RuntimeError: If _init_finished is not False after __init__. 

202 TypeError: If __post_init__ is not callable. 

203 """ 

204 _raise_if_dataclass(cls) 

205 

206 instance = super().__call__(*args, **kwargs) 

207 if not isinstance(instance, cls): 

208 return instance 

209 

210 if not hasattr(instance, '_init_finished') or instance._init_finished: 

211 raise RuntimeError(f"Class {cls.__name__} must set attribute " 

212 "_init_finished to False in __init__") 

213 

214 instance._init_finished = True 

215 

216 post_init = getattr(instance, "__post_init__", None) 

217 if post_init: 

218 if not callable(post_init): 

219 raise TypeError(f"__post_init__ must be callable, " 

220 f"got {instance.__post_init__!r}") 

221 try: 

222 post_init() 

223 except Exception as e: 

224 _re_raise_with_context("__post_init__", e) 

225 

226 return instance 

227 

228 

229def _re_raise_with_context(hook_name: str, exc: Exception) -> None: 

230 """Re-raise an exception with added context about the hook. 

231 

232 Args: 

233 hook_name: The hook name where the error occurred (e.g., "__post_init__"). 

234 exc: The original exception caught during hook execution. 

235 

236 Raises: 

237 RuntimeError: If the exception constructor is incompatible. 

238 Exception: The augmented exception with added context. 

239 """ 

240 try: 

241 new_exc = type(exc)(f"Error in {hook_name}: {exc}") 

242 except Exception: 

243 raise RuntimeError( 

244 f"Error in {hook_name} (original error: {type(exc).__name__}: {exc})" 

245 ) from exc 

246 

247 raise new_exc from exc 

248 

249 

250def _raise_if_dataclass(cls: Type) -> None: 

251 """Forbid GuardedInitMeta on dataclasses due to incompatible lifecycle. 

252 

253 This check runs in two places: 

254 1. In GuardedInitMeta.__init__ - catches inheritance from dataclasses. 

255 2. In GuardedInitMeta.__call__ - catches @dataclass decorator on the class itself. 

256 

257 Args: 

258 cls: The class to check. 

259 

260 Raises: 

261 TypeError: If the class is a dataclass. 

262 """ 

263 if is_dataclass(cls): 

264 raise TypeError( 

265 f"GuardedInitMeta cannot be used with dataclass class {cls.__name__} " 

266 "because dataclasses already manage __post_init__ with different " 

267 "object lifecycle assumptions.")