1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
|
# event/api.py
# Copyright (C) 2005-2023 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
"""Public API functions for the event system.
"""
from __future__ import annotations
from typing import Any
from typing import Callable
from .base import _registrars
from .registry import _ET
from .registry import _EventKey
from .registry import _ListenerFnType
from .. import exc
from .. import util
CANCEL = util.symbol("CANCEL")
NO_RETVAL = util.symbol("NO_RETVAL")
def _event_key(
target: _ET, identifier: str, fn: _ListenerFnType
) -> _EventKey[_ET]:
for evt_cls in _registrars[identifier]:
tgt = evt_cls._accept_with(target, identifier)
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: Any, identifier: str, fn: Callable[..., Any], *args: Any, **kw: Any
) -> None:
"""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)
:param bool insert: The default behavior for event handlers is to append
the decorated user defined function to an internal list of registered
event listeners upon discovery. If a user registers a function with
``insert=True``, SQLAlchemy will insert (prepend) the function to the
internal list upon discovery. This feature is not typically used or
recommended by the SQLAlchemy maintainers, but is provided to ensure
certain user defined functions can run before others, such as when
:ref:`Changing the sql_mode in MySQL <mysql_sql_mode>`.
:param bool named: When using named argument passing, the names listed in
the function argument specification will be used as keys in the
dictionary.
See :ref:`event_named_argument_styles`.
:param bool once: Private/Internal API usage. Deprecated. This parameter
would provide that an event function would run only once per given
target. It does not however imply automatic de-registration of the
listener function; associating an arbitrarily high number of listeners
without explicitly removing them will cause memory to grow unbounded even
if ``once=True`` is specified.
:param bool propagate: The ``propagate`` kwarg is available when working
with ORM instrumentation and mapping events.
See :class:`_ormevent.MapperEvents` and
:meth:`_ormevent.MapperEvents.before_mapper_configured` for examples.
:param bool retval: This flag applies only to specific event listeners,
each of which includes documentation explaining when it should be used.
By default, no listener ever requires a return value.
However, some listeners do support special behaviors for return values,
and include in their documentation that the ``retval=True`` flag is
necessary for a return value to be processed.
Event listener suites that make use of :paramref:`_event.listen.retval`
include :class:`_events.ConnectionEvents` and
:class:`_ormevent.AttributeEvents`.
.. 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.
.. seealso::
:func:`.listens_for`
:func:`.remove`
"""
_event_key(target, identifier, fn).listen(*args, **kw)
def listens_for(
target: Any, identifier: str, *args: Any, **kw: Any
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
"""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`.
This function generally shares the same kwargs as :func:`.listens`.
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()
.. 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: Callable[..., Any]) -> Callable[..., Any]:
listen(target, identifier, fn, *args, **kw)
return fn
return decorate
def remove(target: Any, identifier: str, fn: Callable[..., Any]) -> None:
"""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.
.. 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.
.. seealso::
:func:`.listen`
"""
_event_key(target, identifier, fn).remove()
def contains(target: Any, identifier: str, fn: Callable[..., Any]) -> bool:
"""Return True if the given target/ident/fn is set up to listen."""
return _event_key(target, identifier, fn).contains()
|
