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 # Special methods are only ever looked up on the type of the wrapped 

105 # object, never on the instance itself, so the special methods to add 

106 # are determined from dir() of the type of the wrapped object. For a 

107 # class the type is its metaclass, as the special methods a class 

108 # defines in its body apply to its instances, not to the class 

109 # itself. 

110 

111 wrapped_attrs = dir(type(wrapped)) 

112 class_attrs = set(dir(cls)) 

113 

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

115 namespace["__call__"] = __wrapper_call__ 

116 

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

118 namespace["__iter__"] = __wrapper_iter__ 

119 

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

121 namespace["__next__"] = __wrapper_next__ 

122 

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

124 namespace["__aiter__"] = __wrapper_aiter__ 

125 

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

127 namespace["__anext__"] = __wrapper_anext__ 

128 

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

130 namespace["__length_hint__"] = __wrapper_length_hint__ 

131 

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

133 namespace["__fspath__"] = __wrapper_fspath__ 

134 

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

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

137 # in 3.8. 

138 

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

140 namespace["__await__"] = __wrapper_await__ 

141 

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

143 namespace["__get__"] = __wrapper_get__ 

144 

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

146 namespace["__set__"] = __wrapper_set__ 

147 

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

149 namespace["__delete__"] = __wrapper_delete__ 

150 

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

152 namespace["__set_name__"] = __wrapper_set_name__ 

153 

154 name = cls.__name__ 

155 

156 if cls is AutoObjectProxy: 

157 name = BaseObjectProxy.__name__ 

158 

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

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

161 # created subclass. 

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

163 

164 def __wrapped_setattr_fixups__(self): 

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

166 wrapped object, when `__wrapped__` is changed. 

167 """ 

168 

169 cls = type(self) 

170 class_attrs = set(dir(cls)) 

171 

172 # As for __new__() above, special methods are only ever looked up on 

173 # the type of the wrapped object, never on the instance itself, so 

174 # check for them on the type of the wrapped object. 

175 

176 wrapped_type = type(self.__wrapped__) 

177 

178 if callable(self.__wrapped__): 

179 if "__call__" not in class_attrs: 

180 cls.__call__ = __wrapper_call__ 

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

182 delattr(cls, "__call__") 

183 

184 if hasattr(wrapped_type, "__iter__"): 

185 if "__iter__" not in class_attrs: 

186 cls.__iter__ = __wrapper_iter__ 

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

188 delattr(cls, "__iter__") 

189 

190 if hasattr(wrapped_type, "__next__"): 

191 if "__next__" not in class_attrs: 

192 cls.__next__ = __wrapper_next__ 

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

194 delattr(cls, "__next__") 

195 

196 if hasattr(wrapped_type, "__aiter__"): 

197 if "__aiter__" not in class_attrs: 

198 cls.__aiter__ = __wrapper_aiter__ 

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

200 delattr(cls, "__aiter__") 

201 

202 if hasattr(wrapped_type, "__anext__"): 

203 if "__anext__" not in class_attrs: 

204 cls.__anext__ = __wrapper_anext__ 

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

206 delattr(cls, "__anext__") 

207 

208 if hasattr(wrapped_type, "__length_hint__"): 

209 if "__length_hint__" not in class_attrs: 

210 cls.__length_hint__ = __wrapper_length_hint__ 

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

212 delattr(cls, "__length_hint__") 

213 

214 if hasattr(wrapped_type, "__fspath__"): 

215 if "__fspath__" not in class_attrs: 

216 cls.__fspath__ = __wrapper_fspath__ 

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

218 delattr(cls, "__fspath__") 

219 

220 if hasattr(wrapped_type, "__await__"): 

221 if "__await__" not in class_attrs: 

222 cls.__await__ = __wrapper_await__ 

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

224 delattr(cls, "__await__") 

225 

226 if hasattr(wrapped_type, "__get__"): 

227 if "__get__" not in class_attrs: 

228 cls.__get__ = __wrapper_get__ 

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

230 delattr(cls, "__get__") 

231 

232 if hasattr(wrapped_type, "__set__"): 

233 if "__set__" not in class_attrs: 

234 cls.__set__ = __wrapper_set__ 

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

236 delattr(cls, "__set__") 

237 

238 if hasattr(wrapped_type, "__delete__"): 

239 if "__delete__" not in class_attrs: 

240 cls.__delete__ = __wrapper_delete__ 

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

242 delattr(cls, "__delete__") 

243 

244 if hasattr(wrapped_type, "__set_name__"): 

245 if "__set_name__" not in class_attrs: 

246 cls.__set_name__ = __wrapper_set_name__ 

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

248 delattr(cls, "__set_name__") 

249 

250 

251class LazyObjectProxy(AutoObjectProxy): 

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

253 when it is first needed. 

254 """ 

255 

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

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

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

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

260 and are ignored. 

261 """ 

262 

263 if interface is ...: 

264 interface = type(None) 

265 

266 namespace = {} 

267 

268 interface_attrs = dir(interface) 

269 class_attrs = set(dir(cls)) 

270 

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

272 namespace["__call__"] = __wrapper_call__ 

273 

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

275 namespace["__iter__"] = __wrapper_iter__ 

276 

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

278 namespace["__next__"] = __wrapper_next__ 

279 

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

281 namespace["__aiter__"] = __wrapper_aiter__ 

282 

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

284 namespace["__anext__"] = __wrapper_anext__ 

285 

286 if ( 

287 "__length_hint__" in interface_attrs 

288 and "__length_hint__" not in class_attrs 

289 ): 

290 namespace["__length_hint__"] = __wrapper_length_hint__ 

291 

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

293 namespace["__fspath__"] = __wrapper_fspath__ 

294 

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

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

297 # in 3.8. 

298 

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

300 namespace["__await__"] = __wrapper_await__ 

301 

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

303 namespace["__get__"] = __wrapper_get__ 

304 

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

306 namespace["__set__"] = __wrapper_set__ 

307 

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

309 namespace["__delete__"] = __wrapper_delete__ 

310 

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

312 namespace["__set_name__"] = __wrapper_set_name__ 

313 

314 name = cls.__name__ 

315 

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

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

318 # created subclass. 

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

320 

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

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

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

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

325 when first needed. 

326 """ 

327 

328 if callback is not None: 

329 self.__wrapped_factory__ = callback 

330 

331 super().__init__(None) 

332 

333 __wrapped_get_called__ = False 

334 

335 def __wrapped_factory__(self): 

336 return None 

337 

338 def __wrapped_get__(self): 

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

340 

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

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

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

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

345 

346 with synchronized(type(self)): 

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

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

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

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

351 # create it. 

352 

353 if self.__wrapped_get_called__: 

354 return self.__wrapped__ 

355 

356 self.__wrapped__ = self.__wrapped_factory__() 

357 

358 self.__wrapped_get_called__ = True 

359 

360 return self.__wrapped__ 

361 

362 

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

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

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

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

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

368 from the module. 

369 """ 

370 

371 if attribute is not None: 

372 if interface is ...: 

373 interface = Callable 

374 else: 

375 if interface is ...: 

376 interface = ModuleType 

377 

378 def _import(): 

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

380 

381 if attribute is not None: 

382 return getattr(module, attribute) 

383 

384 return module 

385 

386 return LazyObjectProxy(_import, interface=interface)