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()