Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:45
📄 __init__.py # event/__init__.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php from .api import CANCEL # noqa from .api import contains # noqa from .api import listen # noqa from .api import listens_for # noqa from .api import NO_RETVAL # noqa from .api import remove # noqa from .attr import RefCollection # noqa from .base import dispatcher # noqa from .base import Events # noqa from .legacy import _legacy_signature # noqa	596 B	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 api.py # event/api.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Public API functions for the event system. """ from __future__ import absolute_import from .base import _registrars from .registry import _EventKey from .. import exc from .. import util CANCEL = util.symbol("CANCEL") NO_RETVAL = util.symbol("NO_RETVAL") def _event_key(target, identifier, fn): for evt_cls in _registrars[identifier]: tgt = evt_cls._accept_with(target) if tgt is not None: return _EventKey(target, identifier, fn, tgt) else: raise exc.InvalidRequestError( "No such event '%s' for target '%s'" % (identifier, target) ) def listen(target, identifier, fn, args, kw): """Register a listener function for the given target. The :func:`.listen` function is part of the primary interface for the SQLAlchemy event system, documented at :ref:`event_toplevel`. e.g.:: from sqlalchemy import event from sqlalchemy.schema import UniqueConstraint def unique_constraint_name(const, table): const.name = "uq_%s_%s" % ( table.name, list(const.columns)[0].name ) event.listen( UniqueConstraint, "after_parent_attach", unique_constraint_name) A given function can also be invoked for only the first invocation of the event using the ``once`` argument:: def on_config(): do_config() event.listen(Mapper, "before_configure", on_config, once=True) .. versionadded:: 0.9.4 Added ``once=True`` to :func:`.event.listen` and :func:`.event.listens_for`. .. warning:: The ``once`` argument does not imply automatic de-registration of the listener function after it has been invoked a first time; a listener entry will remain associated with the target object. Associating an arbitrarily high number of listeners without explicitly removing them will cause memory to grow unbounded even if ``once=True`` is specified. .. note:: The :func:`.listen` function cannot be called at the same time that the target event is being run. This has implications for thread safety, and also means an event cannot be added from inside the listener function for itself. The list of events to be run are present inside of a mutable collection that can't be changed during iteration. Event registration and removal is not intended to be a "high velocity" operation; it is a configurational operation. For systems that need to quickly associate and deassociate with events at high scale, use a mutable structure that is handled from inside of a single listener. .. versionchanged:: 1.0.0 - a ``collections.deque()`` object is now used as the container for the list of events, which explicitly disallows collection mutation while the collection is being iterated. .. seealso:: :func:`.listens_for` :func:`.remove` """ _event_key(target, identifier, fn).listen(args, *kw) def listens_for(target, identifier, args, *kw): """Decorate a function as a listener for the given target + identifier. The :func:`.listens_for` decorator is part of the primary interface for the SQLAlchemy event system, documented at :ref:`event_toplevel`. e.g.:: from sqlalchemy import event from sqlalchemy.schema import UniqueConstraint @event.listens_for(UniqueConstraint, "after_parent_attach") def unique_constraint_name(const, table): const.name = "uq_%s_%s" % ( table.name, list(const.columns)[0].name ) A given function can also be invoked for only the first invocation of the event using the ``once`` argument:: @event.listens_for(Mapper, "before_configure", once=True) def on_config(): do_config() .. versionadded:: 0.9.4 Added ``once=True`` to :func:`.event.listen` and :func:`.event.listens_for`. .. warning:: The ``once`` argument does not imply automatic de-registration of the listener function after it has been invoked a first time; a listener entry will remain associated with the target object. Associating an arbitrarily high number of listeners without explicitly removing them will cause memory to grow unbounded even if ``once=True`` is specified. .. seealso:: :func:`.listen` - general description of event listening """ def decorate(fn): listen(target, identifier, fn, args, *kw) return fn return decorate def remove(target, identifier, fn): """Remove an event listener. The arguments here should match exactly those which were sent to :func:`.listen`; all the event registration which proceeded as a result of this call will be reverted by calling :func:`.remove` with the same arguments. e.g.:: # if a function was registered like this... @event.listens_for(SomeMappedClass, "before_insert", propagate=True) def my_listener_function(arg): pass # ... it's removed like this event.remove(SomeMappedClass, "before_insert", my_listener_function) Above, the listener function associated with ``SomeMappedClass`` was also propagated to subclasses of ``SomeMappedClass``; the :func:`.remove` function will revert all of these operations. .. versionadded:: 0.9.0 .. note:: The :func:`.remove` function cannot be called at the same time that the target event is being run. This has implications for thread safety, and also means an event cannot be removed from inside the listener function for itself. The list of events to be run are present inside of a mutable collection that can't be changed during iteration. Event registration and removal is not intended to be a "high velocity" operation; it is a configurational operation. For systems that need to quickly associate and deassociate with events at high scale, use a mutable structure that is handled from inside of a single listener. .. versionchanged:: 1.0.0 - a ``collections.deque()`` object is now used as the container for the list of events, which explicitly disallows collection mutation while the collection is being iterated. .. seealso:: :func:`.listen` """ _event_key(target, identifier, fn).remove() def contains(target, identifier, fn): """Return True if the given target/ident/fn is set up to listen. .. versionadded:: 0.9.0 """ return _event_key(target, identifier, fn).contains()	6.92 KB	2026-01-20 12:58
📄 attr.py # event/attr.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Attribute implementation for _Dispatch classes. The various listener targets for a particular event class are represented as attributes, which refer to collections of listeners to be fired off. These collections can exist at the class level as well as at the instance level. An event is fired off using code like this:: some_object.dispatch.first_connect(arg1, arg2) Above, ``some_object.dispatch`` would be an instance of ``_Dispatch`` and ``first_connect`` is typically an instance of ``_ListenerCollection`` if event listeners are present, or ``_EmptyListener`` if none are present. The attribute mechanics here spend effort trying to ensure listener functions are available with a minimum of function call overhead, that unnecessary objects aren't created (i.e. many empty per-instance listener collections), as well as that everything is garbage collectable when owning references are lost. Other features such as "propagation" of listener functions across many ``_Dispatch`` instances, "joining" of multiple ``_Dispatch`` instances, as well as support for subclass propagation (e.g. events assigned to ``Pool`` vs. ``QueuePool``) are all implemented here. """ from __future__ import absolute_import from __future__ import with_statement import collections from itertools import chain import weakref from . import legacy from . import registry from .. import exc from .. import util from ..util import threading class RefCollection(util.MemoizedSlots): __slots__ = ("ref",) def _memoized_attr_ref(self): return weakref.ref(self, registry._collection_gced) class _empty_collection(object): def append(self, element): pass def extend(self, other): pass def remove(self, element): pass def __iter__(self): return iter([]) def clear(self): pass class _ClsLevelDispatch(RefCollection): """Class-level events on :class:`._Dispatch` classes.""" __slots__ = ( "name", "arg_names", "has_kw", "legacy_signatures", "_clslevel", "__weakref__", ) def __init__(self, parent_dispatch_cls, fn): self.name = fn.__name__ argspec = util.inspect_getfullargspec(fn) self.arg_names = argspec.args[1:] self.has_kw = bool(argspec.varkw) self.legacy_signatures = list( reversed( sorted( getattr(fn, "_legacy_signatures", []), key=lambda s: s[0] ) ) ) fn.__doc__ = legacy._augment_fn_docs(self, parent_dispatch_cls, fn) self._clslevel = weakref.WeakKeyDictionary() def _adjust_fn_spec(self, fn, named): if named: fn = self._wrap_fn_for_kw(fn) if self.legacy_signatures: try: argspec = util.get_callable_argspec(fn, no_self=True) except TypeError: pass else: fn = legacy._wrap_fn_for_legacy(self, fn, argspec) return fn def _wrap_fn_for_kw(self, fn): def wrap_kw(args, kw): argdict = dict(zip(self.arg_names, args)) argdict.update(kw) return fn(argdict) return wrap_kw def insert(self, event_key, propagate): target = event_key.dispatch_target assert isinstance( target, type ), "Class-level Event targets must be classes." if not getattr(target, "_sa_propagate_class_events", True): raise exc.InvalidRequestError( "Can't assign an event directly to the %s class" % target ) stack = [target] while stack: cls = stack.pop(0) stack.extend(cls.__subclasses__()) if cls is not target and cls not in self._clslevel: self.update_subclass(cls) else: if cls not in self._clslevel: self._assign_cls_collection(cls) self._clslevel[cls].appendleft(event_key._listen_fn) registry._stored_in_collection(event_key, self) def append(self, event_key, propagate): target = event_key.dispatch_target assert isinstance( target, type ), "Class-level Event targets must be classes." if not getattr(target, "_sa_propagate_class_events", True): raise exc.InvalidRequestError( "Can't assign an event directly to the %s class" % target ) stack = [target] while stack: cls = stack.pop(0) stack.extend(cls.__subclasses__()) if cls is not target and cls not in self._clslevel: self.update_subclass(cls) else: if cls not in self._clslevel: self._assign_cls_collection(cls) self._clslevel[cls].append(event_key._listen_fn) registry._stored_in_collection(event_key, self) def _assign_cls_collection(self, target): if getattr(target, "_sa_propagate_class_events", True): self._clslevel[target] = collections.deque() else: self._clslevel[target] = _empty_collection() def update_subclass(self, target): if target not in self._clslevel: self._assign_cls_collection(target) clslevel = self._clslevel[target] for cls in target.__mro__[1:]: if cls in self._clslevel: clslevel.extend( [fn for fn in self._clslevel[cls] if fn not in clslevel] ) def remove(self, event_key): target = event_key.dispatch_target stack = [target] while stack: cls = stack.pop(0) stack.extend(cls.__subclasses__()) if cls in self._clslevel: self._clslevel[cls].remove(event_key._listen_fn) registry._removed_from_collection(event_key, self) def clear(self): """Clear all class level listeners""" to_clear = set() for dispatcher in self._clslevel.values(): to_clear.update(dispatcher) dispatcher.clear() registry._clear(self, to_clear) def for_modify(self, obj): """Return an event collection which can be modified. For _ClsLevelDispatch at the class level of a dispatcher, this returns self. """ return self class _InstanceLevelDispatch(RefCollection): __slots__ = () def _adjust_fn_spec(self, fn, named): return self.parent._adjust_fn_spec(fn, named) class _EmptyListener(_InstanceLevelDispatch): """Serves as a proxy interface to the events served by a _ClsLevelDispatch, when there are no instance-level events present. Is replaced by _ListenerCollection when instance-level events are added. """ propagate = frozenset() listeners = () __slots__ = "parent", "parent_listeners", "name" def __init__(self, parent, target_cls): if target_cls not in parent._clslevel: parent.update_subclass(target_cls) self.parent = parent # _ClsLevelDispatch self.parent_listeners = parent._clslevel[target_cls] self.name = parent.name def for_modify(self, obj): """Return an event collection which can be modified. For _EmptyListener at the instance level of a dispatcher, this generates a new _ListenerCollection, applies it to the instance, and returns it. """ result = _ListenerCollection(self.parent, obj._instance_cls) if getattr(obj, self.name) is self: setattr(obj, self.name, result) else: assert isinstance(getattr(obj, self.name), _JoinedListener) return result def _needs_modify(self, args, *kw): raise NotImplementedError("need to call for_modify()") exec_once = ( exec_once_unless_exception ) = insert = append = remove = clear = _needs_modify def __call__(self, args, *kw): """Execute this event.""" for fn in self.parent_listeners: fn(args, *kw) def __len__(self): return len(self.parent_listeners) def __iter__(self): return iter(self.parent_listeners) def __bool__(self): return bool(self.parent_listeners) __nonzero__ = __bool__ class _CompoundListener(_InstanceLevelDispatch): __slots__ = "_exec_once_mutex", "_exec_once" def _memoized_attr__exec_once_mutex(self): return threading.Lock() def _exec_once_impl(self, retry_on_exception, args, *kw): with self._exec_once_mutex: if not self._exec_once: try: self(args, *kw) exception = False except: exception = True raise finally: if not exception or not retry_on_exception: self._exec_once = True def exec_once(self, args, *kw): """Execute this event, but only if it has not been executed already for this collection.""" if not self._exec_once: self._exec_once_impl(False, args, *kw) def exec_once_unless_exception(self, args, *kw): """Execute this event, but only if it has not been executed already for this collection, or was called by a previous exec_once_unless_exception call and raised an exception. If exec_once was already called, then this method will never run the callable regardless of whether it raised or not. .. versionadded:: 1.3.8 """ if not self._exec_once: self._exec_once_impl(True, args, *kw) def __call__(self, args, *kw): """Execute this event.""" for fn in self.parent_listeners: fn(args, *kw) for fn in self.listeners: fn(args, **kw) def __len__(self): return len(self.parent_listeners) + len(self.listeners) def __iter__(self): return chain(self.parent_listeners, self.listeners) def __bool__(self): return bool(self.listeners or self.parent_listeners) __nonzero__ = __bool__ class _ListenerCollection(_CompoundListener): """Instance-level attributes on instances of :class:`._Dispatch`. Represents a collection of listeners. As of 0.7.9, _ListenerCollection is only first created via the _EmptyListener.for_modify() method. """ __slots__ = ( "parent_listeners", "parent", "name", "listeners", "propagate", "__weakref__", ) def __init__(self, parent, target_cls): if target_cls not in parent._clslevel: parent.update_subclass(target_cls) self._exec_once = False self.parent_listeners = parent._clslevel[target_cls] self.parent = parent self.name = parent.name self.listeners = collections.deque() self.propagate = set() def for_modify(self, obj): """Return an event collection which can be modified. For _ListenerCollection at the instance level of a dispatcher, this returns self. """ return self def _update(self, other, only_propagate=True): """Populate from the listeners in another :class:`_Dispatch` object.""" existing_listeners = self.listeners existing_listener_set = set(existing_listeners) self.propagate.update(other.propagate) other_listeners = [ l for l in other.listeners if l not in existing_listener_set and not only_propagate or l in self.propagate ] existing_listeners.extend(other_listeners) to_associate = other.propagate.union(other_listeners) registry._stored_in_collection_multi(self, other, to_associate) def insert(self, event_key, propagate): if event_key.prepend_to_list(self, self.listeners): if propagate: self.propagate.add(event_key._listen_fn) def append(self, event_key, propagate): if event_key.append_to_list(self, self.listeners): if propagate: self.propagate.add(event_key._listen_fn) def remove(self, event_key): self.listeners.remove(event_key._listen_fn) self.propagate.discard(event_key._listen_fn) registry._removed_from_collection(event_key, self) def clear(self): registry._clear(self, self.listeners) self.propagate.clear() self.listeners.clear() class _JoinedListener(_CompoundListener): __slots__ = "parent", "name", "local", "parent_listeners" def __init__(self, parent, name, local): self._exec_once = False self.parent = parent self.name = name self.local = local self.parent_listeners = self.local @property def listeners(self): return getattr(self.parent, self.name) def _adjust_fn_spec(self, fn, named): return self.local._adjust_fn_spec(fn, named) def for_modify(self, obj): self.local = self.parent_listeners = self.local.for_modify(obj) return self def insert(self, event_key, propagate): self.local.insert(event_key, propagate) def append(self, event_key, propagate): self.local.append(event_key, propagate) def remove(self, event_key): self.local.remove(event_key) def clear(self): raise NotImplementedError()	13.52 KB	2026-01-20 12:58
📄 base.py # event/base.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Base implementation classes. The public-facing ``Events`` serves as the base class for an event interface; its public attributes represent different kinds of events. These attributes are mirrored onto a ``_Dispatch`` class, which serves as a container for collections of listener functions. These collections are represented both at the class level of a particular ``_Dispatch`` class as well as within instances of ``_Dispatch``. """ from __future__ import absolute_import import weakref from .attr import _ClsLevelDispatch from .attr import _EmptyListener from .attr import _JoinedListener from .. import util _registrars = util.defaultdict(list) def _is_event_name(name): # _sa_event prefix is special to support internal-only event names. # most event names are just plain method names that aren't # underscored. return ( not name.startswith("_") and name != "dispatch" ) or name.startswith("_sa_event") class _UnpickleDispatch(object): """Serializable callable that re-generates an instance of :class:`_Dispatch` given a particular :class:`.Events` subclass. """ def __call__(self, _instance_cls): for cls in _instance_cls.__mro__: if "dispatch" in cls.__dict__: return cls.__dict__["dispatch"].dispatch._for_class( _instance_cls ) else: raise AttributeError("No class with a 'dispatch' member present.") class _Dispatch(object): """Mirror the event listening definitions of an Events class with listener collections. Classes which define a "dispatch" member will return a non-instantiated :class:`._Dispatch` subclass when the member is accessed at the class level. When the "dispatch" member is accessed at the instance level of its owner, an instance of the :class:`._Dispatch` class is returned. A :class:`._Dispatch` class is generated for each :class:`.Events` class defined, by the :func:`._create_dispatcher_class` function. The original :class:`.Events` classes remain untouched. This decouples the construction of :class:`.Events` subclasses from the implementation used by the event internals, and allows inspecting tools like Sphinx to work in an unsurprising way against the public API. """ # In one ORM edge case, an attribute is added to _Dispatch, # so __dict__ is used in just that case and potentially others. __slots__ = "_parent", "_instance_cls", "__dict__", "_empty_listeners" _empty_listener_reg = weakref.WeakKeyDictionary() def __init__(self, parent, instance_cls=None): self._parent = parent self._instance_cls = instance_cls if instance_cls: try: self._empty_listeners = self._empty_listener_reg[instance_cls] except KeyError: self._empty_listeners = self._empty_listener_reg[ instance_cls ] = { ls.name: _EmptyListener(ls, instance_cls) for ls in parent._event_descriptors } else: self._empty_listeners = {} def __getattr__(self, name): # Assign EmptyListeners as attributes on demand # to reduce startup time for new dispatch objects. try: ls = self._empty_listeners[name] except KeyError: raise AttributeError(name) else: setattr(self, ls.name, ls) return ls @property def _event_descriptors(self): for k in self._event_names: # Yield _ClsLevelDispatch related # to relevant event name. yield getattr(self, k) @property def _listen(self): return self._events._listen def _for_class(self, instance_cls): return self.__class__(self, instance_cls) def _for_instance(self, instance): instance_cls = instance.__class__ return self._for_class(instance_cls) def _join(self, other): """Create a 'join' of this :class:`._Dispatch` and another. This new dispatcher will dispatch events to both :class:`._Dispatch` objects. """ if "_joined_dispatch_cls" not in self.__class__.__dict__: cls = type( "Joined%s" % self.__class__.__name__, (_JoinedDispatcher,), {"__slots__": self._event_names}, ) self.__class__._joined_dispatch_cls = cls return self._joined_dispatch_cls(self, other) def __reduce__(self): return _UnpickleDispatch(), (self._instance_cls,) def _update(self, other, only_propagate=True): """Populate from the listeners in another :class:`_Dispatch` object.""" for ls in other._event_descriptors: if isinstance(ls, _EmptyListener): continue getattr(self, ls.name).for_modify(self)._update( ls, only_propagate=only_propagate ) def _clear(self): for ls in self._event_descriptors: ls.for_modify(self).clear() class _EventMeta(type): """Intercept new Event subclasses and create associated _Dispatch classes.""" def __init__(cls, classname, bases, dict_): _create_dispatcher_class(cls, classname, bases, dict_) type.__init__(cls, classname, bases, dict_) def _create_dispatcher_class(cls, classname, bases, dict_): """Create a :class:`._Dispatch` class corresponding to an :class:`.Events` class.""" # there's all kinds of ways to do this, # i.e. make a Dispatch class that shares the '_listen' method # of the Event class, this is the straight monkeypatch. if hasattr(cls, "dispatch"): dispatch_base = cls.dispatch.__class__ else: dispatch_base = _Dispatch event_names = [k for k in dict_ if _is_event_name(k)] dispatch_cls = type( "%sDispatch" % classname, (dispatch_base,), {"__slots__": event_names} ) dispatch_cls._event_names = event_names dispatch_inst = cls._set_dispatch(cls, dispatch_cls) for k in dispatch_cls._event_names: setattr(dispatch_inst, k, _ClsLevelDispatch(cls, dict_[k])) _registrars[k].append(cls) for super_ in dispatch_cls.__bases__: if issubclass(super_, _Dispatch) and super_ is not _Dispatch: for ls in super_._events.dispatch._event_descriptors: setattr(dispatch_inst, ls.name, ls) dispatch_cls._event_names.append(ls.name) if getattr(cls, "_dispatch_target", None): cls._dispatch_target.dispatch = dispatcher(cls) def _remove_dispatcher(cls): for k in cls.dispatch._event_names: _registrars[k].remove(cls) if not _registrars[k]: del _registrars[k] class Events(util.with_metaclass(_EventMeta, object)): """Define event listening functions for a particular target type.""" @staticmethod def _set_dispatch(cls, dispatch_cls): # This allows an Events subclass to define additional utility # methods made available to the target via # "self.dispatch._events.<utilitymethod>" # @staticmethod to allow easy "super" calls while in a metaclass # constructor. cls.dispatch = dispatch_cls(None) dispatch_cls._events = cls return cls.dispatch @classmethod def _accept_with(cls, target): def dispatch_is(*types): return all(isinstance(target.dispatch, t) for t in types) def dispatch_parent_is(t): return isinstance(target.dispatch.parent, t) # Mapper, ClassManager, Session override this to # also accept classes, scoped_sessions, sessionmakers, etc. if hasattr(target, "dispatch"): if ( dispatch_is(cls.dispatch.__class__) or dispatch_is(type, cls.dispatch.__class__) or ( dispatch_is(_JoinedDispatcher) and dispatch_parent_is(cls.dispatch.__class__) ) ): return target @classmethod def _listen(cls, event_key, propagate=False, insert=False, named=False): event_key.base_listen(propagate=propagate, insert=insert, named=named) @classmethod def _remove(cls, event_key): event_key.remove() @classmethod def _clear(cls): cls.dispatch._clear() class _JoinedDispatcher(object): """Represent a connection between two _Dispatch objects.""" __slots__ = "local", "parent", "_instance_cls" def __init__(self, local, parent): self.local = local self.parent = parent self._instance_cls = self.local._instance_cls def __getattr__(self, name): # Assign _JoinedListeners as attributes on demand # to reduce startup time for new dispatch objects. ls = getattr(self.local, name) jl = _JoinedListener(self.parent, ls.name, ls) setattr(self, ls.name, jl) return jl @property def _listen(self): return self.parent._listen @property def _events(self): return self.parent._events class dispatcher(object): """Descriptor used by target classes to deliver the _Dispatch class at the class level and produce new _Dispatch instances for target instances. """ def __init__(self, events): self.dispatch = events.dispatch self.events = events def __get__(self, obj, cls): if obj is None: return self.dispatch obj.__dict__["dispatch"] = disp = self.dispatch._for_instance(obj) return disp	9.72 KB	2026-01-20 12:58
📄 legacy.py # event/legacy.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Routines to handle adaption of legacy call signatures, generation of deprecation notes and docstrings. """ from .. import util def _legacy_signature(since, argnames, converter=None): def leg(fn): if not hasattr(fn, "_legacy_signatures"): fn._legacy_signatures = [] fn._legacy_signatures.append((since, argnames, converter)) return fn return leg def _wrap_fn_for_legacy(dispatch_collection, fn, argspec): for since, argnames, conv in dispatch_collection.legacy_signatures: if argnames[-1] == "*kw": has_kw = True argnames = argnames[0:-1] else: has_kw = False if len(argnames) == len(argspec.args) and has_kw is bool( argspec.varkw ): if conv: assert not has_kw def wrap_leg(args): return fn(conv(args)) else: def wrap_leg(args, kw): argdict = dict(zip(dispatch_collection.arg_names, args)) args = [argdict[name] for name in argnames] if has_kw: return fn(args, *kw) else: return fn(args) return wrap_leg else: return fn def _indent(text, indent): return "\n".join(indent + line for line in text.split("\n")) def _standard_listen_example(dispatch_collection, sample_target, fn): example_kw_arg = _indent( "\n".join( "%(arg)s = kw['%(arg)s']" % {"arg": arg} for arg in dispatch_collection.arg_names[0:2] ), " ", ) if dispatch_collection.legacy_signatures: current_since = max( since for since, args, conv in dispatch_collection.legacy_signatures ) else: current_since = None text = ( "from sqlalchemy import event\n\n" "# standard decorator style%(current_since)s\n" "@event.listens_for(%(sample_target)s, '%(event_name)s')\n" "def receive_%(event_name)s(" "%(named_event_arguments)s%(has_kw_arguments)s):\n" " \"listen for the '%(event_name)s' event\"\n" "\n # ... (event handling logic) ...\n" ) if len(dispatch_collection.arg_names) > 3: text += ( "\n# named argument style (new in 0.9)\n" "@event.listens_for(" "%(sample_target)s, '%(event_name)s', named=True)\n" "def receive_%(event_name)s(kw):\n" " \"listen for the '%(event_name)s' event\"\n" "%(example_kw_arg)s\n" "\n # ... (event handling logic) ...\n" ) text %= { "current_since": " (arguments as of %s)" % current_since if current_since else "", "event_name": fn.__name__, "has_kw_arguments": ", kw" if dispatch_collection.has_kw else "", "named_event_arguments": ", ".join(dispatch_collection.arg_names), "example_kw_arg": example_kw_arg, "sample_target": sample_target, } return text def _legacy_listen_examples(dispatch_collection, sample_target, fn): text = "" for since, args, conv in dispatch_collection.legacy_signatures: text += ( "\n# DEPRECATED calling style (pre-%(since)s, " "will be removed in a future release)\n" "@event.listens_for(%(sample_target)s, '%(event_name)s')\n" "def receive_%(event_name)s(" "%(named_event_arguments)s%(has_kw_arguments)s):\n" " \"listen for the '%(event_name)s' event\"\n" "\n # ... (event handling logic) ...\n" % { "since": since, "event_name": fn.__name__, "has_kw_arguments": " kw" if dispatch_collection.has_kw else "", "named_event_arguments": ", ".join(args), "sample_target": sample_target, } ) return text def _version_signature_changes(parent_dispatch_cls, dispatch_collection): since, args, conv = dispatch_collection.legacy_signatures[0] return ( "\n.. deprecated:: %(since)s\n" " The :class:`.%(clsname)s.%(event_name)s` event now accepts the \n" " arguments ``%(named_event_arguments)s%(has_kw_arguments)s``.\n" " Support for listener functions which accept the previous \n" ' argument signature(s) listed above as "deprecated" will be \n' " removed in a future release." % { "since": since, "clsname": parent_dispatch_cls.__name__, "event_name": dispatch_collection.name, "named_event_arguments": ", ".join(dispatch_collection.arg_names), "has_kw_arguments": ", kw" if dispatch_collection.has_kw else "", } ) def _augment_fn_docs(dispatch_collection, parent_dispatch_cls, fn): header = ( ".. container:: event_signatures\n\n" " Example argument forms::\n" "\n" ) sample_target = getattr(parent_dispatch_cls, "_target_class_doc", "obj") text = header + _indent( _standard_listen_example(dispatch_collection, sample_target, fn), " " * 8, ) if dispatch_collection.legacy_signatures: text += _indent( _legacy_listen_examples(dispatch_collection, sample_target, fn), " " * 8, ) text += _version_signature_changes( parent_dispatch_cls, dispatch_collection ) return util.inject_docstring_text(fn.__doc__, text, 1)	5.77 KB	2026-01-20 12:58
📄 registry.py # event/registry.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Provides managed registration services on behalf of :func:`.listen` arguments. By "managed registration", we mean that event listening functions and other objects can be added to various collections in such a way that their membership in all those collections can be revoked at once, based on an equivalent :class:`._EventKey`. """ from __future__ import absolute_import import collections import types import weakref from .. import exc from .. import util _key_to_collection = collections.defaultdict(dict) """ Given an original listen() argument, can locate all listener collections and the listener fn contained (target, identifier, fn) -> { ref(listenercollection) -> ref(listener_fn) ref(listenercollection) -> ref(listener_fn) ref(listenercollection) -> ref(listener_fn) } """ _collection_to_key = collections.defaultdict(dict) """ Given a _ListenerCollection or _ClsLevelListener, can locate all the original listen() arguments and the listener fn contained ref(listenercollection) -> { ref(listener_fn) -> (target, identifier, fn), ref(listener_fn) -> (target, identifier, fn), ref(listener_fn) -> (target, identifier, fn), } """ def _collection_gced(ref): # defaultdict, so can't get a KeyError if not _collection_to_key or ref not in _collection_to_key: return listener_to_key = _collection_to_key.pop(ref) for key in listener_to_key.values(): if key in _key_to_collection: # defaultdict, so can't get a KeyError dispatch_reg = _key_to_collection[key] dispatch_reg.pop(ref) if not dispatch_reg: _key_to_collection.pop(key) def _stored_in_collection(event_key, owner): key = event_key._key dispatch_reg = _key_to_collection[key] owner_ref = owner.ref listen_ref = weakref.ref(event_key._listen_fn) if owner_ref in dispatch_reg: return False dispatch_reg[owner_ref] = listen_ref listener_to_key = _collection_to_key[owner_ref] listener_to_key[listen_ref] = key return True def _removed_from_collection(event_key, owner): key = event_key._key dispatch_reg = _key_to_collection[key] listen_ref = weakref.ref(event_key._listen_fn) owner_ref = owner.ref dispatch_reg.pop(owner_ref, None) if not dispatch_reg: del _key_to_collection[key] if owner_ref in _collection_to_key: listener_to_key = _collection_to_key[owner_ref] listener_to_key.pop(listen_ref) def _stored_in_collection_multi(newowner, oldowner, elements): if not elements: return oldowner = oldowner.ref newowner = newowner.ref old_listener_to_key = _collection_to_key[oldowner] new_listener_to_key = _collection_to_key[newowner] for listen_fn in elements: listen_ref = weakref.ref(listen_fn) key = old_listener_to_key[listen_ref] dispatch_reg = _key_to_collection[key] if newowner in dispatch_reg: assert dispatch_reg[newowner] == listen_ref else: dispatch_reg[newowner] = listen_ref new_listener_to_key[listen_ref] = key def _clear(owner, elements): if not elements: return owner = owner.ref listener_to_key = _collection_to_key[owner] for listen_fn in elements: listen_ref = weakref.ref(listen_fn) key = listener_to_key[listen_ref] dispatch_reg = _key_to_collection[key] dispatch_reg.pop(owner, None) if not dispatch_reg: del _key_to_collection[key] class _EventKey(object): """Represent :func:`.listen` arguments.""" __slots__ = ( "target", "identifier", "fn", "fn_key", "fn_wrap", "dispatch_target", ) def __init__(self, target, identifier, fn, dispatch_target, _fn_wrap=None): self.target = target self.identifier = identifier self.fn = fn if isinstance(fn, types.MethodType): self.fn_key = id(fn.__func__), id(fn.__self__) else: self.fn_key = id(fn) self.fn_wrap = _fn_wrap self.dispatch_target = dispatch_target @property def _key(self): return (id(self.target), self.identifier, self.fn_key) def with_wrapper(self, fn_wrap): if fn_wrap is self._listen_fn: return self else: return _EventKey( self.target, self.identifier, self.fn, self.dispatch_target, _fn_wrap=fn_wrap, ) def with_dispatch_target(self, dispatch_target): if dispatch_target is self.dispatch_target: return self else: return _EventKey( self.target, self.identifier, self.fn, dispatch_target, _fn_wrap=self.fn_wrap, ) def listen(self, args, kw): once = kw.pop("once", False) once_unless_exception = kw.pop("_once_unless_exception", False) named = kw.pop("named", False) target, identifier, fn = ( self.dispatch_target, self.identifier, self._listen_fn, ) dispatch_collection = getattr(target.dispatch, identifier) adjusted_fn = dispatch_collection._adjust_fn_spec(fn, named) self = self.with_wrapper(adjusted_fn) stub_function = getattr( self.dispatch_target.dispatch._events, self.identifier ) if hasattr(stub_function, "_sa_warn"): stub_function._sa_warn() if once or once_unless_exception: self.with_wrapper( util.only_once( self._listen_fn, retry_on_exception=once_unless_exception ) ).listen(args, *kw) else: self.dispatch_target.dispatch._listen(self, args, **kw) def remove(self): key = self._key if key not in _key_to_collection: raise exc.InvalidRequestError( "No listeners found for event %s / %r / %s " % (self.target, self.identifier, self.fn) ) dispatch_reg = _key_to_collection.pop(key) for collection_ref, listener_ref in dispatch_reg.items(): collection = collection_ref() listener_fn = listener_ref() if collection is not None and listener_fn is not None: collection.remove(self.with_wrapper(listener_fn)) def contains(self): """Return True if this event key is registered to listen.""" return self._key in _key_to_collection def base_listen( self, propagate=False, insert=False, named=False, retval=None ): target, identifier = self.dispatch_target, self.identifier dispatch_collection = getattr(target.dispatch, identifier) if insert: dispatch_collection.for_modify(target.dispatch).insert( self, propagate ) else: dispatch_collection.for_modify(target.dispatch).append( self, propagate ) @property def _listen_fn(self): return self.fn_wrap or self.fn def append_to_list(self, owner, list_): if _stored_in_collection(self, owner): list_.append(self._listen_fn) return True else: return False def remove_from_list(self, owner, list_): _removed_from_collection(self, owner) list_.remove(self._listen_fn) def prepend_to_list(self, owner, list_): if _stored_in_collection(self, owner): list_.appendleft(self._listen_fn) return True else: return False	8.04 KB	2026-01-20 12:58

Terminal