Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/wrapt/proxies.py: 17%

1"""Variants of ObjectProxy for different use cases.""" 

2 

3from collections.abc import Callable 

4from types import ModuleType 

5 

6from .__wrapt__ import BaseObjectProxy 

7from .synchronization import synchronized 

8 

9# Define ObjectProxy which for compatibility adds `__iter__()` support which 

10# has been removed from `BaseObjectProxy`. 

11 

12 

13class ObjectProxy(BaseObjectProxy): 

14 """A generic object proxy which forwards special methods as needed. 

15 For backwards compatibility this class adds support for `__iter__()`. If 

16 you don't need backward compatibility for `__iter__()` support then it is 

17 preferable to use `BaseObjectProxy` directly. If you want automatic 

18 support for special dunder methods for callables, iterators, and async, 

19 then use `AutoObjectProxy`.""" 

20 

21 @property 

22 def __object_proxy__(self): 

23 return ObjectProxy 

24 

25 def __new__(cls, *args, **kwargs): 

26 return super().__new__(cls) 

27 

28 def __iter__(self): 

29 return iter(self.__wrapped__) 

30 

31 

32# Define variant of ObjectProxy which can automatically adjust to the wrapped 

33# object and add special dunder methods. 

34 

35 

36def __wrapper_call__(*args, **kwargs): 

37 def _unpack_self(self, *args): 

38 return self, args 

39 

40 self, args = _unpack_self(*args) 

41 

42 return self.__wrapped__(*args, **kwargs) 

43 

44 

45def __wrapper_iter__(self): 

46 return iter(self.__wrapped__) 

47 

48 

49def __wrapper_next__(self): 

50 return self.__wrapped__.__next__() 

51 

52 

53def __wrapper_aiter__(self): 

54 return self.__wrapped__.__aiter__() 

55 

56 

57async def __wrapper_anext__(self): 

58 return await self.__wrapped__.__anext__() 

59 

60 

61def __wrapper_length_hint__(self): 

62 return self.__wrapped__.__length_hint__() 

63 

64 

65def __wrapper_await__(self): 

66 return (yield from self.__wrapped__.__await__()) 

67 

68 

69def __wrapper_get__(self, instance, owner): 

70 return self.__wrapped__.__get__(instance, owner) 

71 

72 

73def __wrapper_set__(self, instance, value): 

74 return self.__wrapped__.__set__(instance, value) 

75 

76 

77def __wrapper_delete__(self, instance): 

78 return self.__wrapped__.__delete__(instance) 

79 

80 

81def __wrapper_set_name__(self, owner, name): 

82 return self.__wrapped__.__set_name__(owner, name) 

83 

84 

85class AutoObjectProxy(BaseObjectProxy): 

86 """An object proxy which can automatically adjust to the wrapped object 

87 and add special dunder methods as needed. Note that this creates a new 

88 class for each instance, so it has much higher memory overhead than using 

89 `BaseObjectProxy` directly. If you know what special dunder methods you need 

90 then it is preferable to use `BaseObjectProxy` directly and add them to a 

91 subclass as needed. If you only need `__iter__()` support for backwards 

92 compatibility then use `ObjectProxy` instead. 

93 """ 

94 

95 def __new__(cls, wrapped): 

96 """Injects special dunder methods into a dynamically created subclass 

97 as needed based on the wrapped object. 

98 """ 

99 

100 namespace = {} 

101 

102 wrapped_attrs = dir(wrapped) 

103 class_attrs = set(dir(cls)) 

104 

105 if callable(wrapped) and "__call__" not in class_attrs: 

106 namespace["__call__"] = __wrapper_call__ 

107 

108 if "__iter__" in wrapped_attrs and "__iter__" not in class_attrs: 

109 namespace["__iter__"] = __wrapper_iter__ 

110 

111 if "__next__" in wrapped_attrs and "__next__" not in class_attrs: 

112 namespace["__next__"] = __wrapper_next__ 

113 

114 if "__aiter__" in wrapped_attrs and "__aiter__" not in class_attrs: 

115 namespace["__aiter__"] = __wrapper_aiter__ 

116 

117 if "__anext__" in wrapped_attrs and "__anext__" not in class_attrs: 

118 namespace["__anext__"] = __wrapper_anext__ 

119 

120 if "__length_hint__" in wrapped_attrs and "__length_hint__" not in class_attrs: 

121 namespace["__length_hint__"] = __wrapper_length_hint__ 

122 

123 # Note that not providing compatibility with generator-based coroutines 

124 # (PEP 342) here as they are removed in Python 3.11+ and were deprecated 

125 # in 3.8. 

126 

127 if "__await__" in wrapped_attrs and "__await__" not in class_attrs: 

128 namespace["__await__"] = __wrapper_await__ 

129 

