celery.py 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826
  1. # -*- coding: utf-8 -*-
  2. """
  3. The :program:`celery` umbrella command.
  4. .. program:: celery
  5. """
  6. from __future__ import absolute_import, unicode_literals
  7. import anyjson
  8. import numbers
  9. import os
  10. import sys
  11. from functools import partial
  12. from importlib import import_module
  13. from celery.five import string_t, values
  14. from celery.platforms import EX_OK, EX_FAILURE, EX_UNAVAILABLE, EX_USAGE
  15. from celery.utils import term
  16. from celery.utils import text
  17. from celery.utils.timeutils import maybe_iso8601
  18. # Cannot use relative imports here due to a Windows issue (#1111).
  19. from celery.bin.base import Command, Option, Extensions
  20. # Import commands from other modules
  21. from celery.bin.amqp import amqp
  22. from celery.bin.beat import beat
  23. from celery.bin.events import events
  24. from celery.bin.graph import graph
  25. from celery.bin.worker import worker
  26. __all__ = ['CeleryCommand', 'main']
  27. HELP = """
  28. ---- -- - - ---- Commands- -------------- --- ------------
  29. {commands}
  30. ---- -- - - --------- -- - -------------- --- ------------
  31. Type '{prog_name} <command> --help' for help using a specific command.
  32. """
  33. MIGRATE_PROGRESS_FMT = """\
  34. Migrating task {state.count}/{state.strtotal}: \
  35. {body[task]}[{body[id]}]\
  36. """
  37. DEBUG = os.environ.get('C_DEBUG', False)
  38. command_classes = [
  39. ('Main', ['worker', 'events', 'beat', 'shell', 'multi', 'amqp'], 'green'),
  40. ('Remote Control', ['status', 'inspect', 'control'], 'blue'),
  41. ('Utils', ['purge', 'list', 'migrate', 'call', 'result', 'report'], None),
  42. ]
  43. if DEBUG: # pragma: no cover
  44. command_classes.append(
  45. ('Debug', ['graph'], 'red'),
  46. )
  47. def determine_exit_status(ret):
  48. if isinstance(ret, numbers.Integral):
  49. return ret
  50. return EX_OK if ret else EX_FAILURE
  51. def main(argv=None):
  52. # Fix for setuptools generated scripts, so that it will
  53. # work with multiprocessing fork emulation.
  54. # (see multiprocessing.forking.get_preparation_data())
  55. try:
  56. if __name__ != '__main__': # pragma: no cover
  57. sys.modules['__main__'] = sys.modules[__name__]
  58. cmd = CeleryCommand()
  59. cmd.maybe_patch_concurrency()
  60. from billiard import freeze_support
  61. freeze_support()
  62. cmd.execute_from_commandline(argv)
  63. except KeyboardInterrupt:
  64. pass
  65. class multi(Command):
  66. """Start multiple worker instances."""
  67. respects_app_option = False
  68. def get_options(self):
  69. return ()
  70. def run_from_argv(self, prog_name, argv, command=None):
  71. from celery.bin.multi import MultiTool
  72. multi = MultiTool(quiet=self.quiet, no_color=self.no_color)
  73. return multi.execute_from_commandline(
  74. [command] + argv, prog_name,
  75. )
  76. class list_(Command):
  77. """Get info from broker.
  78. Examples::
  79. celery list bindings
  80. NOTE: For RabbitMQ the management plugin is required.
  81. """
  82. args = '[bindings]'
  83. def list_bindings(self, management):
  84. try:
  85. bindings = management.get_bindings()
  86. except NotImplementedError:
  87. raise self.Error('Your transport cannot list bindings.')
  88. fmt = lambda q, e, r: self.out('{0:<28} {1:<28} {2}'.format(q, e, r))
  89. fmt('Queue', 'Exchange', 'Routing Key')
  90. fmt('-' * 16, '-' * 16, '-' * 16)
  91. for b in bindings:
  92. fmt(b['destination'], b['source'], b['routing_key'])
  93. def run(self, what=None, *_, **kw):
  94. topics = {'bindings': self.list_bindings}
  95. available = ', '.join(topics)
  96. if not what:
  97. raise self.UsageError(
  98. 'You must specify one of {0}'.format(available))
  99. if what not in topics:
  100. raise self.UsageError(
  101. 'unknown topic {0!r} (choose one of: {1})'.format(
  102. what, available))
  103. with self.app.connection() as conn:
  104. self.app.amqp.TaskConsumer(conn).declare()
  105. topics[what](conn.manager)
  106. class call(Command):
  107. """Call a task by name.
  108. Examples::
  109. celery call tasks.add --args='[2, 2]'
  110. celery call tasks.add --args='[2, 2]' --countdown=10
  111. """
  112. args = '<task_name>'
  113. option_list = Command.option_list + (
  114. Option('--args', '-a', help='positional arguments (json).'),
  115. Option('--kwargs', '-k', help='keyword arguments (json).'),
  116. Option('--eta', help='scheduled time (ISO-8601).'),
  117. Option('--countdown', type='float',
  118. help='eta in seconds from now (float/int).'),
  119. Option('--expires', help='expiry time (ISO-8601/float/int).'),
  120. Option('--serializer', default='json', help='defaults to json.'),
  121. Option('--queue', help='custom queue name.'),
  122. Option('--exchange', help='custom exchange name.'),
  123. Option('--routing-key', help='custom routing key.'),
  124. )
  125. def run(self, name, *_, **kw):
  126. # Positional args.
  127. args = kw.get('args') or ()
  128. if isinstance(args, string_t):
  129. args = anyjson.loads(args)
  130. # Keyword args.
  131. kwargs = kw.get('kwargs') or {}
  132. if isinstance(kwargs, string_t):
  133. kwargs = anyjson.loads(kwargs)
  134. # Expires can be int/float.
  135. expires = kw.get('expires') or None
  136. try:
  137. expires = float(expires)
  138. except (TypeError, ValueError):
  139. # or a string describing an ISO 8601 datetime.
  140. try:
  141. expires = maybe_iso8601(expires)
  142. except (TypeError, ValueError):
  143. raise
  144. res = self.app.send_task(name, args=args, kwargs=kwargs,
  145. countdown=kw.get('countdown'),
  146. serializer=kw.get('serializer'),
  147. queue=kw.get('queue'),
  148. exchange=kw.get('exchange'),
  149. routing_key=kw.get('routing_key'),
  150. eta=maybe_iso8601(kw.get('eta')),
  151. expires=expires)
  152. self.out(res.id)
  153. class purge(Command):
  154. """Erase all messages from all known task queues.
  155. WARNING: There is no undo operation for this command.
  156. """
  157. warn_prelude = (
  158. '{warning}: This will remove all tasks from {queues}: {names}.\n'
  159. ' There is no undo for this operation!\n\n'
  160. '(to skip this prompt use the -f option)\n'
  161. )
  162. warn_prompt = 'Are you sure you want to delete all tasks'
  163. fmt_purged = 'Purged {mnum} {messages} from {qnum} known task {queues}.'
  164. fmt_empty = 'No messages purged from {qnum} {queues}'
  165. option_list = Command.option_list + (
  166. Option('--force', '-f', action='store_true',
  167. help='Do not prompt for verification'),
  168. )
  169. def run(self, force=False, **kwargs):
  170. names = list(sorted(self.app.amqp.queues.keys()))
  171. qnum = len(names)
  172. if not force:
  173. self.out(self.warn_prelude.format(
  174. warning=self.colored.red('WARNING'),
  175. queues=text.pluralize(qnum, 'queue'), names=', '.join(names),
  176. ))
  177. if self.ask(self.warn_prompt, ('yes', 'no'), 'no') != 'yes':
  178. return
  179. messages = self.app.control.purge()
  180. fmt = self.fmt_purged if messages else self.fmt_empty
  181. self.out(fmt.format(
  182. mnum=messages, qnum=qnum,
  183. messages=text.pluralize(messages, 'message'),
  184. queues=text.pluralize(qnum, 'queue')))
  185. class result(Command):
  186. """Gives the return value for a given task id.
  187. Examples::
  188. celery result 8f511516-e2f5-4da4-9d2f-0fb83a86e500
  189. celery result 8f511516-e2f5-4da4-9d2f-0fb83a86e500 -t tasks.add
  190. celery result 8f511516-e2f5-4da4-9d2f-0fb83a86e500 --traceback
  191. """
  192. args = '<task_id>'
  193. option_list = Command.option_list + (
  194. Option('--task', '-t', help='name of task (if custom backend)'),
  195. Option('--traceback', action='store_true',
  196. help='show traceback instead'),
  197. )
  198. def run(self, task_id, *args, **kwargs):
  199. result_cls = self.app.AsyncResult
  200. task = kwargs.get('task')
  201. traceback = kwargs.get('traceback', False)
  202. if task:
  203. result_cls = self.app.tasks[task].AsyncResult
  204. result = result_cls(task_id)
  205. if traceback:
  206. value = result.traceback
  207. else:
  208. value = result.get()
  209. self.out(self.pretty(value)[1])
  210. class _RemoteControl(Command):
  211. name = None
  212. choices = None
  213. leaf = False
  214. option_list = Command.option_list + (
  215. Option('--timeout', '-t', type='float',
  216. help='Timeout in seconds (float) waiting for reply'),
  217. Option('--destination', '-d',
  218. help='Comma separated list of destination node names.'))
  219. def __init__(self, *args, **kwargs):
  220. self.show_body = kwargs.pop('show_body', True)
  221. self.show_reply = kwargs.pop('show_reply', True)
  222. super(_RemoteControl, self).__init__(*args, **kwargs)
  223. @classmethod
  224. def get_command_info(self, command,
  225. indent=0, prefix='', color=None, help=False):
  226. if help:
  227. help = '|' + text.indent(self.choices[command][1], indent + 4)
  228. else:
  229. help = None
  230. try:
  231. # see if it uses args.
  232. meth = getattr(self, command)
  233. return text.join([
  234. '|' + text.indent('{0}{1} {2}'.format(
  235. prefix, color(command), meth.__doc__), indent),
  236. help,
  237. ])
  238. except AttributeError:
  239. return text.join([
  240. '|' + text.indent(prefix + str(color(command)), indent), help,
  241. ])
  242. @classmethod
  243. def list_commands(self, indent=0, prefix='', color=None, help=False):
  244. color = color if color else lambda x: x
  245. prefix = prefix + ' ' if prefix else ''
  246. return '\n'.join(self.get_command_info(c, indent, prefix, color, help)
  247. for c in sorted(self.choices))
  248. @property
  249. def epilog(self):
  250. return '\n'.join([
  251. '[Commands]',
  252. self.list_commands(indent=4, help=True)
  253. ])
  254. def usage(self, command):
  255. return '%prog {0} [options] {1} <command> [arg1 .. argN]'.format(
  256. command, self.args)
  257. def call(self, *args, **kwargs):
  258. raise NotImplementedError('call')
  259. def run(self, *args, **kwargs):
  260. if not args:
  261. raise self.UsageError(
  262. 'Missing {0.name} method. See --help'.format(self))
  263. return self.do_call_method(args, **kwargs)
  264. def do_call_method(self, args, **kwargs):
  265. method = args[0]
  266. if method == 'help':
  267. raise self.Error("Did you mean '{0.name} --help'?".format(self))
  268. if method not in self.choices:
  269. raise self.UsageError(
  270. 'Unknown {0.name} method {1}'.format(self, method))
  271. if self.app.connection().transport.driver_type == 'sql':
  272. raise self.Error('Broadcast not supported by SQL broker transport')
  273. destination = kwargs.get('destination')
  274. timeout = kwargs.get('timeout') or self.choices[method][0]
  275. if destination and isinstance(destination, string_t):
  276. destination = [dest.strip() for dest in destination.split(',')]
  277. handler = getattr(self, method, self.call)
  278. replies = handler(method, *args[1:], timeout=timeout,
  279. destination=destination,
  280. callback=self.say_remote_command_reply)
  281. if not replies:
  282. raise self.Error('No nodes replied within time constraint.',
  283. status=EX_UNAVAILABLE)
  284. return replies
  285. class inspect(_RemoteControl):
  286. """Inspect the worker at runtime.
  287. Availability: RabbitMQ (amqp), Redis, and MongoDB transports.
  288. Examples::
  289. celery inspect active --timeout=5
  290. celery inspect scheduled -d worker1@example.com
  291. celery inspect revoked -d w1@e.com,w2@e.com
  292. """
  293. name = 'inspect'
  294. choices = {
  295. 'active': (1.0, 'dump active tasks (being processed)'),
  296. 'active_queues': (1.0, 'dump queues being consumed from'),
  297. 'scheduled': (1.0, 'dump scheduled tasks (eta/countdown/retry)'),
  298. 'reserved': (1.0, 'dump reserved tasks (waiting to be processed)'),
  299. 'stats': (1.0, 'dump worker statistics'),
  300. 'revoked': (1.0, 'dump of revoked task ids'),
  301. 'registered': (1.0, 'dump of registered tasks'),
  302. 'ping': (0.2, 'ping worker(s)'),
  303. 'clock': (1.0, 'get value of logical clock'),
  304. 'conf': (1.0, 'dump worker configuration'),
  305. 'report': (1.0, 'get bugreport info'),
  306. 'memsample': (1.0, 'sample memory (requires psutil)'),
  307. 'memdump': (1.0, 'dump memory samples (requires psutil)'),
  308. 'objgraph': (60.0, 'create object graph (requires objgraph)'),
  309. }
  310. def call(self, method, *args, **options):
  311. i = self.app.control.inspect(**options)
  312. return getattr(i, method)(*args)
  313. def objgraph(self, type_='Request', *args, **kwargs):
  314. return self.call('objgraph', type_, **kwargs)
  315. def conf(self, with_defaults=False, *args, **kwargs):
  316. return self.call('conf', with_defaults, **kwargs)
  317. class control(_RemoteControl):
  318. """Workers remote control.
  319. Availability: RabbitMQ (amqp), Redis, and MongoDB transports.
  320. Examples::
  321. celery control enable_events --timeout=5
  322. celery control -d worker1@example.com enable_events
  323. celery control -d w1.e.com,w2.e.com enable_events
  324. celery control -d w1.e.com add_consumer queue_name
  325. celery control -d w1.e.com cancel_consumer queue_name
  326. celery control -d w1.e.com add_consumer queue exchange direct rkey
  327. """
  328. name = 'control'
  329. choices = {
  330. 'enable_events': (1.0, 'tell worker(s) to enable events'),
  331. 'disable_events': (1.0, 'tell worker(s) to disable events'),
  332. 'add_consumer': (1.0, 'tell worker(s) to start consuming a queue'),
  333. 'cancel_consumer': (1.0, 'tell worker(s) to stop consuming a queue'),
  334. 'rate_limit': (
  335. 1.0, 'tell worker(s) to modify the rate limit for a task type'),
  336. 'time_limit': (
  337. 1.0, 'tell worker(s) to modify the time limit for a task type.'),
  338. 'autoscale': (1.0, 'change autoscale settings'),
  339. 'pool_grow': (1.0, 'start more pool processes'),
  340. 'pool_shrink': (1.0, 'use less pool processes'),
  341. }
  342. def call(self, method, *args, **options):
  343. return getattr(self.app.control, method)(*args, reply=True, **options)
  344. def pool_grow(self, method, n=1, **kwargs):
  345. """[N=1]"""
  346. return self.call(method, int(n), **kwargs)
  347. def pool_shrink(self, method, n=1, **kwargs):
  348. """[N=1]"""
  349. return self.call(method, int(n), **kwargs)
  350. def autoscale(self, method, max=None, min=None, **kwargs):
  351. """[max] [min]"""
  352. return self.call(method, int(max), int(min), **kwargs)
  353. def rate_limit(self, method, task_name, rate_limit, **kwargs):
  354. """<task_name> <rate_limit> (e.g. 5/s | 5/m | 5/h)>"""
  355. return self.call(method, task_name, rate_limit, **kwargs)
  356. def time_limit(self, method, task_name, soft, hard=None, **kwargs):
  357. """<task_name> <soft_secs> [hard_secs]"""
  358. return self.call(method, task_name,
  359. float(soft), float(hard), **kwargs)
  360. def add_consumer(self, method, queue, exchange=None,
  361. exchange_type='direct', routing_key=None, **kwargs):
  362. """<queue> [exchange [type [routing_key]]]"""
  363. return self.call(method, queue, exchange,
  364. exchange_type, routing_key, **kwargs)
  365. def cancel_consumer(self, method, queue, **kwargs):
  366. """<queue>"""
  367. return self.call(method, queue, **kwargs)
  368. class status(Command):
  369. """Show list of workers that are online."""
  370. option_list = inspect.option_list
  371. def run(self, *args, **kwargs):
  372. I = inspect(
  373. app=self.app,
  374. no_color=kwargs.get('no_color', False),
  375. stdout=self.stdout, stderr=self.stderr,
  376. show_reply=False, show_body=False, quiet=True,
  377. )
  378. replies = I.run('ping', **kwargs)
  379. if not replies:
  380. raise self.Error('No nodes replied within time constraint',
  381. status=EX_UNAVAILABLE)
  382. nodecount = len(replies)
  383. if not kwargs.get('quiet', False):
  384. self.out('\n{0} {1} online.'.format(
  385. nodecount, text.pluralize(nodecount, 'node')))
  386. class migrate(Command):
  387. """Migrate tasks from one broker to another.
  388. Examples::
  389. celery migrate redis://localhost amqp://guest@localhost//
  390. celery migrate django:// redis://localhost
  391. NOTE: This command is experimental, make sure you have
  392. a backup of the tasks before you continue.
  393. """
  394. args = '<source_url> <dest_url>'
  395. option_list = Command.option_list + (
  396. Option('--limit', '-n', type='int',
  397. help='Number of tasks to consume (int)'),
  398. Option('--timeout', '-t', type='float', default=1.0,
  399. help='Timeout in seconds (float) waiting for tasks'),
  400. Option('--ack-messages', '-a', action='store_true',
  401. help='Ack messages from source broker.'),
  402. Option('--tasks', '-T',
  403. help='List of task names to filter on.'),
  404. Option('--queues', '-Q',
  405. help='List of queues to migrate.'),
  406. Option('--forever', '-F', action='store_true',
  407. help='Continually migrate tasks until killed.'),
  408. )
  409. progress_fmt = MIGRATE_PROGRESS_FMT
  410. def on_migrate_task(self, state, body, message):
  411. self.out(self.progress_fmt.format(state=state, body=body))
  412. def run(self, source, destination, **kwargs):
  413. from kombu import Connection
  414. from celery.contrib.migrate import migrate_tasks
  415. migrate_tasks(Connection(source),
  416. Connection(destination),
  417. callback=self.on_migrate_task,
  418. **kwargs)
  419. class shell(Command): # pragma: no cover
  420. """Start shell session with convenient access to celery symbols.
  421. The following symbols will be added to the main globals:
  422. - celery: the current application.
  423. - chord, group, chain, chunks,
  424. xmap, xstarmap subtask, Task
  425. - all registered tasks.
  426. """
  427. option_list = Command.option_list + (
  428. Option('--ipython', '-I',
  429. action='store_true', dest='force_ipython',
  430. help='force iPython.'),
  431. Option('--bpython', '-B',
  432. action='store_true', dest='force_bpython',
  433. help='force bpython.'),
  434. Option('--python', '-P',
  435. action='store_true', dest='force_python',
  436. help='force default Python shell.'),
  437. Option('--without-tasks', '-T', action='store_true',
  438. help="don't add tasks to locals."),
  439. Option('--eventlet', action='store_true',
  440. help='use eventlet.'),
  441. Option('--gevent', action='store_true', help='use gevent.'),
  442. )
  443. def run(self, force_ipython=False, force_bpython=False,
  444. force_python=False, without_tasks=False, eventlet=False,
  445. gevent=False, **kwargs):
  446. sys.path.insert(0, os.getcwd())
  447. if eventlet:
  448. import_module('celery.concurrency.eventlet')
  449. if gevent:
  450. import_module('celery.concurrency.gevent')
  451. import celery
  452. import celery.task.base
  453. self.app.loader.import_default_modules()
  454. self.locals = {'app': self.app,
  455. 'celery': self.app,
  456. 'Task': celery.Task,
  457. 'chord': celery.chord,
  458. 'group': celery.group,
  459. 'chain': celery.chain,
  460. 'chunks': celery.chunks,
  461. 'xmap': celery.xmap,
  462. 'xstarmap': celery.xstarmap,
  463. 'subtask': celery.subtask,
  464. 'signature': celery.signature}
  465. if not without_tasks:
  466. self.locals.update(dict(
  467. (task.__name__, task) for task in values(self.app.tasks)
  468. if not task.name.startswith('celery.')),
  469. )
  470. if force_python:
  471. return self.invoke_fallback_shell()
  472. elif force_bpython:
  473. return self.invoke_bpython_shell()
  474. elif force_ipython:
  475. return self.invoke_ipython_shell()
  476. return self.invoke_default_shell()
  477. def invoke_default_shell(self):
  478. try:
  479. import IPython # noqa
  480. except ImportError:
  481. try:
  482. import bpython # noqa
  483. except ImportError:
  484. return self.invoke_fallback_shell()
  485. else:
  486. return self.invoke_bpython_shell()
  487. else:
  488. return self.invoke_ipython_shell()
  489. def invoke_fallback_shell(self):
  490. import code
  491. try:
  492. import readline
  493. except ImportError:
  494. pass
  495. else:
  496. import rlcompleter
  497. readline.set_completer(
  498. rlcompleter.Completer(self.locals).complete)
  499. readline.parse_and_bind('tab:complete')
  500. code.interact(local=self.locals)
  501. def invoke_ipython_shell(self):
  502. try:
  503. from IPython.terminal import embed
  504. embed.TerminalInteractiveShell(user_ns=self.locals).mainloop()
  505. except ImportError: # ipython < 0.11
  506. from IPython.Shell import IPShell
  507. IPShell(argv=[], user_ns=self.locals).mainloop()
  508. def invoke_bpython_shell(self):
  509. import bpython
  510. bpython.embed(self.locals)
  511. class help(Command):
  512. """Show help screen and exit."""
  513. def usage(self, command):
  514. return '%prog <command> [options] {0.args}'.format(self)
  515. def run(self, *args, **kwargs):
  516. self.parser.print_help()
  517. self.out(HELP.format(
  518. prog_name=self.prog_name,
  519. commands=CeleryCommand.list_commands(colored=self.colored),
  520. ))
  521. return EX_USAGE
  522. class report(Command):
  523. """Shows information useful to include in bugreports."""
  524. def run(self, *args, **kwargs):
  525. self.out(self.app.bugreport())
  526. return EX_OK
  527. class CeleryCommand(Command):
  528. namespace = 'celery'
  529. ext_fmt = '{self.namespace}.commands'
  530. commands = {
  531. 'amqp': amqp,
  532. 'beat': beat,
  533. 'call': call,
  534. 'control': control,
  535. 'events': events,
  536. 'graph': graph,
  537. 'help': help,
  538. 'inspect': inspect,
  539. 'list': list_,
  540. 'migrate': migrate,
  541. 'multi': multi,
  542. 'purge': purge,
  543. 'report': report,
  544. 'result': result,
  545. 'shell': shell,
  546. 'status': status,
  547. 'worker': worker,
  548. }
  549. enable_config_from_cmdline = True
  550. prog_name = 'celery'
  551. @classmethod
  552. def register_command(cls, fun, name=None):
  553. cls.commands[name or fun.__name__] = fun
  554. return fun
  555. def execute(self, command, argv=None):
  556. try:
  557. cls = self.commands[command]
  558. except KeyError:
  559. cls, argv = self.commands['help'], ['help']
  560. cls = self.commands.get(command) or self.commands['help']
  561. try:
  562. return cls(
  563. app=self.app, on_error=self.on_error,
  564. no_color=self.no_color, quiet=self.quiet,
  565. on_usage_error=partial(self.on_usage_error, command=command),
  566. ).run_from_argv(self.prog_name, argv[1:], command=argv[0])
  567. except self.UsageError as exc:
  568. self.on_usage_error(exc)
  569. return exc.status
  570. except self.Error as exc:
  571. self.on_error(exc)
  572. return exc.status
  573. def on_usage_error(self, exc, command=None):
  574. if command:
  575. helps = '{self.prog_name} {command} --help'
  576. else:
  577. helps = '{self.prog_name} --help'
  578. self.error(self.colored.magenta('Error: {0}'.format(exc)))
  579. self.error("""Please try '{0}'""".format(helps.format(
  580. self=self, command=command,
  581. )))
  582. def _relocate_args_from_start(self, argv, index=0):
  583. if argv:
  584. rest = []
  585. while index < len(argv):
  586. value = argv[index]
  587. if value.startswith('--'):
  588. rest.append(value)
  589. elif value.startswith('-'):
  590. # we eat the next argument even though we don't know
  591. # if this option takes an argument or not.
  592. # instead we will assume what is the command name in the
  593. # return statements below.
  594. try:
  595. nxt = argv[index + 1]
  596. if nxt.startswith('-'):
  597. # is another option
  598. rest.append(value)
  599. else:
  600. # is (maybe) a value for this option
  601. rest.extend([value, nxt])
  602. index += 1
  603. except IndexError:
  604. rest.append(value)
  605. break
  606. else:
  607. break
  608. index += 1
  609. if argv[index:]:
  610. # if there are more arguments left then divide and swap
  611. # we assume the first argument in argv[i:] is the command
  612. # name.
  613. return argv[index:] + rest
  614. # if there are no more arguments then the last arg in rest'
  615. # must be the command.
  616. [rest.pop()] + rest
  617. return []
  618. def prepare_prog_name(self, name):
  619. if name == '__main__.py':
  620. return sys.modules['__main__'].__file__
  621. return name
  622. def handle_argv(self, prog_name, argv):
  623. self.prog_name = self.prepare_prog_name(prog_name)
  624. argv = self._relocate_args_from_start(argv)
  625. _, argv = self.prepare_args(None, argv)
  626. try:
  627. command = argv[0]
  628. except IndexError:
  629. command, argv = 'help', ['help']
  630. return self.execute(command, argv)
  631. def execute_from_commandline(self, argv=None):
  632. argv = sys.argv if argv is None else argv
  633. if 'multi' in argv[1:3]: # Issue 1008
  634. self.respects_app_option = False
  635. try:
  636. sys.exit(determine_exit_status(
  637. super(CeleryCommand, self).execute_from_commandline(argv)))
  638. except KeyboardInterrupt:
  639. sys.exit(EX_FAILURE)
  640. @classmethod
  641. def get_command_info(self, command, indent=0, color=None, colored=None):
  642. colored = term.colored() if colored is None else colored
  643. colored = colored.names[color] if color else lambda x: x
  644. obj = self.commands[command]
  645. cmd = 'celery {0}'.format(colored(command))
  646. if obj.leaf:
  647. return '|' + text.indent(cmd, indent)
  648. return text.join([
  649. ' ',
  650. '|' + text.indent('{0} --help'.format(cmd), indent),
  651. obj.list_commands(indent, 'celery {0}'.format(command), colored),
  652. ])
  653. @classmethod
  654. def list_commands(self, indent=0, colored=None):
  655. colored = term.colored() if colored is None else colored
  656. white = colored.white
  657. ret = []
  658. for cls, commands, color in command_classes:
  659. ret.extend([
  660. text.indent('+ {0}: '.format(white(cls)), indent),
  661. '\n'.join(
  662. self.get_command_info(command, indent + 4, color, colored)
  663. for command in commands),
  664. ''
  665. ])
  666. return '\n'.join(ret).strip()
  667. def with_pool_option(self, argv):
  668. if len(argv) > 1 and 'worker' in argv[0:3]:
  669. # this command supports custom pools
  670. # that may have to be loaded as early as possible.
  671. return (['-P'], ['--pool'])
  672. def on_concurrency_setup(self):
  673. self.load_extension_commands()
  674. def load_extension_commands(self):
  675. names = Extensions(self.ext_fmt.format(self=self),
  676. self.register_command).load()
  677. if names:
  678. command_classes.append(('Extensions', names, 'magenta'))
  679. def command(*args, **kwargs):
  680. """Deprecated: Use classmethod :meth:`CeleryCommand.register_command`
  681. instead."""
  682. _register = CeleryCommand.register_command
  683. return _register(args[0]) if args else _register
  684. if __name__ == '__main__': # pragma: no cover
  685. main()