databases_tool.py

• Pfad: /var/www/mcp-servers/mcp-db/tools/databases_tool.py
• Namespace: -
• Zeilen: 97 | Größe: 3,172 Bytes
• Geändert: 2025-12-28 13:23:39 | Gescannt: 2025-12-31 10:22:15

Code Hygiene Score: 90

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

Keine Issues gefunden.

Dependencies 9

• use time
• use datetime.datetime
• use shared.constants.LOG_ENTRY_MAX_LENGTH
• use shared.constants.LOG_QUERY_MAX_LENGTH
• use shared.constants.MS_PER_SECOND
• use config.Config
• use infrastructure.db_connection.DatabaseConnection
• use shared.domain.LogEntry
• use shared.infrastructure.get_logger

Funktionen 1

• register_databases_tool() Zeile 14

Code

"""Databases Tool - Listet verfuegbare Datenbanken auf."""

import time
from datetime import datetime

from shared.constants import LOG_ENTRY_MAX_LENGTH, LOG_QUERY_MAX_LENGTH, MS_PER_SECOND

from config import Config
from infrastructure.db_connection import DatabaseConnection
from shared.domain import LogEntry
from shared.infrastructure import get_logger


def register_databases_tool(mcp) -> None:
    """Registriert db_databases Tool."""
    logger = get_logger("mcp-db", Config)

    @mcp.tool()
    def db_databases() -> dict:
        """
        Listet alle verfuegbaren Datenbanken auf.

        Returns:
            Dict mit databases (Liste der erlaubten DBs mit Tabellen-Count)
        """
        start = time.time()

        try:
            # Verbinde zu einer erlaubten DB um Queries auszufuehren
            with DatabaseConnection.get_connection("ki_dev") as conn:
                cursor = conn.cursor(dictionary=True, buffered=True)

                databases = []
                for db_name in Config.ALLOWED_DATABASES:
                    cursor.execute(
                        """SELECT COUNT(*) as table_count
                           FROM information_schema.tables
                           WHERE table_schema = %s""",
                        (db_name,),
                    )
                    row = cursor.fetchone()
                    table_count = row["table_count"] if row else 0

                    databases.append({
                        "name": db_name,
                        "tables_count": table_count,
                    })

                cursor.close()

                duration = int((time.time() - start) * MS_PER_SECOND)

                # Log
                try:
                    logger.log(
                        LogEntry(
                            timestamp=datetime.now(),
                            client_name="mcp-db",
                            tool_name="db_databases",
                            request="SHOW DATABASES (filtered)",
                            status="success",
                            duration_ms=duration,
                        )
                    )
                except Exception:
                    pass

                return {
                    "status": "success",
                    "databases": databases,
                    "execution_ms": duration,
                }

        except Exception as e:
            duration = int((time.time() - start) * MS_PER_SECOND)

            try:
                logger.log(
                    LogEntry(
                        timestamp=datetime.now(),
                        client_name="mcp-db",
                        tool_name="db_databases",
                        request="SHOW DATABASES",
                        status="error",
                        duration_ms=duration,
                        error_message=str(e)[:LOG_ENTRY_MAX_LENGTH],
                    )
                )
            except Exception:
                pass

            return {
                "status": "error",
                "error": str(e)[:LOG_QUERY_MAX_LENGTH],
                "execution_ms": duration,
            }