CODEWITHSUNDEEP
pythonredispython-rq
notconfusing
notconfusing

Reputation: 2606

Schedule a redis job that schedules another redis job with python-rq

I have two kinds of jobs: ones that I want to run in serial and ones that I want to run concurrently in parallel. However I want the parallel jobs to get scheduled in serial (if you're still following). That is:

  1. Do A.
  2. Wait for A, do B.
  3. Wait for B, do 2+ versions of C all concurrently.

My thought it to have 2 redis queues, a serial_queue that has just one worker on it. And a parallel_queue which has multiple workers on it.

serial_queue.schedule(
    scheduled_time=datetime.utcnow(),
    func=job_a,
     ...)    
serial_queue.schedule(
    scheduled_time=datetime.utcnow(),
    func=job_b,
     ...)

def parallel_c():
    for task in range(args.n_tasks):
        queue_concurrent.schedule(
            scheduled_time=datetime.utcnow(),
            func=job_c,
            ...)

serial_queue.schedule(
    scheduled_time=datetime.utcnow(),
    func=parallel_c,
     ...)

But this setup currently, gives the error that AttributeError: module '__main__' has no attribute 'schedule_fetch_tweets' . How can I package this function properly for python-rq?

Upvotes: 2

Views: 1067

Answers (1)

notconfusing
notconfusing

Reputation: 2606

The solution requires a bit of gymnastics, in that you have to import the current script as if it were an external module.

So for instance. The contents of schedule_twitter_jobs.py would be:

from redis import Redis
from rq_scheduler import Scheduler
import schedule_twitter_jobs
# we are importing the very module we are executing

def schedule_fetch_tweets(args, queue_name):
    ''' This is the child process to schedule'''

    concurrent_queue = Scheduler(queue_name=queue_name+'_concurrent', connection=Redis())
    # this scheduler is created based on a queue_name that will be passed in
    for task in range(args.n_tasks):
        scheduler_concurrent.schedule(
            scheduled_time=datetime.utcnow(),
            func=app.controller.fetch_twitter_tweets,
            args=[args.statuses_backfill, fill_start_time])

serial_queue = Scheduler(queue_name='myqueue', connection=Redis())
serial_queue.schedule(
'''This is the first schedule.'''
   scheduled_time=datetime.utcnow(),
   func=schedule_twitter_jobs.schedule_fetch_tweets,
   #now we have a fully-qualified reference to the function we need to schedule.
   args=(args, ttl, timeout, queue_name)
   #pass through the args to the child schedule
   )

Upvotes: 2

Related Questions