Python中MySQL客户端类最佳实践及数据库创建问题咨询

核心问题解决：自动创建不存在的目标数据库 #

你当前的代码仅在数据库不存在时打印提示，未实现自动创建并重新连接的逻辑。可以修改__create_connection方法，将「尝试连接目标库→不存在则创建→重新连接」的流程封装在内部，避免重复编写冗余代码。

优化后的完整代码 #

import mysql.connector
from mysql.connector import errorcode, InterfaceError

class DBClient:
    def __init__(self, db_name=None):
        self.conn = None
        self.db_name = db_name
        # 提取连接参数为实例属性，方便后续维护
        self.host = "127.0.0.1"
        self.port = 3306
        self.user = "root"
        self.password = "password"
        self.__create_connection()

    def __create_connection(self):
        try:
            # 优先尝试直接连接目标数据库
            self.conn = mysql.connector.connect(
                host=self.host,
                port=self.port,
                user=self.user,
                password=self.password,
                database=self.db_name
            )
            print(f'成功连接到数据库 {self.db_name}')
        except InterfaceError as e:
            print(f'网络连接失败：{e}')
            exit(1)
        except mysql.connector.Error as err:
            if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
                print("用户名或密码错误")
                exit(1)
            elif err.errno == errorcode.ER_BAD_DB_ERROR:
                print(f'数据库 {self.db_name} 不存在，正在自动创建...')
                # 先连接到MySQL服务（不指定具体数据库）
                temp_conn = mysql.connector.connect(
                    host=self.host,
                    port=self.port,
                    user=self.user,
                    password=self.password
                )
                cursor = temp_conn.cursor()
                # 使用IF NOT EXISTS避免重复创建报错
                cursor.execute(f"CREATE DATABASE IF NOT EXISTS {self.db_name}")
                temp_conn.close()
                # 递归调用自身，重新连接目标数据库
                self.__create_connection()
            else:
                print(f'数据库错误：{err}')
                exit(1)

    def __del__(self):
        # 关闭连接前先检查连接状态，避免无效操作报错
        if self.conn and self.conn.is_connected():
            self.conn.close()
            print('数据库连接已关闭')

    def __commit(self):
        if self.conn.is_connected():
            self.conn.commit()
            print('事务已提交')

    # 示例：创建商品表的方法
    def create_items_table(self):
        if not self.conn.is_connected():
            print('未建立有效数据库连接，无法创建表')
            return
        cursor = self.conn.cursor()
        create_table_sql = """
        CREATE TABLE IF NOT EXISTS items (
            id INT AUTO_INCREMENT PRIMARY KEY,
            name VARCHAR(255) NOT NULL,
            price DECIMAL(10,2) NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """
        cursor.execute(create_table_sql)
        self.__commit()
        cursor.close()
        print('items表创建成功（若不存在）')

关键优化点说明 #

• 封装自动创建逻辑：将「检查数据库存在性→创建→重连」的流程完全封装在__create_connection内部，外部调用无需额外处理，避免代码重复。
• 参数解耦：把host、port等连接参数提取为实例属性，后续可通过配置文件注入，提升代码可维护性。
• 健壮性增强：创建数据库和表时使用IF NOT EXISTS，避免重复操作报错；关闭连接、提交事务前先检查连接状态，防止无效操作引发异常。
• 错误处理明确：针对不同错误类型给出精准提示，严重错误直接终止程序，避免无效运行。

使用示例 #

# 假设config.DB_NAME为你的目标数据库名称
DBClient(config.DB_NAME).create_items_table()

内容的提问来源于stack exchange，提问作者majesty