55 lines
1.7 KiB
Python
55 lines
1.7 KiB
Python
import os
|
|
import pymysql
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.exc import OperationalError
|
|
|
|
# ── 数据库配置 ──────────────────────────────────────────────────────────────
|
|
DB_HOST = os.getenv("DB_HOST", "mysql1.rdsmbk3ednsgnnt.rds.bj.baidubce.com")
|
|
DB_PORT = os.getenv("DB_PORT", "3306")
|
|
DB_NAME = os.getenv("DB_NAME", "case_platform")
|
|
DB_USER = os.getenv("DB_USER", "root_dev")
|
|
DB_PASS = os.getenv("DB_PASS", "Kdse89sd")
|
|
|
|
MYSQL_URL = f"mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}?charset=utf8mb4"
|
|
SQLITE_URL = "sqlite:///./quantum_test.db"
|
|
|
|
def get_engine():
|
|
try:
|
|
# 尝试连接 MySQL
|
|
engine = create_engine(
|
|
MYSQL_URL,
|
|
pool_pre_ping=True,
|
|
pool_recycle=1800,
|
|
pool_size=10,
|
|
max_overflow=20,
|
|
)
|
|
# 测试连接
|
|
with engine.connect() as conn:
|
|
pass
|
|
print("✅ 成功连接到 MySQL (百度云 RDS)")
|
|
return engine
|
|
except Exception as e:
|
|
print(f"⚠️ MySQL 连接失败 (IP白名单限制或凭证错误) - {e}")
|
|
print("🔄 自动降级：使用本地 SQLite 数据库以保证应用正常运行...")
|
|
engine = create_engine(
|
|
SQLITE_URL,
|
|
connect_args={"check_same_thread": False}
|
|
)
|
|
return engine
|
|
|
|
engine = get_engine()
|
|
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
Base = declarative_base()
|
|
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|