tasks.rst 56 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884
  1. .. _guide-tasks:
  2. =====================================================================
  3. Tasks
  4. =====================================================================
  5. Tasks are the building blocks of Celery applications.
  6. A task is a class that can be created out of any callable. It performs
  7. dual roles in that it defines both what happens when a task is
  8. called (sends a message), and what happens when a worker receives that message.
  9. Every task class has a unique name, and this name is referenced in messages
  10. so the worker can find the right function to execute.
  11. A task message is not removed from the queue
  12. until that message has been :term:`acknowledged` by a worker. A worker can reserve
  13. many messages in advance and even if the worker is killed -- by power failure
  14. or some other reason -- the message will be redelivered to another worker.
  15. Ideally task functions should be :term:`idempotent`: meaning
  16. the function won't cause unintended effects even if called
  17. multiple times with the same arguments.
  18. Since the worker cannot detect if your tasks are idempotent, the default
  19. behavior is to acknowledge the message in advance, just before it's executed,
  20. so that a task invocation that already started is never executed again.
  21. If your task is idempotent you can set the :attr:`~Task.acks_late` option
  22. to have the worker acknowledge the message *after* the task returns
  23. instead. See also the FAQ entry :ref:`faq-acks_late-vs-retry`.
  24. Note that the worker will acknowledge the message if the child process executing
  25. the task is terminated (either by the task calling :func:`sys.exit`, or by signal)
  26. even when :attr:`~Task.acks_late` is enabled. This behavior is by purpose
  27. as...
  28. #. We don't want to rerun tasks that forces the kernel to send
  29. a :sig:`SIGSEGV` (segmentation fault) or similar signals to the process.
  30. #. We assume that a system administrator deliberately killing the task
  31. does not want it to automatically restart.
  32. #. A task that allocates too much memory is in danger of triggering the kernel
  33. OOM killer, the same may happen again.
  34. #. A task that always fails when redelivered may cause a high-frequency
  35. message loop taking down the system.
  36. If you really want a task to be redelivered in these scenarios you should
  37. consider enabling the :setting:`task_reject_on_worker_lost` setting.
  38. .. warning::
  39. A task that blocks indefinitely may eventually stop the worker instance
  40. from doing any other work.
  41. If you task does I/O then make sure you add timeouts to these operations,
  42. like adding a timeout to a web request using the :pypi:`requests` library:
  43. .. code-block:: python
  44. connect_timeout, read_timeout = 5.0, 30.0
  45. response = requests.get(URL, timeout=(connect_timeout, read_timeout))
  46. :ref:`Time limits <worker-time-limits>` are convenient for making sure all
  47. tasks return in a timely manner, but a time limit event will actually kill
  48. the process by force so only use them to detect cases where you haven't
  49. used manual timeouts yet.
  50. The default prefork pool scheduler is not friendly to long-running tasks,
  51. so if you have tasks that run for minutes/hours make sure you enable
  52. the -Ofair`` command-line argument to the :program:`celery worker`.
  53. See :ref:`prefork-pool-prefetch` for more information, and for the
  54. best performance route long-running and short-running tasks to
  55. dedicated workers (:ref:`routing-automatic`).
  56. If your worker hangs then please investigate what tasks are running
  57. before submitting an issue, as most likely the hanging is caused
  58. by one or more tasks hanging on a network operation.
  59. --
  60. In this chapter you'll learn all about defining tasks,
  61. and this is the **table of contents**:
  62. .. contents::
  63. :local:
  64. :depth: 1
  65. .. _task-basics:
  66. Basics
  67. ======
  68. You can easily create a task from any callable by using
  69. the :meth:`~@task` decorator:
  70. .. code-block:: python
  71. from .models import User
  72. @app.task
  73. def create_user(username, password):
  74. User.objects.create(username=username, password=password)
  75. There are also many :ref:`options <task-options>` that can be set for the task,
  76. these can be specified as arguments to the decorator:
  77. .. code-block:: python
  78. @app.task(serializer='json')
  79. def create_user(username, password):
  80. User.objects.create(username=username, password=password)
  81. .. sidebar:: How do I import the task decorator? And what's "app"?
  82. The task decorator is available on your :class:`@Celery` application instance,
  83. if you don't know what this is then please read :ref:`first-steps`.
  84. If you're using Django (see :ref:`django-first-steps`), or you're the author
  85. of a library then you probably want to use the :func:`@shared_task` decorator:
  86. .. code-block:: python
  87. from celery import shared_task
  88. @shared_task
  89. def add(x, y):
  90. return x + y
  91. .. sidebar:: Multiple decorators
  92. When using multiple decorators in combination with the task
  93. decorator you must make sure that the `task`
  94. decorator is applied last (oddly, in Python this means it must
  95. be first in the list):
  96. .. code-block:: python
  97. @app.task
  98. @decorator2
  99. @decorator1
  100. def add(x, y):
  101. return x + y
  102. Bound tasks
  103. -----------
  104. A task being bound means the first argument to the task will always
  105. be the task instance (``self``), just like Python bound methods:
  106. .. code-block:: python
  107. logger = get_task_logger(__name__)
  108. @task(bind=True)
  109. def add(self, x, y):
  110. logger.info(self.request.id)
  111. Bound tasks are needed for retries (using :meth:`Task.retry() <@Task.retry>`),
  112. for accessing information about the current task request, and for any
  113. additional functionality you add to custom task base classes.
  114. Task inheritance
  115. ----------------
  116. The ``base`` argument to the task decorator specifies the base class of the task:
  117. .. code-block:: python
  118. import celery
  119. class MyTask(celery.Task):
  120. def on_failure(self, exc, task_id, args, kwargs, einfo):
  121. print('{0!r} failed: {1!r}'.format(task_id, exc)
  122. @task(base=MyTask)
  123. def add(x, y):
  124. raise KeyError()
  125. .. _task-names:
  126. Names
  127. =====
  128. Every task must have a unique name.
  129. If no explicit name is provided the task decorator will generate one for you,
  130. and this name will be based on 1) the module the task is defined in, and 2)
  131. the name of the task function.
  132. Example setting explicit name:
  133. .. code-block:: pycon
  134. >>> @app.task(name='sum-of-two-numbers')
  135. >>> def add(x, y):
  136. ... return x + y
  137. >>> add.name
  138. 'sum-of-two-numbers'
  139. A best practice is to use the module name as a name-space,
  140. this way names won't collide if there's already a task with that name
  141. defined in another module.
  142. .. code-block:: pycon
  143. >>> @app.task(name='tasks.add')
  144. >>> def add(x, y):
  145. ... return x + y
  146. You can tell the name of the task by investigating its ``.name`` attribute:
  147. .. code-block:: pycon
  148. >>> add.name
  149. 'tasks.add'
  150. The name we specified here (``tasks.add``) is exactly the name that would've
  151. been automatically generated for us if the task was defined in a module
  152. named :file:`tasks.py`:
  153. :file:`tasks.py`:
  154. .. code-block:: python
  155. @app.task
  156. def add(x, y):
  157. return x + y
  158. .. code-block:: pycon
  159. >>> from tasks import add
  160. >>> add.name
  161. 'tasks.add'
  162. .. _task-naming-relative-imports:
  163. Automatic naming and relative imports
  164. -------------------------------------
  165. .. sidebar:: Absolute Imports
  166. The best practice for developers targetting Python 2 is to add the
  167. following to the top of **every module**:
  168. .. code-block:: python
  169. from __future__ import absolute_import
  170. This will force you to always use absolute imports so you will
  171. never have any problems with tasks using relative names.
  172. Absolute imports are the default in Python 3 so you don't need this
  173. if you target that version.
  174. Relative imports and automatic name generation don't go well together,
  175. so if you're using relative imports you should set the name explicitly.
  176. For example if the client imports the module ``"myapp.tasks"``
  177. as ``".tasks"``, and the worker imports the module as ``"myapp.tasks"``,
  178. the generated names won't match and an :exc:`~@NotRegistered` error will
  179. be raised by the worker.
  180. This is also the case when using Django and using ``project.myapp``-style
  181. naming in ``INSTALLED_APPS``:
  182. .. code-block:: python
  183. INSTALLED_APPS = ['project.myapp']
  184. If you install the app under the name ``project.myapp`` then the
  185. tasks module will be imported as ``project.myapp.tasks``,
  186. so you must make sure you always import the tasks using the same name:
  187. .. code-block:: pycon
  188. >>> from project.myapp.tasks import mytask # << GOOD
  189. >>> from myapp.tasks import mytask # << BAD!!!
  190. The second example will cause the task to be named differently
  191. since the worker and the client imports the modules under different names:
  192. .. code-block:: pycon
  193. >>> from project.myapp.tasks import mytask
  194. >>> mytask.name
  195. 'project.myapp.tasks.mytask'
  196. >>> from myapp.tasks import mytask
  197. >>> mytask.name
  198. 'myapp.tasks.mytask'
  199. For this reason you must be consistent in how you
  200. import modules, and that is also a Python best practice.
  201. Similarly, you shouldn't use old-style relative imports:
  202. .. code-block:: python
  203. from module import foo # BAD!
  204. from proj.module import foo # GOOD!
  205. New-style relative imports are fine and can be used:
  206. .. code-block:: python
  207. from .module import foo # GOOD!
  208. If you want to use Celery with a project already using these patterns
  209. extensively and you don't have the time to refactor the existing code
  210. then you can consider specifying the names explicitly instead of relying
  211. on the automatic naming:
  212. .. code-block:: python
  213. @task(name='proj.tasks.add')
  214. def add(x, y):
  215. return x + y
  216. .. _task-name-generator-info:
  217. Changing the automatic naming behavior
  218. --------------------------------------
  219. .. versionadded:: 4.0
  220. There are some cases when the default automatic naming isn't suitable.
  221. Consider you have many tasks within many different modules::
  222. project/
  223. /__init__.py
  224. /celery.py
  225. /moduleA/
  226. /__init__.py
  227. /tasks.py
  228. /moduleB/
  229. /__init__.py
  230. /tasks.py
  231. Using the default automatic naming, each task will have a generated name
  232. like `moduleA.tasks.taskA`, `moduleA.tasks.taskB`, `moduleB.tasks.test`,
  233. and so on. You may want to get rid of having `tasks` in all task names.
  234. As pointed above, you can explicitly give names for all tasks, or you
  235. can change the automatic naming behavior by overriding
  236. :meth:`@gen_task_name`. Continuing with the example, `celery.py`
  237. may contain:
  238. .. code-block:: python
  239. from celery import Celery
  240. class MyCelery(Celery):
  241. def gen_task_name(self, name, module):
  242. if module.endswith('.tasks'):
  243. module = module[:-6]
  244. return super().gen_task_name(name, module)
  245. app = MyCelery('main')
  246. So each task will have a name like `moduleA.taskA`, `moduleA.taskB` and
  247. `moduleB.test`.
  248. .. warning::
  249. Make sure that your :meth:`@gen_task_name` is a pure function: meaning
  250. that for the same input it must always return the same output.
  251. .. _task-request-info:
  252. Task Request
  253. ============
  254. :attr:`Task.request <@Task.request>` contains information and state
  255. related to the currently executing task.
  256. The request defines the following attributes:
  257. :id: The unique id of the executing task.
  258. :group: The unique id of the task's :ref:`group <canvas-group>`, if this task is a member.
  259. :chord: The unique id of the chord this task belongs to (if the task
  260. is part of the header).
  261. :correlation_id: Custom ID used for things like de-duplication.
  262. :args: Positional arguments.
  263. :kwargs: Keyword arguments.
  264. :origin: Name of host that sent this task.
  265. :retries: How many times the current task has been retried.
  266. An integer starting at `0`.
  267. :is_eager: Set to :const:`True` if the task is executed locally in
  268. the client, not by a worker.
  269. :eta: The original ETA of the task (if any).
  270. This is in UTC time (depending on the :setting:`enable_utc`
  271. setting).
  272. :expires: The original expiry time of the task (if any).
  273. This is in UTC time (depending on the :setting:`enable_utc`
  274. setting).
  275. :hostname: Node name of the worker instance executing the task.
  276. :delivery_info: Additional message delivery information. This is a mapping
  277. containing the exchange and routing key used to deliver this
  278. task. Used by for example :meth:`Task.retry() <@Task.retry>`
  279. to resend the task to the same destination queue.
  280. Availability of keys in this dict depends on the
  281. message broker used.
  282. :reply-to: Name of queue to send replies back to (used with RPC result
  283. backend for example).
  284. :called_directly: This flag is set to true if the task wasn't
  285. executed by the worker.
  286. :timelimit: A tuple of the current ``(soft, hard)`` time limits active for
  287. this task (if any).
  288. :callbacks: A list of signatures to be called if this task returns successfully.
  289. :errback: A list of signatures to be called if this task fails.
  290. :utc: Set to true the caller has UTC enabled (:setting:`enable_utc`).
  291. .. versionadded:: 3.1
  292. :headers: Mapping of message headers sent with this task message
  293. (may be :const:`None`).
  294. :reply_to: Where to send reply to (queue name).
  295. :correlation_id: Usually the same as the task id, often used in amqp
  296. to keep track of what a reply is for.
  297. .. versionadded:: 4.0
  298. :root_id: The unique id of the first task in the workflow this task
  299. is part of (if any).
  300. :parent_id: The unique id of the task that called this task (if any).
  301. :chain: Reversed list of tasks that form a chain (if any).
  302. The last item in this list will be the next task to succeed the
  303. current task. If using version one of the task protocol the chain
  304. tasks will be in ``request.callbacks`` instead.
  305. Example
  306. -------
  307. An example task accessing information in the context is:
  308. .. code-block:: python
  309. @app.task(bind=True)
  310. def dump_context(self, x, y):
  311. print('Executing task id {0.id}, args: {0.args!r} kwargs: {0.kwargs!r}'.format(
  312. self.request))
  313. The ``bind`` argument means that the function will be a "bound method" so
  314. that you can access attributes and methods on the task type instance.
  315. .. _task-logging:
  316. Logging
  317. =======
  318. The worker will automatically set up logging for you, or you can
  319. configure logging manually.
  320. A special logger is available named "celery.task", you can inherit
  321. from this logger to automatically get the task name and unique id as part
  322. of the logs.
  323. The best practice is to create a common logger
  324. for all of your tasks at the top of your module:
  325. .. code-block:: python
  326. from celery.utils.log import get_task_logger
  327. logger = get_task_logger(__name__)
  328. @app.task
  329. def add(x, y):
  330. logger.info('Adding {0} + {1}'.format(x, y))
  331. return x + y
  332. Celery uses the standard Python logger library,
  333. and the documentation can be found :mod:`here <logging>`.
  334. You can also use :func:`print`, as anything written to standard
  335. out/-err will be redirected to the logging system (you can disable this,
  336. see :setting:`worker_redirect_stdouts`).
  337. .. note::
  338. The worker won't update the redirection if you create a logger instance
  339. somewhere in your task or task module.
  340. If you want to redirect ``sys.stdout`` and ``sys.stderr`` to a custom
  341. logger you have to enable this manually, for example:
  342. .. code-block:: python
  343. import sys
  344. logger = get_task_logger(__name__)
  345. @app.task(bind=True)
  346. def add(self, x, y):
  347. old_outs = sys.stdout, sys.stderr
  348. rlevel = self.app.conf.worker_redirect_stdouts_level
  349. try:
  350. self.app.log.redirect_stdouts_to_logger(logger, rlevel)
  351. print('Adding {0} + {1}'.format(x, y))
  352. return x + y
  353. finally:
  354. sys.stdout, sys.stderr = old_outs
  355. .. _task-argument-checking:
  356. Argument checking
  357. -----------------
  358. .. versionadded:: 4.0
  359. Celery will verify the arguments passed when you call the task, just
  360. like Python does when calling a normal function:
  361. .. code-block:: pycon
  362. >>> @app.task
  363. ... def add(x, y):
  364. ... return x + y
  365. # Calling the task with two arguments works:
  366. >>> add.delay(8, 8)
  367. <AsyncResult: f59d71ca-1549-43e0-be41-4e8821a83c0c>
  368. # Calling the task with only one argument fails:
  369. >>> add.delay(8)
  370. Traceback (most recent call last):
  371. File "<stdin>", line 1, in <module>
  372. File "celery/app/task.py", line 376, in delay
  373. return self.apply_async(args, kwargs)
  374. File "celery/app/task.py", line 485, in apply_async
  375. check_arguments(*(args or ()), **(kwargs or {}))
  376. TypeError: add() takes exactly 2 arguments (1 given)
  377. You can disable the argument checking for any task by setting its
  378. :attr:`~@Task.typing` attribute to :const:`False`:
  379. .. code-block:: pycon
  380. >>> @app.task(typing=False)
  381. ... def add(x, y):
  382. ... return x + y
  383. # Works locally, but the worker reciving the task will raise an error.
  384. >>> add.delay(8)
  385. <AsyncResult: f59d71ca-1549-43e0-be41-4e8821a83c0c>
  386. Hiding sensitive information in arguments
  387. -----------------------------------------
  388. .. versionadded:: 4.0
  389. When using :setting:`task_protocol` 2 or higher (default since 4.0), you can
  390. override how positional arguments and keyword arguments are represented in logs
  391. and monitoring events using the ``argsrepr`` and ``kwargsrepr`` calling
  392. arguments:
  393. .. code-block:: pycon
  394. >>> add.apply_async((2, 3), argsrepr='(<secret-x>, <secret-y>)')
  395. >>> charge.s(account, card='1234 5678 1234 5678').set(
  396. ... kwargsrepr=repr({'card': '**** **** **** 5678'})
  397. ... ).delay()
  398. .. warning::
  399. Sensitive information will still be accessible to anyone able
  400. to read your task message from the broker, or otherwise able intercept it.
  401. For this reason you should probably encrypt your message if it contains
  402. sensitive information, or in this example with a credit card number
  403. the actual number could be stored encrypted in a secure store that you retrieve
  404. and decrypt in the task itself.
  405. .. _task-retry:
  406. Retrying
  407. ========
  408. :meth:`Task.retry() <@Task.retry>` can be used to re-execute the task,
  409. for example in the event of recoverable errors.
  410. When you call ``retry`` it'll send a new message, using the same
  411. task-id, and it'll take care to make sure the message is delivered
  412. to the same queue as the originating task.
  413. When a task is retried this is also recorded as a task state,
  414. so that you can track the progress of the task using the result
  415. instance (see :ref:`task-states`).
  416. Here's an example using ``retry``:
  417. .. code-block:: python
  418. @app.task(bind=True)
  419. def send_twitter_status(self, oauth, tweet):
  420. try:
  421. twitter = Twitter(oauth)
  422. twitter.update_status(tweet)
  423. except (Twitter.FailWhaleError, Twitter.LoginError) as exc:
  424. raise self.retry(exc=exc)
  425. .. note::
  426. The :meth:`Task.retry() <@Task.retry>` call will raise an exception so any
  427. code after the retry won't be reached. This is the :exc:`~@Retry`
  428. exception, it isn't handled as an error but rather as a semi-predicate
  429. to signify to the worker that the task is to be retried,
  430. so that it can store the correct state when a result backend is enabled.
  431. This is normal operation and always happens unless the
  432. ``throw`` argument to retry is set to :const:`False`.
  433. The bind argument to the task decorator will give access to ``self`` (the
  434. task type instance).
  435. The ``exc`` method is used to pass exception information that's
  436. used in logs, and when storing task results.
  437. Both the exception and the traceback will
  438. be available in the task state (if a result backend is enabled).
  439. If the task has a ``max_retries`` value the current exception
  440. will be re-raised if the max number of retries has been exceeded,
  441. but this won't happen if:
  442. - An ``exc`` argument wasn't given.
  443. In this case the :exc:`~@MaxRetriesExceededError`
  444. exception will be raised.
  445. - There's no current exception
  446. If there's no original exception to re-raise the ``exc``
  447. argument will be used instead, so:
  448. .. code-block:: python
  449. self.retry(exc=Twitter.LoginError())
  450. will raise the ``exc`` argument given.
  451. .. _task-retry-custom-delay:
  452. Using a custom retry delay
  453. --------------------------
  454. When a task is to be retried, it can wait for a given amount of time
  455. before doing so, and the default delay is defined by the
  456. :attr:`~@Task.default_retry_delay`
  457. attribute. By default this is set to 3 minutes. Note that the
  458. unit for setting the delay is in seconds (int or float).
  459. You can also provide the `countdown` argument to :meth:`~@Task.retry` to
  460. override this default.
  461. .. code-block:: python
  462. @app.task(bind=True, default_retry_delay=30 * 60) # retry in 30 minutes.
  463. def add(self, x, y):
  464. try:
  465. something_raising()
  466. except Exception as exc:
  467. # overrides the default delay to retry after 1 minute
  468. raise self.retry(exc=exc, countdown=60)
  469. .. _task-autoretry:
  470. Automatic retry for known exceptions
  471. ------------------------------------
  472. .. versionadded:: 4.0
  473. Sometimes you just want to retry a task whenever a particular exception
  474. is raised.
  475. Fortunately, you can tell Celery to automatically retry a task using
  476. `autoretry_for` argument in `~@Celery.task` decorator:
  477. .. code-block:: python
  478. from twitter.exceptions import FailWhaleError
  479. @app.task(autoretry_for=(FailWhaleError,))
  480. def refresh_timeline(user):
  481. return twitter.refresh_timeline(user)
  482. If you want to specify custom arguments for internal `~@Task.retry`
  483. call, pass `retry_kwargs` argument to `~@Celery.task` decorator:
  484. .. code-block:: python
  485. @app.task(autoretry_for=(FailWhaleError,),
  486. retry_kwargs={'max_retries': 5})
  487. def refresh_timeline(user):
  488. return twitter.refresh_timeline(user)
  489. This is provided as an alternative to manually handling the exceptions,
  490. and the example above will do the same as wrapping the task body
  491. in a :keyword:`try` ... :keyword:`except` statement:
  492. .. code-block:: python
  493. @app.task
  494. def refresh_timeline(user):
  495. try:
  496. twitter.refresh_timeline(user)
  497. except FailWhaleError as exc:
  498. raise div.retry(exc=exc, max_retries=5)
  499. If you want to automatically retry on any error, simply use:
  500. .. code-block:: python
  501. @app.task(autoretry_for=(Exception,))
  502. def x():
  503. ...
  504. .. _task-options:
  505. List of Options
  506. ===============
  507. The task decorator can take a number of options that change the way
  508. the task behaves, for example you can set the rate limit for a task
  509. using the :attr:`rate_limit` option.
  510. Any keyword argument passed to the task decorator will actually be set
  511. as an attribute of the resulting task class, and this is a list
  512. of the built-in attributes.
  513. General
  514. -------
  515. .. _task-general-options:
  516. .. attribute:: Task.name
  517. The name the task is registered as.
  518. You can set this name manually, or a name will be
  519. automatically generated using the module and class name.
  520. See also :ref:`task-names`.
  521. .. attribute:: Task.request
  522. If the task is being executed this will contain information
  523. about the current request. Thread local storage is used.
  524. See :ref:`task-request-info`.
  525. .. attribute:: Task.max_retries
  526. Only applies if the task calls ``self.retry`` or if the task is decorated
  527. with the :ref:`autoretry_for <task-autoretry>` argument.
  528. The maximum number of attempted retries before giving up.
  529. If the number of retries exceeds this value a :exc:`~@MaxRetriesExceededError`
  530. exception will be raised.
  531. .. note::
  532. You have to call :meth:`~@Task.retry`
  533. manually, as it won't automatically retry on exception..
  534. The default is ``3``.
  535. A value of :const:`None` will disable the retry limit and the
  536. task will retry forever until it succeeds.
  537. .. attribute:: Task.throws
  538. Optional tuple of expected error classes that shouldn't be regarded
  539. as an actual error.
  540. Errors in this list will be reported as a failure to the result backend,
  541. but the worker won't log the event as an error, and no traceback will
  542. be included.
  543. Example:
  544. .. code-block:: python
  545. @task(throws=(KeyError, HttpNotFound)):
  546. def get_foo():
  547. something()
  548. Error types:
  549. - Expected errors (in ``Task.throws``)
  550. Logged with severity ``INFO``, traceback excluded.
  551. - Unexpected errors
  552. Logged with severity ``ERROR``, with traceback included.
  553. .. attribute:: Task.default_retry_delay
  554. Default time in seconds before a retry of the task
  555. should be executed. Can be either :class:`int` or :class:`float`.
  556. Default is a three minute delay.
  557. .. attribute:: Task.rate_limit
  558. Set the rate limit for this task type (limits the number of tasks
  559. that can be run in a given time frame). Tasks will still complete when
  560. a rate limit is in effect, but it may take some time before it's allowed to
  561. start.
  562. If this is :const:`None` no rate limit is in effect.
  563. If it is an integer or float, it is interpreted as "tasks per second".
  564. The rate limits can be specified in seconds, minutes or hours
  565. by appending `"/s"`, `"/m"` or `"/h"` to the value. Tasks will be evenly
  566. distributed over the specified time frame.
  567. Example: `"100/m"` (hundred tasks a minute). This will enforce a minimum
  568. delay of 600ms between starting two tasks on the same worker instance.
  569. Default is the :setting:`task_default_rate_limit` setting:
  570. if not specified means rate limiting for tasks is disabled by default.
  571. Note that this is a *per worker instance* rate limit, and not a global
  572. rate limit. To enforce a global rate limit (e.g., for an API with a
  573. maximum number of requests per second), you must restrict to a given
  574. queue.
  575. .. attribute:: Task.time_limit
  576. The hard time limit, in seconds, for this task.
  577. When not set the workers default is used.
  578. .. attribute:: Task.soft_time_limit
  579. The soft time limit for this task.
  580. When not set the workers default is used.
  581. .. attribute:: Task.ignore_result
  582. Don't store task state. Note that this means you can't use
  583. :class:`~celery.result.AsyncResult` to check if the task is ready,
  584. or get its return value.
  585. .. attribute:: Task.store_errors_even_if_ignored
  586. If :const:`True`, errors will be stored even if the task is configured
  587. to ignore results.
  588. .. attribute:: Task.serializer
  589. A string identifying the default serialization
  590. method to use. Defaults to the :setting:`task_serializer`
  591. setting. Can be `pickle`, `json`, `yaml`, or any custom
  592. serialization methods that have been registered with
  593. :mod:`kombu.serialization.registry`.
  594. Please see :ref:`calling-serializers` for more information.
  595. .. attribute:: Task.compression
  596. A string identifying the default compression scheme to use.
  597. Defaults to the :setting:`task_compression` setting.
  598. Can be `gzip`, or `bzip2`, or any custom compression schemes
  599. that have been registered with the :mod:`kombu.compression` registry.
  600. Please see :ref:`calling-compression` for more information.
  601. .. attribute:: Task.backend
  602. The result store backend to use for this task. An instance of one of the
  603. backend classes in `celery.backends`. Defaults to `app.backend`,
  604. defined by the :setting:`result_backend` setting.
  605. .. attribute:: Task.acks_late
  606. If set to :const:`True` messages for this task will be acknowledged
  607. **after** the task has been executed, not *just before* (the default
  608. behavior).
  609. Note: This means the task may be executed multiple times should the worker
  610. crash in the middle of execution. Make sure your tasks are
  611. :term:`idempotent`.
  612. The global default can be overridden by the :setting:`task_acks_late`
  613. setting.
  614. .. _task-track-started:
  615. .. attribute:: Task.track_started
  616. If :const:`True` the task will report its status as "started"
  617. when the task is executed by a worker.
  618. The default value is :const:`False` as the normal behavior is to not
  619. report that level of granularity. Tasks are either pending, finished,
  620. or waiting to be retried. Having a "started" status can be useful for
  621. when there are long running tasks and there's a need to report what
  622. task is currently running.
  623. The host name and process id of the worker executing the task
  624. will be available in the state meta-data (e.g., `result.info['pid']`)
  625. The global default can be overridden by the
  626. :setting:`task_track_started` setting.
  627. .. seealso::
  628. The API reference for :class:`~@Task`.
  629. .. _task-states:
  630. States
  631. ======
  632. Celery can keep track of the tasks current state. The state also contains the
  633. result of a successful task, or the exception and traceback information of a
  634. failed task.
  635. There are several *result backends* to choose from, and they all have
  636. different strengths and weaknesses (see :ref:`task-result-backends`).
  637. During its lifetime a task will transition through several possible states,
  638. and each state may have arbitrary meta-data attached to it. When a task
  639. moves into a new state the previous state is
  640. forgotten about, but some transitions can be deducted, (e.g., a task now
  641. in the :state:`FAILED` state, is implied to have been in the
  642. :state:`STARTED` state at some point).
  643. There are also sets of states, like the set of
  644. :state:`FAILURE_STATES`, and the set of :state:`READY_STATES`.
  645. The client uses the membership of these sets to decide whether
  646. the exception should be re-raised (:state:`PROPAGATE_STATES`), or whether
  647. the state can be cached (it can if the task is ready).
  648. You can also define :ref:`custom-states`.
  649. .. _task-result-backends:
  650. Result Backends
  651. ---------------
  652. If you want to keep track of tasks or need the return values, then Celery
  653. must store or send the states somewhere so that they can be retrieved later.
  654. There are several built-in result backends to choose from: SQLAlchemy/Django ORM,
  655. Memcached, RabbitMQ/QPid (``rpc``), and Redis -- or you can define your own.
  656. No backend works well for every use case.
  657. You should read about the strengths and weaknesses of each backend, and choose
  658. the most appropriate for your needs.
  659. .. seealso::
  660. :ref:`conf-result-backend`
  661. RPC Result Backend (RabbitMQ/QPid)
  662. ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  663. The RPC result backend (`rpc://`) is special as it doesn't actually *store*
  664. the states, but rather sends them as messages. This is an important difference as it
  665. means that a result *can only be retrieved once*, and *only by the client
  666. that initiated the task*. Two different processes can't wait for the same result.
  667. Even with that limitation, it is an excellent choice if you need to receive
  668. state changes in real-time. Using messaging means the client doesn't have to
  669. poll for new states.
  670. The messages are transient (non-persistent) by default, so the results will
  671. disappear if the broker restarts. You can configure the result backend to send
  672. persistent messages using the :setting:`result_persistent` setting.
  673. Database Result Backend
  674. ~~~~~~~~~~~~~~~~~~~~~~~
  675. Keeping state in the database can be convenient for many, especially for
  676. web applications with a database already in place, but it also comes with
  677. limitations.
  678. * Polling the database for new states is expensive, and so you should
  679. increase the polling intervals of operations, such as `result.get()`.
  680. * Some databases use a default transaction isolation level that
  681. isn't suitable for polling tables for changes.
  682. In MySQL the default transaction isolation level is `REPEATABLE-READ`:
  683. meaning the transaction won't see changes made by other transactions until
  684. the current transaction is committed.
  685. Changing that to the `READ-COMMITTED` isolation level is recommended.
  686. .. _task-builtin-states:
  687. Built-in States
  688. ---------------
  689. .. state:: PENDING
  690. PENDING
  691. ~~~~~~~
  692. Task is waiting for execution or unknown.
  693. Any task id that's not known is implied to be in the pending state.
  694. .. state:: STARTED
  695. STARTED
  696. ~~~~~~~
  697. Task has been started.
  698. Not reported by default, to enable please see :attr:`@Task.track_started`.
  699. :meta-data: `pid` and `hostname` of the worker process executing
  700. the task.
  701. .. state:: SUCCESS
  702. SUCCESS
  703. ~~~~~~~
  704. Task has been successfully executed.
  705. :meta-data: `result` contains the return value of the task.
  706. :propagates: Yes
  707. :ready: Yes
  708. .. state:: FAILURE
  709. FAILURE
  710. ~~~~~~~
  711. Task execution resulted in failure.
  712. :meta-data: `result` contains the exception occurred, and `traceback`
  713. contains the backtrace of the stack at the point when the
  714. exception was raised.
  715. :propagates: Yes
  716. .. state:: RETRY
  717. RETRY
  718. ~~~~~
  719. Task is being retried.
  720. :meta-data: `result` contains the exception that caused the retry,
  721. and `traceback` contains the backtrace of the stack at the point
  722. when the exceptions was raised.
  723. :propagates: No
  724. .. state:: REVOKED
  725. REVOKED
  726. ~~~~~~~
  727. Task has been revoked.
  728. :propagates: Yes
  729. .. _custom-states:
  730. Custom states
  731. -------------
  732. You can easily define your own states, all you need is a unique name.
  733. The name of the state is usually an uppercase string. As an example
  734. you could have a look at the :mod:`abortable tasks <~celery.contrib.abortable>`
  735. which defines a custom :state:`ABORTED` state.
  736. Use :meth:`~@Task.update_state` to update a task's state:.
  737. .. code-block:: python
  738. @app.task(bind=True)
  739. def upload_files(self, filenames):
  740. for i, file in enumerate(filenames):
  741. if not self.request.called_directly:
  742. self.update_state(state='PROGRESS',
  743. meta={'current': i, 'total': len(filenames)})
  744. Here I created the state `"PROGRESS"`, telling any application
  745. aware of this state that the task is currently in progress, and also where
  746. it is in the process by having `current` and `total` counts as part of the
  747. state meta-data. This can then be used to create progress bars for example.
  748. .. _pickling_exceptions:
  749. Creating pickleable exceptions
  750. ------------------------------
  751. A rarely known Python fact is that exceptions must conform to some
  752. simple rules to support being serialized by the pickle module.
  753. Tasks that raise exceptions that aren't pickleable won't work
  754. properly when Pickle is used as the serializer.
  755. To make sure that your exceptions are pickleable the exception
  756. *MUST* provide the original arguments it was instantiated
  757. with in its ``.args`` attribute. The simplest way
  758. to ensure this is to have the exception call ``Exception.__init__``.
  759. Let's look at some examples that work, and one that doesn't:
  760. .. code-block:: python
  761. # OK:
  762. class HttpError(Exception):
  763. pass
  764. # BAD:
  765. class HttpError(Exception):
  766. def __init__(self, status_code):
  767. self.status_code = status_code
  768. # OK:
  769. class HttpError(Exception):
  770. def __init__(self, status_code):
  771. self.status_code = status_code
  772. Exception.__init__(self, status_code) # <-- REQUIRED
  773. So the rule is:
  774. For any exception that supports custom arguments ``*args``,
  775. ``Exception.__init__(self, *args)`` must be used.
  776. There's no special support for *keyword arguments*, so if you
  777. want to preserve keyword arguments when the exception is unpickled
  778. you have to pass them as regular args:
  779. .. code-block:: python
  780. class HttpError(Exception):
  781. def __init__(self, status_code, headers=None, body=None):
  782. self.status_code = status_code
  783. self.headers = headers
  784. self.body = body
  785. super().__init__(status_code, headers, body)
  786. .. _task-semipredicates:
  787. Semipredicates
  788. ==============
  789. The worker wraps the task in a tracing function that records the final
  790. state of the task. There are a number of exceptions that can be used to
  791. signal this function to change how it treats the return of the task.
  792. .. _task-semipred-ignore:
  793. Ignore
  794. ------
  795. The task may raise :exc:`~@Ignore` to force the worker to ignore the
  796. task. This means that no state will be recorded for the task, but the
  797. message is still acknowledged (removed from queue).
  798. This can be used if you want to implement custom revoke-like
  799. functionality, or manually store the result of a task.
  800. Example keeping revoked tasks in a Redis set:
  801. .. code-block:: python
  802. from celery.exceptions import Ignore
  803. @app.task(bind=True)
  804. def some_task(self):
  805. if redis.ismember('tasks.revoked', self.request.id):
  806. raise Ignore()
  807. Example that stores results manually:
  808. .. code-block:: python
  809. from celery import states
  810. from celery.exceptions import Ignore
  811. @app.task(bind=True)
  812. def get_tweets(self, user):
  813. timeline = twitter.get_timeline(user)
  814. if not self.request.called_directly:
  815. self.update_state(state=states.SUCCESS, meta=timeline)
  816. raise Ignore()
  817. .. _task-semipred-reject:
  818. Reject
  819. ------
  820. The task may raise :exc:`~@Reject` to reject the task message using
  821. AMQPs ``basic_reject`` method. This won't have any effect unless
  822. :attr:`Task.acks_late` is enabled.
  823. Rejecting a message has the same effect as acking it, but some
  824. brokers may implement additional functionality that can be used.
  825. For example RabbitMQ supports the concept of `Dead Letter Exchanges`_
  826. where a queue can be configured to use a dead letter exchange that rejected
  827. messages are redelivered to.
  828. .. _`Dead Letter Exchanges`: http://www.rabbitmq.com/dlx.html
  829. Reject can also be used to re-queue messages, but please be very careful
  830. when using this as it can easily result in an infinite message loop.
  831. Example using reject when a task causes an out of memory condition:
  832. .. code-block:: python
  833. import errno
  834. from celery.exceptions import Reject
  835. @app.task(bind=True, acks_late=True)
  836. def render_scene(self, path):
  837. file = get_file(path)
  838. try:
  839. renderer.render_scene(file)
  840. # if the file is too big to fit in memory
  841. # we reject it so that it's redelivered to the dead letter exchange
  842. # and we can manually inspect the situation.
  843. except MemoryError as exc:
  844. raise Reject(exc, requeue=False)
  845. except OSError as exc:
  846. if exc.errno == errno.ENOMEM:
  847. raise Reject(exc, requeue=False)
  848. raise self.retry(exc, countdown=10)
  849. # For any other error we retry after 10 seconds.
  850. except Exception as exc:
  851. raise self.retry(exc, countdown=10)
  852. Example re-queuing the message:
  853. .. code-block:: python
  854. from celery.exceptions import Reject
  855. @app.task(bind=True, acks_late=True)
  856. def requeues(self):
  857. if not self.request.delivery_info['redelivered']:
  858. raise Reject('no reason', requeue=True)
  859. print('received two times')
  860. Consult your broker documentation for more details about the ``basic_reject``
  861. method.
  862. .. _task-semipred-retry:
  863. Retry
  864. -----
  865. The :exc:`~@Retry` exception is raised by the ``Task.retry`` method
  866. to tell the worker that the task is being retried.
  867. .. _task-custom-classes:
  868. Custom task classes
  869. ===================
  870. All tasks inherit from the :class:`@Task` class.
  871. The :meth:`~@Task.run` method becomes the task body.
  872. As an example, the following code,
  873. .. code-block:: python
  874. @app.task
  875. def add(x, y):
  876. return x + y
  877. will do roughly this behind the scenes:
  878. .. code-block:: python
  879. class _AddTask(app.Task):
  880. def run(self, x, y):
  881. return x + y
  882. add = app.tasks[_AddTask.name]
  883. Instantiation
  884. -------------
  885. A task is **not** instantiated for every request, but is registered
  886. in the task registry as a global instance.
  887. This means that the ``__init__`` constructor will only be called
  888. once per process, and that the task class is semantically closer to an
  889. Actor.
  890. If you have a task,
  891. .. code-block:: python
  892. from celery import Task
  893. class NaiveAuthenticateServer(Task):
  894. def __init__(self):
  895. self.users = {'george': 'password'}
  896. def run(self, username, password):
  897. try:
  898. return self.users[username] == password
  899. except KeyError:
  900. return False
  901. And you route every request to the same process, then it
  902. will keep state between requests.
  903. This can also be useful to cache resources,
  904. For example, a base Task class that caches a database connection:
  905. .. code-block:: python
  906. from celery import Task
  907. class DatabaseTask(Task):
  908. _db = None
  909. @property
  910. def db(self):
  911. if self._db is None:
  912. self._db = Database.connect()
  913. return self._db
  914. that can be added to tasks like this:
  915. .. code-block:: python
  916. @app.task(base=DatabaseTask)
  917. def process_rows():
  918. for row in process_rows.db.table.all():
  919. process_row(row)
  920. The ``db`` attribute of the ``process_rows`` task will then
  921. always stay the same in each process.
  922. Handlers
  923. --------
  924. .. method:: after_return(self, status, retval, task_id, args, kwargs, einfo)
  925. Handler called after the task returns.
  926. :param status: Current task state.
  927. :param retval: Task return value/exception.
  928. :param task_id: Unique id of the task.
  929. :param args: Original arguments for the task that returned.
  930. :param kwargs: Original keyword arguments for the task
  931. that returned.
  932. :keyword einfo: :class:`~billiard.einfo.ExceptionInfo`
  933. instance, containing the traceback (if any).
  934. The return value of this handler is ignored.
  935. .. method:: on_failure(self, exc, task_id, args, kwargs, einfo)
  936. This is run by the worker when the task fails.
  937. :param exc: The exception raised by the task.
  938. :param task_id: Unique id of the failed task.
  939. :param args: Original arguments for the task that failed.
  940. :param kwargs: Original keyword arguments for the task
  941. that failed.
  942. :keyword einfo: :class:`~billiard.einfo.ExceptionInfo`
  943. instance, containing the traceback.
  944. The return value of this handler is ignored.
  945. .. method:: on_retry(self, exc, task_id, args, kwargs, einfo)
  946. This is run by the worker when the task is to be retried.
  947. :param exc: The exception sent to :meth:`~@Task.retry`.
  948. :param task_id: Unique id of the retried task.
  949. :param args: Original arguments for the retried task.
  950. :param kwargs: Original keyword arguments for the retried task.
  951. :keyword einfo: :class:`~billiard.einfo.ExceptionInfo`
  952. instance, containing the traceback.
  953. The return value of this handler is ignored.
  954. .. method:: on_success(self, retval, task_id, args, kwargs)
  955. Run by the worker if the task executes successfully.
  956. :param retval: The return value of the task.
  957. :param task_id: Unique id of the executed task.
  958. :param args: Original arguments for the executed task.
  959. :param kwargs: Original keyword arguments for the executed task.
  960. The return value of this handler is ignored.
  961. .. _task-how-they-work:
  962. How it works
  963. ============
  964. Here come the technical details. This part isn't something you need to know,
  965. but you may be interested.
  966. All defined tasks are listed in a registry. The registry contains
  967. a list of task names and their task classes. You can investigate this registry
  968. yourself:
  969. .. code-block:: pycon
  970. >>> from proj.celery import app
  971. >>> app.tasks
  972. {'celery.chord_unlock':
  973. <@task: celery.chord_unlock>,
  974. 'celery.backend_cleanup':
  975. <@task: celery.backend_cleanup>,
  976. 'celery.chord':
  977. <@task: celery.chord>}
  978. This is the list of tasks built-in to Celery. Note that tasks
  979. will only be registered when the module they're defined in is imported.
  980. The default loader imports any modules listed in the
  981. :setting:`imports` setting.
  982. The :meth:`@task` decorator is responsible for registering your task
  983. in the applications task registry.
  984. When tasks are sent, no actual function code is sent with it, just the name
  985. of the task to execute. When the worker then receives the message it can look
  986. up the name in its task registry to find the execution code.
  987. This means that your workers should always be updated with the same software
  988. as the client. This is a drawback, but the alternative is a technical
  989. challenge that's yet to be solved.
  990. .. _task-best-practices:
  991. Tips and Best Practices
  992. =======================
  993. .. _task-ignore_results:
  994. Ignore results you don't want
  995. -----------------------------
  996. If you don't care about the results of a task, be sure to set the
  997. :attr:`~@Task.ignore_result` option, as storing results
  998. wastes time and resources.
  999. .. code-block:: python
  1000. @app.task(ignore_result=True)
  1001. def mytask():
  1002. something()
  1003. Results can even be disabled globally using the :setting:`task_ignore_result`
  1004. setting.
  1005. More optimization tips
  1006. ----------------------
  1007. You find additional optimization tips in the
  1008. :ref:`Optimizing Guide <guide-optimizing>`.
  1009. .. _task-synchronous-subtasks:
  1010. Avoid launching synchronous subtasks
  1011. ------------------------------------
  1012. Having a task wait for the result of another task is really inefficient,
  1013. and may even cause a deadlock if the worker pool is exhausted.
  1014. Make your design asynchronous instead, for example by using *callbacks*.
  1015. **Bad**:
  1016. .. code-block:: python
  1017. @app.task
  1018. def update_page_info(url):
  1019. page = fetch_page.delay(url).get()
  1020. info = parse_page.delay(url, page).get()
  1021. store_page_info.delay(url, info)
  1022. @app.task
  1023. def fetch_page(url):
  1024. return myhttplib.get(url)
  1025. @app.task
  1026. def parse_page(url, page):
  1027. return myparser.parse_document(page)
  1028. @app.task
  1029. def store_page_info(url, info):
  1030. return PageInfo.objects.create(url, info)
  1031. **Good**:
  1032. .. code-block:: python
  1033. def update_page_info(url):
  1034. # fetch_page -> parse_page -> store_page
  1035. chain = fetch_page.s(url) | parse_page.s() | store_page_info.s(url)
  1036. chain()
  1037. @app.task()
  1038. def fetch_page(url):
  1039. return myhttplib.get(url)
  1040. @app.task()
  1041. def parse_page(page):
  1042. return myparser.parse_document(page)
  1043. @app.task(ignore_result=True)
  1044. def store_page_info(info, url):
  1045. PageInfo.objects.create(url=url, info=info)
  1046. Here I instead created a chain of tasks by linking together
  1047. different :func:`~celery.signature`'s.
  1048. You can read about chains and other powerful constructs
  1049. at :ref:`designing-workflows`.
  1050. .. _task-performance-and-strategies:
  1051. Performance and Strategies
  1052. ==========================
  1053. .. _task-granularity:
  1054. Granularity
  1055. -----------
  1056. The task granularity is the amount of computation needed by each subtask.
  1057. In general it is better to split the problem up into many small tasks rather
  1058. than have a few long running tasks.
  1059. With smaller tasks you can process more tasks in parallel and the tasks
  1060. won't run long enough to block the worker from processing other waiting tasks.
  1061. However, executing a task does have overhead. A message needs to be sent, data
  1062. may not be local, etc. So if the tasks are too fine-grained the
  1063. overhead added probably removes any benefit.
  1064. .. seealso::
  1065. The book `Art of Concurrency`_ has a section dedicated to the topic
  1066. of task granularity [AOC1]_.
  1067. .. _`Art of Concurrency`: http://oreilly.com/catalog/9780596521547
  1068. .. [AOC1] Breshears, Clay. Section 2.2.1, "The Art of Concurrency".
  1069. O'Reilly Media, Inc. May 15, 2009. ISBN-13 978-0-596-52153-0.
  1070. .. _task-data-locality:
  1071. Data locality
  1072. -------------
  1073. The worker processing the task should be as close to the data as
  1074. possible. The best would be to have a copy in memory, the worst would be a
  1075. full transfer from another continent.
  1076. If the data is far away, you could try to run another worker at location, or
  1077. if that's not possible - cache often used data, or preload data you know
  1078. is going to be used.
  1079. The easiest way to share data between workers is to use a distributed cache
  1080. system, like `memcached`_.
  1081. .. seealso::
  1082. The paper `Distributed Computing Economics`_ by Jim Gray is an excellent
  1083. introduction to the topic of data locality.
  1084. .. _`Distributed Computing Economics`:
  1085. http://research.microsoft.com/pubs/70001/tr-2003-24.pdf
  1086. .. _`memcached`: http://memcached.org/
  1087. .. _task-state:
  1088. State
  1089. -----
  1090. Since celery is a distributed system, you can't know which process, or
  1091. on what machine the task will be executed. You can't even know if the task will
  1092. run in a timely manner.
  1093. The ancient async sayings tells us that “asserting the world is the
  1094. responsibility of the task”. What this means is that the world view may
  1095. have changed since the task was requested, so the task is responsible for
  1096. making sure the world is how it should be; If you have a task
  1097. that re-indexes a search engine, and the search engine should only be
  1098. re-indexed at maximum every 5 minutes, then it must be the tasks
  1099. responsibility to assert that, not the callers.
  1100. Another gotcha is Django model objects. They shouldn't be passed on as
  1101. arguments to tasks. It's almost always better to re-fetch the object from
  1102. the database when the task is running instead, as using old data may lead
  1103. to race conditions.
  1104. Imagine the following scenario where you have an article and a task
  1105. that automatically expands some abbreviations in it:
  1106. .. code-block:: python
  1107. class Article(models.Model):
  1108. title = models.CharField()
  1109. body = models.TextField()
  1110. @app.task
  1111. def expand_abbreviations(article):
  1112. article.body.replace('MyCorp', 'My Corporation')
  1113. article.save()
  1114. First, an author creates an article and saves it, then the author
  1115. clicks on a button that initiates the abbreviation task:
  1116. .. code-block:: pycon
  1117. >>> article = Article.objects.get(id=102)
  1118. >>> expand_abbreviations.delay(article)
  1119. Now, the queue is very busy, so the task won't be run for another 2 minutes.
  1120. In the meantime another author makes changes to the article, so
  1121. when the task is finally run, the body of the article is reverted to the old
  1122. version because the task had the old body in its argument.
  1123. Fixing the race condition is easy, just use the article id instead, and
  1124. re-fetch the article in the task body:
  1125. .. code-block:: python
  1126. @app.task
  1127. def expand_abbreviations(article_id):
  1128. article = Article.objects.get(id=article_id)
  1129. article.body.replace('MyCorp', 'My Corporation')
  1130. article.save()
  1131. .. code-block:: pycon
  1132. >>> expand_abbreviations.delay(article_id)
  1133. There might even be performance benefits to this approach, as sending large
  1134. messages may be expensive.
  1135. .. _task-database-transactions:
  1136. Database transactions
  1137. ---------------------
  1138. Let's have a look at another example:
  1139. .. code-block:: python
  1140. from django.db import transaction
  1141. @transaction.commit_on_success
  1142. def create_article(request):
  1143. article = Article.objects.create()
  1144. expand_abbreviations.delay(article.pk)
  1145. This is a Django view creating an article object in the database,
  1146. then passing the primary key to a task. It uses the `commit_on_success`
  1147. decorator, that will commit the transaction when the view returns, or
  1148. roll back if the view raises an exception.
  1149. There's a race condition if the task starts executing
  1150. before the transaction has been committed; The database object doesn't exist
  1151. yet!
  1152. The solution is to *always commit transactions before sending tasks
  1153. depending on state from the current transaction*:
  1154. .. code-block:: python
  1155. @transaction.commit_manually
  1156. def create_article(request):
  1157. try:
  1158. article = Article.objects.create()
  1159. except:
  1160. transaction.rollback()
  1161. raise
  1162. else:
  1163. transaction.commit()
  1164. expand_abbreviations.delay(article.pk)
  1165. .. note::
  1166. Django 1.6 (and later) now enables autocommit mode by default,
  1167. and ``commit_on_success``/``commit_manually`` are deprecated.
  1168. This means each SQL query is wrapped and executed in individual
  1169. transactions, making it less likely to experience the
  1170. problem described above.
  1171. However, enabling ``ATOMIC_REQUESTS`` on the database
  1172. connection will bring back the transaction-per-request model and the
  1173. race condition along with it. In this case, the simple solution is
  1174. using the ``@transaction.non_atomic_requests`` decorator to go back
  1175. to autocommit for that view only.
  1176. .. _task-example:
  1177. Example
  1178. =======
  1179. Let's take a real world example: a blog where comments posted need to be
  1180. filtered for spam. When the comment is created, the spam filter runs in the
  1181. background, so the user doesn't have to wait for it to finish.
  1182. I have a Django blog application allowing comments
  1183. on blog posts. I'll describe parts of the models/views and tasks for this
  1184. application.
  1185. ``blog/models.py``
  1186. ------------------
  1187. The comment model looks like this:
  1188. .. code-block:: python
  1189. from django.db import models
  1190. from django.utils.translation import ugettext_lazy as _
  1191. class Comment(models.Model):
  1192. name = models.CharField(_('name'), max_length=64)
  1193. email_address = models.EmailField(_('email address'))
  1194. homepage = models.URLField(_('home page'),
  1195. blank=True, verify_exists=False)
  1196. comment = models.TextField(_('comment'))
  1197. pub_date = models.DateTimeField(_('Published date'),
  1198. editable=False, auto_add_now=True)
  1199. is_spam = models.BooleanField(_('spam?'),
  1200. default=False, editable=False)
  1201. class Meta:
  1202. verbose_name = _('comment')
  1203. verbose_name_plural = _('comments')
  1204. In the view where the comment is posted, I first write the comment
  1205. to the database, then I launch the spam filter task in the background.
  1206. .. _task-example-blog-views:
  1207. ``blog/views.py``
  1208. -----------------
  1209. .. code-block:: python
  1210. from django import forms
  1211. from django.http import HttpResponseRedirect
  1212. from django.template.context import RequestContext
  1213. from django.shortcuts import get_object_or_404, render_to_response
  1214. from blog import tasks
  1215. from blog.models import Comment
  1216. class CommentForm(forms.ModelForm):
  1217. class Meta:
  1218. model = Comment
  1219. def add_comment(request, slug, template_name='comments/create.html'):
  1220. post = get_object_or_404(Entry, slug=slug)
  1221. remote_addr = request.META.get('REMOTE_ADDR')
  1222. if request.method == 'post':
  1223. form = CommentForm(request.POST, request.FILES)
  1224. if form.is_valid():
  1225. comment = form.save()
  1226. # Check spam asynchronously.
  1227. tasks.spam_filter.delay(comment_id=comment.id,
  1228. remote_addr=remote_addr)
  1229. return HttpResponseRedirect(post.get_absolute_url())
  1230. else:
  1231. form = CommentForm()
  1232. context = RequestContext(request, {'form': form})
  1233. return render_to_response(template_name, context_instance=context)
  1234. To filter spam in comments I use `Akismet`_, the service
  1235. used to filter spam in comments posted to the free blog platform
  1236. `Wordpress`. `Akismet`_ is free for personal use, but for commercial use you
  1237. need to pay. You have to sign up to their service to get an API key.
  1238. To make API calls to `Akismet`_ I use the `akismet.py`_ library written by
  1239. `Michael Foord`_.
  1240. .. _task-example-blog-tasks:
  1241. ``blog/tasks.py``
  1242. -----------------
  1243. .. code-block:: python
  1244. from celery import Celery
  1245. from akismet import Akismet
  1246. from django.core.exceptions import ImproperlyConfigured
  1247. from django.contrib.sites.models import Site
  1248. from blog.models import Comment
  1249. app = Celery(broker='amqp://')
  1250. @app.task
  1251. def spam_filter(comment_id, remote_addr=None):
  1252. logger = spam_filter.get_logger()
  1253. logger.info('Running spam filter for comment %s', comment_id)
  1254. comment = Comment.objects.get(pk=comment_id)
  1255. current_domain = Site.objects.get_current().domain
  1256. akismet = Akismet(settings.AKISMET_KEY, 'http://{0}'.format(domain))
  1257. if not akismet.verify_key():
  1258. raise ImproperlyConfigured('Invalid AKISMET_KEY')
  1259. is_spam = akismet.comment_check(user_ip=remote_addr,
  1260. comment_content=comment.comment,
  1261. comment_author=comment.name,
  1262. comment_author_email=comment.email_address)
  1263. if is_spam:
  1264. comment.is_spam = True
  1265. comment.save()
  1266. return is_spam
  1267. .. _`Akismet`: http://akismet.com/faq/
  1268. .. _`akismet.py`: http://www.voidspace.org.uk/downloads/akismet.py
  1269. .. _`Michael Foord`: http://www.voidspace.org.uk/