Source code for scrapy.signalmanager

from typing import Any, List, Tuple

from pydispatch import dispatcher
from twisted.internet.defer import Deferred

from scrapy.utils import signal as _signal


[docs]class SignalManager: def __init__(self, sender: Any = dispatcher.Anonymous): self.sender: Any = sender
[docs] def connect(self, receiver: Any, signal: Any, **kwargs: Any) -> None: """ Connect a receiver function to a signal. The signal can be any object, although Scrapy comes with some predefined signals that are documented in the :ref:`topics-signals` section. :param receiver: the function to be connected :type receiver: collections.abc.Callable :param signal: the signal to connect to :type signal: object """ kwargs.setdefault("sender", self.sender) dispatcher.connect(receiver, signal, **kwargs)
[docs] def disconnect(self, receiver: Any, signal: Any, **kwargs: Any) -> None: """ Disconnect a receiver function from a signal. This has the opposite effect of the :meth:`connect` method, and the arguments are the same. """ kwargs.setdefault("sender", self.sender) dispatcher.disconnect(receiver, signal, **kwargs)
[docs] def send_catch_log(self, signal: Any, **kwargs: Any) -> List[Tuple[Any, Any]]: """ Send a signal, catch exceptions and log them. The keyword arguments are passed to the signal handlers (connected through the :meth:`connect` method). """ kwargs.setdefault("sender", self.sender) return _signal.send_catch_log(signal, **kwargs)
[docs] def send_catch_log_deferred(self, signal: Any, **kwargs: Any) -> Deferred: """ Like :meth:`send_catch_log` but supports returning :class:`~twisted.internet.defer.Deferred` objects from signal handlers. Returns a Deferred that gets fired once all signal handlers deferreds were fired. Send a signal, catch exceptions and log them. The keyword arguments are passed to the signal handlers (connected through the :meth:`connect` method). """ kwargs.setdefault("sender", self.sender) return _signal.send_catch_log_deferred(signal, **kwargs)
[docs] def disconnect_all(self, signal: Any, **kwargs: Any) -> None: """ Disconnect all receivers from the given signal. :param signal: the signal to disconnect from :type signal: object """ kwargs.setdefault("sender", self.sender) _signal.disconnect_all(signal, **kwargs)