当使用格式化语句将数据读入sqlite数据库时，我可以生成看起来正确的语句，但它不会执行

Question

这是一个 Web 应用程序，我打算作为 cs50 在线编程课程的最终项目提交，使用：

• 蟒蛇
• sqlite
• 烧瓶

该项目的目标是为我的教师提供即时评分功能（我教授一门不相关的课程）：教师登录应用程序，选择他们想要评分的学生团队，在该团队并有一个用于提交成绩的表格。输入和输出是 csv 电子表格。

为了使实现更容易（我希望），我使用该类提供的 SQL 对象： https://github.com/cs50/python-cs50/blob/main/src/cs50/sql.py 这个 SQL 对象的本质是它不会直接与 pandas 交互； dataframe.to_sql 将不会运行。检查源代码，似乎 SQL 对象中不存在必要的 sql 功能。 因此，我手动迭代 csv 的每一行。

每节课的成绩生成四栏：出勤、准备、参与、结果；列名称添加到实验室会话之前（例如 attendence_lab_1）。教职员工提交成绩后，数据将读回临时 csv 文件，然后使用原始 csv 的名称重命名该临时文件，并覆盖它。 对于下一次评分会话，我通过读取 csv 来重新创建数据库。 这意味着我无法对列标题进行硬编码；我需要从 csv 中读取标题，从中生成一个列表，然后从该标题列表中创建一个 INSET INTO 语句。 使用 Python sqlite3, create table columns from a list with help of .format 作为模型，我已经能够将标题放入表中。 但是当我迭代行时我收到错误：

• 语句必须用换行符或分号分隔
• 字符串文字未终止

我认为一般意义上我理解这些类型的错误是什么，但我不明白我是如何生成它们的。

代码：

# https://pandas.pydata.org/docs/reference/io.html
df = pd.read_csv(stpath, header=0)
#df.to_sql(students, con)
headers = df.columns.values.tolist() # draw list of headers from df

# create table via prepared statement that scales with number of columns in table
def createTableStatement(tableName, columnList):
    return f"CREATE TABLE IF NOT EXISTS {tableName} (id INTEGER, {columnList[0]}" + (", {}"*(len(columnList)-1)).format(*(columnList[1:])) + ", PRIMARY KEY(id))"
db.execute(createTableStatement("students", headers))

# read data into table
def createRowStatement(columnList):
    return f'"INSERT INTO students ({columnList[0]}' + (', {}'*(len(columnList)-1)).format(*(columnList[1:])) + ') VALUES (?' + (', ?'*(len(columnList)-1)) + ')", row["Student"]' + (', row["{}"]'*(len(columnList)-1)).format(*(columnList[1:])) + ')'


with open(stpath, "r", encoding='utf-8-sig') as file:
    reader = csv.DictReader(file)
    for row in reader:
        x = createRowStatement(headers)
        print(x)
        db.execute(x)

代码的早期版本在 INSERT INTO 语句中用

{columnList[0]}

替换了

row["Student"]'

。这导致生成的 sql 查询：

"INSERT INTO students (Student, Team, Session, Pod, Image, Lab1_Grade, Lab1_attend, Lab1_prep, Lab1_part, Lab1_disect) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", {columnList[0]}, row["Team"], row["Session"], row["Pod"], row["Image"], row["Lab1_Grade"], row["Lab1_attend"], row["Lab1_prep"], row["Lab1_part"], row["Lab1_disect"])

将

{columnList[0]}

更改为

row["Student"]'

生成了适当的标题列表：

"INSERT INTO students (Student, Team, Session, Pod, Image, Lab1_Grade, Lab1_attend, Lab1_prep, Lab1_part, Lab1_disect) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", row["Student"], row["Team"], row["Session"], row["Pod"], row["Image"], row["Lab1_Grade"], row["Lab1_attend"], row["Lab1_prep"], row["Lab1_part"], row["Lab1_disect"])

我怀疑该错误与此行为有某种关系 - 我是否缺少引号？ 是不是我投入太多了？ 这行代码是否太长了（我应该如何分解它？）？

Answer 1

F-strings 和 string.format() 通常不应该用于生成 SQL 语句，因为如果存在一些格式错误或恶意的列标题或数据，它可能会让您遭受 SQL 注入攻击。 虽然您（目前）信任输入的来源，但应该避免这种情况。 两者的结合也使得这些语句非常难以解析和调试。 但是，可以使用 f 字符串或字符串连接来插入正确数量的

?

占位符。

我在调试代码时发现的另一个问题是

createRowStatement

函数与

"

和

'

字符不匹配，这可能导致您的“字符串文字未终止”错误。

我还怀疑，如果您的任何数据在标题或值中包含空格，则可能会出现问题，因为您构建语句的方式没有将任何插入的值用引号引起来。

这是一个可能的解决方案。请注意，我在

createRowStatement

函数中添加了表名的占位符，因此它可以更加通用。 此外，通过使用占位符创建语句，现在可以在循环之前调用一次，而不必为每一行生成新语句。

# https://pandas.pydata.org/docs/reference/io.html
df = pd.read_csv(stpath, header=0)
#df.to_sql(students, con)
headers = df.columns.values.tolist() # draw list of headers from df

# create table via prepared statement that scales with number of columns in table
def createTableStatement(tableName, columnList):
    column_placeholders = "? TEXT, " * (len(columnList))
    statement = f"CREATE TABLE IF NOT EXISTS ? (id INTEGER, {column_placeholders} PRIMARY KEY(id))"
    return statement, (tableName, *columnList)

statement, values = createTableStatement("students", headers)   
db.execute(statement, *values)   
# Note for anyone not familiar with CS50, the SQL library being used here expects all of the values to be passed as *args, not as a tuple like some of the built-in libraries.

# read data into table
def createRowStatement(columnList)
    column_placeholders = "?" + ", ?" * (len(columnList)-1)
    statement = f"INSERT INTO ? ({column_placeholders}) VALUES ({column_placeholders})"
    return statement


with open(stpath, "r", encoding='utf-8-sig') as file:
    reader = csv.DictReader(file)
    statement = createRowStatement(headers)
    for row in reader:
        db.execute(statement, "students", *row)

这些版本的输出示例：

tableName="Students"
headers = ["First Name", "Last Name", "Grade", "Age"]

statement, values = createTableStatement("students", headers)   
# statement = "CREATE TABLE IF NOT EXISTS ? (id INTEGER, ? TEXT, ? TEXT, ? TEXT, ? TEXT, PRIMARY KEY(id))"
# values = ('Students', 'First Name', 'Last Name', 'Grade', 'Age')

statement = createRowStatement(headers)
# statement = "INSERT INTO ? (?, ?, ?, ?) VALUES (?, ?, ?, ?)"