使用sqlalchemy cursor结合csv.writer如何写入表头行？

核心方案 #

利用游标自带的description属性动态提取列名作为表头，无需额外执行查询，完美适配大数据量场景。

修改后的代码示例 #

from sqlalchemy.engine import URL
from sqlalchemy import create_engine
import csv

serverName = 'FOO'
databaseName = 'BAR'

connection_string = ("Driver={SQL Server};"
            "Server=" + serverName + ";"
            "Database=" + databaseName + ";"
            "Trusted_Connection=yes;")

connection_url = URL.create("mssql+pyodbc", query={"odbc_connect":connection_string })
engine = create_engine(connection_url)
connection = engine.raw_connection()
cursor = connection.cursor()

sql = "select 'hello world' as 'a', 'goodbye world' as 'b'"

with open('foo.csv', 'w', newline='') as file:
    w = csv.writer(file, delimiter='|')
    # 执行查询
    cursor.execute(sql)
    # 从cursor.description提取列名作为表头
    headers = [col[0] for col in cursor.description]
    w.writerow(headers)
    # 逐行写入数据
    for row in cursor:
        w.writerow(row)

cursor.close()
connection.close()

关键说明 #

• cursor.description是pyodbc游标内置属性，每个元素为元组，第一个值对应列的名称，执行一次查询就能同时获取表头和数据，避免重复查询开销。
• 打开文件时添加newline=''，是csv模块官方推荐写法，可避免导出的CSV出现多余空行。
• 遍历目录下.sql文件的场景中，只需对每个查询重复「执行查询→提取表头→写入表头→写入数据」的流程即可，完全适配不同查询的动态列名。

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