script-build-and-framework-tmp/visualization/backend/app/runtime_bridge.py

from __future__ import annotations

import importlib
import os
import sys
from functools import lru_cache
from pathlib import Path
from typing import Any

from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker


def runtime_dir() -> Path:
    configured = os.getenv("PATTERN_RUNTIME_DIR")
    if configured:
        return Path(configured).expanduser().resolve()
    return Path(__file__).resolve().parents[3]


def database_host_override() -> str | None:
    """Return a visualization-only DB host override.

    The value is deliberately host-only: credentials, database name and driver
    continue to come from the parent runtime's existing SQLAlchemy URL.
    """
    value = os.getenv("PATTERN_DB_HOST_OVERRIDE", "").strip()
    if not value:
        return None
    if any(token in value for token in ("/", "@", "?", "#", ":")) or any(char.isspace() for char in value):
        raise ValueError("PATTERN_DB_HOST_OVERRIDE 只能是不含端口的主机名或 IPv4 地址")
    return value


def load_runtime_modules() -> tuple[Any, Any]:
    """Load only the parent's DB manager and ORM models.

    Importing pattern_service would initialize unrelated mining/config modules, so
    the visualization intentionally avoids it.
    """
    root = str(runtime_dir())
    if root not in sys.path:
        sys.path.insert(0, root)
    return importlib.import_module("db_manager"), importlib.import_module("models")


def new_session():
    """Return a session backed by a pool that is read-only at MySQL level.

    V8 does not borrow the parent application's ordinary session because that
    pool is intentionally write-capable.  The visualization always owns a
    separate pool, even when no host override is configured.
    """
    return _read_only_session_factory(
        str(runtime_dir()), database_host_override() or ""
    )()


@lru_cache(maxsize=4)
def _read_only_session_factory(runtime_path: str, host: str):
    """Build one cached visualization-only read-only connection pool."""
    del runtime_path  # Included in the cache key so runtime switches cannot reuse a stale pool.
    db_manager, _ = load_runtime_modules()
    parent_manager = db_manager.DatabaseManager()
    parent_engine = parent_manager.engine
    try:
        override_url = parent_engine.url.set(host=host) if host else parent_engine.url
    finally:
        # The parent engine is lazy and has not connected; do not retain a second pool.
        parent_engine.dispose()

    engine = create_engine(override_url, pool_pre_ping=True, pool_recycle=3600)
    event.listen(engine, "connect", _set_mysql_session_read_only)
    return sessionmaker(bind=engine, autoflush=False, autocommit=False)


def _set_mysql_session_read_only(dbapi_connection, _connection_record) -> None:
    """Make every transaction on the visualization pool read-only at MySQL level."""
    cursor = dbapi_connection.cursor()
    try:
        cursor.execute("SET SESSION TRANSACTION READ ONLY")
    finally:
        cursor.close()