基于Telethon的Telegram机器人SQLite参数化查询问题排查

我正在开发一个基于Telethon的Telegram聊天机器人，支持以/search thompson格式根据姓氏查询客户数据库，但在使用参数化查询保障安全时遇到问题。

遇到两类错误：

• 直接访问query时出现tuple index out of range错误；
• 使用join方法时出现Incorrect number of bindings supplied. The current statement uses 1, and there are 8 supplied.错误。

完整错误回溯 #

Traceback (most recent call last):
  File "/Users/.../.../.../.../script.py", line 155, in select
    test_message = create_message_select_query(res)
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/.../.../.../.../script.py", line 115, in create_message_select_query
    lname = i[1]
            ~^^^
IndexError: tuple index out of range

相关代码 #

# 创建匹配结果消息
def create_message_select_query(ans):
    text = ""
    for i in ans:
        id = i[0]
        lname = i[1]
        fname = i[2]
        creation_date = i[3]
        text += "<b>"+ str(id) +"</b> | " + "<b>"+ str(lname) +"</b> | " + "<b>"+ str(fname)+"</b> | " + "<b>"+ str(creation_date)+"</b>\n"
    message = "客户信息查询结果:\n\n"+text
    return message

@client.on(events.NewMessage(pattern="(?i)/search"))
async def select(event):
    try:
        # 获取消息发送者
        sender = await event.get_sender()
        SENDER = sender.id
        
        # 提取/search命令后的查询文本
        list_of_words = event.message.text.split(" ")
        # 获取第一个查询词
        query = list_of_words[1]
             
        sql = "select (SELECT * from customers where lname = ?)"

        # args = ''.join((query))
        args = query
        
        cursor = conn.execute(sql, (args,))
        
        res = cursor.fetchall() # 获取所有查询结果
        
        # 如果有匹配结果，生成消息发送
        if(res):
            test_message = create_message_select_query(res) 
            await client.send_message(SENDER, test_message, parse_mode='html')
        # 无匹配结果时提示
        else:
            text = "未找到匹配的客户"
            await client.send_message(SENDER, text, parse_mode='html')

1. SQL语句嵌套导致结果格式异常 #

你的SQL语句用了嵌套查询：select (SELECT * from customers where lname = ?)，这种写法会把内部查询的整行结果作为单个字段返回。比如原本customers表的一行是(id, lname, fname, creation_date)，现在会变成((id, lname, fname, creation_date),)这样的嵌套元组——外层元组只有1个元素，所以你在create_message_select_query里访问i[1]时必然触发索引越界。

修复：把SQL语句改成正常格式：

sql = "SELECT * from customers where lname = ?"

2. 未处理用户输入不完整的情况 #

当用户只发送/search而没有输入姓氏时，list_of_words[1]会直接抛出IndexError，你的代码没有做任何边界处理。

修复：添加输入长度检查：

list_of_words = event.message.text.split(" ")
if len(list_of_words) < 2:
    await client.send_message(SENDER, "请输入要查询的姓氏，格式：/search 姓氏", parse_mode='html')
    return
query = list_of_words[1]

3. join方法导致的参数绑定错误 #

你之前尝试的args = ''.join((query))是错误的：这个写法会把字符串query拆成单个字符的元组（比如"thompson"变成('t','h','o','m','p','s','o','n')），而SQL语句只需要1个参数，因此出现绑定数量不匹配的错误。这个写法完全没必要，直接用query作为参数即可，就像你现在的args = query这样。

4. 保持参数化查询的正确姿势 #

当前的conn.execute(sql, (args,))是正确的参数化查询写法，能有效防止SQL注入，要继续保持这种方式，不要直接把query拼接到SQL语句中。

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