db_connection.py

• Pfad: /var/www/mcp-servers/mcp-db/infrastructure/db_connection.py
• Namespace: -
• Zeilen: 59 | Größe: 1,648 Bytes
• Geändert: 2025-12-28 13:08:27 | Gescannt: 2025-12-31 10:22:15

Code Hygiene Score: 100

• Dependencies: 100 (25%)
• LOC: 100 (20%)
• Methods: 100 (20%)
• Secrets: 100 (15%)
• Classes: 100 (10%)
• Magic Numbers: 100 (10%)

Keine Issues gefunden.

Dependencies 4

• use contextlib.contextmanager
• use mysql.connector
• use mysql.connector.pooling
• use config.Config

Klassen 1

• DatabaseConnection class Zeile 11

Code

"""Database Connection Pool"""

from contextlib import contextmanager

import mysql.connector
from mysql.connector import pooling

from config import Config


class DatabaseConnection:
    """Singleton Connection Pool - SRP: Nur Verbindungsmanagement"""

    _pool = None

    @classmethod
    def get_pool(cls):
        """Erstellt oder gibt existierenden Connection Pool zurueck"""
        if cls._pool is None:
            cls._pool = pooling.MySQLConnectionPool(
                pool_name="mcp_pool",
                pool_size=5,
                host=Config.DB_HOST,
                user=Config.DB_USER,
                password=Config.DB_PASSWORD,
                charset="utf8mb4",
                connection_timeout=10,
                autocommit=True,
            )
        return cls._pool

    @classmethod
    @contextmanager
    def get_connection(cls, database: str):
        """
        Context Manager fuer DB Connection.

        Args:
            database: Datenbankname (bereits validiert)
        """
        conn = cls.get_pool().get_connection()
        try:
            # Datenbank wechseln (USE statement statt property)
            cursor = conn.cursor()
            cursor.execute(f"USE {database}")

            # Query Timeout setzen (MariaDB verwendet max_statement_time in Sekunden)
            try:
                cursor.execute(
                    f"SET SESSION max_statement_time = {Config.QUERY_TIMEOUT_SEC}"
                )
            except Exception:
                pass  # Falls Variable nicht unterstuetzt wird
            cursor.close()

            yield conn
        finally:
            conn.close()