Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/SQLAlchemy-1.3.25.dev0-py3.11-linux-x86_64.egg/sqlalchemy/event/base.py: 79%

1# event/base.py 

2# Copyright (C) 2005-2021 the SQLAlchemy authors and contributors 

3# <see AUTHORS file> 

4# 

5# This module is part of SQLAlchemy and is released under 

6# the MIT License: http://www.opensource.org/licenses/mit-license.php 

7 

8"""Base implementation classes. 

9 

10The public-facing ``Events`` serves as the base class for an event interface; 

11its public attributes represent different kinds of events. These attributes 

12are mirrored onto a ``_Dispatch`` class, which serves as a container for 

13collections of listener functions. These collections are represented both 

14at the class level of a particular ``_Dispatch`` class as well as within 

15instances of ``_Dispatch``. 

16 

17""" 

18from __future__ import absolute_import 

19 

20import weakref 

21 

22from .attr import _ClsLevelDispatch 

23from .attr import _EmptyListener 

24from .attr import _JoinedListener 

25from .. import util 

26 

27 

28_registrars = util.defaultdict(list) 

29 

30 

31def _is_event_name(name): 

32 # _sa_event prefix is special to support internal-only event names. 

33 # most event names are just plain method names that aren't 

34 # underscored. 

35 

36 return ( 

37 not name.startswith("_") and name != "dispatch" 

38 ) or name.startswith("_sa_event") 

39 

40 

41class _UnpickleDispatch(object): 

42 """Serializable callable that re-generates an instance of 

43 :class:`_Dispatch` given a particular :class:`.Events` subclass. 

44 

45 """ 

46 

47 def __call__(self, _instance_cls): 

48 for cls in _instance_cls.__mro__: 

49 if "dispatch" in cls.__dict__: 

50 return cls.__dict__["dispatch"].dispatch._for_class( 

51 _instance_cls 

52 ) 

53 else: 

54 raise AttributeError("No class with a 'dispatch' member present.") 

55 

56 

57class _Dispatch(object): 

58 """Mirror the event listening definitions of an Events class with 

59 listener collections. 

60 

61 Classes which define a "dispatch" member will return a 

62 non-instantiated :class:`._Dispatch` subclass when the member 

63 is accessed at the class level. When the "dispatch" member is 

64 accessed at the instance level of its owner, an instance 

65 of the :class:`._Dispatch` class is returned. 

66 

67 A :class:`._Dispatch` class is generated for each :class:`.Events` 

68 class defined, by the :func:`._create_dispatcher_class` function. 

69 The original :class:`.Events` classes remain untouched. 

70 This decouples the construction of :class:`.Events` subclasses from 

71 the implementation used by the event internals, and allows 

72 inspecting tools like Sphinx to work in an unsurprising 

73 way against the public API. 

74 

75 """ 

76 

77 # In one ORM edge case, an attribute is added to _Dispatch, 

78 # so __dict__ is used in just that case and potentially others. 

79 __slots__ = "_parent", "_instance_cls", "__dict__", "_empty_listeners" 

80 

81 _empty_listener_reg = weakref.WeakKeyDictionary() 

82 

83 def __init__(self, parent, instance_cls=None): 

84 self._parent = parent 

85 self._instance_cls = instance_cls 

86 

87 if instance_cls: 

88 try: 

89 self._empty_listeners = self._empty_listener_reg[instance_cls] 

90 except KeyError: 

91 self._empty_listeners = self._empty_listener_reg[ 

92 instance_cls 

93 ] = { 

94 ls.name: _EmptyListener(ls, instance_cls) 

95 for ls in parent._event_descriptors 

96 } 

97 else: 

98 self._empty_listeners = {} 

99 

100 def __getattr__(self, name): 

101 # Assign EmptyListeners as attributes on demand 

102 # to reduce startup time for new dispatch objects. 

103 try: 

104 ls = self._empty_listeners[name] 

105 except KeyError: 

106 raise AttributeError(name) 

107 else: 

108 setattr(self, ls.name, ls) 

109 return ls 

110 

111 @property 

112 def _event_descriptors(self): 

113 for k in self._event_names: 

114 # Yield _ClsLevelDispatch related 

115 # to relevant event name. 

116 yield getattr(self, k) 

117 

118 @property 

119 def _listen(self): 

120 return self._events._listen 

121 

122 def _for_class(self, instance_cls): 

123 return self.__class__(self, instance_cls) 

124 

125 def _for_instance(self, instance): 

126 instance_cls = instance.__class__ 

127 return self._for_class(instance_cls) 

128 

129 def _join(self, other): 

130 """Create a 'join' of this :class:`._Dispatch` and another. 

131 

132 This new dispatcher will dispatch events to both 

133 :class:`._Dispatch` objects. 

134 

135 """ 

136 if "_joined_dispatch_cls" not in self.__class__.__dict__: 

137 cls = type( 

138 "Joined%s" % self.__class__.__name__, 

139 (_JoinedDispatcher,), 

140 {"__slots__": self._event_names}, 

141 ) 

142 

143 self.__class__._joined_dispatch_cls = cls 

144 return self._joined_dispatch_cls(self, other) 

145 

146 def __reduce__(self): 

147 return _UnpickleDispatch(), (self._instance_cls,) 

148 

149 def _update(self, other, only_propagate=True): 

150 """Populate from the listeners in another :class:`_Dispatch` 

151 object.""" 

152 for ls in other._event_descriptors: 

153 if isinstance(ls, _EmptyListener): 

