attr.py 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676
  1. # event/attr.py
  2. # Copyright (C) 2005-2026 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. """Attribute implementation for _Dispatch classes.
  8. The various listener targets for a particular event class are represented
  9. as attributes, which refer to collections of listeners to be fired off.
  10. These collections can exist at the class level as well as at the instance
  11. level. An event is fired off using code like this::
  12. some_object.dispatch.first_connect(arg1, arg2)
  13. Above, ``some_object.dispatch`` would be an instance of ``_Dispatch`` and
  14. ``first_connect`` is typically an instance of ``_ListenerCollection``
  15. if event listeners are present, or ``_EmptyListener`` if none are present.
  16. The attribute mechanics here spend effort trying to ensure listener functions
  17. are available with a minimum of function call overhead, that unnecessary
  18. objects aren't created (i.e. many empty per-instance listener collections),
  19. as well as that everything is garbage collectable when owning references are
  20. lost. Other features such as "propagation" of listener functions across
  21. many ``_Dispatch`` instances, "joining" of multiple ``_Dispatch`` instances,
  22. as well as support for subclass propagation (e.g. events assigned to
  23. ``Pool`` vs. ``QueuePool``) are all implemented here.
  24. """
  25. from __future__ import annotations
  26. import collections
  27. from itertools import chain
  28. import threading
  29. from types import TracebackType
  30. import typing
  31. from typing import Any
  32. from typing import cast
  33. from typing import Collection
  34. from typing import Deque
  35. from typing import FrozenSet
  36. from typing import Generic
  37. from typing import Iterator
  38. from typing import MutableMapping
  39. from typing import MutableSequence
  40. from typing import NoReturn
  41. from typing import Optional
  42. from typing import Sequence
  43. from typing import Set
  44. from typing import Tuple
  45. from typing import Type
  46. from typing import TypeVar
  47. from typing import Union
  48. import weakref
  49. from . import legacy
  50. from . import registry
  51. from .registry import _ET
  52. from .registry import _EventKey
  53. from .registry import _ListenerFnType
  54. from .. import exc
  55. from .. import util
  56. from ..util.concurrency import AsyncAdaptedLock
  57. from ..util.typing import Protocol
  58. _T = TypeVar("_T", bound=Any)
  59. if typing.TYPE_CHECKING:
  60. from .base import _Dispatch
  61. from .base import _DispatchCommon
  62. from .base import _HasEventsDispatch
  63. class RefCollection(util.MemoizedSlots, Generic[_ET]):
  64. __slots__ = ("ref",)
  65. ref: weakref.ref[RefCollection[_ET]]
  66. def _memoized_attr_ref(self) -> weakref.ref[RefCollection[_ET]]:
  67. return weakref.ref(self, registry._collection_gced)
  68. class _empty_collection(Collection[_T]):
  69. def append(self, element: _T) -> None:
  70. pass
  71. def appendleft(self, element: _T) -> None:
  72. pass
  73. def extend(self, other: Sequence[_T]) -> None:
  74. pass
  75. def remove(self, element: _T) -> None:
  76. pass
  77. def __contains__(self, element: Any) -> bool:
  78. return False
  79. def __iter__(self) -> Iterator[_T]:
  80. return iter([])
  81. def clear(self) -> None:
  82. pass
  83. def __len__(self) -> int:
  84. return 0
  85. _ListenerFnSequenceType = Union[Deque[_T], _empty_collection[_T]]
  86. class _ClsLevelDispatch(RefCollection[_ET]):
  87. """Class-level events on :class:`._Dispatch` classes."""
  88. __slots__ = (
  89. "clsname",
  90. "name",
  91. "arg_names",
  92. "has_kw",
  93. "legacy_signatures",
  94. "_clslevel",
  95. "__weakref__",
  96. )
  97. clsname: str
  98. name: str
  99. arg_names: Sequence[str]
  100. has_kw: bool
  101. legacy_signatures: MutableSequence[legacy._LegacySignatureType]
  102. _clslevel: MutableMapping[
  103. Type[_ET], _ListenerFnSequenceType[_ListenerFnType]
  104. ]
  105. def __init__(
  106. self,
  107. parent_dispatch_cls: Type[_HasEventsDispatch[_ET]],
  108. fn: _ListenerFnType,
  109. ):
  110. self.name = fn.__name__
  111. self.clsname = parent_dispatch_cls.__name__
  112. argspec = util.inspect_getfullargspec(fn)
  113. self.arg_names = argspec.args[1:]
  114. self.has_kw = bool(argspec.varkw)
  115. self.legacy_signatures = list(
  116. reversed(
  117. sorted(
  118. getattr(fn, "_legacy_signatures", []), key=lambda s: s[0]
  119. )
  120. )
  121. )
  122. fn.__doc__ = legacy._augment_fn_docs(self, parent_dispatch_cls, fn)
  123. self._clslevel = weakref.WeakKeyDictionary()
  124. def _adjust_fn_spec(
  125. self, fn: _ListenerFnType, named: bool
  126. ) -> _ListenerFnType:
  127. if named:
  128. fn = self._wrap_fn_for_kw(fn)
  129. if self.legacy_signatures:
  130. try:
  131. argspec = util.get_callable_argspec(fn, no_self=True)
  132. except TypeError:
  133. pass
  134. else:
  135. fn = legacy._wrap_fn_for_legacy(self, fn, argspec)
  136. return fn
  137. def _wrap_fn_for_kw(self, fn: _ListenerFnType) -> _ListenerFnType:
  138. def wrap_kw(*args: Any, **kw: Any) -> Any:
  139. argdict = dict(zip(self.arg_names, args))
  140. argdict.update(kw)
  141. return fn(**argdict)
  142. return wrap_kw
  143. def _do_insert_or_append(
  144. self, event_key: _EventKey[_ET], is_append: bool
  145. ) -> None:
  146. target = event_key.dispatch_target
  147. assert isinstance(
  148. target, type
  149. ), "Class-level Event targets must be classes."
  150. if not getattr(target, "_sa_propagate_class_events", True):
  151. raise exc.InvalidRequestError(
  152. f"Can't assign an event directly to the {target} class"
  153. )
  154. cls: Type[_ET]
  155. for cls in util.walk_subclasses(target):
  156. if cls is not target and cls not in self._clslevel:
  157. self.update_subclass(cls)
  158. else:
  159. if cls not in self._clslevel:
  160. self.update_subclass(cls)
  161. if is_append:
  162. self._clslevel[cls].append(event_key._listen_fn)
  163. else:
  164. self._clslevel[cls].appendleft(event_key._listen_fn)
  165. registry._stored_in_collection(event_key, self)
  166. def insert(self, event_key: _EventKey[_ET], propagate: bool) -> None:
  167. self._do_insert_or_append(event_key, is_append=False)
  168. def append(self, event_key: _EventKey[_ET], propagate: bool) -> None:
  169. self._do_insert_or_append(event_key, is_append=True)
  170. def update_subclass(self, target: Type[_ET]) -> None:
  171. if target not in self._clslevel:
  172. if getattr(target, "_sa_propagate_class_events", True):
  173. self._clslevel[target] = collections.deque()
  174. else:
  175. self._clslevel[target] = _empty_collection()
  176. clslevel = self._clslevel[target]
  177. cls: Type[_ET]
  178. for cls in target.__mro__[1:]:
  179. if cls in self._clslevel:
  180. clslevel.extend(
  181. [fn for fn in self._clslevel[cls] if fn not in clslevel]
  182. )
  183. def remove(self, event_key: _EventKey[_ET]) -> None:
  184. target = event_key.dispatch_target
  185. cls: Type[_ET]
  186. for cls in util.walk_subclasses(target):
  187. if cls in self._clslevel:
  188. self._clslevel[cls].remove(event_key._listen_fn)
  189. registry._removed_from_collection(event_key, self)
  190. def clear(self) -> None:
  191. """Clear all class level listeners"""
  192. to_clear: Set[_ListenerFnType] = set()
  193. for dispatcher in self._clslevel.values():
  194. to_clear.update(dispatcher)
  195. dispatcher.clear()
  196. registry._clear(self, to_clear)
  197. def for_modify(self, obj: _Dispatch[_ET]) -> _ClsLevelDispatch[_ET]:
  198. """Return an event collection which can be modified.
  199. For _ClsLevelDispatch at the class level of
  200. a dispatcher, this returns self.
  201. """
  202. return self
  203. class _InstanceLevelDispatch(RefCollection[_ET], Collection[_ListenerFnType]):
  204. __slots__ = ()
  205. parent: _ClsLevelDispatch[_ET]
  206. def _adjust_fn_spec(
  207. self, fn: _ListenerFnType, named: bool
  208. ) -> _ListenerFnType:
  209. return self.parent._adjust_fn_spec(fn, named)
  210. def __contains__(self, item: Any) -> bool:
  211. raise NotImplementedError()
  212. def __len__(self) -> int:
  213. raise NotImplementedError()
  214. def __iter__(self) -> Iterator[_ListenerFnType]:
  215. raise NotImplementedError()
  216. def __bool__(self) -> bool:
  217. raise NotImplementedError()
  218. def exec_once(self, *args: Any, **kw: Any) -> None:
  219. raise NotImplementedError()
  220. def exec_once_unless_exception(self, *args: Any, **kw: Any) -> None:
  221. raise NotImplementedError()
  222. def _exec_w_sync_on_first_run(self, *args: Any, **kw: Any) -> None:
  223. raise NotImplementedError()
  224. def __call__(self, *args: Any, **kw: Any) -> None:
  225. raise NotImplementedError()
  226. def insert(self, event_key: _EventKey[_ET], propagate: bool) -> None:
  227. raise NotImplementedError()
  228. def append(self, event_key: _EventKey[_ET], propagate: bool) -> None:
  229. raise NotImplementedError()
  230. def remove(self, event_key: _EventKey[_ET]) -> None:
  231. raise NotImplementedError()
  232. def for_modify(
  233. self, obj: _DispatchCommon[_ET]
  234. ) -> _InstanceLevelDispatch[_ET]:
  235. """Return an event collection which can be modified.
  236. For _ClsLevelDispatch at the class level of
  237. a dispatcher, this returns self.
  238. """
  239. return self
  240. class _EmptyListener(_InstanceLevelDispatch[_ET]):
  241. """Serves as a proxy interface to the events
  242. served by a _ClsLevelDispatch, when there are no
  243. instance-level events present.
  244. Is replaced by _ListenerCollection when instance-level
  245. events are added.
  246. """
  247. __slots__ = "parent", "parent_listeners", "name"
  248. propagate: FrozenSet[_ListenerFnType] = frozenset()
  249. listeners: Tuple[()] = ()
  250. parent: _ClsLevelDispatch[_ET]
  251. parent_listeners: _ListenerFnSequenceType[_ListenerFnType]
  252. name: str
  253. def __init__(self, parent: _ClsLevelDispatch[_ET], target_cls: Type[_ET]):
  254. if target_cls not in parent._clslevel:
  255. parent.update_subclass(target_cls)
  256. self.parent = parent
  257. self.parent_listeners = parent._clslevel[target_cls]
  258. self.name = parent.name
  259. def for_modify(
  260. self, obj: _DispatchCommon[_ET]
  261. ) -> _ListenerCollection[_ET]:
  262. """Return an event collection which can be modified.
  263. For _EmptyListener at the instance level of
  264. a dispatcher, this generates a new
  265. _ListenerCollection, applies it to the instance,
  266. and returns it.
  267. """
  268. obj = cast("_Dispatch[_ET]", obj)
  269. assert obj._instance_cls is not None
  270. existing = getattr(obj, self.name)
  271. with util.mini_gil:
  272. if existing is self or isinstance(existing, _JoinedListener):
  273. result = _ListenerCollection(self.parent, obj._instance_cls)
  274. else:
  275. # this codepath is an extremely rare race condition
  276. # that has been observed in test_pool.py->test_timeout_race
  277. # with freethreaded.
  278. assert isinstance(existing, _ListenerCollection)
  279. return existing
  280. if existing is self:
  281. setattr(obj, self.name, result)
  282. return result
  283. def _needs_modify(self, *args: Any, **kw: Any) -> NoReturn:
  284. raise NotImplementedError("need to call for_modify()")
  285. def exec_once(self, *args: Any, **kw: Any) -> NoReturn:
  286. self._needs_modify(*args, **kw)
  287. def exec_once_unless_exception(self, *args: Any, **kw: Any) -> NoReturn:
  288. self._needs_modify(*args, **kw)
  289. def insert(self, *args: Any, **kw: Any) -> NoReturn:
  290. self._needs_modify(*args, **kw)
  291. def append(self, *args: Any, **kw: Any) -> NoReturn:
  292. self._needs_modify(*args, **kw)
  293. def remove(self, *args: Any, **kw: Any) -> NoReturn:
  294. self._needs_modify(*args, **kw)
  295. def clear(self, *args: Any, **kw: Any) -> NoReturn:
  296. self._needs_modify(*args, **kw)
  297. def __call__(self, *args: Any, **kw: Any) -> None:
  298. """Execute this event."""
  299. for fn in self.parent_listeners:
  300. fn(*args, **kw)
  301. def __contains__(self, item: Any) -> bool:
  302. return item in self.parent_listeners
  303. def __len__(self) -> int:
  304. return len(self.parent_listeners)
  305. def __iter__(self) -> Iterator[_ListenerFnType]:
  306. return iter(self.parent_listeners)
  307. def __bool__(self) -> bool:
  308. return bool(self.parent_listeners)
  309. class _MutexProtocol(Protocol):
  310. def __enter__(self) -> bool: ...
  311. def __exit__(
  312. self,
  313. exc_type: Optional[Type[BaseException]],
  314. exc_val: Optional[BaseException],
  315. exc_tb: Optional[TracebackType],
  316. ) -> Optional[bool]: ...
  317. class _CompoundListener(_InstanceLevelDispatch[_ET]):
  318. __slots__ = (
  319. "_exec_once_mutex",
  320. "_exec_once",
  321. "_exec_w_sync_once",
  322. "_is_asyncio",
  323. )
  324. _exec_once_mutex: Optional[_MutexProtocol]
  325. parent_listeners: Collection[_ListenerFnType]
  326. listeners: Collection[_ListenerFnType]
  327. _exec_once: bool
  328. _exec_w_sync_once: bool
  329. def __init__(self, *arg: Any, **kw: Any):
  330. super().__init__(*arg, **kw)
  331. self._is_asyncio = False
  332. def _set_asyncio(self) -> None:
  333. self._is_asyncio = True
  334. def _get_exec_once_mutex(self) -> _MutexProtocol:
  335. with util.mini_gil:
  336. if self._exec_once_mutex is not None:
  337. return self._exec_once_mutex
  338. if self._is_asyncio:
  339. mutex = AsyncAdaptedLock()
  340. else:
  341. mutex = threading.Lock() # type: ignore[assignment]
  342. self._exec_once_mutex = mutex
  343. return mutex
  344. def _exec_once_impl(
  345. self, retry_on_exception: bool, *args: Any, **kw: Any
  346. ) -> None:
  347. with self._get_exec_once_mutex():
  348. if not self._exec_once:
  349. try:
  350. self(*args, **kw)
  351. exception = False
  352. except:
  353. exception = True
  354. raise
  355. finally:
  356. if not exception or not retry_on_exception:
  357. self._exec_once = True
  358. def exec_once(self, *args: Any, **kw: Any) -> None:
  359. """Execute this event, but only if it has not been
  360. executed already for this collection."""
  361. if not self._exec_once:
  362. self._exec_once_impl(False, *args, **kw)
  363. def exec_once_unless_exception(self, *args: Any, **kw: Any) -> None:
  364. """Execute this event, but only if it has not been
  365. executed already for this collection, or was called
  366. by a previous exec_once_unless_exception call and
  367. raised an exception.
  368. If exec_once was already called, then this method will never run
  369. the callable regardless of whether it raised or not.
  370. .. versionadded:: 1.3.8
  371. """
  372. if not self._exec_once:
  373. self._exec_once_impl(True, *args, **kw)
  374. def _exec_w_sync_on_first_run(self, *args: Any, **kw: Any) -> None:
  375. """Execute this event, and use a mutex if it has not been
  376. executed already for this collection, or was called
  377. by a previous _exec_w_sync_on_first_run call and
  378. raised an exception.
  379. If _exec_w_sync_on_first_run was already called and didn't raise an
  380. exception, then a mutex is not used. It's not guaranteed
  381. the mutex won't be used more than once in the case of very rare
  382. race conditions.
  383. .. versionadded:: 1.4.11
  384. """
  385. if not self._exec_w_sync_once:
  386. with self._get_exec_once_mutex():
  387. try:
  388. self(*args, **kw)
  389. except:
  390. raise
  391. else:
  392. self._exec_w_sync_once = True
  393. else:
  394. self(*args, **kw)
  395. def __call__(self, *args: Any, **kw: Any) -> None:
  396. """Execute this event."""
  397. for fn in self.parent_listeners:
  398. fn(*args, **kw)
  399. for fn in self.listeners:
  400. fn(*args, **kw)
  401. def __contains__(self, item: Any) -> bool:
  402. return item in self.parent_listeners or item in self.listeners
  403. def __len__(self) -> int:
  404. return len(self.parent_listeners) + len(self.listeners)
  405. def __iter__(self) -> Iterator[_ListenerFnType]:
  406. return chain(self.parent_listeners, self.listeners)
  407. def __bool__(self) -> bool:
  408. return bool(self.listeners or self.parent_listeners)
  409. class _ListenerCollection(_CompoundListener[_ET]):
  410. """Instance-level attributes on instances of :class:`._Dispatch`.
  411. Represents a collection of listeners.
  412. As of 0.7.9, _ListenerCollection is only first
  413. created via the _EmptyListener.for_modify() method.
  414. """
  415. __slots__ = (
  416. "parent_listeners",
  417. "parent",
  418. "name",
  419. "listeners",
  420. "propagate",
  421. "__weakref__",
  422. )
  423. parent_listeners: Collection[_ListenerFnType]
  424. parent: _ClsLevelDispatch[_ET]
  425. name: str
  426. listeners: Deque[_ListenerFnType]
  427. propagate: Set[_ListenerFnType]
  428. def __init__(self, parent: _ClsLevelDispatch[_ET], target_cls: Type[_ET]):
  429. super().__init__()
  430. if target_cls not in parent._clslevel:
  431. parent.update_subclass(target_cls)
  432. self._exec_once = False
  433. self._exec_w_sync_once = False
  434. self._exec_once_mutex = None
  435. self.parent_listeners = parent._clslevel[target_cls]
  436. self.parent = parent
  437. self.name = parent.name
  438. self.listeners = collections.deque()
  439. self.propagate = set()
  440. def for_modify(
  441. self, obj: _DispatchCommon[_ET]
  442. ) -> _ListenerCollection[_ET]:
  443. """Return an event collection which can be modified.
  444. For _ListenerCollection at the instance level of
  445. a dispatcher, this returns self.
  446. """
  447. return self
  448. def _update(
  449. self, other: _ListenerCollection[_ET], only_propagate: bool = True
  450. ) -> None:
  451. """Populate from the listeners in another :class:`_Dispatch`
  452. object."""
  453. existing_listeners = self.listeners
  454. existing_listener_set = set(existing_listeners)
  455. self.propagate.update(other.propagate)
  456. other_listeners = [
  457. l
  458. for l in other.listeners
  459. if l not in existing_listener_set
  460. and not only_propagate
  461. or l in self.propagate
  462. ]
  463. existing_listeners.extend(other_listeners)
  464. if other._is_asyncio:
  465. self._set_asyncio()
  466. to_associate = other.propagate.union(other_listeners)
  467. registry._stored_in_collection_multi(self, other, to_associate)
  468. def insert(self, event_key: _EventKey[_ET], propagate: bool) -> None:
  469. if event_key.prepend_to_list(self, self.listeners):
  470. if propagate:
  471. self.propagate.add(event_key._listen_fn)
  472. def append(self, event_key: _EventKey[_ET], propagate: bool) -> None:
  473. if event_key.append_to_list(self, self.listeners):
  474. if propagate:
  475. self.propagate.add(event_key._listen_fn)
  476. def remove(self, event_key: _EventKey[_ET]) -> None:
  477. self.listeners.remove(event_key._listen_fn)
  478. self.propagate.discard(event_key._listen_fn)
  479. registry._removed_from_collection(event_key, self)
  480. def clear(self) -> None:
  481. registry._clear(self, self.listeners)
  482. self.propagate.clear()
  483. self.listeners.clear()
  484. class _JoinedListener(_CompoundListener[_ET]):
  485. __slots__ = "parent_dispatch", "name", "local", "parent_listeners"
  486. parent_dispatch: _DispatchCommon[_ET]
  487. name: str
  488. local: _InstanceLevelDispatch[_ET]
  489. parent_listeners: Collection[_ListenerFnType]
  490. def __init__(
  491. self,
  492. parent_dispatch: _DispatchCommon[_ET],
  493. name: str,
  494. local: _EmptyListener[_ET],
  495. ):
  496. self._exec_once = False
  497. self._exec_w_sync_once = False
  498. self._exec_once_mutex = None
  499. self.parent_dispatch = parent_dispatch
  500. self.name = name
  501. self.local = local
  502. self.parent_listeners = self.local
  503. if not typing.TYPE_CHECKING:
  504. # first error, I don't really understand:
  505. # Signature of "listeners" incompatible with
  506. # supertype "_CompoundListener" [override]
  507. # the name / return type are exactly the same
  508. # second error is getattr_isn't typed, the cast() here
  509. # adds too much method overhead
  510. @property
  511. def listeners(self) -> Collection[_ListenerFnType]:
  512. return getattr(self.parent_dispatch, self.name)
  513. def _adjust_fn_spec(
  514. self, fn: _ListenerFnType, named: bool
  515. ) -> _ListenerFnType:
  516. return self.local._adjust_fn_spec(fn, named)
  517. def for_modify(self, obj: _DispatchCommon[_ET]) -> _JoinedListener[_ET]:
  518. self.local = self.parent_listeners = self.local.for_modify(obj)
  519. return self
  520. def insert(self, event_key: _EventKey[_ET], propagate: bool) -> None:
  521. self.local.insert(event_key, propagate)
  522. def append(self, event_key: _EventKey[_ET], propagate: bool) -> None:
  523. self.local.append(event_key, propagate)
  524. def remove(self, event_key: _EventKey[_ET]) -> None:
  525. self.local.remove(event_key)
  526. def clear(self) -> None:
  527. raise NotImplementedError()