python - 从工作进程访问数据库

Question

我正在尝试使用 Python、Redis 和 PostgreSQL 放置需要访问数据库的作业。我正在做以下事情:

使用 RQ 将作业放入 Redis 队列中:

def queue_data(self, json_data):
    Queue().enqueue(process_data, json.dumps(json_data))

在 process_data ，使用 psycopg2 对 PostgreSQL 数据库执行查询:

def process_data(json_data):
    with psycopg2.connect("dbname=pgtest2db user=pgtest2user") as conn:
        with conn.cursor() as cursor:
            # I'm not actually doing this query, but you get the idea
            cursor.execute("SELECT * FROM some_table")
    conn.close()

显然这远非最佳，因为每次调用 process_data正在创建新连接。实现这一目标的最佳实践是什么？谁应该负责管理数据库连接池？

请注意，我强制自己不要使用 ORM，因为我这样做是为了自学，我想从更纯粹的角度理解这些模式。

编辑

我最终使用了这样的自定义 worker :

import os
import node

import redis
from rq import Worker, Queue, Connection
from psycopg2.pool      import ThreadedConnectionPool

listen = ['high', 'default', 'low']
redis_url = os.getenv('REDISTOGO_URL', 'redis://localhost:6379')
redis_conn = redis.from_url(redis_url)
pool = ThreadedConnectionPool(minconn = 1, maxconn = 10, dsn = "dbname=pgtest2db user=pgtest2user")

def process_data(json_data):
    dbconn = pool.getconn()
    result = perform_some_db_actions(dbconn, json_data)
    pool.putconn(dbconn)
    return result

if __name__ == '__main__':
    with Connection(redis_conn):
        print "Setting up Redis"
        worker = Worker(map(Queue, listen))
        worker.work()

Answer 1

我知道现在已经太晚了，但是每个想知道如何在 RQ 中使用 psycopg2 的人(尤其是当您收到 SSL 错误时:解密失败或记录错误 mac - 只需在工作人员内部打开连接(即任务。最好是创建单例而不是每次都打开连接。不是在 worker.work() 函数之前。