Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/event/attr.py: 66%

1# event/attr.py 

2# Copyright (C) 2005-2025 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: https://www.opensource.org/licenses/mit-license.php 

7 

8"""Attribute implementation for _Dispatch classes. 

9 

10The various listener targets for a particular event class are represented 

11as attributes, which refer to collections of listeners to be fired off. 

12These collections can exist at the class level as well as at the instance 

13level. An event is fired off using code like this:: 

14 

15 some_object.dispatch.first_connect(arg1, arg2) 

16 

17Above, ``some_object.dispatch`` would be an instance of ``_Dispatch`` and 

18``first_connect`` is typically an instance of ``_ListenerCollection`` 

19if event listeners are present, or ``_EmptyListener`` if none are present. 

20 

21The attribute mechanics here spend effort trying to ensure listener functions 

22are available with a minimum of function call overhead, that unnecessary 

23objects aren't created (i.e. many empty per-instance listener collections), 

24as well as that everything is garbage collectable when owning references are 

25lost. Other features such as "propagation" of listener functions across 

26many ``_Dispatch`` instances, "joining" of multiple ``_Dispatch`` instances, 

27as well as support for subclass propagation (e.g. events assigned to 

28``Pool`` vs. ``QueuePool``) are all implemented here. 

29 

30""" 

31from __future__ import annotations 

32 

33import collections 

34from itertools import chain 

35import threading 

36from types import TracebackType 

37import typing 

38from typing import Any 

39from typing import cast 

40from typing import Collection 

41from typing import Deque 

42from typing import FrozenSet 

43from typing import Generic 

44from typing import Iterator 

45from typing import MutableMapping 

46from typing import MutableSequence 

47from typing import NoReturn 

48from typing import Optional 

49from typing import Protocol 

50from typing import Sequence 

51from typing import Set 

52from typing import Tuple 

53from typing import Type 

54from typing import TypeVar 

55from typing import Union 

56import weakref 

57 

58from . import legacy 

59from . import registry 

60from .registry import _ET 

61from .registry import _EventKey 

62from .registry import _ListenerFnType 

63from .. import exc 

64from .. import util 

65from ..util.concurrency import AsyncAdaptedLock 

66 

67_T = TypeVar("_T", bound=Any) 

68 

69if typing.TYPE_CHECKING: 

70 from .base import _Dispatch 

71 from .base import _DispatchCommon 

72 from .base import _HasEventsDispatch 

73 

74 

75class RefCollection(util.MemoizedSlots, Generic[_ET]): 

76 __slots__ = ("ref",) 

77 

78 ref: weakref.ref[RefCollection[_ET]] 

79 

80 def _memoized_attr_ref(self) -> weakref.ref[RefCollection[_ET]]: 

81 return weakref.ref(self, registry._collection_gced) 

82 

83 

84class _empty_collection(Collection[_T]): 

85 def append(self, element: _T) -> None: 

86 pass 

87 

88 def appendleft(self, element: _T) -> None: 

89 pass 

90 

91 def extend(self, other: Sequence[_T]) -> None: 

92 pass 

93 

94 def remove(self, element: _T) -> None: 

95 pass 

96 

97 def __contains__(self, element: Any) -> bool: 

98 return False 

99 

100 def __iter__(self) -> Iterator[_T]: 

101 return iter([]) 

102 

103 def clear(self) -> None: 

104 pass 

105 

106 def __len__(self) -> int: 

107 return 0 

108 

109 

110_ListenerFnSequenceType = Union[Deque[_T], _empty_collection[_T]] 

111 

112 

113class _ClsLevelDispatch(RefCollection[_ET]): 

114 """Class-level events on :class:`._Dispatch` classes.""" 

115 

116 __slots__ = ( 

117 "clsname", 

118 "name", 

119 "arg_names", 

120 "has_kw", 

121 "legacy_signatures", 

122 "_clslevel", 

123 "__weakref__", 

124 ) 

