db_healthcheck.py 455 B

1234567891011121314151617
  1. from sqlalchemy import text
  2. from sqlalchemy.exc import OperationalError
  3. from app.config.config import Config
  4. from app import db
  5. import logging
  6. # Set up logger
  7. logger = logging.getLogger(__name__)
  8. def check_db_connection():
  9. try:
  10. with db.engine.connect() as conn:
  11. conn.execute(text("SELECT 1"))
  12. return True
  13. except OperationalError as e:
  14. logger.error(f"数据库连接失败: {str(e)}")
  15. return False