signals.rst 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706
  1. .. _signals:
  2. =======
  3. Signals
  4. =======
  5. .. contents::
  6. :local:
  7. Signals allows decoupled applications to receive notifications when
  8. certain actions occur elsewhere in the application.
  9. Celery ships with many signals that your application can hook into
  10. to augment behavior of certain actions.
  11. .. _signal-basics:
  12. Basics
  13. ======
  14. Several kinds of events trigger signals, you can connect to these signals
  15. to perform actions as they trigger.
  16. Example connecting to the :signal:`after_task_publish` signal:
  17. .. code-block:: python
  18. from celery.signals import after_task_publish
  19. @after_task_publish.connect
  20. def task_sent_handler(sender=None, headers=None, body=None, **kwargs):
  21. # information about task are located in headers for task messages
  22. # using the task protocol version 2.
  23. info = headers if 'task' in headers else body
  24. print('after_task_publish for task id {info[id]}'.format(
  25. info=info,
  26. ))
  27. Some signals also have a sender which you can filter by. For example the
  28. :signal:`after_task_publish` signal uses the task name as a sender, so by
  29. providing the ``sender`` argument to
  30. :class:`~celery.utils.dispatch.signal.Signal.connect` you can
  31. connect your handler to be called every time a task with name `"proj.tasks.add"`
  32. is published:
  33. .. code-block:: python
  34. @after_task_publish.connect(sender='proj.tasks.add')
  35. def task_sent_handler(sender=None, headers=None, body=None, **kwargs):
  36. # information about task are located in headers for task messages
  37. # using the task protocol version 2.
  38. info = headers if 'task' in headers else body
  39. print('after_task_publish for task id {info[id]}'.format(
  40. info=info,
  41. ))
  42. Signals use the same implementation as django.core.dispatch. As a result other
  43. keyword parameters (e.g. signal) are passed to all signal handlers by default.
  44. The best practice for signal handlers is to accept arbitrary keyword
  45. arguments (i.e. ``**kwargs``). That way new celery versions can add additional
  46. arguments without breaking user code.
  47. .. _signal-ref:
  48. Signals
  49. =======
  50. Task Signals
  51. ------------
  52. .. signal:: before_task_publish
  53. before_task_publish
  54. ~~~~~~~~~~~~~~~~~~~
  55. .. versionadded:: 3.1
  56. Dispatched before a task is published.
  57. Note that this is executed in the process sending the task.
  58. Sender is the name of the task being sent.
  59. Provides arguments:
  60. * body
  61. Task message body.
  62. This is a mapping containing the task message fields
  63. (see :ref:`message-protocol-task-v1`).
  64. * exchange
  65. Name of the exchange to send to or a :class:`~kombu.Exchange` object.
  66. * routing_key
  67. Routing key to use when sending the message.
  68. * headers
  69. Application headers mapping (can be modified).
  70. * properties
  71. Message properties (can be modified)
  72. * declare
  73. List of entities (:class:`~kombu.Exchange`,
  74. :class:`~kombu.Queue` or :class:~`kombu.binding` to declare before
  75. publishing the message. Can be modified.
  76. * retry_policy
  77. Mapping of retry options. Can be any argument to
  78. :meth:`kombu.Connection.ensure` and can be modified.
  79. .. signal:: after_task_publish
  80. after_task_publish
  81. ~~~~~~~~~~~~~~~~~~
  82. Dispatched when a task has been sent to the broker.
  83. Note that this is executed in the process that sent the task.
  84. Sender is the name of the task being sent.
  85. Provides arguments:
  86. * headers
  87. The task message headers, see :ref:`message-protocol-task-v2`
  88. and :ref:`message-protocol-task-v1`.
  89. for a reference of possible fields that can be defined.
  90. * body
  91. The task message body, see :ref:`message-protocol-task-v2`
  92. and :ref:`message-protocol-task-v1`.
  93. for a reference of possible fields that can be defined.
  94. * exchange
  95. Name of the exchange or :class:`~kombu.Exchange` object used.
  96. * routing_key
  97. Routing key used.
  98. .. signal:: task_prerun
  99. task_prerun
  100. ~~~~~~~~~~~
  101. Dispatched before a task is executed.
  102. Sender is the task object being executed.
  103. Provides arguments:
  104. * task_id
  105. Id of the task to be executed.
  106. * task
  107. The task being executed.
  108. * args
  109. the tasks positional arguments.
  110. * kwargs
  111. The tasks keyword arguments.
  112. .. signal:: task_postrun
  113. task_postrun
  114. ~~~~~~~~~~~~
  115. Dispatched after a task has been executed.
  116. Sender is the task object executed.
  117. Provides arguments:
  118. * task_id
  119. Id of the task to be executed.
  120. * task
  121. The task being executed.
  122. * args
  123. The tasks positional arguments.
  124. * kwargs
  125. The tasks keyword arguments.
  126. * retval
  127. The return value of the task.
  128. * state
  129. Name of the resulting state.
  130. .. signal:: task_retry
  131. task_retry
  132. ~~~~~~~~~~
  133. Dispatched when a task will be retried.
  134. Sender is the task object.
  135. Provides arguments:
  136. * request
  137. The current task request.
  138. * reason
  139. Reason for retry (usually an exception instance, but can always be
  140. coerced to :class:`str`).
  141. * einfo
  142. Detailed exception information, including traceback
  143. (a :class:`billiard.einfo.ExceptionInfo` object).
  144. .. signal:: task_success
  145. task_success
  146. ~~~~~~~~~~~~
  147. Dispatched when a task succeeds.
  148. Sender is the task object executed.
  149. Provides arguments
  150. * result
  151. Return value of the task.
  152. .. signal:: task_failure
  153. task_failure
  154. ~~~~~~~~~~~~
  155. Dispatched when a task fails.
  156. Sender is the task object executed.
  157. Provides arguments:
  158. * task_id
  159. Id of the task.
  160. * exception
  161. Exception instance raised.
  162. * args
  163. Positional arguments the task was called with.
  164. * kwargs
  165. Keyword arguments the task was called with.
  166. * traceback
  167. Stack trace object.
  168. * einfo
  169. The :class:`celery.datastructures.ExceptionInfo` instance.
  170. .. signal:: task_revoked
  171. task_revoked
  172. ~~~~~~~~~~~~
  173. Dispatched when a task is revoked/terminated by the worker.
  174. Sender is the task object revoked/terminated.
  175. Provides arguments:
  176. * request
  177. This is a :class:`~celery.worker.request.Request` instance, and not
  178. ``task.request``. When using the prefork pool this signal
  179. is dispatched in the parent process, so ``task.request`` is not available
  180. and should not be used. Use this object instead, which should have many
  181. of the same fields.
  182. * terminated
  183. Set to :const:`True` if the task was terminated.
  184. * signum
  185. Signal number used to terminate the task. If this is :const:`None` and
  186. terminated is :const:`True` then :sig:`TERM` should be assumed.
  187. * expired
  188. Set to :const:`True` if the task expired.
  189. .. signal:: task_unknown
  190. task_unknown
  191. ~~~~~~~~~~~~
  192. Dispatched when a worker receives a message for a task that is not registered.
  193. Sender is the worker :class:`~celery.worker.consumer.Consumer`.
  194. Provides arguments:
  195. * message
  196. Raw message object.
  197. * exc
  198. The error that occurred.
  199. .. signal:: task_rejected
  200. task_rejected
  201. ~~~~~~~~~~~~~
  202. Dispatched when a worker receives an unknown type of message to one of its
  203. task queues.
  204. Sender is the worker :class:`~celery.worker.consumer.Consumer`.
  205. Provides arguments:
  206. * message
  207. Raw message object.
  208. * exc
  209. The error that occurred (if any).
  210. App Signals
  211. -----------
  212. .. signal:: import_modules
  213. import_modules
  214. ~~~~~~~~~~~~~~
  215. This signal is sent when a program (worker, beat, shell) etc, asks
  216. for modules in the :setting:`include` and :setting:`imports`
  217. settings to be imported.
  218. Sender is the app instance.
  219. Worker Signals
  220. --------------
  221. .. signal:: celeryd_after_setup
  222. celeryd_after_setup
  223. ~~~~~~~~~~~~~~~~~~~
  224. This signal is sent after the worker instance is set up,
  225. but before it calls run. This means that any queues from the :option:`-Q`
  226. option is enabled, logging has been set up and so on.
  227. It can be used to e.g. add custom queues that should always be consumed
  228. from, disregarding the :option:`-Q` option. Here's an example
  229. that sets up a direct queue for each worker, these queues can then be
  230. used to route a task to any specific worker:
  231. .. code-block:: python
  232. from celery.signals import celeryd_after_setup
  233. @celeryd_after_setup.connect
  234. def setup_direct_queue(sender, instance, **kwargs):
  235. queue_name = '{0}.dq'.format(sender) # sender is the nodename of the worker
  236. instance.app.amqp.queues.select_add(queue_name)
  237. Provides arguments:
  238. * sender
  239. Hostname of the worker.
  240. * instance
  241. This is the :class:`celery.apps.worker.Worker` instance to be initialized.
  242. Note that only the :attr:`app` and :attr:`hostname` (nodename) attributes have been
  243. set so far, and the rest of ``__init__`` has not been executed.
  244. * conf
  245. The configuration of the current app.
  246. .. signal:: celeryd_init
  247. celeryd_init
  248. ~~~~~~~~~~~~
  249. This is the first signal sent when :program:`celery worker` starts up.
  250. The ``sender`` is the host name of the worker, so this signal can be used
  251. to setup worker specific configuration:
  252. .. code-block:: python
  253. from celery.signals import celeryd_init
  254. @celeryd_init.connect(sender='worker12@example.com')
  255. def configure_worker12(conf=None, **kwargs):
  256. conf.task_default_rate_limit = '10/m'
  257. or to set up configuration for multiple workers you can omit specifying a
  258. sender when you connect:
  259. .. code-block:: python
  260. from celery.signals import celeryd_init
  261. @celeryd_init.connect
  262. def configure_workers(sender=None, conf=None, **kwargs):
  263. if sender in ('worker1@example.com', 'worker2@example.com'):
  264. conf.task_default_rate_limit = '10/m'
  265. if sender == 'worker3@example.com':
  266. conf.worker_prefetch_multiplier = 0
  267. Provides arguments:
  268. * sender
  269. Nodename of the worker.
  270. * instance
  271. This is the :class:`celery.apps.worker.Worker` instance to be initialized.
  272. Note that only the :attr:`app` and :attr:`hostname` (nodename) attributes have been
  273. set so far, and the rest of ``__init__`` has not been executed.
  274. * conf
  275. The configuration of the current app.
  276. * options
  277. Options passed to the worker from command-line arguments (including
  278. defaults).
  279. .. signal:: worker_init
  280. worker_init
  281. ~~~~~~~~~~~
  282. Dispatched before the worker is started.
  283. .. signal:: worker_ready
  284. worker_ready
  285. ~~~~~~~~~~~~
  286. Dispatched when the worker is ready to accept work.
  287. .. signal:: worker_process_init
  288. worker_process_init
  289. ~~~~~~~~~~~~~~~~~~~
  290. Dispatched in all pool child processes when they start.
  291. Note that handlers attached to this signal must not be blocking
  292. for more than 4 seconds, or the process will be killed assuming
  293. it failed to start.
  294. .. signal:: worker_process_shutdown
  295. worker_process_shutdown
  296. ~~~~~~~~~~~~~~~~~~~~~~~
  297. Dispatched in all pool child processes just before they exit.
  298. Note: There is no guarantee that this signal will be dispatched,
  299. similarly to finally blocks it's impossible to guarantee that handlers
  300. will be called at shutdown, and if called it may be interrupted during.
  301. Provides arguments:
  302. * pid
  303. The pid of the child process that is about to shutdown.
  304. * exitcode
  305. The exitcode that will be used when the child process exits.
  306. .. signal:: worker_shutdown
  307. worker_shutdown
  308. ~~~~~~~~~~~~~~~
  309. Dispatched when the worker is about to shut down.
  310. Beat Signals
  311. ------------
  312. .. signal:: beat_init
  313. beat_init
  314. ~~~~~~~~~
  315. Dispatched when :program:`celery beat` starts (either standalone or embedded).
  316. Sender is the :class:`celery.beat.Service` instance.
  317. .. signal:: beat_embedded_init
  318. beat_embedded_init
  319. ~~~~~~~~~~~~~~~~~~
  320. Dispatched in addition to the :signal:`beat_init` signal when :program:`celery
  321. beat` is started as an embedded process. Sender is the
  322. :class:`celery.beat.Service` instance.
  323. Eventlet Signals
  324. ----------------
  325. .. signal:: eventlet_pool_started
  326. eventlet_pool_started
  327. ~~~~~~~~~~~~~~~~~~~~~
  328. Sent when the eventlet pool has been started.
  329. Sender is the :class:`celery.concurrency.eventlet.TaskPool` instance.
  330. .. signal:: eventlet_pool_preshutdown
  331. eventlet_pool_preshutdown
  332. ~~~~~~~~~~~~~~~~~~~~~~~~~
  333. Sent when the worker shutdown, just before the eventlet pool
  334. is requested to wait for remaining workers.
  335. Sender is the :class:`celery.concurrency.eventlet.TaskPool` instance.
  336. .. signal:: eventlet_pool_postshutdown
  337. eventlet_pool_postshutdown
  338. ~~~~~~~~~~~~~~~~~~~~~~~~~~
  339. Sent when the pool has been joined and the worker is ready to shutdown.
  340. Sender is the :class:`celery.concurrency.eventlet.TaskPool` instance.
  341. .. signal:: eventlet_pool_apply
  342. eventlet_pool_apply
  343. ~~~~~~~~~~~~~~~~~~~
  344. Sent whenever a task is applied to the pool.
  345. Sender is the :class:`celery.concurrency.eventlet.TaskPool` instance.
  346. Provides arguments:
  347. * target
  348. The target function.
  349. * args
  350. Positional arguments.
  351. * kwargs
  352. Keyword arguments.
  353. Logging Signals
  354. ---------------
  355. .. signal:: setup_logging
  356. setup_logging
  357. ~~~~~~~~~~~~~
  358. Celery won't configure the loggers if this signal is connected,
  359. so you can use this to completely override the logging configuration
  360. with your own.
  361. If you would like to augment the logging configuration setup by
  362. Celery then you can use the :signal:`after_setup_logger` and
  363. :signal:`after_setup_task_logger` signals.
  364. Provides arguments:
  365. * loglevel
  366. The level of the logging object.
  367. * logfile
  368. The name of the logfile.
  369. * format
  370. The log format string.
  371. * colorize
  372. Specify if log messages are colored or not.
  373. .. signal:: after_setup_logger
  374. after_setup_logger
  375. ~~~~~~~~~~~~~~~~~~
  376. Sent after the setup of every global logger (not task loggers).
  377. Used to augment logging configuration.
  378. Provides arguments:
  379. * logger
  380. The logger object.
  381. * loglevel
  382. The level of the logging object.
  383. * logfile
  384. The name of the logfile.
  385. * format
  386. The log format string.
  387. * colorize
  388. Specify if log messages are colored or not.
  389. .. signal:: after_setup_task_logger
  390. after_setup_task_logger
  391. ~~~~~~~~~~~~~~~~~~~~~~~
  392. Sent after the setup of every single task logger.
  393. Used to augment logging configuration.
  394. Provides arguments:
  395. * logger
  396. The logger object.
  397. * loglevel
  398. The level of the logging object.
  399. * logfile
  400. The name of the logfile.
  401. * format
  402. The log format string.
  403. * colorize
  404. Specify if log messages are colored or not.
  405. Command signals
  406. ---------------
  407. .. signal:: user_preload_options
  408. user_preload_options
  409. ~~~~~~~~~~~~~~~~~~~~
  410. This signal is sent after any of the Celery command line programs
  411. are finished parsing the user preload options.
  412. It can be used to add additional command-line arguments to the
  413. :program:`celery` umbrella command:
  414. .. code-block:: python
  415. from celery import Celery
  416. from celery import signals
  417. from celery.bin.base import Option
  418. app = Celery()
  419. app.user_options['preload'].add(Option(
  420. '--monitoring', action='store_true',
  421. help='Enable our external monitoring utility, blahblah',
  422. ))
  423. @signals.user_preload_options.connect
  424. def handle_preload_options(options, **kwargs):
  425. if options['monitoring']:
  426. enable_monitoring()
  427. Sender is the :class:`~celery.bin.base.Command` instance, which depends
  428. on what program was called (e.g. for the umbrella command it will be
  429. a :class:`~celery.bin.celery.CeleryCommand`) object).
  430. Provides arguments:
  431. * app
  432. The app instance.
  433. * options
  434. Mapping of the parsed user preload options (with default values).
  435. Deprecated Signals
  436. ------------------
  437. .. signal:: task_sent
  438. task_sent
  439. ~~~~~~~~~
  440. This signal is deprecated, please use :signal:`after_task_publish` instead.