DataOps-platform/app/services/db_healthcheck.py

import logging

from flask import current_app
from sqlalchemy import create_engine, text
from sqlalchemy.exc import OperationalError

from app import db

# 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