125 

126 clsname: str 

127 name: str 

128 arg_names: Sequence[str] 

129 has_kw: bool 

130 legacy_signatures: MutableSequence[legacy._LegacySignatureType] 

131 _clslevel: MutableMapping[ 

132 Type[_ET], _ListenerFnSequenceType[_ListenerFnType] 

133 ] 

134 

135 def __init__( 

136 self, 

137 parent_dispatch_cls: Type[_HasEventsDispatch[_ET]], 

138 fn: _ListenerFnType, 

139 ): 

140 self.name = fn.__name__ 

141 self.clsname = parent_dispatch_cls.__name__ 

142 argspec = util.inspect_getfullargspec(fn) 

143 self.arg_names = argspec.args[1:] 

144 self.has_kw = bool(argspec.varkw) 

145 self.legacy_signatures = list( 

146 reversed( 

147 sorted( 

148 getattr(fn, "_legacy_signatures", []), key=lambda s: s[0] 

149 ) 

150 ) 

151 ) 

152 fn.__doc__ = legacy._augment_fn_docs(self, parent_dispatch_cls, fn) 

153 

154 self._clslevel = weakref.WeakKeyDictionary() 

155 

156 def _adjust_fn_spec( 

157 self, fn: _ListenerFnType, named: bool 

158 ) -> _ListenerFnType: 

159 if named: 

160 fn = self._wrap_fn_for_kw(fn) 

161 if self.legacy_signatures: 

162 try: 

163 argspec = util.get_callable_argspec(fn, no_self=True) 

164 except TypeError: 

165 pass 

166 else: 

167 fn = legacy._wrap_fn_for_legacy(self, fn, argspec) 

168 return fn 

169 

170 def _wrap_fn_for_kw(self, fn: _ListenerFnType) -> _ListenerFnType: 

171 def wrap_kw(*args: Any, **kw: Any) -> Any: 

172 argdict = dict(zip(self.arg_names, args)) 

173 argdict.update(kw) 

174 return fn(**argdict) 

175 

176 return wrap_kw 

177 

178 def _do_insert_or_append( 

179 self, event_key: _EventKey[_ET], is_append: bool 

180 ) -> None: 

181 target = event_key.dispatch_target 

182 assert isinstance( 

183 target, type 

184 ), "Class-level Event targets must be classes." 

185 if not getattr(target, "_sa_propagate_class_events", True): 

186 raise exc.InvalidRequestError( 

187 f"Can't assign an event directly to the {target} class" 

188 ) 

189 

190 cls: Type[_ET] 

191 

192 for cls in util.walk_subclasses(target): 

193 if cls is not target and cls not in self._clslevel: 

194 self.update_subclass(cls) 

195 else: 

196 if cls not in self._clslevel: 

197 self.update_subclass(cls) 

198 if is_append: 

199 self._clslevel[cls].append(event_key._listen_fn) 

200 else: 

201 self._clslevel[cls].appendleft(event_key._listen_fn) 

202 registry._stored_in_collection(event_key, self) 

203 

204 def insert(self, event_key: _EventKey[_ET], propagate: bool) -> None: 

205 self._do_insert_or_append(event_key, is_append=False) 

206 

207 def append(self, event_key: _EventKey[_ET], propagate: bool) -> None: 

208 self._do_insert_or_append(event_key, is_append=True) 

209 

210 def update_subclass(self, target: Type[_ET]) -> None: 

211 if target not in self._clslevel: 

212 if getattr(target, "_sa_propagate_class_events", True): 

213 self._clslevel[target] = collections.deque() 

214 else: 

215 self._clslevel[target] = _empty_collection() 

216 

217 clslevel = self._clslevel[target] 

218 cls: Type[_ET] 

219 for cls in target.__mro__[1:]: 

220 if cls in self._clslevel: 

221 clslevel.extend( 

222 [fn for fn in self._clslevel[cls] if fn not in clslevel] 

223 ) 

