utils.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362
  1. # -*- coding: utf-8 -*-
  2. """
  3. celery.app.utils
  4. ~~~~~~~~~~~~~~~~
  5. App utilities: Compat settings, bug-report tool, pickling apps.
  6. """
  7. from __future__ import absolute_import, unicode_literals
  8. import os
  9. import platform as _platform
  10. import re
  11. from collections import Mapping, namedtuple
  12. from copy import deepcopy
  13. from types import ModuleType
  14. from kombu.utils.url import maybe_sanitize_url
  15. from celery.datastructures import ConfigurationView
  16. from celery.exceptions import ImproperlyConfigured
  17. from celery.five import items, keys, string_t, values
  18. from celery.platforms import pyimplementation
  19. from celery.utils.text import pretty
  20. from celery.utils.imports import import_from_cwd, symbol_by_name, qualname
  21. from .defaults import (
  22. _TO_NEW_KEY, _TO_OLD_KEY, _OLD_DEFAULTS, _OLD_SETTING_KEYS,
  23. DEFAULTS, SETTING_KEYS, find,
  24. )
  25. __all__ = ['Settings', 'appstr', 'bugreport',
  26. 'filter_hidden_settings', 'find_app']
  27. #: Format used to generate bug-report information.
  28. BUGREPORT_INFO = """
  29. software -> celery:{celery_v} kombu:{kombu_v} py:{py_v}
  30. billiard:{billiard_v} {driver_v}
  31. platform -> system:{system} arch:{arch} imp:{py_i}
  32. loader -> {loader}
  33. settings -> transport:{transport} results:{results}
  34. {human_settings}
  35. """
  36. HIDDEN_SETTINGS = re.compile(
  37. 'API|TOKEN|KEY|SECRET|PASS|PROFANITIES_LIST|SIGNATURE|DATABASE',
  38. re.IGNORECASE,
  39. )
  40. E_MIX_OLD_INTO_NEW = """
  41. Cannot mix new and old setting keys, please rename the
  42. following settings to the new format:
  43. {renames}
  44. """
  45. E_MIX_NEW_INTO_OLD = """
  46. Cannot mix new setting names with old setting names, please
  47. rename the following settings to use the old format:
  48. {renames}
  49. Or change all of the settings to use the new format :)
  50. """
  51. FMT_REPLACE_SETTING = '{replace:<36} -> {with_}'
  52. def appstr(app):
  53. """String used in __repr__ etc, to id app instances."""
  54. return '{0}:{1:#x}'.format(app.main or '__main__', id(app))
  55. class Settings(ConfigurationView):
  56. """Celery settings object.
  57. .. seealso:
  58. :ref:`configuration` for a full list of configuration keys.
  59. """
  60. @property
  61. def broker_read_url(self):
  62. return (
  63. os.environ.get('CELERY_BROKER_READ_URL') or
  64. self.get('broker_read_url') or
  65. self.broker_url
  66. )
  67. @property
  68. def broker_write_url(self):
  69. return (
  70. os.environ.get('CELERY_BROKER_WRITE_URL') or
  71. self.get('broker_write_url') or
  72. self.broker_url
  73. )
  74. @property
  75. def broker_url(self):
  76. return (
  77. os.environ.get('CELERY_BROKER_URL') or
  78. self.first('broker_url', 'broker_host')
  79. )
  80. @property
  81. def timezone(self):
  82. # this way we also support django's time zone.
  83. return self.first('timezone', 'time_zone')
  84. def without_defaults(self):
  85. """Return the current configuration, but without defaults."""
  86. # the last stash is the default settings, so just skip that
  87. return Settings({}, self._order[:-1])
  88. def value_set_for(self, key):
  89. return key in self.without_defaults()
  90. def find_option(self, name, namespace=''):
  91. """Search for option by name.
  92. Will return ``(namespace, key, type)`` tuple, e.g.::
  93. >>> from proj.celery import app
  94. >>> app.conf.find_option('disable_rate_limits')
  95. ('worker', 'prefetch_multiplier',
  96. <Option: type->bool default->False>))
  97. :param name: Name of option, cannot be partial.
  98. :keyword namespace: Preferred name-space (``None`` by default).
  99. """
  100. return find(name, namespace)
  101. def find_value_for_key(self, name, namespace='celery'):
  102. """Shortcut to ``get_by_parts(*find_option(name)[:-1])``"""
  103. return self.get_by_parts(*self.find_option(name, namespace)[:-1])
  104. def get_by_parts(self, *parts):
  105. """Return the current value for setting specified as a path.
  106. Example::
  107. >>> from proj.celery import app
  108. >>> app.conf.get_by_parts('worker', 'disable_rate_limits')
  109. False
  110. """
  111. return self['_'.join(part for part in parts if part)]
  112. def table(self, with_defaults=False, censored=True):
  113. filt = filter_hidden_settings if censored else lambda v: v
  114. return filt({
  115. k: v for k, v in items(
  116. self if with_defaults else self.without_defaults())
  117. if not k.startswith('_')
  118. })
  119. def humanize(self, with_defaults=False, censored=True):
  120. """Return a human readable string showing changes to the
  121. configuration."""
  122. return '\n'.join(
  123. '{0}: {1}'.format(key, pretty(value, width=50))
  124. for key, value in items(self.table(with_defaults, censored)))
  125. def _new_key_to_old(key, convert=_TO_OLD_KEY.get):
  126. return convert(key, key)
  127. def _old_key_to_new(key, convert=_TO_NEW_KEY.get):
  128. return convert(key, key)
  129. _settings_info_t = namedtuple('settings_info_t', (
  130. 'defaults', 'convert', 'key_t', 'mix_error',
  131. ))
  132. _settings_info = _settings_info_t(
  133. DEFAULTS, _TO_NEW_KEY, _old_key_to_new, E_MIX_OLD_INTO_NEW,
  134. )
  135. _old_settings_info = _settings_info_t(
  136. _OLD_DEFAULTS, _TO_OLD_KEY, _new_key_to_old, E_MIX_NEW_INTO_OLD,
  137. )
  138. def detect_settings(conf, preconf={}, ignore_keys=set(), prefix=None,
  139. all_keys=SETTING_KEYS, old_keys=_OLD_SETTING_KEYS):
  140. source = conf
  141. if conf is None:
  142. source, conf = preconf, {}
  143. have = set(keys(source)) - ignore_keys
  144. is_in_new = have.intersection(all_keys)
  145. is_in_old = have.intersection(old_keys)
  146. info = None
  147. if is_in_new:
  148. # have new setting names
  149. info, left = _settings_info, is_in_old
  150. if is_in_old and len(is_in_old) > len(is_in_new):
  151. # Majority of the settings are old.
  152. info, left = _old_settings_info, is_in_new
  153. if is_in_old:
  154. # have old setting names, or a majority of the names are old.
  155. if not info:
  156. info, left = _old_settings_info, is_in_new
  157. if is_in_new and len(is_in_new) > len(is_in_old):
  158. # Majority of the settings are new
  159. info, left = _settings_info, is_in_old
  160. else:
  161. # no settings, just use new format.
  162. info, left = _settings_info, is_in_old
  163. if prefix:
  164. # always use new format if prefix is used.
  165. info, left = _settings_info, set()
  166. # only raise error for keys that the user did not provide two keys
  167. # for (e.g. both ``result_expires`` and ``CELERY_TASK_RESULT_EXPIRES``).
  168. really_left = {key for key in left if info.convert[key] not in have}
  169. if really_left:
  170. # user is mixing old/new, or new/old settings, give renaming
  171. # suggestions.
  172. raise ImproperlyConfigured(info.mix_error.format(renames='\n'.join(
  173. FMT_REPLACE_SETTING.format(replace=key, with_=info.convert[key])
  174. for key in sorted(really_left)
  175. )))
  176. preconf = {info.convert.get(k, k): v for k, v in items(preconf)}
  177. defaults = dict(deepcopy(info.defaults), **preconf)
  178. return Settings(preconf, [conf, defaults], info.key_t, prefix=prefix)
  179. class AppPickler(object):
  180. """Old application pickler/unpickler (< 3.1)."""
  181. def __call__(self, cls, *args):
  182. kwargs = self.build_kwargs(*args)
  183. app = self.construct(cls, **kwargs)
  184. self.prepare(app, **kwargs)
  185. return app
  186. def prepare(self, app, **kwargs):
  187. app.conf.update(kwargs['changes'])
  188. def build_kwargs(self, *args):
  189. return self.build_standard_kwargs(*args)
  190. def build_standard_kwargs(self, main, changes, loader, backend, amqp,
  191. events, log, control, accept_magic_kwargs,
  192. config_source=None):
  193. return dict(main=main, loader=loader, backend=backend, amqp=amqp,
  194. changes=changes, events=events, log=log, control=control,
  195. set_as_current=False,
  196. config_source=config_source)
  197. def construct(self, cls, **kwargs):
  198. return cls(**kwargs)
  199. def _unpickle_app(cls, pickler, *args):
  200. """Rebuild app for versions 2.5+"""
  201. return pickler()(cls, *args)
  202. def _unpickle_app_v2(cls, kwargs):
  203. """Rebuild app for versions 3.1+"""
  204. kwargs['set_as_current'] = False
  205. return cls(**kwargs)
  206. def filter_hidden_settings(conf):
  207. def maybe_censor(key, value, mask='*' * 8):
  208. if isinstance(value, Mapping):
  209. return filter_hidden_settings(value)
  210. if isinstance(key, string_t):
  211. if HIDDEN_SETTINGS.search(key):
  212. return mask
  213. elif 'broker_url' in key.lower():
  214. from kombu import Connection
  215. return Connection(value).as_uri(mask=mask)
  216. elif 'backend' in key.lower():
  217. return maybe_sanitize_url(value, mask=mask)
  218. return value
  219. return {k: maybe_censor(k, v) for k, v in items(conf)}
  220. def bugreport(app):
  221. """Return a string containing information useful in bug-reports."""
  222. import billiard
  223. import celery
  224. import kombu
  225. try:
  226. conn = app.connection()
  227. driver_v = '{0}:{1}'.format(conn.transport.driver_name,
  228. conn.transport.driver_version())
  229. transport = conn.transport_cls
  230. except Exception:
  231. transport = driver_v = ''
  232. return BUGREPORT_INFO.format(
  233. system=_platform.system(),
  234. arch=', '.join(x for x in _platform.architecture() if x),
  235. py_i=pyimplementation(),
  236. celery_v=celery.VERSION_BANNER,
  237. kombu_v=kombu.__version__,
  238. billiard_v=billiard.__version__,
  239. py_v=_platform.python_version(),
  240. driver_v=driver_v,
  241. transport=transport,
  242. results=maybe_sanitize_url(app.conf.result_backend or 'disabled'),
  243. human_settings=app.conf.humanize(),
  244. loader=qualname(app.loader.__class__),
  245. )
  246. def find_app(app, symbol_by_name=symbol_by_name, imp=import_from_cwd):
  247. from .base import Celery
  248. try:
  249. sym = symbol_by_name(app, imp=imp)
  250. except AttributeError:
  251. # last part was not an attribute, but a module
  252. sym = imp(app)
  253. if isinstance(sym, ModuleType) and ':' not in app:
  254. try:
  255. found = sym.app
  256. if isinstance(found, ModuleType):
  257. raise AttributeError()
  258. except AttributeError:
  259. try:
  260. found = sym.celery
  261. if isinstance(found, ModuleType):
  262. raise AttributeError()
  263. except AttributeError:
  264. if getattr(sym, '__path__', None):
  265. try:
  266. return find_app(
  267. '{0}.celery'.format(app),
  268. symbol_by_name=symbol_by_name, imp=imp,
  269. )
  270. except ImportError:
  271. pass
  272. for suspect in values(vars(sym)):
  273. if isinstance(suspect, Celery):
  274. return suspect
  275. raise
  276. else:
  277. return found
  278. else:
  279. return found
  280. return sym