DEV Community

Cover image for Crear Pool de Conexiones con Psycopg2 y Python
Luis_Zapata_Yamo
Luis_Zapata_Yamo

Posted on • Edited on

Crear Pool de Conexiones con Psycopg2 y Python

#postgres #python #psycopg2

Psycopg2

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**
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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}')
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

Top comments (1)

Collapse
 
andyrcr profile image
Andy Canales Roman •

Muy gozu eres webon