import pandas as pd import sqlalchemy as sa engine = sa.create_engine("mssql+pyodbc://@mssqlLocal64") def dump_tran_test_table(conn): print(conn.execute(sa.text("SELECT * FROM tran_test")).fetchall()) # set up test environment with engine.begin() as conn: conn.exec_driver_sql("DROP TABLE IF EXISTS tran_test") conn.exec_driver_sql( "CREATE TABLE tran_test " "(txt varchar(10), id int primary key)" ) conn.exec_driver_sql( "INSERT INTO tran_test (txt, id) VALUES " "('old_foo', 1), ('old_bar', 2)" ) # test with engine.connect() as conn: tran = conn.begin() df = pd.DataFrame([("new_baz", 3)], columns=["txt", "id"]) df.to_sql("tran_test", conn, index=False, if_exists="append") dump_tran_test_table(conn) """console output: [('old_foo', 1), ('old_bar', 2), ('new_baz', 3)] """ tran.rollback() dump_tran_test_table(conn) """console output: [('old_foo', 1), ('old_bar', 2)] """
Preview:
downloadDownload PNG
downloadDownload JPEG
downloadDownload SVG
Tip: You can change the style, width & colours of the snippet with the inspect tool before clicking Download!
Click to optimize width for Twitter