FM / bin

/opt/imh-python/lib/python3.9/site-packages/celery/bin/

/opt/imh-python/lib/python3.9/site-packages/celery/bin

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
amqp.py	10023	0644	edit dl rm
base.py	9173	0644	edit dl rm
beat.py	2592	0644	edit dl rm
call.py	2370	0644	edit dl rm
celery.py	7440	0644	edit dl rm
control.py	8645	0644	edit dl rm
events.py	2794	0644	edit dl rm
graph.py	5796	0644	edit dl rm
list.py	1058	0644	edit dl rm
logtool.py	4267	0644	edit dl rm
migrate.py	2108	0644	edit dl rm
multi.py	15374	0644	edit dl rm
purge.py	2608	0644	edit dl rm
result.py	976	0644	edit dl rm
shell.py	4839	0644	edit dl rm
upgrade.py	3064	0644	edit dl rm
worker.py	12886	0644	edit dl rm
__init__.py	0	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/celery/bin/call.py (2370B)

"""The ``celery call`` program used to send tasks from the command-line."""
import click

from celery.bin.base import (ISO8601, ISO8601_OR_FLOAT, JSON_ARRAY, JSON_OBJECT, CeleryCommand, CeleryOption,
                             handle_preload_options)

@click.command(cls=CeleryCommand)
@click.argument('name')
@click.option('-a',
              '--args',
              cls=CeleryOption,
              type=JSON_ARRAY,
              default='[]',
              help_group="Calling Options",
              help="Positional arguments.")
@click.option('-k',
              '--kwargs',
              cls=CeleryOption,
              type=JSON_OBJECT,
              default='{}',
              help_group="Calling Options",
              help="Keyword arguments.")
@click.option('--eta',
              cls=CeleryOption,
              type=ISO8601,
              help_group="Calling Options",
              help="scheduled time.")
@click.option('--countdown',
              cls=CeleryOption,
              type=float,
              help_group="Calling Options",
              help="eta in seconds from now.")
@click.option('--expires',
              cls=CeleryOption,
              type=ISO8601_OR_FLOAT,
              help_group="Calling Options",
              help="expiry time.")
@click.option('--serializer',
              cls=CeleryOption,
              default='json',
              help_group="Calling Options",
              help="task serializer.")
@click.option('--queue',
              cls=CeleryOption,
              help_group="Routing Options",
              help="custom queue name.")
@click.option('--exchange',
              cls=CeleryOption,
              help_group="Routing Options",
              help="custom exchange name.")
@click.option('--routing-key',
              cls=CeleryOption,
              help_group="Routing Options",
              help="custom routing key.")
@click.pass_context
@handle_preload_options
def call(ctx, name, args, kwargs, eta, countdown, expires, serializer, queue, exchange, routing_key):
    """Call a task by name."""
    task_id = ctx.obj.app.send_task(
        name,
        args=args, kwargs=kwargs,
        countdown=countdown,
        serializer=serializer,
        queue=queue,
        exchange=exchange,
        routing_key=routing_key,
        eta=eta,
        expires=expires
    ).id
    ctx.obj.echo(task_id)

"""The ``celery call`` program used to send tasks from the command-line.""" import click from celery.bin.base import (ISO8601, ISO8601_OR_FLOAT, JSON_ARRAY, JSON_OBJECT, CeleryCommand, CeleryOption, handle_preload_options) @click.command(cls=CeleryCommand) @click.argument('name') @click.option('-a', '--args', cls=CeleryOption, type=JSON_ARRAY, default='[]', help_group="Calling Options", help="Positional arguments.") @click.option('-k', '--kwargs', cls=CeleryOption, type=JSON_OBJECT, default='{}', help_group="Calling Options", help="Keyword arguments.") @click.option('--eta', cls=CeleryOption, type=ISO8601, help_group="Calling Options", help="scheduled time.") @click.option('--countdown', cls=CeleryOption, type=float, help_group="Calling Options", help="eta in seconds from now.") @click.option('--expires', cls=CeleryOption, type=ISO8601_OR_FLOAT, help_group="Calling Options", help="expiry time.") @click.option('--serializer', cls=CeleryOption, default='json', help_group="Calling Options", help="task serializer.") @click.option('--queue', cls=CeleryOption, help_group="Routing Options", help="custom queue name.") @click.option('--exchange', cls=CeleryOption, help_group="Routing Options", help="custom exchange name.") @click.option('--routing-key', cls=CeleryOption, help_group="Routing Options", help="custom routing key.") @click.pass_context @handle_preload_options def call(ctx, name, args, kwargs, eta, countdown, expires, serializer, queue, exchange, routing_key): """Call a task by name.""" task_id = ctx.obj.app.send_task( name, args=args, kwargs=kwargs, countdown=countdown, serializer=serializer, queue=queue, exchange=exchange, routing_key=routing_key, eta=eta, expires=expires ).id ctx.obj.echo(task_id)