95 lines
3.5 KiB
Python
95 lines
3.5 KiB
Python
import psycopg2
|
|
from psycopg2 import sql
|
|
from psycopg2.extras import execute_values
|
|
|
|
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 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 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}") |