config.py

• Pfad: /var/www/mcp-servers/mcp-db/config.py
• Namespace: -
• Zeilen: 104 | Größe: 3,432 Bytes
• Geändert: 2025-12-28 12:21:49 | 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 5

• extends type
• use json
• use os
• use pathlib.Path
• use typing.List

Klassen 2

• _ConfigMeta class Zeile 22
• Config class Zeile 79

Funktionen 1

• _load_json_config() Zeile 11

Code

"""Zentrale Konfiguration für MCP-DB Server mit Hot-Reload aus JSON"""
import json
import os
from pathlib import Path
from typing import List

# Pfad zur JSON-Config
CONFIG_PATH = Path("/etc/mcp_db/config.json")


def _load_json_config() -> dict:
    """Lädt JSON-Config bei jedem Aufruf (Hot-Reload)."""
    if CONFIG_PATH.exists():
        try:
            with open(CONFIG_PATH, "r") as f:
                return json.load(f)
        except (json.JSONDecodeError, IOError):
            pass
    return {}


class _ConfigMeta(type):
    """Metaclass für dynamische Property-Zugriffe auf Config."""

    @property
    def ALLOWED_DATABASES(cls) -> List[str]:
        """Erlaubte Datenbanken - aus JSON geladen."""
        cfg = _load_json_config()
        return cfg.get("allowed_databases", ["ki_dev", "ki_content"])

    @property
    def ALLOWED_TABLES(cls) -> List[str]:
        """Erlaubte Tabellen - flache Liste aus allen DBs."""
        cfg = _load_json_config()
        tables_dict = cfg.get("allowed_tables", {})
        # Flatten: alle Tabellen aus allen DBs
        all_tables = []
        for db_tables in tables_dict.values():
            if isinstance(db_tables, list):
                all_tables.extend(db_tables)
        return all_tables if all_tables else [
            "mcp_log", "protokoll", "tasks", "contracts",
            "dokumentation", "prompts", "critics"
        ]

    @property
    def BLOCKED_KEYWORDS(cls) -> List[str]:
        """Blockierte SQL-Keywords."""
        cfg = _load_json_config()
        return cfg.get("blocked_keywords", [
            "DROP", "DELETE", "INSERT", "UPDATE", "TRUNCATE",
            "ALTER", "CREATE", "RENAME", "GRANT", "REVOKE",
            "LOAD_FILE", "INTO OUTFILE", "INTO DUMPFILE",
            "BENCHMARK", "SLEEP"
        ])

    @property
    def MAX_QUERY_LENGTH(cls) -> int:
        """Maximale Query-Länge."""
        cfg = _load_json_config()
        limits = cfg.get("limits", {})
        return limits.get("max_query_length", 2000)

    @property
    def MAX_ROWS(cls) -> int:
        """Maximale Ergebniszeilen."""
        cfg = _load_json_config()
        limits = cfg.get("limits", {})
        return limits.get("max_rows", 100)

    @property
    def QUERY_TIMEOUT_SEC(cls) -> int:
        """Query-Timeout in Sekunden."""
        cfg = _load_json_config()
        limits = cfg.get("limits", {})
        return limits.get("query_timeout_sec", 30)


class Config(metaclass=_ConfigMeta):
    """
    Konfiguration mit Hot-Reload aus /etc/mcp_db/config.json.

    Änderungen an der JSON-Datei werden sofort wirksam,
    ohne den MCP-Server neu starten zu müssen.

    Statische Werte aus Environment Variables:
    - DB_HOST, DB_USER, DB_PASSWORD (Query-User)
    - LOG_DB_HOST, LOG_DB_NAME, LOG_DB_USER, LOG_DB_PASSWORD (Logger)

    Dynamische Werte aus JSON:
    - ALLOWED_DATABASES, ALLOWED_TABLES, BLOCKED_KEYWORDS
    - MAX_QUERY_LENGTH, MAX_ROWS, QUERY_TIMEOUT_SEC
    """

    # Statische Werte aus Environment (nur bei Start geladen)
    DB_HOST: str = os.getenv("DB_HOST", "localhost")
    DB_USER: str = os.getenv("DB_USER", "mcp_readonly")
    DB_PASSWORD: str = os.getenv("DB_PASSWORD", "")

    LOG_DB_HOST: str = os.getenv("LOG_DB_HOST", "localhost")
    LOG_DB_NAME: str = os.getenv("LOG_DB_NAME", "ki_dev")
    LOG_DB_USER: str = os.getenv("LOG_DB_USER", "mcp_logger")
    LOG_DB_PASSWORD: str = os.getenv("LOG_DB_PASSWORD", "")