File: //opt/af/src/database/operations.py
import logging
import pandas as pd
from .connection import get_mysql_connection
# Logging konfigurieren (übernimmt Konfiguration aus dem Hauptskript oder Standard)
log = logging.getLogger(__name__)
def execute_select_query(query: str, params: tuple = None, return_type: str = 'dataframe'):
"""
Führt eine SELECT-SQL-Abfrage sicher aus und gibt die Ergebnisse zurück.
Args:
query (str): Die SQL-SELECT-Abfrage.
params (tuple, optional): Parameter für die Abfrage zur Vermeidung von SQL-Injection.
Standardmäßig None.
return_type (str): Der gewünschte Rückgabetyp. 'dataframe' oder 'dict'.
Standardmäßig 'dataframe'.
Returns:
pandas.DataFrame oder list[dict] oder None: Die Abfrageergebnisse im gewünschten Format
oder None bei einem Fehler.
"""
conn = None
cursor = None
try:
conn = get_mysql_connection()
if not conn or not conn.is_connected():
log.error("Keine gültige MySQL-Verbindung vorhanden.")
return None
# Verwende dictionary=True, um Ergebnisse als Dictionaries zu erhalten
cursor = conn.cursor(dictionary=True)
log.info(f"Führe Abfrage aus: {query}" + (f" mit Parametern: {params}" if params else ""))
cursor.execute(query, params)
results = cursor.fetchall()
log.info(f"{len(results)} Zeilen von der Abfrage zurückgegeben.")
if not results:
log.info("Abfrage lieferte keine Ergebnisse.")
# Leeren DataFrame oder leere Liste zurückgeben, je nach Anforderung
if return_type == 'dataframe':
# Spaltennamen extrahieren, auch wenn keine Zeilen vorhanden sind
column_names = [desc[0] for desc in cursor.description] if cursor.description else []
return pd.DataFrame([], columns=column_names)
else:
return []
if return_type == 'dataframe':
# Konvertiere die Liste von Dictionaries in einen Pandas DataFrame
df = pd.DataFrame(results)
log.info(f"Ergebnisse als DataFrame zurückgegeben mit Spalten: {list(df.columns)}")
return df
elif return_type == 'dict':
log.info("Ergebnisse als Liste von Dictionaries zurückgegeben.")
return results
else:
log.warning(f"Ungültiger return_type '{return_type}'. Gebe Liste von Dictionaries zurück.")
return results
except Exception as e:
log.error(f"Fehler beim Ausführen der SELECT-Abfrage: {e}", exc_info=True)
# Überlege, ob die Exception weitergegeben werden soll oder ob None ausreicht
return None
finally:
if cursor:
cursor.close()
if conn and conn.is_connected():
conn.close()
log.info("MySQL-Verbindung für Abfrage geschlossen.")
def get_tables() -> list[str] | None:
"""
Ruft eine Liste aller Tabellennamen aus der konfigurierten Datenbank ab.
Returns:
list[str] | None: Eine Liste der Tabellennamen oder None bei einem Fehler.
"""
conn = None
cursor = None
try:
conn = get_mysql_connection()
if not conn or not conn.is_connected():
log.error("Keine gültige MySQL-Verbindung für get_tables vorhanden.")
return None
cursor = conn.cursor()
log.info("Führe 'SHOW TABLES' aus...")
cursor.execute("SHOW TABLES")
tables = [table[0] for table in cursor.fetchall()]
log.info(f"Gefundene Tabellen: {tables}")
return tables
except Exception as e:
log.error(f"Fehler beim Abrufen der Tabellenliste: {e}", exc_info=True)
return None
finally:
if cursor:
cursor.close()
if conn and conn.is_connected():
conn.close()
log.info("MySQL-Verbindung für get_tables geschlossen.")
def get_table_schema(table_name: str) -> dict[str, str] | None:
"""
Ruft das Schema (Spaltennamen und Datentypen) für eine bestimmte Tabelle ab.
Verwendet DESCRIBE zur Sicherheit, um nur Schema-Infos zu erhalten.
Args:
table_name (str): Der Name der Tabelle.
Returns:
dict[str, str] | None: Ein Dictionary {spaltenname: datentyp} oder None bei Fehler.
"""
conn = None
cursor = None
# Sicherheit: Verhindere einfache Injections im Tabellennamen (rudimentär)
# Eine robustere Lösung würde eine Allow-List oder komplexere Validierung erfordern.
if not table_name.isalnum() and '_' not in table_name:
log.error(f"Ungültiger Tabellenname für Schema-Abfrage: {table_name}")
return None
try:
conn = get_mysql_connection()
if not conn or not conn.is_connected():
log.error(f"Keine gültige MySQL-Verbindung für get_table_schema({table_name}) vorhanden.")
return None
# Verwende dictionary=True für einfacheren Zugriff auf Spaltennamen
cursor = conn.cursor(dictionary=True)
# Verwende DESCRIBE statt SELECT * FROM ... LIMIT 0, da es sicherer ist
# und direkt die Schema-Informationen liefert. Parameterisierung ist hier nicht
# direkt möglich für Tabellennamen, daher der Check oben.
query = f"DESCRIBE `{table_name}`" # Backticks für mögliche reservierte Wörter
log.info(f"Führe Schema-Abfrage aus: {query}")
cursor.execute(query)
schema_info = cursor.fetchall()
if not schema_info:
log.warning(f"Keine Schema-Informationen für Tabelle '{table_name}' gefunden (existiert sie?).")
return None # Oder leeres Dict {}? None signalisiert eher einen Fehler/Nichtexistenz
schema = {row['Field']: row['Type'] for row in schema_info}
log.info(f"Schema für Tabelle '{table_name}': {schema}")
return schema
except Exception as e:
# Speziell den Fehler abfangen, wenn die Tabelle nicht existiert
if "1146" in str(e): # MySQL error code for "Table doesn't exist"
log.warning(f"Tabelle '{table_name}' nicht gefunden bei Schema-Abfrage.")
return None
log.error(f"Fehler beim Abrufen des Schemas für Tabelle '{table_name}': {e}", exc_info=True)
return None
finally:
if cursor:
cursor.close()
if conn and conn.is_connected():
conn.close()
log.info(f"MySQL-Verbindung für get_table_schema({table_name}) geschlossen.")
# Beispielhafte Verwendung (kann für Tests aktiviert werden)
# if __name__ == '__main__':
# from dotenv import load_dotenv
# import os
# # Stelle sicher, dass .env geladen ist
# project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))
# load_dotenv(dotenv_path=os.path.join(project_root, '.env'))
# logging.basicConfig(level=logging.INFO)
#
# # Testabfrage (passe sie an dein Schema an)
# # test_query = "SELECT * FROM your_table_name LIMIT 10"
# # results_df = execute_select_query(test_query, return_type='dataframe')
# # if results_df is not None:
# # print("Ergebnisse als DataFrame:")
# # print(results_df)
# #
# # results_dict = execute_select_query(test_query, return_type='dict')
# # if results_dict is not None:
# # print("\nErgebnisse als Liste von Dictionaries:")
# # print(results_dict)
#
# # Beispiel mit Parametern (sicherer)
# # user_id = 123
# # query_with_params = "SELECT user_name, email FROM users WHERE id = %s"
# # user_data = execute_select_query(query_with_params, params=(user_id,), return_type='dict')
# # if user_data:
# # print(f"\nBenutzerdaten für ID {user_id}:")
# # print(user_data)
#
# # Teste die neuen Funktionen
# print("\nTeste get_tables():")
# all_tables = get_tables()
# if all_tables is not None:
# print(f"Gefundene Tabellen: {all_tables}")
# if all_tables:
# test_table = all_tables[0] # Nimm die erste gefundene Tabelle zum Testen
# print(f"\nTeste get_table_schema für Tabelle: {test_table}")
# schema = get_table_schema(test_table)
# if schema is not None:
# print(f"Schema: {schema}")
# else:
# print("Konnte Schema nicht abrufen.")
# else:
# print("Konnte Tabellenliste nicht abrufen.")