使用SQLAlchemy execu将varchar转换为datetime时出现问题

Question

我可以使用以下脚本从jupyter笔记本成功连接到SQL Server Management Studio：

from sqlalchemy import create_engine
import pyodbc 
import csv
import time
import urllib

params = urllib.parse.quote_plus('''DRIVER={SQL Server Native Client 11.0};
                                    SERVER=SV;
                                    DATABASE=DB;
                                    TRUSTED_CONNECTION=YES;''')

engine = create_engine("mssql+pyodbc:///?odbc_connect=%s" % params)

举个例子，下面的脚本非常有效：

engine.execute("delete from table_name_X")

但是，我没能让下面的脚本工作。作为参考，当我在SQL Server Management Studio中执行其自适应时，它可以工作：

cde = 5
reportDate = df.loc[df.index[0],'Report Date'] # when you execute reportDate it returns 2019-11-15 00:00:00

req = "DELETE table_name_Y "
req+= "WHERE code = " + str(cde)
req+= " AND report_date = '" + str(reportDate.strftime('%Y-%m-%d')) + "'"

engine.execute(req)

根据错误消息，varchar到datetime的转换有问题，这导致值超出范围。但是，独立执行的脚本str(reportDate.strftime('%Y-%m-%d'))可以工作。你知道吗

你能帮我理解为什么之前的脚本不起作用吗？你知道吗

Answer 1

正如@Ilja在对问题的评论中提到的，您确实不应该使用动态SQL来构造语句。它容易出错并且可能不安全。如果您使用适当的参数化查询许多问题就会消失。你知道吗

值得一提的是，这对我很有用：

import pandas as pd
import sqlalchemy as sa

# ...

# set up test environment
engine.execute(sa.text("CREATE TABLE #table_name_Y (code int, report_date date)"))
engine.execute(sa.text("INSERT INTO #table_name_Y (code, report_date) VALUES (5, '2019-11-15')"))
# verify test environment
result = engine.execute(sa.text("SELECT * FROM #table_name_Y")).fetchall()
print(result)  # [(5, datetime.date(2019, 11, 15))]

# test code
df = pd.DataFrame([(5, datetime.datetime(2019, 11, 15),), ], columns=['code', 'Report Date'])
cde = int(df.loc[df.index[0],'code'])
print(type(cde))  # <class 'int'>
reportDate = df.loc[df.index[0],'Report Date']
print(type(reportDate))  # <class 'pandas._libs.tslibs.timestamps.Timestamp'>
sql = sa.text("DELETE FROM #table_name_Y WHERE code = :p0 AND report_date = :p1")
params = {'p0': cde, 'p1': reportDate}
engine.execute(sql, params)

# verify outcome
result = engine.execute(sa.text("SELECT * FROM #table_name_Y")).fetchall()
print(result)  # []