db_core.py

• Pfad: /var/www/scripts/pipeline/db_core.py
• Namespace: pipeline
• Zeilen: 71 | Größe: 1,972 Bytes
• Geändert: 2025-12-28 08:55:51 | 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 3

• use mysql.connector
• use mysql.connector.Error
• use config.DB_CONFIG

Klassen 1

• DatabaseCore class Zeile 14

Code

"""
Database Core - Connection Management

Single Responsibility: Database connection lifecycle and query execution.
All other database modules (mixins) inherit from this base class.
"""

import mysql.connector
from mysql.connector import Error

from config import DB_CONFIG


class DatabaseCore:
    """Base database class with connection management.

    Provides:
    - Connection lifecycle (connect, disconnect)
    - Query execution (execute, commit)

    All database mixins expect these methods to be available via self.
    """

    def __init__(self):
        """Initialize database instance without connection."""
        self.connection = None

    def connect(self) -> bool:
        """Establish database connection.

        Returns:
            True if connection successful, False otherwise.
        """
        try:
            self.connection = mysql.connector.connect(**DB_CONFIG)
            return True
        except Error as e:
            print(f"Database connection error: {e}")
            return False

    def disconnect(self):
        """Close database connection if open."""
        if self.connection and self.connection.is_connected():
            self.connection.close()

    def execute(self, query: str, params: tuple = None):
        """Execute a query and return the cursor.

        Args:
            query: SQL query string
            params: Optional tuple of query parameters

        Returns:
            Cursor with query results (dictionary mode)
        """
        cursor = self.connection.cursor(dictionary=True)
        cursor.execute(query, params or ())
        return cursor

    def commit(self):
        """Commit the current transaction."""
        self.connection.commit()

    def is_connected(self) -> bool:
        """Check if database connection is active.

        Returns:
            True if connected, False otherwise.
        """
        return self.connection is not None and self.connection.is_connected()