130 if "__get__" in wrapped_attrs and "__get__" not in class_attrs: 

131 namespace["__get__"] = __wrapper_get__ 

132 

133 if "__set__" in wrapped_attrs and "__set__" not in class_attrs: 

134 namespace["__set__"] = __wrapper_set__ 

135 

136 if "__delete__" in wrapped_attrs and "__delete__" not in class_attrs: 

137 namespace["__delete__"] = __wrapper_delete__ 

138 

139 if "__set_name__" in wrapped_attrs and "__set_name__" not in class_attrs: 

140 namespace["__set_name__"] = __wrapper_set_name__ 

141 

142 name = cls.__name__ 

143 

144 if cls is AutoObjectProxy: 

145 name = BaseObjectProxy.__name__ 

146 

147 # Explicit class in super() is required here to ensure __new__ 

148 # is called on the parent of AutoObjectProxy, not the dynamically 

149 # created subclass. 

150 return super(AutoObjectProxy, cls).__new__(type(name, (cls,), namespace)) 

151 

152 def __wrapped_setattr_fixups__(self): 

153 """Adjusts special dunder methods on the class as needed based on the 

154 wrapped object, when `__wrapped__` is changed. 

155 """ 

156 

157 cls = type(self) 

158 class_attrs = set(dir(cls)) 

159 

160 if callable(self.__wrapped__): 

161 if "__call__" not in class_attrs: 

162 cls.__call__ = __wrapper_call__ 

163 elif getattr(cls, "__call__", None) is __wrapper_call__: 

164 delattr(cls, "__call__") 

165 

166 if hasattr(self.__wrapped__, "__iter__"): 

167 if "__iter__" not in class_attrs: 

168 cls.__iter__ = __wrapper_iter__ 

169 elif getattr(cls, "__iter__", None) is __wrapper_iter__: 

170 delattr(cls, "__iter__") 

171 

172 if hasattr(self.__wrapped__, "__next__"): 

173 if "__next__" not in class_attrs: 

174 cls.__next__ = __wrapper_next__ 

175 elif getattr(cls, "__next__", None) is __wrapper_next__: 

176 delattr(cls, "__next__") 

177 

178 if hasattr(self.__wrapped__, "__aiter__"): 

179 if "__aiter__" not in class_attrs: 

180 cls.__aiter__ = __wrapper_aiter__ 

181 elif getattr(cls, "__aiter__", None) is __wrapper_aiter__: 

182 delattr(cls, "__aiter__") 

183 

184 if hasattr(self.__wrapped__, "__anext__"): 

185 if "__anext__" not in class_attrs: 

186 cls.__anext__ = __wrapper_anext__ 

187 elif getattr(cls, "__anext__", None) is __wrapper_anext__: 

188 delattr(cls, "__anext__") 

189 

190 if hasattr(self.__wrapped__, "__length_hint__"): 

191 if "__length_hint__" not in class_attrs: 

192 cls.__length_hint__ = __wrapper_length_hint__ 

193 elif getattr(cls, "__length_hint__", None) is __wrapper_length_hint__: 

194 delattr(cls, "__length_hint__") 

195 

196 if hasattr(self.__wrapped__, "__await__"): 

197 if "__await__" not in class_attrs: 

198 cls.__await__ = __wrapper_await__ 

199 elif getattr(cls, "__await__", None) is __wrapper_await__: 

200 delattr(cls, "__await__") 

201 

202 if hasattr(self.__wrapped__, "__get__"): 

203 if "__get__" not in class_attrs: 

204 cls.__get__ = __wrapper_get__ 

205 elif getattr(cls, "__get__", None) is __wrapper_get__: 

206 delattr(cls, "__get__") 

207 

208 if hasattr(self.__wrapped__, "__set__"): 

209 if "__set__" not in class_attrs: 

210 cls.__set__ = __wrapper_set__ 

211 elif getattr(cls, "__set__", None) is __wrapper_set__: 

212 delattr(cls, "__set__") 

213 

214 if hasattr(self.__wrapped__, "__delete__"): 

215 if "__delete__" not in class_attrs: 

216 cls.__delete__ = __wrapper_delete__ 

217 elif getattr(cls, "__delete__", None) is __wrapper_delete__: 

218 delattr(cls, "__delete__") 

219 

220 if hasattr(self.__wrapped__, "__set_name__"): 

221 if "__set_name__" not in class_attrs: 

222 cls.__set_name__ = __wrapper_set_name__ 

223 elif getattr(cls, "__set_name__", None) is __wrapper_set_name__: 

224 delattr(cls, "__set_name__") 

225 

226 

227class LazyObjectProxy(AutoObjectProxy): 

228 """An object proxy which can generate/create the wrapped object on demand 

229 when it is first needed. 

230 """ 

231 

232 def __new__(cls, callback=None, *, interface=...): 

