76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
from app.services.database_service import DatabaseService
|
|
import os
|
|
|
|
class DatabaseManager:
|
|
"""Manager für verschiedene Datenbankverbindungen"""
|
|
|
|
def __init__(self):
|
|
self.connections = {}
|
|
self._load_default_connections()
|
|
|
|
def _load_default_connections(self):
|
|
"""Lade Standard-Datenbankverbindungen"""
|
|
|
|
# Oracle-Datenbank (wenn konfiguriert)
|
|
oracle_host = os.environ.get('ORACLE_HOST')
|
|
if oracle_host:
|
|
self.connections['oracle'] = {
|
|
'name': 'Oracle Database',
|
|
'type': 'oracle',
|
|
'description': 'Oracle-Produktionsdatenbank',
|
|
'host': oracle_host,
|
|
'port': int(os.environ.get('ORACLE_PORT', 1521)),
|
|
'service_name': os.environ.get('ORACLE_SERVICE_NAME', 'ORCL'),
|
|
'username': os.environ.get('ORACLE_USERNAME'),
|
|
'password': os.environ.get('ORACLE_PASSWORD')
|
|
}
|
|
|
|
# PostgreSQL (wenn konfiguriert)
|
|
postgres_host = os.environ.get('POSTGRES_HOST')
|
|
if postgres_host:
|
|
self.connections['postgres'] = {
|
|
'name': 'PostgreSQL Database',
|
|
'type': 'postgresql',
|
|
'description': 'PostgreSQL-Datenbank',
|
|
'host': postgres_host,
|
|
'port': int(os.environ.get('POSTGRES_PORT', 5432)),
|
|
'database': os.environ.get('POSTGRES_DATABASE', 'postgres'),
|
|
'username': os.environ.get('POSTGRES_USERNAME'),
|
|
'password': os.environ.get('POSTGRES_PASSWORD')
|
|
}
|
|
|
|
def get_database_service(self, connection_name='demo'):
|
|
"""Erstelle DatabaseService für spezifische Verbindung"""
|
|
if connection_name not in self.connections:
|
|
raise ValueError(f"Datenbankverbindung '{connection_name}' nicht gefunden")
|
|
|
|
config = self.connections[connection_name]
|
|
|
|
# Nur Oracle-Verbindungen unterstützt
|
|
return DatabaseService(config)
|
|
|
|
def get_available_connections(self):
|
|
"""Hole alle verfügbaren Datenbankverbindungen"""
|
|
return [
|
|
{
|
|
'key': key,
|
|
'name': config['name'],
|
|
'type': config['type'],
|
|
'description': config['description']
|
|
}
|
|
for key, config in self.connections.items()
|
|
]
|
|
|
|
def test_connection(self, connection_name):
|
|
"""Teste eine Datenbankverbindung"""
|
|
try:
|
|
db_service = self.get_database_service(connection_name)
|
|
|
|
# Oracle Test-Query
|
|
test_query = "SELECT 1 FROM DUAL"
|
|
|
|
result = db_service.execute_query(test_query)
|
|
return {'success': True, 'message': 'Verbindung erfolgreich'}
|
|
|
|
except Exception as e:
|
|
return {'success': False, 'message': str(e)} |