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

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

2 

3from .__wrapt__ import BaseObjectProxy 

4from .decorators import synchronized 

5 

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

7# has been removed from `BaseObjectProxy`. 

8 

9 

10class ObjectProxy(BaseObjectProxy): 

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

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

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

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

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

16 then use `AutoObjectProxy`.""" 

17 

18 @property 

19 def __object_proxy__(self): 

20 return ObjectProxy 

21 

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

23 return super().__new__(cls) 

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__(self, *args, **kwargs): 

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

35 

36 

37def __wrapper_iter__(self): 

38 return iter(self.__wrapped__) 

39 

40 

41def __wrapper_next__(self): 

42 return self.__wrapped__.__next__() 

43 

44 

45def __wrapper_aiter__(self): 

46 return self.__wrapped__.__aiter__() 

47 

48 

49async def __wrapper_anext__(self): 

50 return await self.__wrapped__.__anext__() 

51 

52 

53def __wrapper_length_hint__(self): 

54 return self.__wrapped__.__length_hint__() 

55 

56 

57def __wrapper_await__(self): 

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

59 

60 

61def __wrapper_get__(self, instance, owner): 

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

63 

64 

65def __wrapper_set__(self, instance, value): 

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

67 

68 

69def __wrapper_delete__(self, instance): 

70 return self.__wrapped__.__delete__(instance) 

71 

72 

73def __wrapper_set_name__(self, owner, name): 

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

75 

76 

77class AutoObjectProxy(BaseObjectProxy): 

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

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

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

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

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

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

84 compatibility then use `ObjectProxy` instead. 

85 """ 

86 

87 def __new__(cls, wrapped): 

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

89 as needed based on the wrapped object. 

90 """ 

91 

92 namespace = {} 

93 

94 wrapped_attrs = dir(wrapped) 

95 class_attrs = set(dir(cls)) 

96 

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

98 namespace["__call__"] = __wrapper_call__ 

99 

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

101 namespace["__iter__"] = __wrapper_iter__ 

102 

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

104 namespace["__next__"] = __wrapper_next__ 

105 

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

107 namespace["__aiter__"] = __wrapper_aiter__ 

108 

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

110 namespace["__anext__"] = __wrapper_anext__ 

111 

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

113 namespace["__length_hint__"] = __wrapper_length_hint__ 

114 

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

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

117 # in 3.8. 

118 

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

120 namespace["__await__"] = __wrapper_await__ 

121 

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

123 namespace["__get__"] = __wrapper_get__ 

124 

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

126 namespace["__set__"] = __wrapper_set__ 

127 

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

129 namespace["__delete__"] = __wrapper_delete__ 

130 

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

132 namespace["__set_name__"] = __wrapper_set_name__ 

133 

134 name = cls.__name__ 

135 

136 if cls is AutoObjectProxy: 

137 name = BaseObjectProxy.__name__ 

138 

139 return super().__new__(type(name, (cls,), namespace)) 

140 

141 def __wrapped_setattr_fixups__(self): 

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

143 wrapped object, when `__wrapped__` is changed. 

144 """ 

145 

146 cls = type(self) 

147 class_attrs = set(dir(cls)) 

148 

149 if callable(self.__wrapped__): 

150 if "__call__" not in class_attrs: 

151 cls.__call__ = __wrapper_call__ 

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

153 delattr(cls, "__call__") 

154 

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

156 if "__iter__" not in class_attrs: 

157 cls.__iter__ = __wrapper_iter__ 

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

159 delattr(cls, "__iter__") 

160 

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

162 if "__next__" not in class_attrs: 

163 cls.__next__ = __wrapper_next__ 

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

165 delattr(cls, "__next__") 

166 

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

168 if "__aiter__" not in class_attrs: 

169 cls.__aiter__ = __wrapper_aiter__ 

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

171 delattr(cls, "__aiter__") 

172 

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

174 if "__anext__" not in class_attrs: 

175 cls.__anext__ = __wrapper_anext__ 

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

177 delattr(cls, "__anext__") 

178 

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

180 if "__length_hint__" not in class_attrs: 

181 cls.__length_hint__ = __wrapper_length_hint__ 

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

183 delattr(cls, "__length_hint__") 

184 

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

186 if "__await__" not in class_attrs: 

187 cls.__await__ = __wrapper_await__ 

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

189 delattr(cls, "__await__") 

190 

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

192 if "__get__" not in class_attrs: 

193 cls.__get__ = __wrapper_get__ 

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

195 delattr(cls, "__get__") 

196 

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

198 if "__set__" not in class_attrs: 

199 cls.__set__ = __wrapper_set__ 

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

201 delattr(cls, "__set__") 

202 

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

204 if "__delete__" not in class_attrs: 

205 cls.__delete__ = __wrapper_delete__ 

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

207 delattr(cls, "__delete__") 

208 

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

210 if "__set_name__" not in class_attrs: 

211 cls.__set_name__ = __wrapper_set_name__ 

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

213 delattr(cls, "__set_name__") 

214 

215 

216class LazyObjectProxy(AutoObjectProxy): 

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

218 when it is first needed. 

219 """ 

220 

221 def __new__(cls, callback=None): 

222 return super().__new__(cls, None) 

223 

224 def __init__(self, callback=None): 

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

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

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

228 when first needed. 

229 """ 

230 

231 if callback is not None: 

232 self.__wrapped_factory__ = callback 

233 

234 super().__init__(None) 

235 

236 __wrapped_initialized__ = False 

237 

238 def __wrapped_factory__(self): 

239 return None 

240 

241 def __wrapped_get__(self): 

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

243 

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

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

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

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

248 

249 with synchronized(type(self)): 

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

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

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

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

254 # create it. 

255 

256 if self.__wrapped_initialized__: 

257 return self.__wrapped__ 

258 

259 self.__wrapped__ = self.__wrapped_factory__() 

260 

261 self.__wrapped_initialized__ = True 

262 

263 return self.__wrapped__ 

264 

265 

266def lazy_import(name, attribute=None): 

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

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

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

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

271 from the module. 

272 """ 

273 

274 def _import(): 

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

276 

277 if attribute is not None: 

278 return getattr(module, attribute) 

279 

280 return module 

281 

282 return LazyObjectProxy(_import)