util
/
rq
mirror of https://github.com/peter4431/rq.git
4
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Cleaned up cli.py (#1876)

Browse Source
main
Selwin Ong 2 years ago committed by GitHub
parent ed7a171460
commit b639ee6406
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 98 additions and 80 deletions
Show Stats Download Patch File Download Diff File

75
rq/cli/cli.py
Unescape Escape View File

@ -2,7 +2,6 @@
RQ command line tool RQ command line tool
""" """
from functools import update_wrapper
import os import os
import sys import sys
import warnings import warnings
@ -18,22 +17,18 @@ from rq.cli.helpers import (
show_both, show_both,
show_queues, show_queues,
show_workers, show_workers,
CliConfig,
parse_function_args, parse_function_args,
parse_schedule, parse_schedule,
pass_cli_config,
) )
from rq.contrib.legacy import cleanup_ghosts from rq.contrib.legacy import cleanup_ghosts
from rq.defaults import ( from rq.defaults import (
DEFAULT_CONNECTION_CLASS,
DEFAULT_JOB_CLASS,
DEFAULT_QUEUE_CLASS,
DEFAULT_WORKER_CLASS,
DEFAULT_RESULT_TTL, DEFAULT_RESULT_TTL,
DEFAULT_WORKER_TTL, DEFAULT_WORKER_TTL,
DEFAULT_JOB_MONITORING_INTERVAL, DEFAULT_JOB_MONITORING_INTERVAL,
DEFAULT_LOGGING_FORMAT, DEFAULT_LOGGING_FORMAT,
DEFAULT_LOGGING_DATE_FORMAT, DEFAULT_LOGGING_DATE_FORMAT,
DEFAULT_SERIALIZER_CLASS, DEFAULT_MAINTENANCE_TASK_INTERVAL, DEFAULT_MAINTENANCE_TASK_INTERVAL,
) )
from rq.exceptions import InvalidJobOperationError from rq.exceptions import InvalidJobOperationError
from rq.registry import FailedJobRegistry, clean_registries from rq.registry import FailedJobRegistry, clean_registries
@ -45,50 +40,6 @@ from rq.job import JobStatus
blue = make_colorizer('darkblue') blue = make_colorizer('darkblue')
# 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),
click.option(
'--serializer',
'-S',
default=DEFAULT_SERIALIZER_CLASS,
help='Path to serializer, defaults to rq.serializers.DefaultSerializer',
),
]
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.group()
@click.version_option(version) @click.version_option(version)
def main(): def main():
@ -105,8 +56,10 @@ def empty(cli_config, all, queues, serializer, **options):
if all: if all:
queues = cli_config.queue_class.all( queues = cli_config.queue_class.all(
connection=cli_config.connection, job_class=cli_config.job_class, connection=cli_config.connection,
death_penalty_class=cli_config.death_penalty_class, serializer=serializer job_class=cli_config.job_class,
death_penalty_class=cli_config.death_penalty_class,
serializer=serializer,
) )
else: else:
queues = [ queues = [
@ -206,7 +159,7 @@ def info(cli_config, interval, raw, only_queues, only_workers, by_queue, queues,
'--maintenance-interval', '--maintenance-interval',
type=int, type=int,
default=DEFAULT_MAINTENANCE_TASK_INTERVAL, default=DEFAULT_MAINTENANCE_TASK_INTERVAL,
help='Maintenance task interval (in seconds) to be used' help='Maintenance task interval (in seconds) to be used',
) )
@click.option( @click.option(
'--job-monitoring-interval', '--job-monitoring-interval',
@ -227,7 +180,9 @@ def info(cli_config, interval, raw, only_queues, only_workers, by_queue, queues,
@click.option('--max-idle-time', type=int, default=None, help='Maximum seconds to stay alive without jobs to execute') @click.option('--max-idle-time', type=int, default=None, help='Maximum seconds to stay alive without jobs to execute')
@click.option('--with-scheduler', '-s', is_flag=True, help='Run worker with scheduler') @click.option('--with-scheduler', '-s', is_flag=True, help='Run worker with scheduler')
@click.option('--serializer', '-S', default=None, help='Run worker with custom serializer') @click.option('--serializer', '-S', default=None, help='Run worker with custom serializer')
@click.option('--dequeue-strategy', '-ds', default='default', help='Sets a custom stratey to dequeue from multiple queues') @click.option(
'--dequeue-strategy', '-ds', default='default', help='Sets a custom stratey to dequeue from multiple queues'
)
@click.argument('queues', nargs=-1) @click.argument('queues', nargs=-1)
@pass_cli_config @pass_cli_config
def worker( def worker(
@ -274,12 +229,14 @@ def worker(
worker_name = cli_config.worker_class.__qualname__ worker_name = cli_config.worker_class.__qualname__
if worker_name in ["RoundRobinWorker", "RandomWorker"]: if worker_name in ["RoundRobinWorker", "RandomWorker"]:
strategy_alternative = "random" if worker_name == "RandomWorker" else "round_robin" strategy_alternative = "random" if worker_name == "RandomWorker" else "round_robin"
msg = f"WARNING: The {worker_name} is deprecated. Use the --dequeue-strategy / -ds option with the {strategy_alternative} argument to set the strategy." msg = f"WARNING: {worker_name} is deprecated. Use `--dequeue-strategy {strategy_alternative}` instead."
warnings.warn(msg, DeprecationWarning) warnings.warn(msg, DeprecationWarning)
click.secho(msg, fg='yellow') click.secho(msg, fg='yellow')
if dequeue_strategy not in ("default", "random", "round_robin"): if dequeue_strategy not in ("default", "random", "round_robin"):
click.secho("ERROR: Dequeue Strategy can only be one of `default`, `random` or `round_robin`.", err=True, fg='red') click.secho(
"ERROR: Dequeue Strategy can only be one of `default`, `random` or `round_robin`.", err=True, fg='red'
)
sys.exit(1) sys.exit(1)
setup_loghandlers_from_args(verbose, quiet, date_format, log_format) setup_loghandlers_from_args(verbose, quiet, date_format, log_format)
@ -313,7 +270,7 @@ def worker(
exception_handlers=exception_handlers or None, exception_handlers=exception_handlers or None,
disable_default_exception_handler=disable_default_exception_handler, disable_default_exception_handler=disable_default_exception_handler,
log_job_description=not disable_job_desc_logging, log_job_description=not disable_job_desc_logging,
serializer=serializer serializer=serializer,
) )
# Should we configure Sentry? # Should we configure Sentry?
@ -335,7 +292,7 @@ def worker(
max_jobs=max_jobs, max_jobs=max_jobs,
max_idle_time=max_idle_time, max_idle_time=max_idle_time,
with_scheduler=with_scheduler, with_scheduler=with_scheduler,
dequeue_strategy=dequeue_strategy dequeue_strategy=dequeue_strategy,
) )
except ConnectionError as e: except ConnectionError as e:
print(e) print(e)

103
rq/cli/helpers.py
Unescape Escape View File

@ -2,7 +2,8 @@ import sys
import importlib import importlib
import time import time
import os import os
from functools import partial
from functools import partial, update_wrapper
from enum import Enum from enum import Enum
from datetime import datetime, timezone, timedelta from datetime import datetime, timezone, timedelta
@ -13,8 +14,15 @@ from shutil import get_terminal_size
import click import click
from redis import Redis from redis import Redis
from redis.sentinel import Sentinel from redis.sentinel import Sentinel
from rq.defaults import DEFAULT_CONNECTION_CLASS, DEFAULT_JOB_CLASS, DEFAULT_QUEUE_CLASS, DEFAULT_WORKER_CLASS, \
DEFAULT_DEATH_PENALTY_CLASS from rq.defaults import (
DEFAULT_CONNECTION_CLASS,
DEFAULT_DEATH_PENALTY_CLASS,
DEFAULT_JOB_CLASS,
DEFAULT_QUEUE_CLASS,
DEFAULT_WORKER_CLASS,
DEFAULT_SERIALIZER_CLASS,
)
from rq.logutils import setup_loghandlers from rq.logutils import setup_loghandlers
from rq.utils import import_attribute, parse_timeout from rq.utils import import_attribute, parse_timeout
from rq.worker import WorkerStatus from rq.worker import WorkerStatus
@ -42,7 +50,7 @@ def get_redis_from_config(settings, connection_class=Redis):
elif settings.get('SENTINEL') is not None: elif settings.get('SENTINEL') is not None:
instances = settings['SENTINEL'].get('INSTANCES', [('localhost', 26379)]) instances = settings['SENTINEL'].get('INSTANCES', [('localhost', 26379)])
master_name = settings['SENTINEL'].get('MASTER_NAME', 'mymaster') master_name = settings['SENTINEL'].get('MASTER_NAME', 'mymaster')
connection_kwargs = { connection_kwargs = {
'db': settings['SENTINEL'].get('DB', 0), 'db': settings['SENTINEL'].get('DB', 0),
'username': settings['SENTINEL'].get('USERNAME', None), 'username': settings['SENTINEL'].get('USERNAME', None),
@ -52,10 +60,8 @@ def get_redis_from_config(settings, connection_class=Redis):
} }
connection_kwargs.update(settings['SENTINEL'].get('CONNECTION_KWARGS', {})) connection_kwargs.update(settings['SENTINEL'].get('CONNECTION_KWARGS', {}))
sentinel_kwargs = settings['SENTINEL'].get('SENTINEL_KWARGS', {}) sentinel_kwargs = settings['SENTINEL'].get('SENTINEL_KWARGS', {})
sn = Sentinel( sn = Sentinel(instances, sentinel_kwargs=sentinel_kwargs, **connection_kwargs)
instances, sentinel_kwargs=sentinel_kwargs, **connection_kwargs
)
return sn.master_for(master_name) return sn.master_for(master_name)
ssl = settings.get('REDIS_SSL', False) ssl = settings.get('REDIS_SSL', False)
@ -124,13 +130,22 @@ def show_queues(queues, raw, by_queue, queue_class, worker_class):
count = counts[q] count = counts[q]
if not raw: if not raw:
chart = green('|' + '█' * int(ratio * count)) chart = green('|' + '█' * int(ratio * count))
line = '%-12s %s %d, %d executing, %d finished, %d failed' \ line = '%-12s %s %d, %d executing, %d finished, %d failed' % (
% (q.name, chart, count, q.started_job_registry.count, \ q.name,
q.finished_job_registry.count, q.failed_job_registry.count) chart,
count,
q.started_job_registry.count,
q.finished_job_registry.count,
q.failed_job_registry.count,
)
else: else:
line = 'queue %s %d, %d executing, %d finished, %d failed' \ line = 'queue %s %d, %d executing, %d finished, %d failed' % (
% (q.name, count, q.started_job_registry.count, \ q.name,
q.finished_job_registry.count, q.failed_job_registry.count) count,
q.started_job_registry.count,
q.finished_job_registry.count,
q.failed_job_registry.count,
)
click.echo(line) click.echo(line)
num_jobs += count num_jobs += count
@ -155,14 +170,22 @@ def show_workers(queues, raw, by_queue, queue_class, worker_class):
queue_names = ', '.join(worker.queue_names()) queue_names = ', '.join(worker.queue_names())
name = '%s (%s %s %s)' % (worker.name, worker.hostname, worker.ip_address, worker.pid) name = '%s (%s %s %s)' % (worker.name, worker.hostname, worker.ip_address, worker.pid)
if not raw: if not raw:
line = '%s: %s %s. jobs: %d finished, %d failed' \ line = '%s: %s %s. jobs: %d finished, %d failed' % (
% (name, state_symbol(worker.get_state()), queue_names, \ name,
worker.successful_job_count, worker.failed_job_count) state_symbol(worker.get_state()),
queue_names,
worker.successful_job_count,
worker.failed_job_count,
)
click.echo(line) click.echo(line)
else: else:
line = 'worker %s %s %s. jobs: %d finished, %d failed' \ line = 'worker %s %s %s. jobs: %d finished, %d failed' % (
% (name, worker.get_state(), queue_names,\ name,
worker.successful_job_count, worker.failed_job_count) worker.get_state(),
queue_names,
worker.successful_job_count,
worker.failed_job_count,
)
click.echo(line) click.echo(line)
else: else:
@ -318,7 +341,7 @@ class CliConfig:
connection_class=DEFAULT_CONNECTION_CLASS, connection_class=DEFAULT_CONNECTION_CLASS,
path=None, path=None,
*args, *args,
**kwargs **kwargs,
): ):
self._connection = None self._connection = None
self.url = url self.url = url
@ -363,3 +386,41 @@ class CliConfig:
else: else:
self._connection = get_redis_from_config(os.environ, self.connection_class) self._connection = get_redis_from_config(os.environ, self.connection_class)
return self._connection return self._connection
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),
click.option(
'--serializer',
'-S',
default=DEFAULT_SERIALIZER_CLASS,
help='Path to serializer, defaults to rq.serializers.DefaultSerializer',
),
]
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)

Write Preview
Loading…
Cancel
Save