DataOps-platform/app/services/db_healthcheck.py

from flask import current_app
from sqlalchemy import create_engine
from sqlalchemy.exc import OperationalError
import logging

# Set up logger
logger = logging.getLogger(__name__)

def check_database_connection():
    """检查数据库连接状态"""
    try:
        engine = create_engine(current_app.config['SQLALCHEMY_DATABASE_URI'])
        connection = engine.connect()
        connection.close()
        return True
    except OperationalError as e:
        logger.error(f"数据库连接失败: {str(e)}")
        return False

def check_db_connection():
    try:
        with db.engine.connect() as conn:
            conn.execute(text("SELECT 1"))
            return True
    except OperationalError as e:
        logger.error(f"数据库连接失败: {str(e)}")
        return False