使用Pandas高效更新SQL表列数据教程

import pandas as pd
import pyodbc as odbc

# 数据库连接字符串，请根据实际情况替换
# 例如：'DRIVER={ODBC Driver 17 for SQL Server};SERVER=your_server;DATABASE=your_database;UID=your_user;PWD=your_password'
connection_string = "<your_connection_stuff>"
sql_conn = odbc.connect(connection_string)

# 1. 从数据库读取数据到DataFrame
query = "SELECT id, myColumn FROM myTable" # 确保查询包含主键列 (id)
df = pd.read_sql(query, sql_conn)

# 2. 在DataFrame中更新目标列
# 假设我们有一个新的值列表，长度与DataFrame行数相同
myNewValueList = [11, 12, 13, 14, 15, 16, 17, 18, 19, 20] # 示例值，实际应根据业务逻辑生成
# 确保 myNewValueList 的长度与 df 的行数匹配
if len(myNewValueList) != len(df):
    raise ValueError("新值列表的长度必须与DataFrame的行数匹配")
df['myColumn'] = myNewValueList

# 3. 逐行更新数据库
cursor = sql_conn.cursor()
# SQL UPDATE 语句，使用问号 (?) 作为参数占位符
# 必须包含 WHERE 子句和主键，以确保只更新当前行
update_sql = "UPDATE myTable SET myColumn = ? WHERE id = ?"

try:
    for index, row in df.iterrows():
        # 执行更新操作，参数顺序与 SQL 语句中的占位符顺序一致
        cursor.execute(update_sql, (row['myColumn'], row['id']))

    # 提交事务以保存更改
    sql_conn.commit()
    print("数据库逐行更新成功！")

except Exception as e:
    sql_conn.rollback() # 发生错误时回滚事务
    print(f"数据库更新失败: {e}")

finally:
    # 关闭游标和连接
    cursor.close()
    sql_conn.close()

import pandas as pd
import pyodbc as odbc
from sqlalchemy import create_engine, text

# 数据库连接字符串，请根据实际情况替换
# 对于SQLAlchemy，连接字符串格式通常为：
# 'mssql+pyodbc://<username>:<password>@<server>/<database>?driver=ODBC+Driver+17+for+SQL+Server'
# 或 'sqlite:///your_database.db' 等
sqlalchemy_connection_string = "mssql+pyodbc://<your_connection_stuff_for_sqlalchemy>"
engine = create_engine(sqlalchemy_connection_string)

# 也可以使用 pyodbc 进行初始数据读取，如果已有的连接方式更方便
pyodbc_connection_string = "<your_connection_stuff_for_pyodbc>"
sql_conn = odbc.connect(pyodbc_connection_string)

# 1. 从数据库读取数据到DataFrame
query = "SELECT id, myColumn FROM myTable" # 确保查询包含主键列 (id)
df = pd.read_sql(query, sql_conn)
sql_conn.close() # 读取完毕后可以关闭 pyodbc 连接

# 2. 在DataFrame中更新目标列
myNewValueList = [11, 12, 13, 14, 15, 16, 17, 18, 19, 20] # 示例值
if len(myNewValueList) != len(df):
    raise ValueError("新值列表的长度必须与DataFrame的行数匹配")
df['myColumn_new_values'] = myNewValueList # 使用一个新列名来存储更新后的值

# 定义临时表名
temp_table_name = 'temp_myTable_update_data'

try:
    # 3. 将修改后的DataFrame写入临时表
    # if_exists='replace' 会在每次运行时重新创建表
    df.to_sql(temp_table_name, engine, if_exists='replace', index=False)
    print(f"DataFrame成功写入临时表 '{temp_table_name}'。")

    # 4. 执行SQL查询，通过JOIN临时表来更新原始表
    with engine.connect() as conn:
        # 使用 f-string 构造 UPDATE 语句，注意 SQL 注入风险，这里假设表名和列名是受控的
        # 假设 'id' 是主键列，用于连接原始表和临时表
        update_query = text(f"""
        UPDATE myTable
        SET myColumn = temp.myColumn_new_values
        FROM myTable
        INNER JOIN {temp_table_name} AS temp
        ON myTable.id = temp.id;
        """)
        conn.execute(update_query)
        conn.commit() # 提交更新操作
        print("数据库批量更新成功！")

        # 5. 删除临时表
        drop_table_query = text(f"DROP TABLE {temp_table_name};")
        conn.execute(drop_table_query)
        conn.commit() # 提交删除操作
        print(f"临时表 '{temp_table_name}' 已删除。")

except Exception as e:
    print(f"数据库批量更新失败: {e}")
    # 尝试删除可能残留的临时表
    try:
        with engine.connect() as conn:
            conn.execute(text(f"DROP TABLE IF EXISTS {temp_table_name};"))
            conn.commit()
            print(f"发生错误时，尝试删除临时表 '{temp_table_name}'。")
    except Exception as cleanup_e:
        print(f"清理临时表失败: {cleanup_e}")

finally:
    engine.dispose() # 关闭 SQLAlchemy 引擎连接池