Python / rq - monitoring worker status

Question

If this is an idiotic question, I apologize and will go hide my head in shame, but:

I'm using rq to queue jobs in Python. I want it to work like this:

1. Job A starts. Job A grabs data via web API and stores it.
2. Job A runs.
3. Job A completes.
4. Upon completion of A, job B starts. Job B checks each record stored by job A and adds some additional response data.
5. Upon completion of job B, user gets a happy e-mail saying their report's ready.

My code so far:

redis_conn = Redis()
use_connection(redis_conn)
q = Queue('normal', connection=redis_conn) # this is terrible, I know - fixing later
w = Worker(q)
job = q.enqueue(getlinksmod.lsGet, theURL,total,domainid)
w.work()

I assumed my best solution was to have 2 workers, one for job A and one for B. The job B worker could monitor job A and, when job A was done, get started on job B.

What I can't figure out to save my life is how I get one worker to monitor the status of another. I can grab the job ID from job A with job.id. I can grab the worker name with w.name. But haven't the foggiest as to how I pass any of that information to the other worker.

Or, is there a much simpler way to do this that I'm totally missing?

Answer 1

Update januari 2015, this pull request is now merged, and the parameter is renamed to depends_on, ie:

second_job = q.enqueue(email_customer, depends_on=first_job)

The original post left intact for people running older versions and such:

I have submitted a pull request (https://github.com/nvie/rq/pull/207) to handle job dependencies in RQ. When this pull request gets merged in, you'll be able to do:

def generate_report():
    pass

def email_customer():
    pass

first_job = q.enqueue(generate_report)
second_job = q.enqueue(email_customer, after=first_job)
# In the second enqueue call, job is created,
# but only moved into queue after first_job finishes

For now, I suggest writing a wrapper function to sequentially run your jobs. For example:

def generate_report():
     pass

def email_customer():
    pass

def generate_report_and_email():
    generate_report()
    email_customer() # You can also enqueue this function, if you really want to

# Somewhere else
q.enqueue(generate_report_and_email)

Answer 2

From this page on the rq docs, it looks like each job object has a result attribute, callable by job.result, which you can check. If the job hasn't finished, it'll be None, but if you ensure that your job returns some value (even just "Done"), then you can have your other worker check the result of the first job and then begin working only when job.result has a value, meaning the first worker was completed.