tasks.rst 37 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286
  1. .. _guide-tasks:
  2. =======
  3. Tasks
  4. =======
  5. .. contents::
  6. :local:
  7. :depth: 1
  8. .. _task-basics:
  9. Basics
  10. ======
  11. A task is a class that encapsulates a function and its execution options.
  12. Given a function ``create_user`` taking two arguments: `username` and
  13. `password`, you can easily create a task from any function by using
  14. the task decorator:
  15. .. code-block:: python
  16. from django.contrib.auth import User
  17. @celery.task()
  18. def create_user(username, password):
  19. User.objects.create(username=username, password=password)
  20. Task options can be specified as arguments to the decorator:
  21. .. code-block:: python
  22. @celery.task(serializer="json")
  23. def create_user(username, password):
  24. User.objects.create(username=username, password=password)
  25. .. sidebar:: How do I import the task decorator?
  26. The task decorator is available on your :class:`@Celery` instance,
  27. if you don't know what that is then please read :ref:`first-steps`.
  28. If you're using Django or are still using the "old" module based celery API,
  29. then you can import the task decorator like this::
  30. from celery import task
  31. @task()
  32. def add(x, y):
  33. return x + y
  34. .. sidebar:: Multiple decorators
  35. When using multiple decorators in combination with the task
  36. decorator you must make sure that the `task`
  37. decorator is applied last (which in Python oddly means that it must
  38. be the first in the list):
  39. .. code-block:: python
  40. @celery.task()
  41. @decorator2
  42. @decorator1
  43. def add(x, y):
  44. return x + y
  45. .. _task-request-info:
  46. Context
  47. =======
  48. :attr:`~@Task.request` contains information and state related to
  49. the executing task.
  50. The request defines the following attributes:
  51. :id: The unique id of the executing task.
  52. :taskset: The unique id of the taskset this task is a member of (if any).
  53. :args: Positional arguments.
  54. :kwargs: Keyword arguments.
  55. :retries: How many times the current task has been retried.
  56. An integer starting at `0`.
  57. :is_eager: Set to :const:`True` if the task is executed locally in
  58. the client, and not by a worker.
  59. :logfile: The file the worker logs to. See `Logging`_.
  60. :loglevel: The current log level used.
  61. :hostname: Hostname of the worker instance executing the task.
  62. :delivery_info: Additional message delivery information. This is a mapping
  63. containing the exchange and routing key used to deliver this
  64. task. Used by e.g. :meth:`~@Task.retry`
  65. to resend the task to the same destination queue.
  66. An example task accessing information in the context is:
  67. .. code-block:: python
  68. @celery.task()
  69. def dump_context(x, y):
  70. print("Executing task id %r, args: %r kwargs: %r" % (
  71. add.request.id, add.request.args, add.request.kwargs))
  72. .. _task-logging:
  73. Logging
  74. =======
  75. The worker will automatically set up logging for you, or you can
  76. configure logging manually.
  77. A special logger is available named "celery.task", you can inherit
  78. from this logger to automatically get the task name and unique id as part
  79. of the logs.
  80. The best practice is to create a common logger
  81. for all of your tasks at the top of your module:
  82. .. code-block:: python
  83. from celery.utils.log import get_task_logger
  84. logger = get_task_logger(__name__)
  85. @celery.task()
  86. def add(x, y):
  87. logger.info("Adding %s + %s" % (x, y))
  88. return x + y
  89. Celery uses the standard Python logger library,
  90. for which documentation can be found in the :mod:`logging`
  91. module.
  92. You can also simply use :func:`print`, as anything written to standard
  93. out/-err will be redirected to the workers logs by default (see
  94. :setting:`CELERY_REDIRECT_STDOUTS`).
  95. .. _task-retry:
  96. Retrying a task if something fails
  97. ==================================
  98. :meth:`~@Task.retry` can be used to re-execute the task in the event
  99. of temporary failure, or for any other reason.
  100. A new message will the be sent using the same task-id, and sent
  101. to the same queue as the current task. This also means that you
  102. can track retries using the task's result instance (if a result
  103. backend is enabled).
  104. An example:
  105. .. code-block:: python
  106. @celery.task()
  107. def send_twitter_status(oauth, tweet):
  108. try:
  109. twitter = Twitter(oauth)
  110. twitter.update_status(tweet)
  111. except (Twitter.FailWhaleError, Twitter.LoginError), exc:
  112. raise send_twitter_status.retry(exc=exc)
  113. Here we used the `exc` argument to pass the current exception to
  114. :meth:`~@Task.retry`. At each step of the retry this exception
  115. is available as the tombstone (result) of the task. When
  116. :attr:`~@Task.max_retries` has been exceeded this is the
  117. exception raised. However, if an `exc` argument is not provided the
  118. :exc:`~@RetryTaskError` exception is raised instead.
  119. .. note::
  120. The :meth:`~@Task.retry` call will raise an exception so any code after the retry
  121. will not be reached. This is the :exc:`~@RetryTaskError`
  122. exception, it is not handled as an error but rather as a semi-predicate
  123. to signify to the worker that the task is to be retried,
  124. so that it can store the correct state when a result backend is enabled.
  125. This is normal operation and always happens unless the
  126. ``throw`` argument to retry is set to :const:`False`.
  127. .. _task-retry-custom-delay:
  128. Using a custom retry delay
  129. --------------------------
  130. When a task is to be retried, it can wait for a given amount of time
  131. before doing so, and the default delay is defined by the
  132. :attr:`~@Task.default_retry_delay`
  133. attribute. By default this is set to 3 minutes. Note that the
  134. unit for setting the delay is in seconds (int or float).
  135. You can also provide the `countdown` argument to :meth:`~@Task.retry` to
  136. override this default.
  137. .. code-block:: python
  138. @celery.task(default_retry_delay=30 * 60) # retry in 30 minutes.
  139. def add(x, y):
  140. try:
  141. ...
  142. except Exception, exc:
  143. raise add.retry(exc=exc, countdown=60) # override the default and
  144. # retry in 1 minute
  145. .. _task-options:
  146. Task options
  147. ============
  148. General
  149. -------
  150. .. _task-general-options:
  151. .. attribute:: Task.name
  152. The name the task is registered as.
  153. You can set this name manually, or a name will be
  154. automatically generated using the module and class name. See
  155. :ref:`task-names`.
  156. .. attribute:: Task.request
  157. If the task is being executed this will contain information
  158. about the current request. Thread local storage is used.
  159. See :ref:`task-request-info`.
  160. .. attribute:: Task.abstract
  161. Abstract classes are not registered, but are used as the
  162. base class for new task types.
  163. .. attribute:: Task.max_retries
  164. The maximum number of attempted retries before giving up.
  165. If the number of retries exceeds this value a :exc:`~@MaxRetriesExceeded`
  166. exception will be raised. *NOTE:* You have to call :meth:`~@Task.retry`
  167. manually, as it will not automatically retry on exception..
  168. .. attribute:: Task.default_retry_delay
  169. Default time in seconds before a retry of the task
  170. should be executed. Can be either :class:`int` or :class:`float`.
  171. Default is a 3 minute delay.
  172. .. attribute:: Task.rate_limit
  173. Set the rate limit for this task type, i.e. how many times in
  174. a given period of time is the task allowed to run.
  175. If this is :const:`None` no rate limit is in effect.
  176. If it is an integer, it is interpreted as "tasks per second".
  177. The rate limits can be specified in seconds, minutes or hours
  178. by appending `"/s"`, `"/m"` or `"/h"` to the value.
  179. Example: `"100/m"` (hundred tasks a minute). Default is the
  180. :setting:`CELERY_DEFAULT_RATE_LIMIT` setting, which if not specified means
  181. rate limiting for tasks is disabled by default.
  182. .. attribute:: Task.time_limit
  183. The hard time limit for this task. If not set then the workers default
  184. will be used.
  185. .. attribute:: Task.soft_time_limit
  186. The soft time limit for this task. If not set then the workers default
  187. will be used.
  188. .. attribute:: Task.ignore_result
  189. Don't store task state. Note that this means you can't use
  190. :class:`~celery.result.AsyncResult` to check if the task is ready,
  191. or get its return value.
  192. .. attribute:: Task.store_errors_even_if_ignored
  193. If :const:`True`, errors will be stored even if the task is configured
  194. to ignore results.
  195. .. attribute:: Task.send_error_emails
  196. Send an email whenever a task of this type fails.
  197. Defaults to the :setting:`CELERY_SEND_TASK_ERROR_EMAILS` setting.
  198. See :ref:`conf-error-mails` for more information.
  199. .. attribute:: Task.ErrorMail
  200. If the sending of error emails is enabled for this task, then
  201. this is the class defining the logic to send error mails.
  202. .. attribute:: Task.serializer
  203. A string identifying the default serialization
  204. method to use. Defaults to the :setting:`CELERY_TASK_SERIALIZER`
  205. setting. Can be `pickle` `json`, `yaml`, or any custom
  206. serialization methods that have been registered with
  207. :mod:`kombu.serialization.registry`.
  208. Please see :ref:`calling-serializers` for more information.
  209. .. attribute:: Task.backend
  210. The result store backend to use for this task. Defaults to the
  211. :setting:`CELERY_RESULT_BACKEND` setting.
  212. .. attribute:: Task.acks_late
  213. If set to :const:`True` messages for this task will be acknowledged
  214. **after** the task has been executed, not *just before*, which is
  215. the default behavior.
  216. Note that this means the task may be executed twice if the worker
  217. crashes in the middle of execution, which may be acceptable for some
  218. applications.
  219. The global default can be overridden by the :setting:`CELERY_ACKS_LATE`
  220. setting.
  221. .. _task-track-started:
  222. .. attribute:: Task.track_started
  223. If :const:`True` the task will report its status as "started"
  224. when the task is executed by a worker.
  225. The default value is :const:`False` as the normal behaviour is to not
  226. report that level of granularity. Tasks are either pending, finished,
  227. or waiting to be retried. Having a "started" status can be useful for
  228. when there are long running tasks and there is a need to report which
  229. task is currently running.
  230. The host name and process id of the worker executing the task
  231. will be available in the state metadata (e.g. `result.info["pid"]`)
  232. The global default can be overridden by the
  233. :setting:`CELERY_TRACK_STARTED` setting.
  234. .. seealso::
  235. The API reference for :class:`~@Task`.
  236. .. _task-names:
  237. Task names
  238. ==========
  239. The task type is identified by the *task name*.
  240. If not provided a name will be automatically generated using the module
  241. and class name.
  242. For example:
  243. .. code-block:: python
  244. >>> @celery.task(name="sum-of-two-numbers")
  245. >>> def add(x, y):
  246. ... return x + y
  247. >>> add.name
  248. 'sum-of-two-numbers'
  249. A best practice is to use the module name as a prefix to classify the
  250. tasks using namespaces. This way the name won't collide with the name from
  251. another module:
  252. .. code-block:: python
  253. >>> @celery.task(name="tasks.add")
  254. >>> def add(x, y):
  255. ... return x + y
  256. You can tell the name of the task by investigating its name attribute::
  257. >>> add.name
  258. 'tasks.add'
  259. Which is exactly the name automatically generated for this
  260. task if the module name is "tasks.py":
  261. .. code-block:: python
  262. >>> @celery.task()
  263. >>> def add(x, y):
  264. ... return x + y
  265. >>> add.name
  266. 'tasks.add'
  267. .. _task-naming-relative-imports:
  268. Automatic naming and relative imports
  269. -------------------------------------
  270. Relative imports and automatic name generation does not go well together,
  271. so if you're using relative imports you should set the name explicitly.
  272. For example if the client imports the module "myapp.tasks" as ".tasks", and
  273. the worker imports the module as "myapp.tasks", the generated names won't match
  274. and an :exc:`~@NotRegistered` error will be raised by the worker.
  275. This is also the case if using Django and using `project.myapp`::
  276. INSTALLED_APPS = ("project.myapp", )
  277. The worker will have the tasks registered as "project.myapp.tasks.*",
  278. while this is what happens in the client if the module is imported as
  279. "myapp.tasks":
  280. .. code-block:: python
  281. >>> from myapp.tasks import add
  282. >>> add.name
  283. 'myapp.tasks.add'
  284. For this reason you should never use "project.app", but rather
  285. add the project directory to the Python path::
  286. import os
  287. import sys
  288. sys.path.append(os.getcwd())
  289. INSTALLED_APPS = ("myapp", )
  290. This makes more sense from the reusable app perspective anyway.
  291. .. _task-states:
  292. Task States
  293. ===========
  294. Celery can keep track of the tasks current state. The state also contains the
  295. result of a successful task, or the exception and traceback information of a
  296. failed task.
  297. There are several *result backends* to choose from, and they all have
  298. different strengths and weaknesses (see :ref:`task-result-backends`).
  299. During its lifetime a task will transition through several possible states,
  300. and each state may have arbitrary metadata attached to it. When a task
  301. moves into a new state the previous state is
  302. forgotten about, but some transitions can be deducted, (e.g. a task now
  303. in the :state:`FAILED` state, is implied to have been in the
  304. :state:`STARTED` state at some point).
  305. There are also sets of states, like the set of
  306. :state:`FAILURE_STATES`, and the set of :state:`READY_STATES`.
  307. The client uses the membership of these sets to decide whether
  308. the exception should be re-raised (:state:`PROPAGATE_STATES`), or whether
  309. the state can be cached (it can if the task is ready).
  310. You can also define :ref:`custom-states`.
  311. .. _task-result-backends:
  312. Result Backends
  313. ---------------
  314. Celery needs to store or send the states somewhere. There are several
  315. built-in backends to choose from: SQLAlchemy/Django ORM, Memcached, Redis,
  316. AMQP, MongoDB, Tokyo Tyrant and Redis -- or you can define your own.
  317. No backend works well for every use case.
  318. You should read about the strengths and weaknesses of each backend, and choose
  319. the most appropriate for your needs.
  320. .. seealso::
  321. :ref:`conf-result-backend`
  322. AMQP Result Backend
  323. ~~~~~~~~~~~~~~~~~~~
  324. The AMQP result backend is special as it does not actually *store* the states,
  325. but rather sends them as messages. This is an important difference as it
  326. means that a result *can only be retrieved once*; If you have two processes
  327. waiting for the same result, one of the processes will never receive the
  328. result!
  329. Even with that limitation, it is an excellent choice if you need to receive
  330. state changes in real-time. Using messaging means the client does not have to
  331. poll for new states.
  332. There are several other pitfalls you should be aware of when using the AMQP
  333. backend:
  334. * Every new task creates a new queue on the server, with thousands of tasks
  335. the broker may be overloaded with queues and this will affect performance in
  336. negative ways. If you're using RabbitMQ then each queue will be a separate
  337. Erlang process, so if you're planning to keep many results simultaneously you
  338. may have to increase the Erlang process limit, and the maximum number of file
  339. descriptors your OS allows.
  340. * Old results will be cleaned automatically, based on the
  341. :setting:`CELERY_TASK_RESULT_EXPIRES` setting. By default this is set to
  342. expire after 1 day: if you have a very busy cluster you should lower
  343. this value.
  344. For a list of options supported by the AMQP result backend, please see
  345. :ref:`conf-amqp-result-backend`.
  346. Database Result Backend
  347. ~~~~~~~~~~~~~~~~~~~~~~~
  348. Keeping state in the database can be convenient for many, especially for
  349. web applications with a database already in place, but it also comes with
  350. limitations.
  351. * Polling the database for new states is expensive, and so you should
  352. increase the polling intervals of operations such as `result.get()`.
  353. * Some databases use a default transaction isolation level that
  354. is not suitable for polling tables for changes.
  355. In MySQL the default transaction isolation level is `REPEATABLE-READ`, which
  356. means the transaction will not see changes by other transactions until the
  357. transaction is committed. It is recommended that you change to the
  358. `READ-COMMITTED` isolation level.
  359. .. _task-builtin-states:
  360. Built-in States
  361. ---------------
  362. .. state:: PENDING
  363. PENDING
  364. ~~~~~~~
  365. Task is waiting for execution or unknown.
  366. Any task id that is not known is implied to be in the pending state.
  367. .. state:: STARTED
  368. STARTED
  369. ~~~~~~~
  370. Task has been started.
  371. Not reported by default, to enable please see :attr:`@Task.track_started`.
  372. :metadata: `pid` and `hostname` of the worker process executing
  373. the task.
  374. .. state:: SUCCESS
  375. SUCCESS
  376. ~~~~~~~
  377. Task has been successfully executed.
  378. :metadata: `result` contains the return value of the task.
  379. :propagates: Yes
  380. :ready: Yes
  381. .. state:: FAILURE
  382. FAILURE
  383. ~~~~~~~
  384. Task execution resulted in failure.
  385. :metadata: `result` contains the exception occurred, and `traceback`
  386. contains the backtrace of the stack at the point when the
  387. exception was raised.
  388. :propagates: Yes
  389. .. state:: RETRY
  390. RETRY
  391. ~~~~~
  392. Task is being retried.
  393. :metadata: `result` contains the exception that caused the retry,
  394. and `traceback` contains the backtrace of the stack at the point
  395. when the exceptions was raised.
  396. :propagates: No
  397. .. state:: REVOKED
  398. REVOKED
  399. ~~~~~~~
  400. Task has been revoked.
  401. :propagates: Yes
  402. .. _custom-states:
  403. Custom states
  404. -------------
  405. You can easily define your own states, all you need is a unique name.
  406. The name of the state is usually an uppercase string. As an example
  407. you could have a look at :mod:`abortable tasks <~celery.contrib.abortable>`
  408. which defines its own custom :state:`ABORTED` state.
  409. Use :meth:`~@Task.update_state` to update a task's state::
  410. from celery import current_task
  411. @celery.task()
  412. def upload_files(filenames):
  413. for i, file in enumerate(filenames):
  414. current_task.update_state(state="PROGRESS",
  415. meta={"current": i, "total": len(filenames)})
  416. Here we created the state `"PROGRESS"`, which tells any application
  417. aware of this state that the task is currently in progress, and also where
  418. it is in the process by having `current` and `total` counts as part of the
  419. state metadata. This can then be used to create e.g. progress bars.
  420. .. _pickling_exceptions:
  421. Creating pickleable exceptions
  422. ------------------------------
  423. A little known Python fact is that exceptions must behave a certain
  424. way to support being pickled.
  425. Tasks that raise exceptions that are not pickleable will not work
  426. properly when Pickle is used as the serializer.
  427. To make sure that your exceptions are pickleable the exception
  428. *MUST* provide the original arguments it was instantiated
  429. with in its ``.args`` attribute. The simplest way
  430. to ensure this is to have the exception call ``Exception.__init__``.
  431. Let's look at some examples that work, and one that doesn't:
  432. .. code-block:: python
  433. # OK:
  434. class HttpError(Exception):
  435. pass
  436. # BAD:
  437. class HttpError(Exception):
  438. def __init__(self, status_code):
  439. self.status_code = status_code
  440. # OK:
  441. class HttpError(Exception):
  442. def __init__(self, status_code):
  443. self.status_code = status_code
  444. Exception.__init__(self, status_code) # <-- REQUIRED
  445. So the rule is:
  446. For any exception that supports custom arguments ``*args``,
  447. ``Exception.__init__(self, *args)`` must be used.
  448. There is no special support for *keyword arguments*, so if you
  449. want to preserve keyword arguments when the exception is unpickled
  450. you have to pass them as regular args:
  451. .. code-block:: python
  452. class HttpError(Exception):
  453. def __init__(self, status_code, headers=None, body=None):
  454. self.status_code = status_code
  455. self.headers = headers
  456. self.body = body
  457. super(HttpError, self).__init__(status_code, headers, body)
  458. .. _task-custom-classes:
  459. Creating custom task classes
  460. ============================
  461. All tasks inherit from the :class:`@Task` class.
  462. The :meth:`~@Task.run` method becomes the task body.
  463. As an example, the following code,
  464. .. code-block:: python
  465. @celery.task()
  466. def add(x, y):
  467. return x + y
  468. will do roughly this behind the scenes:
  469. .. code-block:: python
  470. @celery.task()
  471. class AddTask(Task):
  472. def run(self, x, y):
  473. return x + y
  474. add = registry.tasks[AddTask.name]
  475. Instantiation
  476. -------------
  477. A task is **not** instantiated for every request, but is registered
  478. in the task registry as a global instance.
  479. This means that the ``__init__`` constructor will only be called
  480. once per process, and that the task class is semantically closer to an
  481. Actor.
  482. If you have a task,
  483. .. code-block:: python
  484. from celery import Task
  485. class NaiveAuthenticateServer(Task):
  486. def __init__(self):
  487. self.users = {"george": "password"}
  488. def run(self, username, password):
  489. try:
  490. return self.users[username] == password
  491. except KeyError:
  492. return False
  493. And you route every request to the same process, then it
  494. will keep state between requests.
  495. This can also be useful to keep cached resources::
  496. class DatabaseTask(Task):
  497. _db = None
  498. @property
  499. def db(self):
  500. if self._db = None:
  501. self._db = Database.connect()
  502. return self._db
  503. Abstract classes
  504. ----------------
  505. Abstract classes are not registered, but are used as the
  506. base class for new task types.
  507. .. code-block:: python
  508. from celery import Task
  509. class DebugTask(Task):
  510. abstract = True
  511. def after_return(self, *args, **kwargs):
  512. print("Task returned: %r" % (self.request, ))
  513. @celery.task(base=DebugTask)
  514. def add(x, y):
  515. return x + y
  516. Handlers
  517. --------
  518. .. method:: after_return(self, status, retval, task_id, args, kwargs, einfo)
  519. Handler called after the task returns.
  520. :param status: Current task state.
  521. :param retval: Task return value/exception.
  522. :param task_id: Unique id of the task.
  523. :param args: Original arguments for the task that failed.
  524. :param kwargs: Original keyword arguments for the task
  525. that failed.
  526. :keyword einfo: :class:`~celery.datastructures.ExceptionInfo`
  527. instance, containing the traceback (if any).
  528. The return value of this handler is ignored.
  529. .. method:: on_failure(self, exc, task_id, args, kwargs, einfo)
  530. This is run by the worker when the task fails.
  531. :param exc: The exception raised by the task.
  532. :param task_id: Unique id of the failed task.
  533. :param args: Original arguments for the task that failed.
  534. :param kwargs: Original keyword arguments for the task
  535. that failed.
  536. :keyword einfo: :class:`~celery.datastructures.ExceptionInfo`
  537. instance, containing the traceback.
  538. The return value of this handler is ignored.
  539. .. method:: on_retry(self, exc, task_id, args, kwargs, einfo)
  540. This is run by the worker when the task is to be retried.
  541. :param exc: The exception sent to :meth:`~@Task.retry`.
  542. :param task_id: Unique id of the retried task.
  543. :param args: Original arguments for the retried task.
  544. :param kwargs: Original keyword arguments for the retried task.
  545. :keyword einfo: :class:`~celery.datastructures.ExceptionInfo`
  546. instance, containing the traceback.
  547. The return value of this handler is ignored.
  548. .. method:: on_success(self, retval, task_id, args, kwargs)
  549. Run by the worker if the task executes successfully.
  550. :param retval: The return value of the task.
  551. :param task_id: Unique id of the executed task.
  552. :param args: Original arguments for the executed task.
  553. :param kwargs: Original keyword arguments for the executed task.
  554. The return value of this handler is ignored.
  555. on_retry
  556. ~~~~~~~~
  557. .. _task-how-they-work:
  558. How it works
  559. ============
  560. Here comes the technical details, this part isn't something you need to know,
  561. but you may be interested.
  562. All defined tasks are listed in a registry. The registry contains
  563. a list of task names and their task classes. You can investigate this registry
  564. yourself:
  565. .. code-block:: python
  566. >>> from celery import current_app
  567. >>> current_app.tasks
  568. {'celery.chord_unlock':
  569. <@task: celery.chord_unlock>,
  570. 'celery.backend_cleanup':
  571. <@task: celery.backend_cleanup>,
  572. 'celery.chord':
  573. <@task: celery.chord>}
  574. This is the list of tasks built-in to celery. Note that tasks
  575. will only be registered when the module they are defined in is imported.
  576. The default loader imports any modules listed in the
  577. :setting:`CELERY_IMPORTS` setting.
  578. The entity responsible for registering your task in the registry is the
  579. metaclass: :class:`~celery.task.base.TaskType`.
  580. If you want to register your task manually you can mark the
  581. task as :attr:`~@Task.abstract`:
  582. .. code-block:: python
  583. class MyTask(Task):
  584. abstract = True
  585. This way the task won't be registered, but any task inheriting from
  586. it will be.
  587. When tasks are sent, we don't send any actual function code, just the name
  588. of the task to execute. When the worker then receives the message it can look
  589. up the name in its task registry to find the execution code.
  590. This means that your workers should always be updated with the same software
  591. as the client. This is a drawback, but the alternative is a technical
  592. challenge that has yet to be solved.
  593. .. _task-best-practices:
  594. Tips and Best Practices
  595. =======================
  596. .. _task-ignore_results:
  597. Ignore results you don't want
  598. -----------------------------
  599. If you don't care about the results of a task, be sure to set the
  600. :attr:`~@Task.ignore_result` option, as storing results
  601. wastes time and resources.
  602. .. code-block:: python
  603. @celery.task(ignore_result=True)
  604. def mytask(...)
  605. something()
  606. Results can even be disabled globally using the :setting:`CELERY_IGNORE_RESULT`
  607. setting.
  608. .. _task-disable-rate-limits:
  609. Disable rate limits if they're not used
  610. ---------------------------------------
  611. Disabling rate limits altogether is recommended if you don't have
  612. any tasks using them. This is because the rate limit subsystem introduces
  613. quite a lot of complexity.
  614. Set the :setting:`CELERY_DISABLE_RATE_LIMITS` setting to globally disable
  615. rate limits:
  616. .. code-block:: python
  617. CELERY_DISABLE_RATE_LIMITS = True
  618. You find additional optimization tips in the
  619. :ref:`Optimizing Guide <guide-optimizing>`.
  620. .. _task-synchronous-subtasks:
  621. Avoid launching synchronous subtasks
  622. ------------------------------------
  623. Having a task wait for the result of another task is really inefficient,
  624. and may even cause a deadlock if the worker pool is exhausted.
  625. Make your design asynchronous instead, for example by using *callbacks*.
  626. **Bad**:
  627. .. code-block:: python
  628. @celery.task()
  629. def update_page_info(url):
  630. page = fetch_page.delay(url).get()
  631. info = parse_page.delay(url, page).get()
  632. store_page_info.delay(url, info)
  633. @celery.task()
  634. def fetch_page(url):
  635. return myhttplib.get(url)
  636. @celery.task()
  637. def parse_page(url, page):
  638. return myparser.parse_document(page)
  639. @celery.task()
  640. def store_page_info(url, info):
  641. return PageInfo.objects.create(url, info)
  642. **Good**:
  643. .. code-block:: python
  644. def update_page_info(url):
  645. # fetch_page -> parse_page -> store_page
  646. chain = fetch_page.s() | parse_page.s(url) | store_page_info.s(url)
  647. chain.apply_async()
  648. @celery.task(ignore_result=True)
  649. def fetch_page(url):
  650. return myhttplib.get(url)
  651. @celery.task(ignore_result=True)
  652. def parse_page(url, page):
  653. return myparser.parse_document(page)
  654. @celery.task(ignore_result=True)
  655. def store_page_info(url, info):
  656. PageInfo.objects.create(url, info)
  657. Here we instead create a chain of tasks by linking together
  658. different :func:`~celery.subtask`'s.
  659. You can read about chains and other powerful constructs
  660. at :ref:`designing-workflows`.
  661. .. _task-performance-and-strategies:
  662. Performance and Strategies
  663. ==========================
  664. .. _task-granularity:
  665. Granularity
  666. -----------
  667. The task granularity is the amount of computation needed by each subtask.
  668. In general it is better to split the problem up into many small tasks, than
  669. have a few long running tasks.
  670. With smaller tasks you can process more tasks in parallel and the tasks
  671. won't run long enough to block the worker from processing other waiting tasks.
  672. However, executing a task does have overhead. A message needs to be sent, data
  673. may not be local, etc. So if the tasks are too fine-grained the additional
  674. overhead may not be worth it in the end.
  675. .. seealso::
  676. The book `Art of Concurrency`_ has a whole section dedicated to the topic
  677. of task granularity.
  678. .. _`Art of Concurrency`: http://oreilly.com/catalog/9780596521547
  679. .. _task-data-locality:
  680. Data locality
  681. -------------
  682. The worker processing the task should be as close to the data as
  683. possible. The best would be to have a copy in memory, the worst would be a
  684. full transfer from another continent.
  685. If the data is far away, you could try to run another worker at location, or
  686. if that's not possible - cache often used data, or preload data you know
  687. is going to be used.
  688. The easiest way to share data between workers is to use a distributed cache
  689. system, like `memcached`_.
  690. .. seealso::
  691. The paper `Distributed Computing Economics`_ by Jim Gray is an excellent
  692. introduction to the topic of data locality.
  693. .. _`Distributed Computing Economics`:
  694. http://research.microsoft.com/pubs/70001/tr-2003-24.pdf
  695. .. _`memcached`: http://memcached.org/
  696. .. _task-state:
  697. State
  698. -----
  699. Since celery is a distributed system, you can't know in which process, or
  700. on what machine the task will be executed. You can't even know if the task will
  701. run in a timely manner.
  702. The ancient async sayings tells us that “asserting the world is the
  703. responsibility of the task”. What this means is that the world view may
  704. have changed since the task was requested, so the task is responsible for
  705. making sure the world is how it should be; If you have a task
  706. that re-indexes a search engine, and the search engine should only be
  707. re-indexed at maximum every 5 minutes, then it must be the tasks
  708. responsibility to assert that, not the callers.
  709. Another gotcha is Django model objects. They shouldn't be passed on as
  710. arguments to tasks. It's almost always better to re-fetch the object from
  711. the database when the task is running instead, as using old data may lead
  712. to race conditions.
  713. Imagine the following scenario where you have an article and a task
  714. that automatically expands some abbreviations in it:
  715. .. code-block:: python
  716. class Article(models.Model):
  717. title = models.CharField()
  718. body = models.TextField()
  719. @celery.task()
  720. def expand_abbreviations(article):
  721. article.body.replace("MyCorp", "My Corporation")
  722. article.save()
  723. First, an author creates an article and saves it, then the author
  724. clicks on a button that initiates the abbreviation task::
  725. >>> article = Article.objects.get(id=102)
  726. >>> expand_abbreviations.delay(model_object)
  727. Now, the queue is very busy, so the task won't be run for another 2 minutes.
  728. In the meantime another author makes changes to the article, so
  729. when the task is finally run, the body of the article is reverted to the old
  730. version because the task had the old body in its argument.
  731. Fixing the race condition is easy, just use the article id instead, and
  732. re-fetch the article in the task body:
  733. .. code-block:: python
  734. @celery.task()
  735. def expand_abbreviations(article_id):
  736. article = Article.objects.get(id=article_id)
  737. article.body.replace("MyCorp", "My Corporation")
  738. article.save()
  739. >>> expand_abbreviations(article_id)
  740. There might even be performance benefits to this approach, as sending large
  741. messages may be expensive.
  742. .. _task-database-transactions:
  743. Database transactions
  744. ---------------------
  745. Let's have a look at another example:
  746. .. code-block:: python
  747. from django.db import transaction
  748. @transaction.commit_on_success
  749. def create_article(request):
  750. article = Article.objects.create(....)
  751. expand_abbreviations.delay(article.pk)
  752. This is a Django view creating an article object in the database,
  753. then passing the primary key to a task. It uses the `commit_on_success`
  754. decorator, which will commit the transaction when the view returns, or
  755. roll back if the view raises an exception.
  756. There is a race condition if the task starts executing
  757. before the transaction has been committed; The database object does not exist
  758. yet!
  759. The solution is to *always commit transactions before sending tasks
  760. depending on state from the current transaction*:
  761. .. code-block:: python
  762. @transaction.commit_manually
  763. def create_article(request):
  764. try:
  765. article = Article.objects.create(...)
  766. except:
  767. transaction.rollback()
  768. raise
  769. else:
  770. transaction.commit()
  771. expand_abbreviations.delay(article.pk)
  772. .. _task-example:
  773. Example
  774. =======
  775. Let's take a real wold example; A blog where comments posted needs to be
  776. filtered for spam. When the comment is created, the spam filter runs in the
  777. background, so the user doesn't have to wait for it to finish.
  778. We have a Django blog application allowing comments
  779. on blog posts. We'll describe parts of the models/views and tasks for this
  780. application.
  781. blog/models.py
  782. --------------
  783. The comment model looks like this:
  784. .. code-block:: python
  785. from django.db import models
  786. from django.utils.translation import ugettext_lazy as _
  787. class Comment(models.Model):
  788. name = models.CharField(_("name"), max_length=64)
  789. email_address = models.EmailField(_("email address"))
  790. homepage = models.URLField(_("home page"),
  791. blank=True, verify_exists=False)
  792. comment = models.TextField(_("comment"))
  793. pub_date = models.DateTimeField(_("Published date"),
  794. editable=False, auto_add_now=True)
  795. is_spam = models.BooleanField(_("spam?"),
  796. default=False, editable=False)
  797. class Meta:
  798. verbose_name = _("comment")
  799. verbose_name_plural = _("comments")
  800. In the view where the comment is posted, we first write the comment
  801. to the database, then we launch the spam filter task in the background.
  802. .. _task-example-blog-views:
  803. blog/views.py
  804. -------------
  805. .. code-block:: python
  806. from django import forms
  807. from django.http import HttpResponseRedirect
  808. from django.template.context import RequestContext
  809. from django.shortcuts import get_object_or_404, render_to_response
  810. from blog import tasks
  811. from blog.models import Comment
  812. class CommentForm(forms.ModelForm):
  813. class Meta:
  814. model = Comment
  815. def add_comment(request, slug, template_name="comments/create.html"):
  816. post = get_object_or_404(Entry, slug=slug)
  817. remote_addr = request.META.get("REMOTE_ADDR")
  818. if request.method == "post":
  819. form = CommentForm(request.POST, request.FILES)
  820. if form.is_valid():
  821. comment = form.save()
  822. # Check spam asynchronously.
  823. tasks.spam_filter.delay(comment_id=comment.id,
  824. remote_addr=remote_addr)
  825. return HttpResponseRedirect(post.get_absolute_url())
  826. else:
  827. form = CommentForm()
  828. context = RequestContext(request, {"form": form})
  829. return render_to_response(template_name, context_instance=context)
  830. To filter spam in comments we use `Akismet`_, the service
  831. used to filter spam in comments posted to the free weblog platform
  832. `Wordpress`. `Akismet`_ is free for personal use, but for commercial use you
  833. need to pay. You have to sign up to their service to get an API key.
  834. To make API calls to `Akismet`_ we use the `akismet.py`_ library written by
  835. `Michael Foord`_.
  836. .. _task-example-blog-tasks:
  837. blog/tasks.py
  838. -------------
  839. .. code-block:: python
  840. import celery
  841. from akismet import Akismet
  842. from django.core.exceptions import ImproperlyConfigured
  843. from django.contrib.sites.models import Site
  844. from blog.models import Comment
  845. @celery.task()
  846. def spam_filter(comment_id, remote_addr=None):
  847. logger = spam_filter.get_logger()
  848. logger.info("Running spam filter for comment %s" % comment_id)
  849. comment = Comment.objects.get(pk=comment_id)
  850. current_domain = Site.objects.get_current().domain
  851. akismet = Akismet(settings.AKISMET_KEY, "http://%s" % domain)
  852. if not akismet.verify_key():
  853. raise ImproperlyConfigured("Invalid AKISMET_KEY")
  854. is_spam = akismet.comment_check(user_ip=remote_addr,
  855. comment_content=comment.comment,
  856. comment_author=comment.name,
  857. comment_author_email=comment.email_address)
  858. if is_spam:
  859. comment.is_spam = True
  860. comment.save()
  861. return is_spam
  862. .. _`Akismet`: http://akismet.com/faq/
  863. .. _`akismet.py`: http://www.voidspace.org.uk/downloads/akismet.py
  864. .. _`Michael Foord`: http://www.voidspace.org.uk/