接続プールは、この種のことにはうまく機能します。私は本番環境で(主にDjangoまたはSQLAlchemyを使用して)使用していませんが、psycopg2.pool
いくつかの異なる実装が含まれています(SimpleConnectionPool
またはPersistentConnectionPool
)それはおそらくあなたのニーズに合うでしょう。一般的に、プールは共有リソースとしての接続の管理に役立つだけでなく、必要に応じて接続をテストして再初期化するのにも役立ちます。
from psycopg2 import pool
conn_pool = pool.PersistentConnectionPool(minconn, maxconn, **dbopts)
def work_method():
conn = conn_pool.getconn()
with conn.cursor() as stmt:
stmt.execute(sql)
conn_pool.putconn(conn)
putconn
は非常に重要であるため、接続がまだ使用されていると考えて例外がプールを離れることはありません。コンテキストマネージャーとして処理するとよいでしょう:
import contextlib
@contextlib.contextmanager
def get_db_connection():
conn = conn_pool.getconn()
yield conn
conn_pool.putconn(conn)
def work_method():
with get_db_connection() as conn:
with conn.cursor() as stmt:
stmt.execute(sql)
お役に立てば幸いです。