224 

225 def remove(self, event_key: _EventKey[_ET]) -> None: 

226 target = event_key.dispatch_target 

227 cls: Type[_ET] 

228 for cls in util.walk_subclasses(target): 

229 if cls in self._clslevel: 

230 self._clslevel[cls].remove(event_key._listen_fn) 

231 registry._removed_from_collection(event_key, self) 

232 

233 def clear(self) -> None: 

234 """Clear all class level listeners""" 

235 

236 to_clear: Set[_ListenerFnType] = set() 

237 for dispatcher in self._clslevel.values(): 

238 to_clear.update(dispatcher) 

239 dispatcher.clear() 

240 registry._clear(self, to_clear) 

241 

242 def for_modify(self, obj: _Dispatch[_ET]) -> _ClsLevelDispatch[_ET]: 

243 """Return an event collection which can be modified. 

244 

245 For _ClsLevelDispatch at the class level of 

246 a dispatcher, this returns self. 

247 

248 """ 

249 return self 

250 

251 

252class _InstanceLevelDispatch(RefCollection[_ET], Collection[_ListenerFnType]): 

253 __slots__ = () 

254 

255 parent: _ClsLevelDispatch[_ET] 

256 

257 def _adjust_fn_spec( 

258 self, fn: _ListenerFnType, named: bool 

259 ) -> _ListenerFnType: 

260 return self.parent._adjust_fn_spec(fn, named) 

261 

262 def __contains__(self, item: Any) -> bool: 

263 raise NotImplementedError() 

264 

265 def __len__(self) -> int: 

266 raise NotImplementedError() 

267 

268 def __iter__(self) -> Iterator[_ListenerFnType]: 

269 raise NotImplementedError() 

270 

271 def __bool__(self) -> bool: 

272 raise NotImplementedError() 

273 

274 def exec_once(self, *args: Any, **kw: Any) -> None: 

275 raise NotImplementedError() 

276 

277 def exec_once_unless_exception(self, *args: Any, **kw: Any) -> None: 

278 raise NotImplementedError() 

279 

280 def _exec_w_sync_on_first_run(self, *args: Any, **kw: Any) -> None: 

281 raise NotImplementedError() 

282 

283 def __call__(self, *args: Any, **kw: Any) -> None: 

284 raise NotImplementedError() 

285 

286 def insert(self, event_key: _EventKey[_ET], propagate: bool) -> None: 

287 raise NotImplementedError() 

288 

289 def append(self, event_key: _EventKey[_ET], propagate: bool) -> None: 

290 raise NotImplementedError() 

291 

292 def remove(self, event_key: _EventKey[_ET]) -> None: 

293 raise NotImplementedError() 

294 

295 def for_modify( 

296 self, obj: _DispatchCommon[_ET] 

297 ) -> _InstanceLevelDispatch[_ET]: 

298 """Return an event collection which can be modified. 

299 

300 For _ClsLevelDispatch at the class level of 

301 a dispatcher, this returns self. 

302 

303 """ 

304 return self 

305 

306 

307class _EmptyListener(_InstanceLevelDispatch[_ET]): 

308 """Serves as a proxy interface to the events 

309 served by a _ClsLevelDispatch, when there are no 

310 instance-level events present. 

311 

312 Is replaced by _ListenerCollection when instance-level 

313 events are added. 

314 

315 """ 

316 

317 __slots__ = "parent", "parent_listeners", "name" 

318 

319 propagate: FrozenSet[_ListenerFnType] = frozenset() 

320 listeners: Tuple[()] = () 

321 parent: _ClsLevelDispatch[_ET] 

322 parent_listeners: _ListenerFnSequenceType[_ListenerFnType] 

323 name: str 

324 

325 def __init__(self, parent: _ClsLevelDispatch[_ET], target_cls: Type[_ET]): 

326 if target_cls not in parent._clslevel: 

327 parent.update_subclass(target_cls) 

