1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
|
import click
import time
from django.core.management.base import BaseCommand, CommandError
from ...utils import get_statistics
class Command(BaseCommand):
"""
Print RQ statistics
"""
help = __doc__
_separator: str
def add_arguments(self, parser):
# TODO: convert this to @click.command like rq does
parser.add_argument(
'-j', '--json',
action='store_true',
dest='json',
help='Output statistics as JSON',
)
parser.add_argument(
'-y', '--yaml',
action='store_true',
dest='yaml',
help='Output statistics as YAML',
)
parser.add_argument(
'-i', '--interval',
dest='interval',
type=float,
help='Poll statistics every N seconds',
)
def _print_separator(self):
try:
click.echo(self._separator)
except AttributeError:
self._separator = "-" * self.table_width
click.echo(self._separator)
def _print_stats_dashboard(self, statistics):
if self.interval:
click.clear()
click.echo()
click.echo("Django RQ CLI Dashboard")
click.echo()
self._print_separator()
# Header
click.echo(
"""| %-15s|%10s |%10s |%10s |%10s |%10s |%10s |""" %
("Name", "Queued", "Active", "Deferred", "Finished", "Failed", "Workers")
)
self._print_separator()
# Print every queues in a row
for queue in statistics["queues"]:
click.echo(
"""| %-15s|%10s |%10s |%10s |%10s |%10s |%10s |""" %
(queue["name"], queue["jobs"],
queue["started_jobs"], queue["deferred_jobs"],
queue["finished_jobs"],queue["failed_jobs"],
queue["workers"])
)
self._print_separator()
if self.interval:
click.echo()
click.echo("Press 'Ctrl+c' to quit")
def handle(self, *args, **options):
if options.get("json"):
import json
click.echo(json.dumps(get_statistics()))
return
if options.get("yaml"):
try:
import yaml
except ImportError as ex:
raise CommandError("PyYAML is not installed.") from ex
# Disable YAML alias
yaml.Dumper.ignore_aliases = lambda *args: True # type: ignore[method-assign]
click.echo(yaml.dump(get_statistics(), default_flow_style=False))
return
self.interval = options.get("interval")
# Arbitrary
self.table_width = 90
# Do not continuously poll
if not self.interval:
self._print_stats_dashboard(get_statistics())
return
# Abuse clicks to 'live' render CLI dashboard TODO: Use curses instead
try:
while True:
self._print_stats_dashboard(get_statistics())
time.sleep(self.interval)
except KeyboardInterrupt:
pass
|
