MySQL动态列名中的特殊字符处理与最佳实践：以[]为例

from sqlalchemy import create_engine, text
from sqlalchemy.exc import ProgrammingError
from sqlalchemy.dialects import mysql

# 模拟数据库连接对象，以便在没有实际数据库连接的情况下演示错误和解决方案
class MockConnection:
    def execute(self, statement):
        # 编译SQL语句以获取其字符串形式
        sql = statement.compile(dialect=self.dialect).string
        print(f"Executing SQL: {sql}")

        # 模拟MySQL在未转义特殊字符时抛出ProgrammingError
        # 这里简化判断逻辑：如果SQL中包含`[`但没有反引号包裹，则模拟报错
        if '[' in sql and '`' not in sql:
            raise ProgrammingError(
                "(pymysql.err.ProgrammingError) (1064, 'You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near \''Camera[1]' VARCHAR(100)\' at line 1')"
            )
        print("SQL executed successfully.")

    @property
    def dialect(self):
        return mysql.dialect()

# 实例化模拟连接
connection = MockConnection()

# --- 原始错误演示 ---
gear_type_bad = "camera[1]"
print(f"
--- 尝试添加列: {gear_type_bad} (未转义) ---")
try:
    connection.execute(text(f"ALTER TABLE shotlist ADD COLUMN {gear_type_bad} VARCHAR(100)"))
except ProgrammingError as e:
    print(f"捕获到错误: {e}")
    print("错误原因：列名中包含特殊字符`[]`，且未进行转义，导致MySQL语法错误。")

# --- 正确的转义方式演示 ---
gear_type_good = "camera[1]"
print(f"
--- 尝试添加列: {gear_type_good} (使用反引号转义) ---")
try:
    connection.execute(text(f"ALTER TABLE shotlist ADD COLUMN `{gear_type_good}` VARCHAR(100)"))
    print("使用反引号转义后，SQL语句成功执行。")
except ProgrammingError as e:
    # 如果转义正确，这里不应该捕获到错误
    print(f"捕获到意外错误: {e}")

# --- 动态生成多个类似列（使用反引号转义） ---
print("
--- 动态添加多个列 (使用反引号转义) ---")
for i in range(1, 3):
    dynamic_gear_type = f"camera[{i}]"
    print(f"
动态添加列: {dynamic_gear_type}")
    connection.execute(text(f"ALTER TABLE shotlist ADD COLUMN `{dynamic_gear_type}` VARCHAR(100)"))

from sqlalchemy import create_engine, text
from sqlalchemy.dialects import mysql

# 模拟数据库连接对象
class MockConnection:
    def execute(self, statement):
        sql = statement.compile(dialect=self.dialect).string
        print(f"Executing SQL: {sql}")
        print("SQL executed successfully.")

    @property
    def dialect(self):
        return mysql.dialect()

connection = MockConnection()

print("
--- 采用更规范的命名约定 ---")
for i in range(1, 3):
    # 使用下划线代替方括号，生成安全的列名
    dynamic_gear_type_safe = f"camera_{i}"
    print(f"
动态添加列 (安全命名): {dynamic_gear_type_safe}")
    # 注意：这里不需要反引号，因为列名不包含特殊字符
    connection.execute(text(f"ALTER TABLE shotlist ADD COLUMN {dynamic_gear_type_safe} VARCHAR(100)"))