Python SQL Server: Transaction

Summary: in this tutorial, you will learn how to perform an SQL Server transaction in Python.

This tutorial begins where the Deleting data from a table tutorial left off.

Setting up sample data #

Step 1. Launch the SQL Server Management Studio (SSMS) and connect to the SQL Server.

Step 2. Create a query and execute the following statements to insert data into the Books, BookAuthors, and Inventories tables:

INSERT INTO Books (Title, Publisher, ISBN, PublishedDate) VALUES
('Mastering SQL: A Comprehensive Guide', 'Tech Books Publishing', '978-1234567890', '2022-01-15'),
('The Art of Database Design', 'Expert Press', '978-0987654321', '2021-06-10'),
('SQL Queries for Mere Mortals', 'Practical SQL Publishing', '978-1122334455', '2023-03-21'),
('Advanced SQL Programming Techniques', 'Pro Code Press', '978-6677889900', '2020-09-30'),
('Database Systems: Theory and Practice', 'Academic Press', '978-5566778899', '2022-11-05');

DECLARE @BookID1 INT, @BookID2 INT, @BookID3 INT, @BookID4 INT, @BookID5 INT;

SELECT @BookID1 = BookID FROM Books WHERE ISBN = '978-1234567890';
SELECT @BookID2 = BookID FROM Books WHERE ISBN = '978-0987654321';
SELECT @BookID3 = BookID FROM Books WHERE ISBN = '978-1122334455';
SELECT @BookID4 = BookID FROM Books WHERE ISBN = '978-6677889900';
SELECT @BookID5 = BookID FROM Books WHERE ISBN = '978-5566778899';

-- Insert records into the BookAuthors table
INSERT INTO BookAuthors (BookID, AuthorID)
VALUES (@BookID1, 1),
       (@BookID2, 1),
       (@BookID3, 1),
       (@BookID4, 1),
       (@BookID5, 1);

INSERT INTO Inventories (BookID, Qty)
SELECT BookID, ABS(CHECKSUM(NEWID()) % 101) + 100
FROM Books;

Performing a transaction #

Step 1. Create a new module transaction.py within the project directory with the following code:

import logging
from connect import create_connection


def create_order(customer_id:int , book_id:int, quantity:int, price:float, order_date:str) -> bool:
    # connect to the SQL Server
    conn = create_connection()
    
    if conn is None:
        return False

    with (conn, conn.cursor() as cursor):
        try:
            # check inventory
            cursor.execute("SELECT Qty FROM Inventories WHERE BookId = %s", (book_id,))
            row = cursor.fetchone()
            if row is None or row[0] < quantity:
                raise Exception("Insufficient inventory")

            # insert into orders
            cursor.execute("INSERT INTO Orders (OrderDate, CustomerId, TotalAmount) VALUES (%s, %s, %s)", (order_date, customer_id, price*quantity))
            order_id = cursor.lastrowid

            # insert into orderDetails
            cursor.execute("INSERT INTO OrderDetails (OrderId, BookId, Quantity, Price) VALUES (%s, %s, %s, %s)", (order_id, book_id, quantity, price))

            # update inventories
            cursor.execute("UPDATE Inventories SET Qty = Qty - %s WHERE BookId = %s", (quantity, book_id))
            
            conn.commit()        
            return True
        except Exception as e:
            logging.error(f"Error creating order: {e}")
            conn.rollback()

    return False

How it works.

First, import the built-in logging module:

import logging

Second, import the create_connection function from the connect module:

from connect import create_connection

Third, define the create_order() function that creates a new order:

def create_order(customer_id:int , book_id:int, quantity:int, price:float, order_date:str) -> bool:

Fourth, connect to SQL Server by calling the create_connection() function and return 0 if the connection fails:

conn = create_connection()
if conn is None:
    return 0

Fifth, manage the Connection and Cursor objects using the with statement:

with (conn, conn.cursor() as cursor):

Sixth, check the ordered quantity with the inventory quantity of the book:

cursor.execute("SELECT Qty FROM Inventories WHERE BookId = %s", (book_id,))
row = cursor.fetchone()
if row is None or row[0] < quantity:
    raise Exception("Insufficient inventory")

Seventh, insert a new row into the Orders table, get the inserted order ID, and use it to insert a new row into the OrderDetails table:

# insert into orders
cursor.execute("INSERT INTO Orders (OrderDate, CustomerId, TotalAmount) VALUES (%s, %s, %s)", (order_date, customer_id, price*quantity))
order_id = cursor.lastrowid

# insert into orderDetails
cursor.execute("INSERT INTO OrderDetails (OrderId, BookId, Quantity, Price) VALUES (%s, %s, %s, %s)", (order_id, book_id, quantity, price))

Eighth, update the inventory by reducing the inventory quantity by the order quantity:

cursor.execute("UPDATE Inventories SET Qty = Qty - %s WHERE BookId = %s", (quantity, book_id))

Ninth, commit the transaction:

conn.commit()

Tenth, return True if the order was created successfully:

return True

Eleventh, if an error occurred during order creation, log the error details and roll back the transaction:

logging.error(f"Error creating order: {e}")
conn.rollback()

Step 2. Modify the main.py to use the create_order function:

import logging, sys
from transaction import create_order

# config logging to console
logging.basicConfig(
    stream=sys.stdout, 
    encoding='utf-8', 
    format='%(levelname)s:%(message)s',
    level=logging.DEBUG
)

# create an order
try:
    create_order(
        customer_id=10,
        book_id=1,
        quantity=15, 
        price=15.99,
        order_date='2024-07-24'
    );
except Exception as e:
    logging.error(f"Error fetching customers: {e}")

Step 3. Open your terminal and run the main.py file:

python main.py

Download the project source code #

Download the project source code

Summary #

Call the commit() method of the Connection object to commit a transaction.
Use the rollback() method of the Connection object to roll back a transaction.

Was this tutorial helpful?