328 self.parent = parent 

329 self.parent_listeners = parent._clslevel[target_cls] 

330 self.name = parent.name 

331 

332 def for_modify( 

333 self, obj: _DispatchCommon[_ET] 

334 ) -> _ListenerCollection[_ET]: 

335 """Return an event collection which can be modified. 

336 

337 For _EmptyListener at the instance level of 

338 a dispatcher, this generates a new 

339 _ListenerCollection, applies it to the instance, 

340 and returns it. 

341 

342 """ 

343 obj = cast("_Dispatch[_ET]", obj) 

344 

345 assert obj._instance_cls is not None 

346 result = _ListenerCollection(self.parent, obj._instance_cls) 

347 if getattr(obj, self.name) is self: 

348 setattr(obj, self.name, result) 

349 else: 

350 assert isinstance(getattr(obj, self.name), _JoinedListener) 

351 return result 

352 

353 def _needs_modify(self, *args: Any, **kw: Any) -> NoReturn: 

354 raise NotImplementedError("need to call for_modify()") 

355 

356 def exec_once(self, *args: Any, **kw: Any) -> NoReturn: 

357 self._needs_modify(*args, **kw) 

358 

359 def exec_once_unless_exception(self, *args: Any, **kw: Any) -> NoReturn: 

360 self._needs_modify(*args, **kw) 

361 

362 def insert(self, *args: Any, **kw: Any) -> NoReturn: 

363 self._needs_modify(*args, **kw) 

364 

365 def append(self, *args: Any, **kw: Any) -> NoReturn: 

366 self._needs_modify(*args, **kw) 

367 

368 def remove(self, *args: Any, **kw: Any) -> NoReturn: 

369 self._needs_modify(*args, **kw) 

370 

371 def clear(self, *args: Any, **kw: Any) -> NoReturn: 

372 self._needs_modify(*args, **kw) 

373 

374 def __call__(self, *args: Any, **kw: Any) -> None: 

375 """Execute this event.""" 

376 

377 for fn in self.parent_listeners: 

378 fn(*args, **kw) 

379 

380 def __contains__(self, item: Any) -> bool: 

381 return item in self.parent_listeners 

382 

383 def __len__(self) -> int: 

384 return len(self.parent_listeners) 

385 

386 def __iter__(self) -> Iterator[_ListenerFnType]: 

387 return iter(self.parent_listeners) 

388 

389 def __bool__(self) -> bool: 

390 return bool(self.parent_listeners) 

391 

392 

393class _MutexProtocol(Protocol): 

394 def __enter__(self) -> bool: ... 

395 

396 def __exit__( 

397 self, 

398 exc_type: Optional[Type[BaseException]], 

399 exc_val: Optional[BaseException], 

400 exc_tb: Optional[TracebackType], 

401 ) -> Optional[bool]: ... 

402 

403 

404class _CompoundListener(_InstanceLevelDispatch[_ET]): 

405 __slots__ = ( 

406 "_exec_once_mutex", 

407 "_exec_once", 

408 "_exec_w_sync_once", 

409 "_is_asyncio", 

410 ) 

411 

412 _exec_once_mutex: _MutexProtocol 

413 parent_listeners: Collection[_ListenerFnType] 

414 listeners: Collection[_ListenerFnType] 

415 _exec_once: bool 

416 _exec_w_sync_once: bool 

417 

418 def __init__(self, *arg: Any, **kw: Any): 

419 super().__init__(*arg, **kw) 

420 self._is_asyncio = False 

421 

422 def _set_asyncio(self) -> None: 

423 self._is_asyncio = True 

424 

425 def _memoized_attr__exec_once_mutex(self) -> _MutexProtocol: 

426 if self._is_asyncio: 

427 return AsyncAdaptedLock() 

428 else: 

429 return threading.Lock() 

430 

431 def _exec_once_impl( 

432 self, retry_on_exception: bool, *args: Any, **kw: Any 

433 ) -> None: 

