48 lines
1.8 KiB
Python
48 lines
1.8 KiB
Python
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()
|
|
|