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

Question

我可以使用此脚本从我的 jupyter notebook 成功连接到 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

# ...

with engine.begin() as conn:    
    # set up test environment
    conn.execute(sa.text("CREATE TABLE #table_name_Y (code int, report_date date)"))
    conn.execute(sa.text("INSERT INTO #table_name_Y (code, report_date) VALUES (5, '2019-11-15')"))
    # verify test environment
    result = conn.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}
    conn.execute(sql, params)
    
    # verify outcome
    result = conn.execute(sa.text("SELECT * FROM #table_name_Y")).fetchall()
    print(result)  # []

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

Problem converting varchar to datetime with SQLAlchemy execute

python

sql

sql-server

sqlalchemy

pyodbc