feat(db): add MySQL connection module

app/mysql_database.py

@@ -0,0 +1,37 @@
+"""
+数据缓冲平台 - MySQL 数据库连接
+"""
+import logging
+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:
+        url = f"mysql+pymysql://{MYSQL_USER}:{MYSQL_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()