MCPcopy Create free account
hub / github.com/sqlalchemy/sqlalchemy / _EventKey

Class _EventKey

lib/sqlalchemy/event/registry.py:220–391  ·  view source on GitHub ↗

Represent :func:`.listen` arguments.

Source from the content-addressed store, hash-verified

218
219
220class _EventKey(Generic[_ET]):
221 """Represent :func:`.listen` arguments."""
222
223 __slots__ = (
224 "target",
225 "identifier",
226 "fn",
227 "fn_key",
228 "fn_wrap",
229 "dispatch_target",
230 )
231
232 target: _ET
233 identifier: str
234 fn: _ListenerFnType
235 fn_key: _ListenerFnKeyType
236 dispatch_target: Any
237 _fn_wrap: Optional[_ListenerFnType]
238
239 def __init__(
240 self,
241 target: _ET,
242 identifier: str,
243 fn: _ListenerFnType,
244 dispatch_target: Any,
245 _fn_wrap: Optional[_ListenerFnType] = None,
246 ):
247 self.target = target
248 self.identifier = identifier
249 self.fn = fn
250 if isinstance(fn, types.MethodType):
251 self.fn_key = id(fn.__func__), id(fn.__self__)
252 else:
253 self.fn_key = id(fn)
254 self.fn_wrap = _fn_wrap
255 self.dispatch_target = dispatch_target
256
257 @property
258 def _key(self) -> _EventKeyTupleType:
259 return (id(self.target), self.identifier, self.fn_key)
260
261 def with_wrapper(self, fn_wrap: _ListenerFnType) -> _EventKey[_ET]:
262 if fn_wrap is self._listen_fn:
263 return self
264 else:
265 return _EventKey(
266 self.target,
267 self.identifier,
268 self.fn,
269 self.dispatch_target,
270 _fn_wrap=fn_wrap,
271 )
272
273 def with_dispatch_target(self, dispatch_target: Any) -> _EventKey[_ET]:
274 if dispatch_target is self.dispatch_target:
275 return self
276 else:
277 return _EventKey(

Callers 3

_event_keyFunction · 0.85
with_wrapperMethod · 0.85
with_dispatch_targetMethod · 0.85

Calls

no outgoing calls

Tested by

no test coverage detected