/opt/imh-python/lib/python3.9/site-packages/celery/bin
NameSizeModeActions
__pycache__/-0755rm
amqp.py100230644editdlrm
base.py91730644editdlrm
beat.py25920644editdlrm
call.py23700644editdlrm
celery.py74400644editdlrm
control.py86450644editdlrm
events.py27940644editdlrm
graph.py57960644editdlrm
list.py10580644editdlrm
logtool.py42670644editdlrm
migrate.py21080644editdlrm
multi.py153740644editdlrm
purge.py26080644editdlrm
result.py9760644editdlrm
shell.py48390644editdlrm
upgrade.py30640644editdlrm
worker.py128860644editdlrm
__init__.py00644editdlrm
Edit: /opt/imh-python/lib/python3.9/site-packages/celery/bin/result.py (976B)
"""The ``celery result`` program, used to inspect task results.""" import click from celery.bin.base import CeleryCommand, CeleryOption, handle_preload_options @click.command(cls=CeleryCommand) @click.argument('task_id') @click.option('-t', '--task', cls=CeleryOption, help_group='Result Options', help="Name of task (if custom backend).") @click.option('--traceback', cls=CeleryOption, is_flag=True, help_group='Result Options', help="Show traceback instead.") @click.pass_context @handle_preload_options def result(ctx, task_id, task, traceback): """Print the return value for a given task id.""" app = ctx.obj.app result_cls = app.tasks[task].AsyncResult if task else app.AsyncResult task_result = result_cls(task_id) value = task_result.traceback if traceback else task_result.get() # TODO: Prettify result ctx.obj.echo(value)