buffer_platform/app/mysql_database.py

"""
数据缓冲平台 - MySQL 数据库连接
"""
import logging
import urllib.parse
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from app.config import MYSQL_HOST, MYSQL_PORT, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE

logger = logging.getLogger(__name__)

mysql_engine = None
MySQLSessionLocal = None


def init_mysql():
    """初始化 MySQL 连接引擎，失败时返回 None"""
    global mysql_engine, MySQLSessionLocal
    try:
        encoded_password = urllib.parse.quote_plus(MYSQL_PASSWORD)
        url = f"mysql+pymysql://{MYSQL_USER}:{encoded_password}@{MYSQL_HOST}:{MYSQL_PORT}/{MYSQL_DATABASE}?charset=utf8mb4"
        mysql_engine = create_engine(url, pool_pre_ping=True, pool_recycle=3600)
        with mysql_engine.connect() as conn:
            conn.execute(text("SELECT 1"))
        MySQLSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=mysql_engine)
        logger.info("MySQL 连接初始化成功")
        return mysql_engine
    except Exception as e:
        logger.warning(f"MySQL 初始化失败: {e}")
        mysql_engine = None
        MySQLSessionLocal = None
        return None


def get_mysql_session():
    """获取 MySQL 会话"""
    if MySQLSessionLocal is None:
        raise RuntimeError("MySQL 未初始化")
    return MySQLSessionLocal()