_state.py 2.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110
  1. # -*- coding: utf-8 -*-
  2. """
  3. celery._state
  4. ~~~~~~~~~~~~~~~
  5. This is an internal module containing thread state
  6. like the ``current_app``, and ``current_task``.
  7. This module shouldn't be used directly.
  8. """
  9. from __future__ import absolute_import, print_function
  10. import os
  11. import sys
  12. import threading
  13. import weakref
  14. from celery.local import Proxy
  15. from celery.utils.threads import LocalStack
  16. __all__ = ['set_default_app', 'get_current_app', 'get_current_task',
  17. 'get_current_worker_task', 'current_app', 'current_task']
  18. #: Global default app used when no current app.
  19. default_app = None
  20. #: List of all app instances (weakrefs), must not be used directly.
  21. _apps = set()
  22. class _TLS(threading.local):
  23. #: Apps with the :attr:`~celery.app.base.BaseApp.set_as_current` attribute
  24. #: sets this, so it will always contain the last instantiated app,
  25. #: and is the default app returned by :func:`app_or_default`.
  26. current_app = None
  27. _tls = _TLS()
  28. _task_stack = LocalStack()
  29. def set_default_app(app):
  30. global default_app
  31. default_app = app
  32. def _get_current_app():
  33. if default_app is None:
  34. #: creates the global fallback app instance.
  35. from celery.app import Celery
  36. set_default_app(Celery(
  37. 'default',
  38. loader=os.environ.get('CELERY_LOADER') or 'default',
  39. set_as_current=False, accept_magic_kwargs=True,
  40. ))
  41. return _tls.current_app or default_app
  42. C_STRICT_APP = os.environ.get('C_STRICT_APP')
  43. if os.environ.get('C_STRICT_APP'): # pragma: no cover
  44. def get_current_app():
  45. raise Exception('USES CURRENT APP')
  46. import traceback
  47. print('-- USES CURRENT_APP', file=sys.stderr) # noqa+
  48. traceback.print_stack(file=sys.stderr)
  49. return _get_current_app()
  50. else:
  51. get_current_app = _get_current_app
  52. def get_current_task():
  53. """Currently executing task."""
  54. return _task_stack.top
  55. def get_current_worker_task():
  56. """Currently executing task, that was applied by the worker.
  57. This is used to differentiate between the actual task
  58. executed by the worker and any task that was called within
  59. a task (using ``task.__call__`` or ``task.apply``)
  60. """
  61. for task in reversed(_task_stack.stack):
  62. if not task.request.called_directly:
  63. return task
  64. #: Proxy to current app.
  65. current_app = Proxy(get_current_app)
  66. #: Proxy to current task.
  67. current_task = Proxy(get_current_task)
  68. def _register_app(app):
  69. _apps.add(weakref.ref(app))
  70. def _get_active_apps():
  71. dirty = []
  72. try:
  73. for appref in _apps:
  74. app = appref()
  75. if app is None:
  76. dirty.append(appref)
  77. else:
  78. yield app
  79. finally:
  80. while dirty:
  81. _apps.discard(dirty.pop())