434 with self._exec_once_mutex: 

435 if not self._exec_once: 

436 try: 

437 self(*args, **kw) 

438 exception = False 

439 except: 

440 exception = True 

441 raise 

442 finally: 

443 if not exception or not retry_on_exception: 

444 self._exec_once = True 

445 

446 def exec_once(self, *args: Any, **kw: Any) -> None: 

447 """Execute this event, but only if it has not been 

448 executed already for this collection.""" 

449 

450 if not self._exec_once: 

451 self._exec_once_impl(False, *args, **kw) 

452 

453 def exec_once_unless_exception(self, *args: Any, **kw: Any) -> None: 

454 """Execute this event, but only if it has not been 

455 executed already for this collection, or was called 

456 by a previous exec_once_unless_exception call and 

457 raised an exception. 

458 

459 If exec_once was already called, then this method will never run 

460 the callable regardless of whether it raised or not. 

461 

462 """ 

463 if not self._exec_once: 

464 self._exec_once_impl(True, *args, **kw) 

465 

466 def _exec_w_sync_on_first_run(self, *args: Any, **kw: Any) -> None: 

467 """Execute this event, and use a mutex if it has not been 

468 executed already for this collection, or was called 

469 by a previous _exec_w_sync_on_first_run call and 

470 raised an exception. 

471 

472 If _exec_w_sync_on_first_run was already called and didn't raise an 

473 exception, then a mutex is not used. 

474 

475 .. versionadded:: 1.4.11 

476 

477 """ 

478 if not self._exec_w_sync_once: 

479 with self._exec_once_mutex: 

480 try: 

481 self(*args, **kw) 

482 except: 

483 raise 

484 else: 

485 self._exec_w_sync_once = True 

486 else: 

487 self(*args, **kw) 

488 

489 def __call__(self, *args: Any, **kw: Any) -> None: 

490 """Execute this event.""" 

491 

492 for fn in self.parent_listeners: 

493 fn(*args, **kw) 

494 for fn in self.listeners: 

495 fn(*args, **kw) 

496 

497 def __contains__(self, item: Any) -> bool: 

498 return item in self.parent_listeners or item in self.listeners 

499 

500 def __len__(self) -> int: 

501 return len(self.parent_listeners) + len(self.listeners) 

502 

503 def __iter__(self) -> Iterator[_ListenerFnType]: 

504 return chain(self.parent_listeners, self.listeners) 

505 

506 def __bool__(self) -> bool: 

507 return bool(self.listeners or self.parent_listeners) 

508 

509 

510class _ListenerCollection(_CompoundListener[_ET]): 

511 """Instance-level attributes on instances of :class:`._Dispatch`. 

512 

513 Represents a collection of listeners. 

514 

515 As of 0.7.9, _ListenerCollection is only first 

516 created via the _EmptyListener.for_modify() method. 

517 

518 """ 

519 

520 __slots__ = ( 

521 "parent_listeners", 

522 "parent", 

523 "name", 

524 "listeners", 

525 "propagate", 

526 "__weakref__", 

527 ) 

528 

529 parent_listeners: Collection[_ListenerFnType] 

530 parent: _ClsLevelDispatch[_ET] 

531 name: str 

532 listeners: Deque[_ListenerFnType] 

533 propagate: Set[_ListenerFnType] 

534 

535 def __init__(self, parent: _ClsLevelDispatch[_ET], target_cls: Type[_ET]): 

536 super().__init__() 

537 if target_cls not in parent._clslevel: 

538 parent.update_subclass(target_cls) 

539 self._exec_once = False 

540 self._exec_w_sync_once = False 

541 self.parent_listeners = parent._clslevel[target_cls] 

542 self.parent = parent 

543 self.name = parent.name 

544 self.listeners = collections.deque() 

545 self.propagate = set() 

546 

547 def for_modify( 

548 self, obj: _DispatchCommon[_ET] 

549 ) -> _ListenerCollection[_ET]: 

