formha/forms_py/cls_db_usr.py

import psycopg2
from psycopg2 import sql
from psycopg2.extras import execute_values
from psycopg2.extras import RealDictCursor


class DBForma:
    def __init__(self, db_obj: dict):
        """
        Inicializa la conexión a la base de datos.

        :param db_obj: Diccionario con las credenciales de la base de datos.
        Debe contener: host, port, database, user, password
        """
        self.db_obj = db_obj

    def _get_connection(self):
        """
        Crea y retorna una nueva conexión a la base de datos.
        """
        return psycopg2.connect(
            host=self.db_obj['host'],
            port=self.db_obj['port'],
            database=self.db_obj['database'],
            user=self.db_obj['user'],
            password=self.db_obj['password']
        )

    def login(self, email: str):
        """
        Verifica las credenciales de un usuario.
        
        :param email: Email del usuario a verificar
        :return: Tupla con la contraseña si el usuario existe, None si no existe
        :raises: RuntimeError si hay algún error en la consulta
        """
        # Corrección 1: Usar parámetros correctamente para evitar SQL injection
        query = "SELECT pswd FROM users WHERE email = %s;"
        
        try:
            with self._get_connection() as conn:
                with conn.cursor() as cursor:
                    # Corrección 2: Pasar parámetros como tupla (aunque sea uno solo)
                    cursor.execute(query, (email,))
                    return cursor.fetchone()
        except Exception as e:
            raise RuntimeError(f"Error al verificar credenciales: {e}")
    
    def reset_pswd(self, email: str):
        query = "SELECT email FROM users WHERE email = %s;"
        try:
            with self._get_connection() as conn:
                with conn.cursor() as cursor:
                    cursor.execute(query, (email,))
                    return cursor.fetchone()
        except Exception as e:
            raise RuntimeError(f"Error al verificar credenciales: {e}")
        

    def get_id(self, email: str):
        query = "SELECT id FROM users WHERE email = %s;"
        try:
            with self._get_connection() as conn:
                with conn.cursor() as cursor:
                    cursor.execute(query, (email,))
                    return cursor.fetchone()
        except Exception as e:
            raise RuntimeError(f"Error al verificar credenciales: {e}")
    
    def get_data(self, query: str, data_tuple: tuple):
        try:
            with self._get_connection() as conn:
                with conn.cursor() as cursor:
                    cursor.execute(query, data_tuple)
                    result = cursor.fetchone()
                    return result
        except psycopg2.DatabaseError as e:
            conn.rollback()  # Asegura que la conexión no quede en un estado erróneo
            raise RuntimeError(f"Error al ejecutar la consulta: {e}")
        except Exception as e:
            raise RuntimeError(f"Error inesperado: {e}")
        
    def get_all_data(self, query: str, data_tuple: tuple = ()):
        try:
            with self._get_connection() as conn:
                with conn.cursor() as cursor:
                    cursor.execute(query, data_tuple)
                    result = cursor.fetchall()
                    return result
        except psycopg2.DatabaseError as e:
            conn.rollback()  # Asegura que la conexión no quede en un estado erróneo
            raise RuntimeError(f"Error al ejecutar la consulta: {e}")
        except Exception as e:
            raise RuntimeError(f"Error inesperado: {e}")

    def get_all_data_dict(self, query):
        try:
            with self._get_connection() as conn:
                with conn.cursor(cursor_factory=RealDictCursor) as cursor:
                    cursor.execute(query)
                    return cursor.fetchall()
        except psycopg2.DatabaseError as e:
            raise RuntimeError(f"Error al ejecutar la consulta: {e}")
        except Exception as e:
            raise RuntimeError(f"Error inesperado: {e}")

    def update_data(self, query: str, data_tuple: tuple) -> bool:
        """
        """
        try:
            with self._get_connection() as conn:
                with conn.cursor() as cursor:
                    cursor.execute(query, data_tuple)
                    if cursor.rowcount == 0:
                        return False
                    conn.commit()  # Confirmar explícitamente la transacción
                    return True
        except Exception as e:
            conn.rollback()  # Revertir en caso de error
            raise RuntimeError(f"Error al actualizar la contraseña: {e}")

        
    def update_pswd(self, pswd: str, email: str) -> bool:
        """
        Actualiza la contraseña de un usuario en la base de datos.
        
        Args:
            pswd: Nueva contraseña (debería estar hasheada)
            email: Email del usuario a actualizar
            
        Returns:
            bool: True si la actualización fue exitosa, False si no se actualizó ningún registro
            
        Raises:
            RuntimeError: Si ocurre un error durante la operación
        """
        query = "UPDATE users SET pswd = %s, is_pswd_reseted = true WHERE email = %s;"
        try:
            with self._get_connection() as conn:
                with conn.cursor() as cursor:
                    cursor.execute(query, (pswd, email))
                    # Verificar si realmente se actualizó algún registro
                    if cursor.rowcount == 0:
                        return False
                    conn.commit()  # Confirmar explícitamente la transacción
                    return True
        except Exception as e:
            conn.rollback()  # Revertir en caso de error
            raise RuntimeError(f"Error al actualizar la contraseña: {e}")