Managing Transaction in Python

Summary: in this tutorial, you will learn how to use the python-oracledb API to manage transactions in Python.

Transaction management #

When you call the cursor.execute() to insert, update, or delete data from a table, Python does not automatically commit the change to the database.

To apply the change to the database, you need to call the connection.commit() method:

cursor.execute('<DML statement>')
connection.commit()Code language: Python (python)

Or to undo the change, you call the connection.rollback() method:

cursor.execute('<DML statement>')
connection.rollback()Code language: Python (python)

By default, an uncommitted transaction is rolled back if the database connection is closed.

Autocommitting mode #

The Connection object has an attribute called autocommit that allows you to commit the transaction automatically.

By default, its value is set to False. To instruct python to commit the transaction automatically, you set the value of the connection.autocommit to True as follows:

connection.autocommit = TrueCode language: Python (python)

Different from the connection.commit(), setting connection.autocommit to True does not require an additional roundtrip to the Oracle Database.

Therefore, it is more efficient when you know how to use it appropriately.

Starting a transaction explicitly #

A transaction starts implicitly. To explicitly start a transaction, you use the connection.begin() method.

A transaction can be local or global. The connection.begin() method without any parameter starts a local transaction.

Transaction example #

The following code illustrates how to manage transactions in Python:

• First, insert a row into the billing_headers table and return the generated billing_no.
• Second, insert some rows into the billing_items table.
• Third, if the two steps succeed, commit the transaction. Otherwise, roll it back.

Here is the transaction.py code:

import oracledb
from connect import connect
from datetime import datetime
import logging

def insert_billing_header(billing_header_data : tuple, cursor: oracledb.Cursor) -> int:

    sql = '''INSERT INTO billing_headers(billing_date, amount, customer_id) 
             VALUES(:billing_date, :amount, :customer_id)  
             RETURNING billing_no INTO :billing_no '''

    bind_params = {
        'billing_date': billing_header_data[0],
        'amount': billing_header_data[1],
        'customer_id': billing_header_data[2],
        'billing_no': cursor.var(oracledb.NUMBER) 
    }

    try:
        cursor.execute(sql, bind_params)
        inserted_billing_no = bind_params['billing_no'].getvalue()[0]
        return inserted_billing_no
    except oracledb.Error as e:
        logging.error(f"Error inserting billing header: {e}")
        raise 

def insert_billing_items(billing_no: int, billing_items: list, cursor: oracledb.Cursor) -> None:
   
    sql = '''INSERT INTO billing_items(billing_no, product_id, price) 
             VALUES(:billing_no, :product_id, :price)'''

    items = []
    for item in billing_items:
        items.append({
            'billing_no': billing_no,
            'product_id': item[0],
            'price': item[1]
        })

    if not items:
        logging.warning(f"No billing items provided for billing_no: {billing_no}. Skipping item insertion.")
        return

    try:
        cursor.executemany(sql, items)
    except oracledb.Error as e:
        logging.error(f"Error inserting billing items for billing_no {billing_no}: {e}")
        raise 

def insert_billing_doc(billing_header_data: tuple, billing_items: list) -> bool: 
   
    connection = None
    try:
        connection = connect()

        with connection.cursor() as cursor:
            # Insert the billing header
            billing_no = insert_billing_header(billing_header_data, cursor)

            if billing_no is None:
                # This check is mostly for robustness, insert_billing_header should re-raise on failure
                logging.error("Failed to generate billing number. Rolling back transaction.")
                connection.rollback()
                return False

            # Insert the billing items
            insert_billing_items(billing_no, billing_items, cursor)

            # Commit the transaction if all operations were successful
            connection.commit()
            logging.info(f"Billing document (billing_no: {billing_no}) and items committed successfully.")
            return True

    except oracledb.Error as e:
        logging.error(f"Transaction failed due to Oracle database error: {e}")
        if connection:
            logging.info("Attempting to roll back the transaction.")
            connection.rollback()
            logging.info("Transaction rolled back.")
        return False
    except Exception as e:
        logging.critical(f"An unexpected error occurred during billing document insertion: {e}")
        if connection:
            logging.info("Attempting to roll back the transaction.")
            connection.rollback()
            logging.info("Transaction rolled back.")
        return False
    finally:
        # Ensure the connection is closed even if an error occurs
        if connection:
            connection.close()
            logging.info("Database connection closed.")



if __name__ == '__main__':
    insert_billing_doc(
        [datetime.now(), 1, 1000],
        [(1, 200),
        (2, 300),
        (3, 500)])Code language: Python (python)

In this tutorial, you have learned how to use python-oracledb API to manage Oracle Database transactions in Python.