RQ – Simple Job Queues for Python
Reposting from a while back in case it solved a problem for someone.<p>I use Postgres SKIP LOCKED as a queue. Postgres gives me everything I want. I can also do priority queueing and sorting.<p>All the other queueing mechanisms I investigated were dramatically more complex and heavyweight than Postgres SKIP LOCKED.<p>Here is a complete implementation - nothing needed but Postgres, Python and psycopg2 driver:<p><pre><code>    import psycopg2
    import psycopg2.extras
    import random

    db_params = {
        &#x27;database&#x27;: &#x27;jobs&#x27;,
        &#x27;user&#x27;: &#x27;jobsuser&#x27;,
        &#x27;password&#x27;: &#x27;superSecret&#x27;,
        &#x27;host&#x27;: &#x27;127.0.0.1&#x27;,
        &#x27;port&#x27;: &#x27;5432&#x27;,
    }
    
    conn = psycopg2.connect(**db_params)
    cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
    
    def do_some_work(job_data):
        if random.choice([True, False]):
            print(&#x27;do_some_work FAILED&#x27;)
            raise Exception
        else:
            print(&#x27;do_some_work SUCCESS&#x27;)
    
    def process_job():
    
        sql = &quot;&quot;&quot;DELETE FROM message_queue 
    WHERE id = (
      SELECT id
      FROM message_queue
      WHERE status = &#x27;new&#x27;
      ORDER BY created ASC 
      FOR UPDATE SKIP LOCKED
      LIMIT 1
    )
    RETURNING *;
    &quot;&quot;&quot;
        cur.execute(sql)
        queue_item = cur.fetchone()
        print(&#x27;message_queue says to process job id: &#x27;, queue_item[&#x27;target_id&#x27;])
        sql = &quot;&quot;&quot;SELECT * FROM jobs WHERE id =%s AND status=&#x27;new_waiting&#x27; AND attempts &lt;= 3 FOR UPDATE;&quot;&quot;&quot;
        cur.execute(sql, (queue_item[&#x27;target_id&#x27;],))
        job_data = cur.fetchone()
        if job_data:
            try:
                do_some_work(job_data)
                sql = &quot;&quot;&quot;UPDATE jobs SET status = &#x27;complete&#x27; WHERE id =%s;&quot;&quot;&quot;
                cur.execute(sql, (queue_item[&#x27;target_id&#x27;],))
            except Exception as e:
                sql = &quot;&quot;&quot;UPDATE jobs SET status = &#x27;failed&#x27;, attempts = attempts + 1 WHERE id =%s;&quot;&quot;&quot;
                # if we want the job to run again, insert a new item to the message queue with this job id
                cur.execute(sql, (queue_item[&#x27;target_id&#x27;],))
        else:
            print(&#x27;no job found, did not get job id: &#x27;, queue_item[&#x27;target_id&#x27;])
        conn.commit()
    
    process_job()
    cur.close()
    conn.close()</code></pre>