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

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

26 return iter(self.__wrapped__) 

27 

28 

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

30# object and add special dunder methods. 

31 

32 

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

34 def _unpack_self(self, *args): 

35 return self, args 

36 

37 self, args = _unpack_self(*args) 

38 

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

40 

41 

42def __wrapper_iter__(self): 

43 return iter(self.__wrapped__) 

44 

45 

46def __wrapper_next__(self): 

47 return self.__wrapped__.__next__() 

48 

49 

50def __wrapper_aiter__(self): 

51 return self.__wrapped__.__aiter__() 

52 

53 

54async def __wrapper_anext__(self): 

55 return await self.__wrapped__.__anext__() 

56 

57 

58def __wrapper_length_hint__(self): 

59 return self.__wrapped__.__length_hint__() 

60 

61 

62def __wrapper_fspath__(self): 

63 return self.__wrapped__.__fspath__() 

64 

65 

66def __wrapper_await__(self): 

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

68 

69 

70def __wrapper_get__(self, instance, owner=None): 

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

72 

73 

74def __wrapper_set__(self, instance, value): 

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

76 

77 

78def __wrapper_delete__(self, instance): 

79 return self.__wrapped__.__delete__(instance) 

80 

81 

82def __wrapper_set_name__(self, owner, name): 

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

84 

85 

86class AutoObjectProxy(BaseObjectProxy): 

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

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

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

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

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

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

93 compatibility then use `ObjectProxy` instead. 

94 """ 

95 

96 def __new__(cls, wrapped, *args, **kwargs): 

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

98 as needed based on the wrapped object. Any further arguments are 

99 those a derived class has added to `__init__()` and are ignored. 

100 """ 

101 

102 namespace = {} 

103 

104 wrapped_attrs = dir(wrapped) 

105 class_attrs = set(dir(cls)) 

106 

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

108 namespace["__call__"] = __wrapper_call__ 

109 

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

111 namespace["__iter__"] = __wrapper_iter__ 

112 

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

114 namespace["__next__"] = __wrapper_next__ 

115 

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

117 namespace["__aiter__"] = __wrapper_aiter__ 

118 

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

120 namespace["__anext__"] = __wrapper_anext__ 

121 

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

123 namespace["__length_hint__"] = __wrapper_length_hint__ 

124 

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

126 namespace["__fspath__"] = __wrapper_fspath__ 

127 

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

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

130 # in 3.8. 

131 

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

133 namespace["__await__"] = __wrapper_await__ 

134 

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

136 namespace["__get__"] = __wrapper_get__ 

137 

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

139 namespace["__set__"] = __wrapper_set__ 

140 

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

142 namespace["__delete__"] = __wrapper_delete__ 

143 

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

145 namespace["__set_name__"] = __wrapper_set_name__ 

146 

147 name = cls.__name__ 

148 

149 if cls is AutoObjectProxy: 

150 name = BaseObjectProxy.__name__ 

151 

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

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

154 # created subclass. 

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

156 

157 def __wrapped_setattr_fixups__(self): 

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

159 wrapped object, when `__wrapped__` is changed. 

160 """ 

161 

162 cls = type(self) 

163 class_attrs = set(dir(cls)) 

164 

165 if callable(self.__wrapped__): 

166 if "__call__" not in class_attrs: 

167 cls.__call__ = __wrapper_call__ 

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

169 delattr(cls, "__call__") 

170 

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

172 if "__iter__" not in class_attrs: 

173 cls.__iter__ = __wrapper_iter__ 

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

175 delattr(cls, "__iter__") 

176 

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

178 if "__next__" not in class_attrs: 

179 cls.__next__ = __wrapper_next__ 

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

181 delattr(cls, "__next__") 

182 

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

184 if "__aiter__" not in class_attrs: 

185 cls.__aiter__ = __wrapper_aiter__ 

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

187 delattr(cls, "__aiter__") 

188 

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

190 if "__anext__" not in class_attrs: 

191 cls.__anext__ = __wrapper_anext__ 

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

193 delattr(cls, "__anext__") 

194 

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

196 if "__length_hint__" not in class_attrs: 

197 cls.__length_hint__ = __wrapper_length_hint__ 

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

199 delattr(cls, "__length_hint__") 

200 

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

202 if "__fspath__" not in class_attrs: 

203 cls.__fspath__ = __wrapper_fspath__ 

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

205 delattr(cls, "__fspath__") 

206 

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

208 if "__await__" not in class_attrs: 

209 cls.__await__ = __wrapper_await__ 

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

211 delattr(cls, "__await__") 

212 

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

214 if "__get__" not in class_attrs: 

215 cls.__get__ = __wrapper_get__ 

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

217 delattr(cls, "__get__") 

218 

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

220 if "__set__" not in class_attrs: 

221 cls.__set__ = __wrapper_set__ 

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

223 delattr(cls, "__set__") 

224 

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

226 if "__delete__" not in class_attrs: 

227 cls.__delete__ = __wrapper_delete__ 

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

229 delattr(cls, "__delete__") 

230 

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

232 if "__set_name__" not in class_attrs: 

233 cls.__set_name__ = __wrapper_set_name__ 

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

235 delattr(cls, "__set_name__") 

236 

237 

238class LazyObjectProxy(AutoObjectProxy): 

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

240 when it is first needed. 

241 """ 

242 

243 def __new__(cls, callback=None, *args, interface=..., **kwargs): 

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

245 as needed based on the declared interface of the wrapped object. Any 

246 further arguments are those a derived class has added to `__init__()` 

247 and are ignored. 

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)