233 """Injects special dunder methods into a dynamically created subclass 

234 as needed based on the wrapped object. 

235 """ 

236 

237 if interface is ...: 

238 interface = type(None) 

239 

240 namespace = {} 

241 

242 interface_attrs = dir(interface) 

243 class_attrs = set(dir(cls)) 

244 

245 if "__call__" in interface_attrs and "__call__" not in class_attrs: 

246 namespace["__call__"] = __wrapper_call__ 

247 

248 if "__iter__" in interface_attrs and "__iter__" not in class_attrs: 

249 namespace["__iter__"] = __wrapper_iter__ 

250 

251 if "__next__" in interface_attrs and "__next__" not in class_attrs: 

252 namespace["__next__"] = __wrapper_next__ 

253 

254 if "__aiter__" in interface_attrs and "__aiter__" not in class_attrs: 

255 namespace["__aiter__"] = __wrapper_aiter__ 

256 

257 if "__anext__" in interface_attrs and "__anext__" not in class_attrs: 

258 namespace["__anext__"] = __wrapper_anext__ 

259 

260 if ( 

261 "__length_hint__" in interface_attrs 

262 and "__length_hint__" not in class_attrs 

263 ): 

264 namespace["__length_hint__"] = __wrapper_length_hint__ 

265 

266 # Note that not providing compatibility with generator-based coroutines 

267 # (PEP 342) here as they are removed in Python 3.11+ and were deprecated 

268 # in 3.8. 

269 

270 if "__await__" in interface_attrs and "__await__" not in class_attrs: 

271 namespace["__await__"] = __wrapper_await__ 

272 

273 if "__get__" in interface_attrs and "__get__" not in class_attrs: 

274 namespace["__get__"] = __wrapper_get__ 

275 

276 if "__set__" in interface_attrs and "__set__" not in class_attrs: 

277 namespace["__set__"] = __wrapper_set__ 

278 

279 if "__delete__" in interface_attrs and "__delete__" not in class_attrs: 

280 namespace["__delete__"] = __wrapper_delete__ 

281 

282 if "__set_name__" in interface_attrs and "__set_name__" not in class_attrs: 

283 namespace["__set_name__"] = __wrapper_set_name__ 

284 

285 name = cls.__name__ 

286 

287 # Explicit class in super() is required here to ensure __new__ 

288 # is called on the parent of AutoObjectProxy, not the dynamically 

289 # created subclass. 

290 return super(AutoObjectProxy, cls).__new__(type(name, (cls,), namespace)) 

291 

292 def __init__(self, callback=None, *, interface=...): 

293 """Initialize the object proxy with wrapped object as `None` but due 

294 to presence of special `__wrapped_factory__` attribute addded first, 

295 this will actually trigger the deferred creation of the wrapped object 

296 when first needed. 

297 """ 

298 

299 if callback is not None: 

300 self.__wrapped_factory__ = callback 

301 

302 super().__init__(None) 

303 

304 __wrapped_get_called__ = False 

305 

306 def __wrapped_factory__(self): 

307 return None 

308 

309 def __wrapped_get__(self): 

310 """Gets the wrapped object, creating it if necessary.""" 

311 

312 # We synchronize on the class type, which will be unique to this instance 

313 # since we inherit from `AutoObjectProxy` which creates a new class 

314 # for each instance. If we synchronize on `self` or the method then 

315 # we can end up in infinite recursion via `__getattr__()`. 

316 

317 with synchronized(type(self)): 

318 # We were called because `__wrapped__` was not set, but because of 

319 # multiple threads we may find that it has been set by the time 

320 # we get the lock. So check again now whether `__wrapped__` is set. 

321 # If it is then just return it, otherwise call the factory to 

322 # create it. 

323 

324 if self.__wrapped_get_called__: 

325 return self.__wrapped__ 

326 

327 self.__wrapped__ = self.__wrapped_factory__() 

328 

329 self.__wrapped_get_called__ = True 

330 

331 return self.__wrapped__ 

332 

333 

334def lazy_import(name, attribute=None, *, interface=...): 

335 """Lazily imports the module `name`, returning a `LazyObjectProxy` which 

336 will import the module when it is first needed. When `name is a dotted name, 

337 then the full dotted name is imported and the last module is taken as the 

338 target. If `attribute` is provided then it is used to retrieve an attribute 

339 from the module. 

340 """ 

341 

342 if attribute is not None: 

343 if interface is ...: 

344 interface = Callable 

345 else: 

346 if interface is ...: 

347 interface = ModuleType 

348 

349 def _import(): 

350 module = __import__(name, fromlist=[""]) 

351 

352 if attribute is not None: 

353 return getattr(module, attribute) 

354 

355 return module 

356 

357 return LazyObjectProxy(_import, interface=interface)