Python+MySQL数据库初始化调用结构是否符合最佳实践？

1. 游标被覆盖时是否自动清理资源？ #

不会自动清理。mysql.connector的Cursor对象被变量覆盖后，Python垃圾回收机制不会立即触发其销毁逻辑，可能导致数据库游标资源被占用，甚至引发连接泄漏。因此必须显式调用cursor.close()关闭游标，更推荐用上下文管理器自动管理资源，避免手动操作遗漏。

2. 代码优化建议 #

（1）用上下文管理器自动管理连接和游标 #

with语句会在代码块结束后自动调用close()，避免资源泄漏，代码更简洁可靠：

def initialize(config):
    db_config = config["DATABASE"]["CONNECTION"].copy()
    db_name = config["DATABASE"]["NAME"]
    
    try:
        # 上下文管理器自动管理连接
        with mysql.connector.connect(**db_config) as connection:
            # 上下文管理器自动管理游标
            with connection.cursor() as cursor:
                # 后续数据库操作逻辑
    except mysql.connector.Error as err:
        print(err)
        # 建议抛自定义异常替代直接exit，让调用者决定后续处理
        raise DatabaseInitError(f"初始化失败: {err}")

（2）安全处理SQL标识符，避免拼接风险 #

直接用format拼接数据库名、表名存在注入风险，可使用官方提供的标识符转义方法：

from mysql.connector.conversion import MySQLConverter

# 转义数据库名，防止注入
escaped_db_name = MySQLConverter.escape_identifier(connection, db_name)
cursor.execute(f"CREATE DATABASE {escaped_db_name} DEFAULT CHARACTER SET 'utf8mb4'")

（3）优化数据库存在性检查逻辑 #

用INFORMATION_SCHEMA查询替代USE语句捕获异常，逻辑更清晰可靠：

cursor.execute("SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA WHERE SCHEMA_NAME = %s", (db_name,))
if not cursor.fetchone():
    # 数据库不存在，执行创建逻辑
    cursor.execute(f"CREATE DATABASE {escaped_db_name} DEFAULT CHARACTER SET 'utf8mb4'")
    connection.database = db_name

（4）提取重复逻辑，减少代码冗余 #

读取SQL文件并执行的逻辑重复出现，封装成单独函数：

def execute_sql_file(cursor, connection, file_path):
    with open(file_path, "r", encoding="utf-8") as f:
        sql_content = f.read()
        # 更严谨拆分SQL（避免拆分字符串中的分号）
        import re
        sql_statements = re.split(r';\s*(?=(?:[^"]*"[^"]*")*[^"]*$)', sql_content)
        
        for stmt in sql_statements:
            stmt = stmt.strip()
            if stmt:
                cursor.execute(stmt)
    # 批量提交，提升性能
    connection.commit()

（5）异常处理更灵活，避免直接终止程序 #

直接exit(1)会终止整个程序，不利于代码复用，建议抛出自定义异常：

class DatabaseInitError(Exception):
    pass

# 替换原exit(1)逻辑
raise DatabaseInitError(f"创建数据库失败: {err}")

（6）优化空表检查逻辑 #

用COUNT(*)替代rowcount判断，逻辑更直观：

cursor.execute("SELECT COUNT(*) FROM account")
count = cursor.fetchone()[0]
if count == 0:
    # 填充测试数据
    execute_sql_file(cursor, connection, config["DATABASE"]["DATA"])

（7）避免修改传入的配置字典 #

直接修改config会影响外部配置对象，产生副作用，可返回新配置：

new_db_config = db_config.copy()
new_db_config["database"] = db_name
# 返回新配置给调用者
return new_db_config

（8）简化文件读取逻辑 #

直接用文本模式读取SQL文件，指定编码，替代二进制读取后解码：

with open(config["DATABASE"]["SCHEMA"], "r", encoding="utf-8") as f:
    sql_content = f.read()

内容的提问来源于stack exchange，提问作者mr.b