mirror of https://github.com/peter4431/rq.git
You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
272 lines
11 KiB
Python
272 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
RQ command line tool
|
|
"""
|
|
from __future__ import (absolute_import, division, print_function,
|
|
unicode_literals)
|
|
|
|
from functools import update_wrapper
|
|
import os
|
|
import sys
|
|
|
|
import click
|
|
from redis.exceptions import ConnectionError
|
|
|
|
from rq import Connection, __version__ as version
|
|
from rq.cli.helpers import (read_config_file, refresh,
|
|
setup_loghandlers_from_args,
|
|
show_both, show_queues, show_workers, CliConfig)
|
|
from rq.contrib.legacy import cleanup_ghosts
|
|
from rq.defaults import (DEFAULT_CONNECTION_CLASS, DEFAULT_JOB_CLASS,
|
|
DEFAULT_QUEUE_CLASS, DEFAULT_WORKER_CLASS,
|
|
DEFAULT_RESULT_TTL, DEFAULT_WORKER_TTL,
|
|
DEFAULT_JOB_MONITORING_INTERVAL,
|
|
DEFAULT_LOGGING_FORMAT, DEFAULT_LOGGING_DATE_FORMAT)
|
|
from rq.exceptions import InvalidJobOperationError
|
|
from rq.registry import FailedJobRegistry
|
|
from rq.utils import import_attribute
|
|
from rq.suspension import (suspend as connection_suspend,
|
|
resume as connection_resume, is_suspended)
|
|
|
|
|
|
# Disable the warning that Click displays (as of Click version 5.0) when users
|
|
# use unicode_literals in Python 2.
|
|
# See http://click.pocoo.org/dev/python3/#unicode-literals for more details.
|
|
click.disable_unicode_literals_warning = True
|
|
|
|
|
|
shared_options = [
|
|
click.option('--url', '-u',
|
|
envvar='RQ_REDIS_URL',
|
|
help='URL describing Redis connection details.'),
|
|
click.option('--config', '-c',
|
|
envvar='RQ_CONFIG',
|
|
help='Module containing RQ settings.'),
|
|
click.option('--worker-class', '-w',
|
|
envvar='RQ_WORKER_CLASS',
|
|
default=DEFAULT_WORKER_CLASS,
|
|
help='RQ Worker class to use'),
|
|
click.option('--job-class', '-j',
|
|
envvar='RQ_JOB_CLASS',
|
|
default=DEFAULT_JOB_CLASS,
|
|
help='RQ Job class to use'),
|
|
click.option('--queue-class',
|
|
envvar='RQ_QUEUE_CLASS',
|
|
default=DEFAULT_QUEUE_CLASS,
|
|
help='RQ Queue class to use'),
|
|
click.option('--connection-class',
|
|
envvar='RQ_CONNECTION_CLASS',
|
|
default=DEFAULT_CONNECTION_CLASS,
|
|
help='Redis client class to use'),
|
|
click.option('--path', '-P',
|
|
default='.',
|
|
help='Specify the import path.',
|
|
multiple=True)
|
|
]
|
|
|
|
|
|
def pass_cli_config(func):
|
|
# add all the shared options to the command
|
|
for option in shared_options:
|
|
func = option(func)
|
|
|
|
# pass the cli config object into the command
|
|
def wrapper(*args, **kwargs):
|
|
ctx = click.get_current_context()
|
|
cli_config = CliConfig(**kwargs)
|
|
return ctx.invoke(func, cli_config, *args[1:], **kwargs)
|
|
|
|
return update_wrapper(wrapper, func)
|
|
|
|
|
|
@click.group()
|
|
@click.version_option(version)
|
|
def main():
|
|
"""RQ command line tool."""
|
|
pass
|
|
|
|
|
|
@main.command()
|
|
@click.option('--all', '-a', is_flag=True, help='Empty all queues')
|
|
@click.argument('queues', nargs=-1)
|
|
@pass_cli_config
|
|
def empty(cli_config, all, queues, **options):
|
|
"""Empty given queues."""
|
|
|
|
if all:
|
|
queues = cli_config.queue_class.all(connection=cli_config.connection,
|
|
job_class=cli_config.job_class)
|
|
else:
|
|
queues = [cli_config.queue_class(queue,
|
|
connection=cli_config.connection,
|
|
job_class=cli_config.job_class)
|
|
for queue in queues]
|
|
|
|
if not queues:
|
|
click.echo('Nothing to do')
|
|
sys.exit(0)
|
|
|
|
for queue in queues:
|
|
num_jobs = queue.empty()
|
|
click.echo('{0} jobs removed from {1} queue'.format(num_jobs, queue.name))
|
|
|
|
|
|
@main.command()
|
|
@click.option('--all', '-a', is_flag=True, help='Requeue all failed jobs')
|
|
@click.option('--queue', required=True, type=str)
|
|
@click.argument('job_ids', nargs=-1)
|
|
@pass_cli_config
|
|
def requeue(cli_config, queue, all, job_class, job_ids, **options):
|
|
"""Requeue failed jobs."""
|
|
|
|
failed_job_registry = FailedJobRegistry(queue,
|
|
connection=cli_config.connection)
|
|
if all:
|
|
job_ids = failed_job_registry.get_job_ids()
|
|
|
|
if not job_ids:
|
|
click.echo('Nothing to do')
|
|
sys.exit(0)
|
|
|
|
click.echo('Requeueing {0} jobs from failed queue'.format(len(job_ids)))
|
|
fail_count = 0
|
|
with click.progressbar(job_ids) as job_ids:
|
|
for job_id in job_ids:
|
|
try:
|
|
failed_job_registry.requeue(job_id)
|
|
except InvalidJobOperationError:
|
|
fail_count += 1
|
|
|
|
if fail_count > 0:
|
|
click.secho('Unable to requeue {0} jobs from failed job registry'.format(fail_count), fg='red')
|
|
|
|
|
|
@main.command()
|
|
@click.option('--interval', '-i', type=float, help='Updates stats every N seconds (default: don\'t poll)')
|
|
@click.option('--raw', '-r', is_flag=True, help='Print only the raw numbers, no bar charts')
|
|
@click.option('--only-queues', '-Q', is_flag=True, help='Show only queue info')
|
|
@click.option('--only-workers', '-W', is_flag=True, help='Show only worker info')
|
|
@click.option('--by-queue', '-R', is_flag=True, help='Shows workers by queue')
|
|
@click.argument('queues', nargs=-1)
|
|
@pass_cli_config
|
|
def info(cli_config, interval, raw, only_queues, only_workers, by_queue, queues,
|
|
**options):
|
|
"""RQ command-line monitor."""
|
|
|
|
if only_queues:
|
|
func = show_queues
|
|
elif only_workers:
|
|
func = show_workers
|
|
else:
|
|
func = show_both
|
|
|
|
try:
|
|
with Connection(cli_config.connection):
|
|
refresh(interval, func, queues, raw, by_queue,
|
|
cli_config.queue_class, cli_config.worker_class)
|
|
except ConnectionError as e:
|
|
click.echo(e)
|
|
sys.exit(1)
|
|
except KeyboardInterrupt:
|
|
click.echo()
|
|
sys.exit(0)
|
|
|
|
|
|
@main.command()
|
|
@click.option('--burst', '-b', is_flag=True, help='Run in burst mode (quit after all work is done)')
|
|
@click.option('--logging_level', type=str, default="INFO", help='Set logging level')
|
|
@click.option('--log-format', type=str, default=DEFAULT_LOGGING_FORMAT, help='Set the format of the logs')
|
|
@click.option('--date-format', type=str, default=DEFAULT_LOGGING_DATE_FORMAT, help='Set the date format of the logs')
|
|
@click.option('--name', '-n', help='Specify a different name')
|
|
@click.option('--results-ttl', type=int, default=DEFAULT_RESULT_TTL , help='Default results timeout to be used')
|
|
@click.option('--worker-ttl', type=int, default=DEFAULT_WORKER_TTL , help='Default worker timeout to be used')
|
|
@click.option('--job-monitoring-interval', type=int, default=DEFAULT_JOB_MONITORING_INTERVAL , help='Default job monitoring interval to be used')
|
|
@click.option('--disable-job-desc-logging', is_flag=True, help='Turn off description logging.')
|
|
@click.option('--verbose', '-v', is_flag=True, help='Show more output')
|
|
@click.option('--quiet', '-q', is_flag=True, help='Show less output')
|
|
@click.option('--sentry-dsn', envvar='RQ_SENTRY_DSN', help='Report exceptions to this Sentry DSN')
|
|
@click.option('--exception-handler', help='Exception handler(s) to use', multiple=True)
|
|
@click.option('--pid', help='Write the process ID number to a file at the specified path')
|
|
@click.option('--disable-default-exception-handler', '-d', is_flag=True, help='Disable RQ\'s default exception handler')
|
|
@click.argument('queues', nargs=-1)
|
|
@pass_cli_config
|
|
def worker(cli_config, burst, logging_level, name, results_ttl,
|
|
worker_ttl, job_monitoring_interval, disable_job_desc_logging, verbose, quiet, sentry_dsn,
|
|
exception_handler, pid, disable_default_exception_handler, queues,
|
|
log_format, date_format, **options):
|
|
"""Starts an RQ worker."""
|
|
settings = read_config_file(cli_config.config) if cli_config.config else {}
|
|
# Worker specific default arguments
|
|
queues = queues or settings.get('QUEUES', ['default'])
|
|
sentry_dsn = sentry_dsn or settings.get('SENTRY_DSN')
|
|
name = name or settings.get('NAME')
|
|
|
|
if pid:
|
|
with open(os.path.expanduser(pid), "w") as fp:
|
|
fp.write(str(os.getpid()))
|
|
|
|
setup_loghandlers_from_args(verbose, quiet, date_format, log_format)
|
|
|
|
try:
|
|
|
|
cleanup_ghosts(cli_config.connection)
|
|
exception_handlers = []
|
|
for h in exception_handler:
|
|
exception_handlers.append(import_attribute(h))
|
|
|
|
if is_suspended(cli_config.connection):
|
|
click.secho('RQ is currently suspended, to resume job execution run "rq resume"', fg='red')
|
|
sys.exit(1)
|
|
|
|
queues = [cli_config.queue_class(queue,
|
|
connection=cli_config.connection,
|
|
job_class=cli_config.job_class)
|
|
for queue in queues]
|
|
worker = cli_config.worker_class(
|
|
queues, name=name, connection=cli_config.connection,
|
|
default_worker_ttl=worker_ttl, default_result_ttl=results_ttl,
|
|
job_monitoring_interval=job_monitoring_interval,
|
|
job_class=cli_config.job_class, queue_class=cli_config.queue_class,
|
|
exception_handlers=exception_handlers or None,
|
|
disable_default_exception_handler=disable_default_exception_handler,
|
|
log_job_description=not disable_job_desc_logging
|
|
)
|
|
|
|
# Should we configure Sentry?
|
|
if sentry_dsn:
|
|
from rq.contrib.sentry import register_sentry
|
|
register_sentry(sentry_dsn)
|
|
|
|
worker.work(burst=burst, logging_level=logging_level, date_format=date_format, log_format=log_format)
|
|
except ConnectionError as e:
|
|
print(e)
|
|
sys.exit(1)
|
|
|
|
|
|
@main.command()
|
|
@click.option('--duration', help='Seconds you want the workers to be suspended. Default is forever.', type=int)
|
|
@pass_cli_config
|
|
def suspend(cli_config, duration, **options):
|
|
"""Suspends all workers, to resume run `rq resume`"""
|
|
|
|
if duration is not None and duration < 1:
|
|
click.echo("Duration must be an integer greater than 1")
|
|
sys.exit(1)
|
|
|
|
connection_suspend(cli_config.connection, duration)
|
|
|
|
if duration:
|
|
msg = """Suspending workers for {0} seconds. No new jobs will be started during that time, but then will
|
|
automatically resume""".format(duration)
|
|
click.echo(msg)
|
|
else:
|
|
click.echo("Suspending workers. No new jobs will be started. But current jobs will be completed")
|
|
|
|
|
|
@main.command()
|
|
@pass_cli_config
|
|
def resume(cli_config, **options):
|
|
"""Resumes processing of queues, that where suspended with `rq suspend`"""
|
|
connection_resume(cli_config.connection)
|
|
click.echo("Resuming workers.")
|