SQLite使用占位符执行UPDATE语句报错：near "?"语法错误求助

问题背景 #

我尝试用SQLite占位符编写安全的UPDATE语句抵御SQL注入，但始终报错。以下是存在注入风险但能正常运行的代码：

def updateUser(self, user_id: int, **kwargs) -> bool:
        for arg, value in kwargs.items():
            try:
                sql = f"UPDATE user SET {arg}='{value}' WHERE user_id = {user_id};"
                self.con.execute(sql)
            except Exception as e:
                print(e)
                self.con.rollback()
                return False
        self.con.commit()
        return True

这种写法适配任意数据类型，但存在SQL注入隐患。我尝试的参数化写法却一直报错：

def updateUser(self, user_id: int, **kwargs) -> bool:
        for arg, value in kwargs.items():
            try:
                self.con.execute("UPDATE user SET ?='?' WHERE user_id = ?;", (arg, value, user_id))
            except Exception as e:
                print(e)
                self.con.rollback()
                return False
        self.con.commit()
        return True

执行ud.updateUser(1, nick="test")时出现错误：

>>> ud.updateUser(1, nick="test")
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<path>/inter.py", line 56, in updateUser
    self.con.execute("UPDATE user SET ?='?' WHERE user_id = ?;", (arg, value, user_id))
sqlite3.OperationalError: near "?": syntax error

错误原因 #

SQLite的参数化占位符?只能用于替换SQL语句中的值，不能用来替换列名、表名这类语法结构。你把列名nick用?占位，SQL解析时会将其视为一个普通值，导致语法错误。另外，手动给value加单引号也是错误操作——参数化查询会自动处理值的引号和转义，不需要手动添加。

正确的安全写法 #

要实现动态更新且避免注入，需要分两步处理：

1. 用白名单验证列名，只允许更新预先定义的合法列，杜绝列名注入
2. 值部分使用参数化占位符，交给SQLite处理转义，避免值注入

基础版本（循环执行单字段更新） #

def updateUser(self, user_id: int, **kwargs) -> bool:
    # 定义允许更新的列名白名单
    allowed_columns = {"nick", "email", "password", "age"}
    try:
        for arg, value in kwargs.items():
            if arg not in allowed_columns:
                raise ValueError(f"不允许更新的列: {arg}")
            # 列名经白名单验证后直接拼接，值用参数化占位符
            sql = f"UPDATE user SET {arg} = ? WHERE user_id = ?;"
            self.con.execute(sql, (value, user_id))
        self.con.commit()
        return True
    except Exception as e:
        print(e)
        self.con.rollback()
        return False

优化版本（单次执行多字段更新） #

减少数据库交互次数，更高效：

def updateUser(self, user_id: int, **kwargs) -> bool:
    allowed_columns = {"nick", "email", "password", "age"}
    # 过滤掉非法列，只保留合法的更新字段
    valid_updates = [(col, val) for col, val in kwargs.items() if col in allowed_columns]
    if not valid_updates:
        return True  # 无更新内容，直接返回成功
    
    try:
        # 动态生成SET子句，每个字段对应一个占位符
        set_clause = ", ".join([f"{col} = ?" for col, _ in valid_updates])
        sql = f"UPDATE user SET {set_clause} WHERE user_id = ?;"
        # 整理参数：所有字段值 + user_id
        params = [val for _, val in valid_updates] + [user_id]
        self.con.execute(sql, params)
        self.con.commit()
        return True
    except Exception as e:
        print(e)
        self.con.rollback()
        return False

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