Crear Pool de Conexiones con Psycopg2 y Python

Definición de Pool de Conexiones:

Es un conjunto limitado de conexiones hacia una base de datos, estas conexiones son reutilizables por los usuarios, este es manejado por un servidor de aplicaciones.

Aplicación:

• Primero crearemos un archivo para la clase conexión , aquí pondremos todo lo referente a la conexión a la base de datos

conexion.py

• Del módulo de psycopg2 importar pool

from psycopg2 import pool

• Crear una clase Conexion, ya con los parámetros como HOST, USERNAME, PASSWORD, etc. Todos los parámetros como privados

class Conexion:
    __DATABASE = 'test_db'
    __USERNAME = 'postgres'
    __PASSWORD = 'admin'
    __DB_PORT = '5432'
    __HOST = '127.0.0.1'
    __MIN_CON = 1
    __MAX_CON = 5
    __pool = None**

• Luego crear una classmethod, obtenerPool(), para el pool de conexiones:

#Obteniendo varias coexiones a la base de datos con pool
@classmethod
def obtenerPool(cls):
    if cls.__pool == None:
        try:
            cls.__pool = pool.SimpleConnectionPool(
                                        cls.__MIN_CON,
                                        cls.__MAX_CON,
                                        host=cls.__HOST,
                                        user=cls.__USERNAME,
                                        password=cls.__PASSWORD,
                                        port=cls.__DB_PORT,
                                        database=cls.__DATABASE)
            logger.debug(f'Creacion pool exitosa: {cls.__pool}')
            return cls.__pool
        except Exception as e:
            logger.error(f'Errror al crear el pool de conexiones: {e}')
            sys.exit()
    else:
        return cls.__pool

• Luego tenemos que crear una classmethod, obtenerConexion(), para obtener del pool una conexión:

#Obteniendo la conexion del pool
@classmethod
def obtenerConexion(cls):
    conexion = cls.obtenerPool().getconn()
    logger.debug(f'Conexión obtenida del pool: {conexion}')
    return conexion

• Creamos otra classmethod, liberarConexion(), para retornar la conexión al pool

#Regresar el objeto conexion al pool
@classmethod
def liberarConexion(cls, conexion):
    cls.obtenerPool().putconn(conexion)

• Utilizaremos otro classmethod, cerrarConexiones(), para cerrar las conexiones

#Cerrando todas las conexiones del pool
@classmethod
def cerrarConexiones(cls):
    cls.obtenerPool().closeall()
    logger.debug(f'Cerramos todas las conexiones del pool: {cls.__pool}')

• Aquí un ejemplo de como utilizar los métodos de clase

if __name__ == "__main__":
    #Obtener una conexion a partir del pool
    conexion1 = Conexion.obtenerConexion()
    conexion2 = Conexion.obtenerConexion()
    #Regresamos las conexiones al pool
    Conexion.liberarConexion(conexion1)
    Conexion.liberarConexion(conexion2)
    #Cerramos el pool
    Conexion.cerrarConexiones()
    #error el pool ya esta cerrado
    #conexion3 = Conexion.obtenerConexion()