pyodbc.Connection无executemany属性报错，实现API数据导入SQL表

错误根源 #

你混淆了pyodbc的Connection连接对象和Cursor游标对象：
pyodbc.connect()返回的是连接对象，本身没有executemany方法，该方法属于连接对象创建的游标对象，你直接把连接对象赋值给了self.cursor变量，调用方法时就触发了属性不存在报错。 #

修复步骤（完全保留原有代码结构） #

1. 调整DatabaseClient类的连接初始化逻辑 #

同时存储连接对象和游标对象，方便后续事务管理：

import pyodbc # 记得提前导入pyodbc依赖

class DatabaseClient:
    def __init__(self):
        self.connect()
    
    def connect(self):
        server = "sql-test-server-vdzbi.database.windows.net"
        database = "sql-test-database-vdzbi"
        username = "vdzbi-admin"
        password = "VD9ffqjHaugvTf2"
        # 先存储连接对象
        self.conn = pyodbc.connect("DRIVER={SQL Server};SERVER="+server+";DATABASE="+database+";UID="+username+";PWD="+password)
        # 通过连接对象创建真正的游标对象
        self.cursor = self.conn.cursor()

2. 修正write_frequency_data方法的执行逻辑 #

补全事务处理、异常回滚逻辑，修正SQL执行方式：

def write_frequency_data(self, data):
        self.cursor.fast_executemany = True # 数据量较大时可开启提升插入性能
        
        stage_query = "INSERT INTO staging.Frequency (FreqId, Title) VALUES (?,?)"
        merge_query = """INSERT INTO mykiosk.Frequency(FreqId, Title) 
            SELECT FreqId, Title
            FROM staging.Frequency stage 
            WHERE stage.FreqId not in (Select FreqId from mykiosk.Frequency)"""
        truncate_query = "TRUNCATE TABLE staging.Frequency"
            
        try:
            # 先清空staging表旧数据
            self.cursor.execute(truncate_query)
            # 批量插入新数据到staging表，只有带占位符的批量操作才需要用executemany
            self.cursor.executemany(stage_query, data)
            # merge语句没有动态参数，不需要executemany，直接execute即可
            self.cursor.execute(merge_query)
            # 提交事务，不提交数据不会真正写入数据库
            self.conn.commit()
        except Exception as e:
            # 异常时回滚事务，避免产生脏数据
            self.conn.rollback()
            print(f"数据写入失败：{e}")
        finally:
            # 操作完成后关闭游标和连接，释放资源
            self.cursor.close()
            self.conn.close()

其他说明 #

现有api.py的逻辑无需修改，返回的元组列表格式完全适配批量插入的参数要求。生产环境建议把数据库账号密码存入环境变量，不要硬编码在代码中避免信息泄露。

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