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_fspath__(self): 

66 return self.__wrapped__.__fspath__() 

67 

68 

69def __wrapper_await__(self): 

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

71 

72 

73def __wrapper_get__(self, instance, owner): 

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

75 

76 

77def __wrapper_set__(self, instance, value): 

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

79 

80 

81def __wrapper_delete__(self, instance): 

82 return self.__wrapped__.__delete__(instance) 

83 

84 

85def __wrapper_set_name__(self, owner, name): 

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

87 

88 

89class AutoObjectProxy(BaseObjectProxy): 

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

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

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

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

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

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

96 compatibility then use `ObjectProxy` instead. 

97 """ 

98 

99 def __new__(cls, wrapped): 

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

101 as needed based on the wrapped object. 

102 """ 

103 

104 namespace = {} 

105 

106 wrapped_attrs = dir(wrapped) 

107 class_attrs = set(dir(cls)) 

108 

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

110 namespace["__call__"] = __wrapper_call__ 

111 

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

113 namespace["__iter__"] = __wrapper_iter__ 

114 

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

116 namespace["__next__"] = __wrapper_next__ 

117 

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

119 namespace["__aiter__"] = __wrapper_aiter__ 

120 

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

122 namespace["__anext__"] = __wrapper_anext__ 

123 

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

125 namespace["__length_hint__"] = __wrapper_length_hint__ 

126 

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

128 namespace["__fspath__"] = __wrapper_fspath__ 

129 

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

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

132 # in 3.8. 

133 

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

135 namespace["__await__"] = __wrapper_await__ 

136 

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

138 namespace["__get__"] = __wrapper_get__ 

139 

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

141 namespace["__set__"] = __wrapper_set__ 

142 

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

144 namespace["__delete__"] = __wrapper_delete__ 

145 

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

147 namespace["__set_name__"] = __wrapper_set_name__ 

148 

149 name = cls.__name__ 

150 

151 if cls is AutoObjectProxy: 

152 name = BaseObjectProxy.__name__ 

153 

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

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

156 # created subclass. 

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

158 

159 def __wrapped_setattr_fixups__(self): 

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

161 wrapped object, when `__wrapped__` is changed. 

162 """ 

163 

164 cls = type(self) 

165 class_attrs = set(dir(cls)) 

166 

167 if callable(self.__wrapped__): 

168 if "__call__" not in class_attrs: 

169 cls.__call__ = __wrapper_call__ 

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

171 delattr(cls, "__call__") 

172 

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

174 if "__iter__" not in class_attrs: 

175 cls.__iter__ = __wrapper_iter__ 

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

177 delattr(cls, "__iter__") 

178 

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

180 if "__next__" not in class_attrs: 

181 cls.__next__ = __wrapper_next__ 

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

183 delattr(cls, "__next__") 

184 

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

186 if "__aiter__" not in class_attrs: 

187 cls.__aiter__ = __wrapper_aiter__ 

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

189 delattr(cls, "__aiter__") 

190 

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

192 if "__anext__" not in class_attrs: 

193 cls.__anext__ = __wrapper_anext__ 

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

195 delattr(cls, "__anext__") 

196 

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

198 if "__length_hint__" not in class_attrs: 

199 cls.__length_hint__ = __wrapper_length_hint__ 

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

201 delattr(cls, "__length_hint__") 

202 

203 if hasattr(self.__wrapped__, "__fspath__"): 

204 if "__fspath__" not in class_attrs: 

205 cls.__fspath__ = __wrapper_fspath__ 

206 elif getattr(cls, "__fspath__", None) is __wrapper_fspath__: 

207 delattr(cls, "__fspath__") 

208 

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

210 if "__await__" not in class_attrs: 

211 cls.__await__ = __wrapper_await__ 

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

213 delattr(cls, "__await__") 

214 

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

216 if "__get__" not in class_attrs: 

217 cls.__get__ = __wrapper_get__ 

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

219 delattr(cls, "__get__") 

220 

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

222 if "__set__" not in class_attrs: 

223 cls.__set__ = __wrapper_set__ 

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

225 delattr(cls, "__set__") 

226 

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

228 if "__delete__" not in class_attrs: 

229 cls.__delete__ = __wrapper_delete__ 

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

231 delattr(cls, "__delete__") 

232 

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

234 if "__set_name__" not in class_attrs: 

235 cls.__set_name__ = __wrapper_set_name__ 

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

237 delattr(cls, "__set_name__") 

238 

239 

240class LazyObjectProxy(AutoObjectProxy): 

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

242 when it is first needed. 

243 """ 

244 

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

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

247 as needed based on the wrapped object. 

248 """ 

249 

250 if interface is ...: 

251 interface = type(None) 

252 

253 namespace = {} 

254 

255 interface_attrs = dir(interface) 

256 class_attrs = set(dir(cls)) 

257 

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

259 namespace["__call__"] = __wrapper_call__ 

260 

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

262 namespace["__iter__"] = __wrapper_iter__ 

263 

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

265 namespace["__next__"] = __wrapper_next__ 

266 

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

268 namespace["__aiter__"] = __wrapper_aiter__ 

269 

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

271 namespace["__anext__"] = __wrapper_anext__ 

272 

273 if ( 

274 "__length_hint__" in interface_attrs 

275 and "__length_hint__" not in class_attrs 

276 ): 

277 namespace["__length_hint__"] = __wrapper_length_hint__ 

278 

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

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

281 # in 3.8. 

282 

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

284 namespace["__await__"] = __wrapper_await__ 

285 

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

287 namespace["__get__"] = __wrapper_get__ 

288 

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

290 namespace["__set__"] = __wrapper_set__ 

291 

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

293 namespace["__delete__"] = __wrapper_delete__ 

294 

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

296 namespace["__set_name__"] = __wrapper_set_name__ 

297 

298 name = cls.__name__ 

299 

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

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

302 # created subclass. 

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

304 

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

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

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

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

309 when first needed. 

310 """ 

311 

312 if callback is not None: 

313 self.__wrapped_factory__ = callback 

314 

315 super().__init__(None) 

316 

317 __wrapped_get_called__ = False 

318 

319 def __wrapped_factory__(self): 

320 return None 

321 

322 def __wrapped_get__(self): 

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

324 

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

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

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

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

329 

330 with synchronized(type(self)): 

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

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

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

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

335 # create it. 

336 

337 if self.__wrapped_get_called__: 

338 return self.__wrapped__ 

339 

340 self.__wrapped__ = self.__wrapped_factory__() 

341 

342 self.__wrapped_get_called__ = True 

343 

344 return self.__wrapped__ 

345 

346 

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

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

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

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

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

352 from the module. 

353 """ 

354 

355 if attribute is not None: 

356 if interface is ...: 

357 interface = Callable 

358 else: 

359 if interface is ...: 

360 interface = ModuleType 

361 

362 def _import(): 

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

364 

365 if attribute is not None: 

366 return getattr(module, attribute) 

367 

368 return module 

369 

370 return LazyObjectProxy(_import, interface=interface)