550 """Return an event collection which can be modified. 

551 

552 For _ListenerCollection at the instance level of 

553 a dispatcher, this returns self. 

554 

555 """ 

556 return self 

557 

558 def _update( 

559 self, other: _ListenerCollection[_ET], only_propagate: bool = True 

560 ) -> None: 

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

562 object.""" 

563 existing_listeners = self.listeners 

564 existing_listener_set = set(existing_listeners) 

565 self.propagate.update(other.propagate) 

566 other_listeners = [ 

567 l 

568 for l in other.listeners 

569 if l not in existing_listener_set 

570 and not only_propagate 

571 or l in self.propagate 

572 ] 

573 

574 existing_listeners.extend(other_listeners) 

575 

576 if other._is_asyncio: 

577 self._set_asyncio() 

578 

579 to_associate = other.propagate.union(other_listeners) 

580 registry._stored_in_collection_multi(self, other, to_associate) 

581 

582 def insert(self, event_key: _EventKey[_ET], propagate: bool) -> None: 

583 if event_key.prepend_to_list(self, self.listeners): 

584 if propagate: 

585 self.propagate.add(event_key._listen_fn) 

586 

587 def append(self, event_key: _EventKey[_ET], propagate: bool) -> None: 

588 if event_key.append_to_list(self, self.listeners): 

589 if propagate: 

590 self.propagate.add(event_key._listen_fn) 

591 

592 def remove(self, event_key: _EventKey[_ET]) -> None: 

593 self.listeners.remove(event_key._listen_fn) 

594 self.propagate.discard(event_key._listen_fn) 

595 registry._removed_from_collection(event_key, self) 

596 

597 def clear(self) -> None: 

598 registry._clear(self, self.listeners) 

599 self.propagate.clear() 

600 self.listeners.clear() 

601 

602 

603class _JoinedListener(_CompoundListener[_ET]): 

604 __slots__ = "parent_dispatch", "name", "local", "parent_listeners" 

605 

606 parent_dispatch: _DispatchCommon[_ET] 

607 name: str 

608 local: _InstanceLevelDispatch[_ET] 

609 parent_listeners: Collection[_ListenerFnType] 

610 

611 def __init__( 

612 self, 

613 parent_dispatch: _DispatchCommon[_ET], 

614 name: str, 

615 local: _EmptyListener[_ET], 

616 ): 

617 self._exec_once = False 

618 self.parent_dispatch = parent_dispatch 

619 self.name = name 

620 self.local = local 

621 self.parent_listeners = self.local 

622 

623 if not typing.TYPE_CHECKING: 

624 # first error, I don't really understand: 

625 # Signature of "listeners" incompatible with 

626 # supertype "_CompoundListener" [override] 

627 # the name / return type are exactly the same 

628 # second error is getattr_isn't typed, the cast() here 

629 # adds too much method overhead 

630 @property 

631 def listeners(self) -> Collection[_ListenerFnType]: 

632 return getattr(self.parent_dispatch, self.name) 

633 

634 def _adjust_fn_spec( 

635 self, fn: _ListenerFnType, named: bool 

636 ) -> _ListenerFnType: 

637 return self.local._adjust_fn_spec(fn, named) 

638 

639 def for_modify(self, obj: _DispatchCommon[_ET]) -> _JoinedListener[_ET]: 

640 self.local = self.parent_listeners = self.local.for_modify(obj) 

641 return self 

642 

643 def insert(self, event_key: _EventKey[_ET], propagate: bool) -> None: 

644 self.local.insert(event_key, propagate) 

645 

646 def append(self, event_key: _EventKey[_ET], propagate: bool) -> None: 

647 self.local.append(event_key, propagate) 

648 

649 def remove(self, event_key: _EventKey[_ET]) -> None: 

650 self.local.remove(event_key) 

651 

652 def clear(self) -> None: 

653 raise NotImplementedError()