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 .decorators 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 return super(AutoObjectProxy, cls).__new__(type(name, (cls,), namespace)) 

148 

149 def __wrapped_setattr_fixups__(self): 

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

151 wrapped object, when `__wrapped__` is changed. 

152 """ 

153 

154 cls = type(self) 

155 class_attrs = set(dir(cls)) 

156 

157 if callable(self.__wrapped__): 

158 if "__call__" not in class_attrs: 

159 cls.__call__ = __wrapper_call__ 

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

161 delattr(cls, "__call__") 

162 

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

164 if "__iter__" not in class_attrs: 

165 cls.__iter__ = __wrapper_iter__ 

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

167 delattr(cls, "__iter__") 

168 

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

170 if "__next__" not in class_attrs: 

171 cls.__next__ = __wrapper_next__ 

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

173 delattr(cls, "__next__") 

174 

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

176 if "__aiter__" not in class_attrs: 

177 cls.__aiter__ = __wrapper_aiter__ 

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

179 delattr(cls, "__aiter__") 

180 

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

182 if "__anext__" not in class_attrs: 

183 cls.__anext__ = __wrapper_anext__ 

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

185 delattr(cls, "__anext__") 

186 

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

188 if "__length_hint__" not in class_attrs: 

189 cls.__length_hint__ = __wrapper_length_hint__ 

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

191 delattr(cls, "__length_hint__") 

192 

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

194 if "__await__" not in class_attrs: 

195 cls.__await__ = __wrapper_await__ 

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

197 delattr(cls, "__await__") 

198 

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

200 if "__get__" not in class_attrs: 

201 cls.__get__ = __wrapper_get__ 

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

203 delattr(cls, "__get__") 

204 

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

206 if "__set__" not in class_attrs: 

207 cls.__set__ = __wrapper_set__ 

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

209 delattr(cls, "__set__") 

210 

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

212 if "__delete__" not in class_attrs: 

213 cls.__delete__ = __wrapper_delete__ 

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

215 delattr(cls, "__delete__") 

216 

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

218 if "__set_name__" not in class_attrs: 

219 cls.__set_name__ = __wrapper_set_name__ 

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

221 delattr(cls, "__set_name__") 

222 

223 

224class LazyObjectProxy(AutoObjectProxy): 

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

226 when it is first needed. 

227 """ 

228 

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

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

231 as needed based on the wrapped object. 

232 """ 

233 

234 if interface is ...: 

235 interface = type(None) 

236 

237 namespace = {} 

238 

239 interface_attrs = dir(interface) 

240 class_attrs = set(dir(cls)) 

241 

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

243 namespace["__call__"] = __wrapper_call__ 

244 

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

246 namespace["__iter__"] = __wrapper_iter__ 

247 

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

249 namespace["__next__"] = __wrapper_next__ 

250 

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

252 namespace["__aiter__"] = __wrapper_aiter__ 

253 

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

255 namespace["__anext__"] = __wrapper_anext__ 

256 

257 if ( 

258 "__length_hint__" in interface_attrs 

259 and "__length_hint__" not in class_attrs 

260 ): 

261 namespace["__length_hint__"] = __wrapper_length_hint__ 

262 

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

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

265 # in 3.8. 

266 

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

268 namespace["__await__"] = __wrapper_await__ 

269 

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

271 namespace["__get__"] = __wrapper_get__ 

272 

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

274 namespace["__set__"] = __wrapper_set__ 

275 

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

277 namespace["__delete__"] = __wrapper_delete__ 

278 

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

280 namespace["__set_name__"] = __wrapper_set_name__ 

281 

282 name = cls.__name__ 

283 

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

285 

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

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

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

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

290 when first needed. 

291 """ 

292 

293 if callback is not None: 

294 self.__wrapped_factory__ = callback 

295 

296 super().__init__(None) 

297 

298 __wrapped_initialized__ = False 

299 

300 def __wrapped_factory__(self): 

301 return None 

302 

303 def __wrapped_get__(self): 

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

305 

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

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

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

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

310 

311 with synchronized(type(self)): 

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

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

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

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

316 # create it. 

317 

318 if self.__wrapped_initialized__: 

319 return self.__wrapped__ 

320 

321 self.__wrapped__ = self.__wrapped_factory__() 

322 

323 self.__wrapped_initialized__ = True 

324 

325 return self.__wrapped__ 

326 

327 

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

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

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

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

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

333 from the module. 

334 """ 

335 

336 if attribute is not None: 

337 if interface is ...: 

338 interface = Callable 

339 else: 

340 if interface is ...: 

341 interface = ModuleType 

342 

343 def _import(): 

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

345 

346 if attribute is not None: 

347 return getattr(module, attribute) 

348 

349 return module 

350 

351 return LazyObjectProxy(_import, interface=interface)