Agent/mysql_test.py

from __future__ import annotations

from typing import Any, Dict, List

from dotenv import find_dotenv, load_dotenv

from examples_how.db_utils.errors import MySQLBaseException
from examples_how.db_utils.mysql_db import MySQLDB
from examples_how.db_utils.mysql_manager import MySQLClientManager


def _print_rows_preview(title: str, rows: List[Dict[str, Any]], preview: int = 5) -> None:
    print(f"\n=== {title} ===")
    print(f"total: {len(rows)}")
    for i, r in enumerate(rows[:preview], start=1):
        print(f"[{i}] {r}")
    if len(rows) > preview:
        print(f"... (show first {preview} rows)")


def main() -> None:
    dotenv_path = find_dotenv()
    if dotenv_path:
        load_dotenv(dotenv_path)

    # 从 .env 的 MYSQL_SOURCES_INFO 自动构建多 source 的 manager
    manager = MySQLClientManager.from_env_sources_info(
        env_var="MYSQL_SOURCES_INFO", dotenv_path=dotenv_path, allow_fallback_single_source=False
    )

    db_default = MySQLDB(manager=manager, source="default")
    db_pattern = MySQLDB(manager=manager, source="pattern")

    # 1) default: post 前 10 条
    post_rows = db_default.select("post", columns="*", limit=10)
    _print_rows_preview("default.post (first 10)", post_rows, preview=10)

    # 2) pattern: topic_pattern_mining_config 全部
    configs_rows = db_pattern.select("topic_pattern_mining_config", columns="*")
    _print_rows_preview(
        "pattern.topic_pattern_mining_config (all, preview)",
        configs_rows,
        preview=5,
    )

    # 3) pattern: topic_build_log 一条
    build_log_row = db_pattern.select_one("topic_build_log", columns="*", where="")
    print("\n=== pattern.topic_build_log (one) ===")
    print(build_log_row)


if __name__ == "__main__":
    try:
        main()
    except MySQLBaseException as e:
        print(f"MySQL 操作失败: {e.message}")
        if getattr(e, "original_error", None):
            print(f"original_error: {e.original_error}")
    except Exception as e:
        print(f"脚本运行失败: {e}")