154 continue 

155 getattr(self, ls.name).for_modify(self)._update( 

156 ls, only_propagate=only_propagate 

157 ) 

158 

159 def _clear(self): 

160 for ls in self._event_descriptors: 

161 ls.for_modify(self).clear() 

162 

163 

164class _EventMeta(type): 

165 """Intercept new Event subclasses and create 

166 associated _Dispatch classes.""" 

167 

168 def __init__(cls, classname, bases, dict_): 

169 _create_dispatcher_class(cls, classname, bases, dict_) 

170 type.__init__(cls, classname, bases, dict_) 

171 

172 

173def _create_dispatcher_class(cls, classname, bases, dict_): 

174 """Create a :class:`._Dispatch` class corresponding to an 

175 :class:`.Events` class.""" 

176 

177 # there's all kinds of ways to do this, 

178 # i.e. make a Dispatch class that shares the '_listen' method 

179 # of the Event class, this is the straight monkeypatch. 

180 if hasattr(cls, "dispatch"): 

181 dispatch_base = cls.dispatch.__class__ 

182 else: 

183 dispatch_base = _Dispatch 

184 

185 event_names = [k for k in dict_ if _is_event_name(k)] 

186 dispatch_cls = type( 

187 "%sDispatch" % classname, (dispatch_base,), {"__slots__": event_names} 

188 ) 

189 

190 dispatch_cls._event_names = event_names 

191 

192 dispatch_inst = cls._set_dispatch(cls, dispatch_cls) 

193 for k in dispatch_cls._event_names: 

194 setattr(dispatch_inst, k, _ClsLevelDispatch(cls, dict_[k])) 

195 _registrars[k].append(cls) 

196 

197 for super_ in dispatch_cls.__bases__: 

198 if issubclass(super_, _Dispatch) and super_ is not _Dispatch: 

199 for ls in super_._events.dispatch._event_descriptors: 

200 setattr(dispatch_inst, ls.name, ls) 

201 dispatch_cls._event_names.append(ls.name) 

202 

203 if getattr(cls, "_dispatch_target", None): 

204 cls._dispatch_target.dispatch = dispatcher(cls) 

205 

206 

207def _remove_dispatcher(cls): 

208 for k in cls.dispatch._event_names: 

209 _registrars[k].remove(cls) 

210 if not _registrars[k]: 

211 del _registrars[k] 

212 

213 

214class Events(util.with_metaclass(_EventMeta, object)): 

215 """Define event listening functions for a particular target type.""" 

216 

217 @staticmethod 

218 def _set_dispatch(cls, dispatch_cls): 

219 # This allows an Events subclass to define additional utility 

220 # methods made available to the target via 

221 # "self.dispatch._events.<utilitymethod>" 

222 # @staticmethod to allow easy "super" calls while in a metaclass 

223 # constructor. 

224 cls.dispatch = dispatch_cls(None) 

225 dispatch_cls._events = cls 

226 return cls.dispatch 

227 

228 @classmethod 

229 def _accept_with(cls, target): 

230 def dispatch_is(*types): 

231 return all(isinstance(target.dispatch, t) for t in types) 

232 

233 def dispatch_parent_is(t): 

234 return isinstance(target.dispatch.parent, t) 

235 

236 # Mapper, ClassManager, Session override this to 

237 # also accept classes, scoped_sessions, sessionmakers, etc. 

238 if hasattr(target, "dispatch"): 

239 if ( 

240 dispatch_is(cls.dispatch.__class__) 

241 or dispatch_is(type, cls.dispatch.__class__) 

242 or ( 

243 dispatch_is(_JoinedDispatcher) 

244 and dispatch_parent_is(cls.dispatch.__class__) 

245 ) 

246 ): 

247 return target 

248 

249 @classmethod 

250 def _listen(cls, event_key, propagate=False, insert=False, named=False): 

251 event_key.base_listen(propagate=propagate, insert=insert, named=named) 

252 

253 @classmethod 

254 def _remove(cls, event_key): 

255 event_key.remove() 

256 

257 @classmethod 

258 def _clear(cls): 

259 cls.dispatch._clear() 

260 

261 

262class _JoinedDispatcher(object): 

263 """Represent a connection between two _Dispatch objects.""" 

264 

265 __slots__ = "local", "parent", "_instance_cls" 

266 

267 def __init__(self, local, parent): 

268 self.local = local 

269 self.parent = parent 

270 self._instance_cls = self.local._instance_cls 

271 

272 def __getattr__(self, name): 

273 # Assign _JoinedListeners as attributes on demand 

274 # to reduce startup time for new dispatch objects. 

275 ls = getattr(self.local, name) 

276 jl = _JoinedListener(self.parent, ls.name, ls) 

277 setattr(self, ls.name, jl) 

278 return jl 

279 

280 @property 

281 def _listen(self): 

282 return self.parent._listen 

283 

284 @property 

285 def _events(self): 

286 return self.parent._events 

287 

288 

289class dispatcher(object): 

290 """Descriptor used by target classes to 

291 deliver the _Dispatch class at the class level 

292 and produce new _Dispatch instances for target 

293 instances. 

294 

295 """ 

296 

297 def __init__(self, events): 

298 self.dispatch = events.dispatch 

299 self.events = events 

300 

301 def __get__(self, obj, cls): 

302 if obj is None: 

303 return self.dispatch 

304 obj.__dict__["dispatch"] = disp = self.dispatch._for_instance(obj) 

305 return disp