在 Python 中捕获 MySQL 数据库连接错误，通常涉及使用 `try-except` 语句来捕获连接过程中可能抛出的异常。无论是使用 `mysql-connector-python` 还是 `PyMySQL`，都可以通过特定的异常类来捕获连接错误，以便进行错误处理和日志记录。 ### 使用 PyMySQL 捕获连接错误 在 `PyMySQL` 中，连接错误通常会抛出 `pymysql.Error` 或其子类异常（如 `pymysql.OperationalError`、`pymysql.DatabaseError` 等）。可以通过捕获这些异常来处理连接失败的情况。 ```python import pymysql try: conn = pymysql.connect( host='localhost', user='your_username', password='your_password', database='your_database' ) print("数据库连接成功") except pymysql.Error as e: print(f"数据库连接失败: {type(e).__name__}: {e}") ``` ### 使用 mysql-connector-python 捕获连接错误 在 `mysql-connector-python` 中，连接错误通常由 `mysql.connector.Error` 或其子类（如 `mysql.connector.InterfaceError`、`mysql.connector.DatabaseError`）表示。可以通过捕获这些异常来处理连接问题。 ```python import mysql.connector from mysql.connector import Error try: conn = mysql.connector.connect( host='localhost', user='your_username', password='your_password', database='your_database' ) if conn.is_connected(): print("数据库连接成功") except Error as e: print(f"数据库连接失败: {type(e).__name__}: {e}") ``` ### 自定义数据库连接类并捕获错误 可以通过封装一个数据库连接类，并在初始化时捕获连接错误，实现更结构化的错误处理机制。以下是一个基于 `PyMySQL` 的示例： ```python import pymysql class DatabaseConnection: def __init__(self): try: self.conn = pymysql.connect( host='localhost', user='your_username', password='your_password', database='your_database' ) print("数据库连接成功") except pymysql.Error as e: print(f"数据库连接失败: {type(e).__name__}: {e}") self.conn = None def get_cursor(self): if self.conn: return self.conn.cursor() else: print("数据库未连接") return None ``` ### 连接池中的错误处理 在使用连接池（如 `pymysql` 的 `ConnectionPool` 或 `SQLAlchemy`）时，也可以通过捕获异常来处理连接失败的情况。连接池可以提高数据库连接的效率，但同样需要处理连接失败的可能性。 ```python from DBUtils.PooledDB import PooledDB import pymysql try: pool = PooledDB( creator=pymysql, host='localhost', user='your_username', password='your_password', database='your_database', maxconnections=5 ) conn = pool.connection() print("从连接池获取数据库连接成功") except pymysql.Error as e: print(f"数据库连接失败: {type(e).__name__}: {e}") ``` ### 错误日志记录与重试机制 在实际应用中，除了捕获错误并输出提示信息外，还可以将错误记录到日志文件中，并实现自动重试机制，以提高系统的健壮性。 ```python import logging import time import pymysql logging.basicConfig(filename='database_errors.log', level=logging.ERROR) for attempt in range(3): try: conn = pymysql.connect( host='localhost', user='your_username', password='your_password', database='your_database' ) print("数据库连接成功") break except pymysql.Error as e: logging.error(f"尝试连接数据库失败 (第 {attempt + 1} 次): {type(e).__name__}: {e}") print(f"尝试重新连接数据库... (第 {attempt + 1} 次)") time.sleep(2) else: print("无法连接到数据库，请检查网络或数据库配置。") ```