62 Zeilen
1.8 KiB
Python
62 Zeilen
1.8 KiB
Python
"""
|
|
Database connection helper for License Server database
|
|
"""
|
|
import psycopg2
|
|
from psycopg2.extras import RealDictCursor
|
|
import logging
|
|
from contextlib import contextmanager
|
|
|
|
# License Server DB configuration
|
|
LICENSE_DB_CONFIG = {
|
|
'host': 'db', # Same container name as in docker network
|
|
'port': 5432,
|
|
'database': 'meinedatenbank', # License Server database name
|
|
'user': 'adminuser',
|
|
'password': 'supergeheimespasswort'
|
|
}
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def get_license_db_connection():
|
|
"""Get a connection to the license server database"""
|
|
try:
|
|
conn = psycopg2.connect(**LICENSE_DB_CONFIG)
|
|
return conn
|
|
except Exception as e:
|
|
logger.error(f"Failed to connect to license server database: {str(e)}")
|
|
raise
|
|
|
|
@contextmanager
|
|
def get_license_db_cursor(dict_cursor=False):
|
|
"""Context manager for license server database cursor"""
|
|
conn = None
|
|
cur = None
|
|
try:
|
|
conn = get_license_db_connection()
|
|
cursor_factory = RealDictCursor if dict_cursor else None
|
|
cur = conn.cursor(cursor_factory=cursor_factory)
|
|
yield cur
|
|
conn.commit()
|
|
except Exception as e:
|
|
if conn:
|
|
conn.rollback()
|
|
logger.error(f"License DB error: {str(e)}")
|
|
raise
|
|
finally:
|
|
if cur:
|
|
cur.close()
|
|
if conn:
|
|
conn.close()
|
|
|
|
def test_license_db_connection():
|
|
"""Test the connection to license server database"""
|
|
try:
|
|
with get_license_db_cursor() as cur:
|
|
cur.execute("SELECT 1")
|
|
result = cur.fetchone()
|
|
if result:
|
|
logger.info("Successfully connected to license server database")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to test license server database connection: {str(e)}")
|
|
return False |