DataOps-platform/deployment/app/services/db_healthcheck.py

import logging

from sqlalchemy import text
from sqlalchemy.exc import OperationalError

from app import db

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


def check_database_connection():
    """检查数据库连接状态"""
    try:
        with db.engine.connect() as connection:
            connection.execute(text("SELECT 1"))
        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