Un pool de connexion fonctionne bien pour ce genre de chose. Je n'ai pas travaillé avec en production (en utilisant principalement Django ou SQLAlchemy), mais psycopg2.pool inclut quelques implémentations différentes (SimpleConnectionPool ou PersistentConnectionPool ) qui répondrait probablement à vos besoins. De manière générale, un pool permet non seulement de gérer les connexions en tant que ressource partagée, mais également de tester et de réinitialiser la connexion lorsque cela est nécessaire.
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)
Le putconn est extrêmement important, afin qu'une exception ne quitte pas le pool en pensant que la connexion est toujours utilisée. Ce serait bien de le gérer en tant que gestionnaire de contexte :
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)
J'espère que ça aide.
