[Python] SQL Alchemy and Big Query not Respecting Atomicity of Transactions

Code


#!/usr/bin/env python3
# encoding: utf-8
import polars as pl
from sqlalchemy import Table, MetaData, func
from sqlalchemy.engine import create_engine
from sqlalchemy.engine.base import Engine
from sqlalchemy.orm import Query, Session
...
frame=pl.DataFrame(data={'data':range(10)})
OUT_TABLE_NAME:str='transaction_test'

with EngineContext(uri=CONNECTION) as engine:
out_table: Table = Table(OUT_TABLE_NAME, MetaData(bind=engine, schema=SCHEMA_NAME), autoload=True)
with Session(bind=engine) as session, session.begin():
for item in frame.to_dicts():
# Insert values from 0 to 9 sequentially
print(item['data'])
if item['data']==5:
# This to ensure premature termination
# and making sure the session is rolled back
raise ValueError
statement=out_table.insert(values=[item])
print(statement)
session.execute(statement=statement)
sleep(1)


Description


Insert a sequence of records (each containing an integer) into big query one by one.

Before


Before I run the above code, assume the table (in Google Bigquery) is empty.

After

Expected Outcome


The table remains empty, as the session should be rolled back by upon the error.

Observed Outcome


The table contains values from 0 to 4, both inclusive.

Pip Libraries


sqlalchemy==1.4.51 # ORM
sqlalchemy-bigquery==1.9.0 # Dialect for bigquery, need latest Ubuntu version
polars==1.2.1 # Tabular data manipulation
google-cloud-bigquery-storage==2.25.0 # Bigquery storage write client used for inserts

Technology Stack

• Python 3.8
• Ubuntu 20.04

Continue reading...