从数据帧列更新slqalchemy orm列的正确方法是什么

Question

我已经加载了一些数据并修改了dataframe中的一列，希望更新DB以反映更改

我试过：

db.session.query(sqlTableName).update({sqlTableName.sql_col_name: pdDataframe.pd_col_name})

但这只是删除了数据库中的列（将每个值设置为默认值“0”）。我尝试了一些其他的数据格式，但没有成功。我猜我把数据类型搞混了，可能是有点怪怪的，或者你不允许直接用这样的变量来更新列

我可以做一个循环，但。。。那真的很糟糕。对不起，基本的问题是，在一个项目中断了很长时间后，我对sqlalchemy的掌握肯定已经减弱了

Answer 1

要将数据帧上载到临时表，然后执行更新，您不需要自己编写SQL，您可以让SQLAlchemy Core为您完成：

import pandas as pd
import sqlalchemy as sa


def update_table_columns_from_df(engine, df, table_name, cols_to_update):
    metadata = sa.MetaData()
    main_table = sa.Table(table_name, metadata, autoload_with=engine)
    pk_columns = [x.name for x in main_table.primary_key.columns]

    df.to_sql("temp_table", engine, index=False, if_exists="replace")

    temp_table = sa.Table("temp_table", metadata, autoload_with=engine)
    with engine.begin() as conn:
        values_clause = {x: temp_table.columns[x] for x in cols_to_update}
        where_clause = sa.and_(
            main_table.columns[x] == temp_table.columns[x] for x in pk_columns
        )
        conn.execute(
            main_table.update().values(values_clause).where(where_clause)
        )
    temp_table.drop(engine)


if __name__ == "__main__":
    test_engine = sa.create_engine(
        "postgresql+psycopg2://scott:tiger@192.168.0.199/test",
        echo=True,  # (for demonstration purposes)
    )
    with test_engine.begin() as test_conn:
        test_conn.exec_driver_sql("DROP TABLE IF EXISTS main_table")
        test_conn.exec_driver_sql(
            """\
            CREATE TABLE main_table ( 
            id1 integer NOT NULL,
            id2 integer NOT NULL,
            txt1 varchar(50),
            txt2 varchar(50),
            CONSTRAINT main_table_pkey PRIMARY KEY (id1, id2)
            )
            """
        )
        test_conn.exec_driver_sql(
            """\
            INSERT INTO main_table (id1, id2, txt1, txt2)
            VALUES (1, 1, 'foo', 'x'), (1, 2, 'bar', 'y'), (1, 3, 'baz', 'z')
            """
        )

    df_updates = pd.DataFrame(
        [
            (1, 1, "new_foo", "new_x"),
            (1, 3, "new_baz", "new_z"),
        ],
        columns=["id1", "id2", "txt1", "txt2"],
    )
    update_table_columns_from_df(
        test_engine, df_updates, "main_table", ["txt1", "txt2"]
    )
    """SQL emitted:
    UPDATE main_table 
    SET txt1=temp_table.txt1, txt2=temp_table.txt2 
    FROM temp_table 
    WHERE main_table.id1 = temp_table.id1 AND main_table.id2 = temp_table.id2
    """

    df_result = pd.read_sql_query(
        "SELECT * FROM main_table ORDER BY id1, id2", test_engine
    )
    print(df_result)
    """
       id1  id2     txt1   txt2
    0    1    1  new_foo  new_x
    1    1    2      bar      y
    2    1    3  new_baz  new_z
    """