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 existing = getattr(obj, self.name) 

347 

348 with util.mini_gil: 

349 if existing is self or isinstance(existing, _JoinedListener): 

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

351 else: 

352 # this codepath is an extremely rare race condition 

353 # that has been observed in test_pool.py->test_timeout_race 

354 # with freethreaded. 

355 assert isinstance(existing, _ListenerCollection) 

356 return existing 

357 

358 if existing is self: 

359 setattr(obj, self.name, result) 

360 return result 

361 

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

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

364 

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

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

367 

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

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

370 

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

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

373 

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

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

376 

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

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

379 

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

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

382 

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

384 """Execute this event.""" 

385 

386 for fn in self.parent_listeners: 

387 fn(*args, **kw) 

388 

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

390 return item in self.parent_listeners 

391 

392 def __len__(self) -> int: 

393 return len(self.parent_listeners) 

394 

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

396 return iter(self.parent_listeners) 

397 

398 def __bool__(self) -> bool: 

399 return bool(self.parent_listeners) 

400 

401 

402class _MutexProtocol(Protocol): 

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

404 

405 def __exit__( 

406 self, 

407 exc_type: Optional[Type[BaseException]], 

408 exc_val: Optional[BaseException], 

409 exc_tb: Optional[TracebackType], 

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

411 

412 

413class _CompoundListener(_InstanceLevelDispatch[_ET]): 

414 __slots__ = ( 

415 "_exec_once_mutex", 

416 "_exec_once", 

417 "_exec_w_sync_once", 

418 "_is_asyncio", 

419 ) 

420 

421 _exec_once_mutex: Optional[_MutexProtocol] 

422 parent_listeners: Collection[_ListenerFnType] 

423 listeners: Collection[_ListenerFnType] 

424 _exec_once: bool 

425 _exec_w_sync_once: bool 

426 

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

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

429 self._is_asyncio = False 

430 

431 def _set_asyncio(self) -> None: 

432 self._is_asyncio = True 

433 

434 def _get_exec_once_mutex(self) -> _MutexProtocol: 

435 with util.mini_gil: 

436 if self._exec_once_mutex is not None: 

437 return self._exec_once_mutex 

438 

439 if self._is_asyncio: 

440 mutex = AsyncAdaptedLock() 

441 else: 

442 mutex = threading.Lock() # type: ignore[assignment] 

443 self._exec_once_mutex = mutex 

444 

445 return mutex 

446 

447 def _exec_once_impl( 

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

449 ) -> None: 

450 with self._get_exec_once_mutex(): 

451 if not self._exec_once: 

452 try: 

453 self(*args, **kw) 

454 exception = False 

455 except: 

456 exception = True 

457 raise 

458 finally: 

459 if not exception or not retry_on_exception: 

460 self._exec_once = True 

461 

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

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

464 executed already for this collection.""" 

465 

466 if not self._exec_once: 

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

468 

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

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

471 executed already for this collection, or was called 

472 by a previous exec_once_unless_exception call and 

473 raised an exception. 

474 

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

476 the callable regardless of whether it raised or not. 

477 

478 """ 

479 if not self._exec_once: 

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

481 

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

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

484 executed already for this collection, or was called 

485 by a previous _exec_w_sync_on_first_run call and 

486 raised an exception. 

487 

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

489 exception, then a mutex is not used. It's not guaranteed 

490 the mutex won't be used more than once in the case of very rare 

491 race conditions. 

492 

493 .. versionadded:: 1.4.11 

494 

495 """ 

496 if not self._exec_w_sync_once: 

497 with self._get_exec_once_mutex(): 

498 try: 

499 self(*args, **kw) 

500 except: 

501 raise 

502 else: 

503 self._exec_w_sync_once = True 

504 else: 

505 self(*args, **kw) 

506 

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

508 """Execute this event.""" 

509 

510 for fn in self.parent_listeners: 

511 fn(*args, **kw) 

512 for fn in self.listeners: 

513 fn(*args, **kw) 

514 

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

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

517 

518 def __len__(self) -> int: 

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

520 

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

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

523 

524 def __bool__(self) -> bool: 

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

526 

527 

528class _ListenerCollection(_CompoundListener[_ET]): 

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

530 

531 Represents a collection of listeners. 

532 

533 As of 0.7.9, _ListenerCollection is only first 

534 created via the _EmptyListener.for_modify() method. 

535 

536 """ 

537 

538 __slots__ = ( 

539 "parent_listeners", 

540 "parent", 

541 "name", 

542 "listeners", 

543 "propagate", 

544 "__weakref__", 

545 ) 

546 

547 parent_listeners: Collection[_ListenerFnType] 

548 parent: _ClsLevelDispatch[_ET] 

549 name: str 

550 listeners: Deque[_ListenerFnType] 

551 propagate: Set[_ListenerFnType] 

552 

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

554 super().__init__() 

555 if target_cls not in parent._clslevel: 

556 parent.update_subclass(target_cls) 

557 self._exec_once = False 

558 self._exec_w_sync_once = False 

559 self._exec_once_mutex = None 

560 self.parent_listeners = parent._clslevel[target_cls] 

561 self.parent = parent 

562 self.name = parent.name 

563 self.listeners = collections.deque() 

564 self.propagate = set() 

565 

566 def for_modify( 

567 self, obj: _DispatchCommon[_ET] 

568 ) -> _ListenerCollection[_ET]: 

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

570 

571 For _ListenerCollection at the instance level of 

572 a dispatcher, this returns self. 

573 

574 """ 

575 return self 

576 

577 def _update( 

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

579 ) -> None: 

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

581 object.""" 

582 existing_listeners = self.listeners 

583 existing_listener_set = set(existing_listeners) 

584 self.propagate.update(other.propagate) 

585 other_listeners = [ 

586 l 

587 for l in other.listeners 

588 if l not in existing_listener_set 

589 and not only_propagate 

590 or l in self.propagate 

591 ] 

592 

593 existing_listeners.extend(other_listeners) 

594 

595 if other._is_asyncio: 

596 self._set_asyncio() 

597 

598 to_associate = other.propagate.union(other_listeners) 

599 registry._stored_in_collection_multi(self, other, to_associate) 

600 

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

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

603 if propagate: 

604 self.propagate.add(event_key._listen_fn) 

605 

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

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

608 if propagate: 

609 self.propagate.add(event_key._listen_fn) 

610 

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

612 self.listeners.remove(event_key._listen_fn) 

613 self.propagate.discard(event_key._listen_fn) 

614 registry._removed_from_collection(event_key, self) 

615 

616 def clear(self) -> None: 

617 registry._clear(self, self.listeners) 

618 self.propagate.clear() 

619 self.listeners.clear() 

620 

621 

622class _JoinedListener(_CompoundListener[_ET]): 

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

624 

625 parent_dispatch: _DispatchCommon[_ET] 

626 name: str 

627 local: _InstanceLevelDispatch[_ET] 

628 parent_listeners: Collection[_ListenerFnType] 

629 

630 def __init__( 

631 self, 

632 parent_dispatch: _DispatchCommon[_ET], 

633 name: str, 

634 local: _EmptyListener[_ET], 

635 ): 

636 self._exec_once = False 

637 self._exec_w_sync_once = False 

638 self._exec_once_mutex = None 

639 self.parent_dispatch = parent_dispatch 

640 self.name = name 

641 self.local = local 

642 self.parent_listeners = self.local 

643 

644 if not typing.TYPE_CHECKING: 

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

646 # Signature of "listeners" incompatible with 

647 # supertype "_CompoundListener" [override] 

648 # the name / return type are exactly the same 

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

650 # adds too much method overhead 

651 @property 

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

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

654 

655 def _adjust_fn_spec( 

656 self, fn: _ListenerFnType, named: bool 

657 ) -> _ListenerFnType: 

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

659 

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

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

662 return self 

663 

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

665 self.local.insert(event_key, propagate) 

666 

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

668 self.local.append(event_key, propagate) 

669 

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

671 self.local.remove(event_key) 

672 

673 def clear(self) -> None: 

674 raise NotImplementedError()