extraccion/lib/database_operations.py

import psycopg2
from psycopg2.extras import DictCursor

def connect_to_database(dbname="sgi", user="postgres", password="h3rcul3s#$", host="200.13.89.8", port="5432"):
    """
    Establece una conexión a la base de datos y la retorna.
    """
    try:
        connection = psycopg2.connect(
            dbname=dbname,
            user=user,
            password=password,
            host=host,
            port=port
        )
        return connection
    except psycopg2.Error as e:
        print(f"No se pudo conectar a la base de datos: {e}")
        exit()

def query_all(sql):
    with connect_to_database() as conn:
        with conn.cursor(cursor_factory=DictCursor) as cur:
            cur.execute(sql)
            return cur.fetchall()

def query_single(sql):
    with connect_to_database() as conn:
        with conn.cursor(cursor_factory=DictCursor) as cur:
            cur.execute(sql)
            return cur.fetchone()  # Returns a dictionary-like object

def execute_query(sql):
    with connect_to_database() as conn:
        with conn.cursor() as cur:
            cur.execute(sql)
            conn.commit()  # Commit to save the insert operation

def log(message, status, error_message=None, clave=None, no_insertadas=[]):
    with connect_to_database(dbname="adcfi", user="postgres", password="Ultr4p0d3r0s0##", host="200.13.89.42", port="5432") as conn:
        with conn.cursor() as cur:
            cur.execute(f"""
                        INSERT INTO calificaciones.calificaciones_log (response_status, error_message, user_ip, additional_info)
                        VALUES ({status}, {f"'{error_message}'" if error_message else 'NULL'}, '200.13.89.42', '{{"clave": "{clave}", "message": "{message}", "no insertadas": "[{', '.join(no_insertada for no_insertada in no_insertadas)}]"}}')
                        """)
            conn.commit()