Coverage for core/management/commands/rqworker.py: 0%

1import logging 

2 

3from django_rq.management.commands.rqworker import Command as _Command 

4 

5from netbox.registry import registry 

6 

7DEFAULT_QUEUES = ('high', 'default', 'low') 

8 

9logger = logging.getLogger('netbox.rqworker') 

10 

11 

12class Command(_Command): 

13 """ 

14 Subclass django_rq's built-in rqworker to listen on all configured queues if none are specified (instead 

15 of only the 'default' queue). 

16 """ 

17 def handle(self, *args, **options): 

18 # Setup system jobs. 

19 for job, kwargs in registry['system_jobs'].items(): 

20 try: 

21 interval = kwargs['interval'] 

22 except KeyError: 

23 raise TypeError("System job must specify an interval (in minutes).") 

24 logger.debug(f"Scheduling system job {job.name} (interval={interval})") 

25 job.enqueue_once(**kwargs) 

26 

27 # Run the worker with scheduler functionality 

28 options['with_scheduler'] = True 

29 

30 # If no queues have been specified on the command line, listen on all configured queues. 

31 if len(args) < 1: 

32 queues = ', '.join(DEFAULT_QUEUES) 

33 logger.warning( 

34 f"No queues have been specified. This process will service the following queues by default: {queues}" 

35 ) 

36 args = DEFAULT_QUEUES 

37 

38 super().handle(*args, **options)