Python SQLite Database Interaction (sqlite3 module)

SQLite is a lightweight, serverless, and self-contained SQL database engine. It's an excellent choice for local data storage, embedded applications, and prototyping, as it reads and writes directly to disk files. Python's built-in sqlite3 module provides a powerful and convenient way to interact with SQLite databases without needing a separate database server. This tutorial will guide you through connecting, creating tables, performing CRUD (Create, Read, Update, Delete) operations, and managing transactions using Python.

Connecting to a database

To begin interacting with an SQLite database in Python, the first step is to establish a connection. The sqlite3.connect() function is your gateway. If the specified database file doesn't exist, sqlite3 will automatically create it for you. This function returns a Connection object, which represents your link to the database.

Example 1: Basic Database Connection

import sqlite3

# Define the database file name
database_file = 'my_first_database.db'

# Connect to the SQLite database
# If 'my_first_database.db' doesn't exist, it will be created.
try:
    conn = sqlite3.connect(database_file)
    print(f"Successfully connected to {database_file}")
except sqlite3.Error as e:
    print(f"Error connecting to database: {e}")
finally:
    # Always ensure the connection is closed
    if conn:
        conn.close()
        print("Database connection closed.")

Explanation

This beginner-friendly example demonstrates the most basic way to connect to an SQLite database. We import the sqlite3 module, define a filename for our database, and then use sqlite3.connect() to establish a connection. A try-except-finally block is used for robust error handling, catching sqlite3.Error for potential database issues, and ensuring the connection is always closed with conn.close() in the finally block, regardless of success or failure. This is a fundamental step for any Python database tutorial or SQLite connection example.

Example 2: In-Memory Database Connection

import sqlite3

# Connect to an in-memory SQLite database
# This database exists only for the duration of the script execution.
# It's useful for testing or temporary data storage.
try:
    conn = sqlite3.connect(':memory:')
    print("Successfully connected to an in-memory SQLite database.")

    # You can perform database operations here
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE temp_data (id INTEGER PRIMARY KEY, value TEXT)")
    print("Table 'temp_data' created in memory.")

except sqlite3.Error as e:
    print(f"Error connecting or creating table in memory: {e}")
finally:
    if conn:
        conn.close()
        print("In-memory database connection closed.")

Explanation

This example shows how to create an in-memory SQLite database by passing ':memory:' to sqlite3.connect(). This creates a temporary database that resides entirely in RAM, offering high performance for transient data. It's perfect for Python unit testing or scenarios where data persistence isn't required. The database and its contents are deleted as soon as the connection is closed or the script finishes.

Example 3: Checking Connection Status and Database Version

import sqlite3

database_file = 'check_db_status.db'
conn = None # Initialize conn to None

try:
    conn = sqlite3.connect(database_file)
    print(f"Connected to {database_file}")

    # Create a cursor object to execute SQL commands
    cursor = conn.cursor()

    # Execute a simple query to get the SQLite version
    cursor.execute("SELECT sqlite_version();")
    sqlite_version = cursor.fetchone()[0]
    print(f"SQLite Version: {sqlite_version}")

except sqlite3.Error as e:
    print(f"Database error: {e}")
except Exception as e:
    print(f"An unexpected error occurred: {e}")
finally:
    if conn:
        conn.close()
        print("Connection closed.")

Explanation

This example expands on connection handling by demonstrating how to check if a connection was successfully established and how to retrieve information from the database, specifically the SQLite version. It utilizes a cursor object, obtained via conn.cursor(), which is essential for executing SQL queries. We use cursor.fetchone() to retrieve a single result from the query, which is a common pattern for fetching data from SQLite in Python.

Example 4: Using a Context Manager for Connection Handling

import sqlite3

database_file = 'context_manager_db.db'

# The 'with' statement ensures the connection is properly closed
# even if errors occur. This is a Pythonic best practice.
try:
    with sqlite3.connect(database_file) as conn:
        print(f"Connected to {database_file} using a context manager.")
        cursor = conn.cursor()
        cursor.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)")
        print("Table 'users' ensured to exist (or created).")
    print("Connection automatically closed by context manager.")

except sqlite3.Error as e:
    print(f"Database operation failed: {e}")

Explanation

This example showcases the highly recommended Python with statement for SQLite connections. Using with sqlite3.connect(database_file) as conn: ensures that conn.close() is automatically called when the block is exited, even if exceptions occur. This simplifies resource management and prevents common issues like unclosed database connections. It's a cornerstone for robust Python database applications.

Example 5: Read-Only Database Connection

import sqlite3
import os

database_file = 'read_only_db.db'

# Ensure the database file exists and has some data for read-only test
if not os.path.exists(database_file):
    with sqlite3.connect(database_file) as conn:
        cursor = conn.cursor()
        cursor.execute("CREATE TABLE IF NOT EXISTS products (id INTEGER PRIMARY KEY, name TEXT)")
        cursor.execute("INSERT INTO products (name) VALUES ('Laptop'), ('Mouse')")
        conn.commit()
    print(f"Created and populated {database_file} for read-only test.")

conn = None # Initialize conn to None

try:
    # Connect in read-only mode using a URI
    conn = sqlite3.connect(f'file:{database_file}?mode=ro', uri=True)
    print(f"Connected to {database_file} in read-only mode.")

    cursor = conn.cursor()
    cursor.execute("SELECT * FROM products")
    products = cursor.fetchall()
    print("Products in database (read-only):", products)

    # Attempt to write (should raise an error)
    cursor.execute("INSERT INTO products (name) VALUES ('Keyboard')")
    conn.commit() # This commit will fail in read-only mode

except sqlite3.OperationalError as e:
    print(f"Attempted write in read-only mode: {e}")
except sqlite3.Error as e:
    print(f"General SQLite error: {e}")
finally:
    if conn:
        conn.close()
        print("Read-only connection closed.")

Explanation

This advanced example demonstrates how to establish a read-only SQLite connection using a URI. By appending ?mode=ro to the file path and setting uri=True in sqlite3.connect(), you can prevent accidental modifications to your database. This is crucial for data integrity in scenarios where an application should only view, not alter, existing data. The example explicitly shows an attempted write operation failing, highlighting the security benefit of this mode.

Creating Tables

After connecting to a database, the next logical step is to define your data structure by creating tables. In SQLite, this is done using the CREATE TABLE SQL statement, executed via the cursor object's execute() method. You define column names and their data types (e.g., INTEGER, TEXT, REAL, BLOB, NULL), and can also add constraints like PRIMARY KEY, NOT NULL, UNIQUE, and FOREIGN KEY for database schema design and data validation.

Example 1: Creating a Simple Table

import sqlite3

database_file = 'inventory.db'

with sqlite3.connect(database_file) as conn:
    cursor = conn.cursor()

    # SQL command to create a simple table named 'items'
    # IF NOT EXISTS is crucial to prevent errors if the table already exists.
    create_table_sql = """
    CREATE TABLE IF NOT EXISTS items (
        item_id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        quantity INTEGER,
        price REAL
    );
    """
    cursor.execute(create_table_sql)
    conn.commit() # Save the changes to the database
    print(f"Table 'items' created successfully in {database_file}.")

Explanation

This basic example illustrates how to create an SQLite table using a CREATE TABLE SQL statement. The IF NOT EXISTS clause is a best practice, ensuring the script won't throw an error if you run it multiple times and the table already exists. We define item_id as an INTEGER PRIMARY KEY (which is auto-incrementing by default in SQLite for INTEGER PRIMARY KEY), name as TEXT NOT NULL, and quantity and price with appropriate types. conn.commit() is vital to persist the table creation to the disk. This is a fundamental step in database initialization in Python.

Example 2: Creating a Table with Constraints (UNIQUE, DEFAULT)

import sqlite3

database_file = 'user_management.db'

with sqlite3.connect(database_file) as conn:
    cursor = conn.cursor()

    # Creating a 'users' table with more advanced constraints
    create_table_sql = """
    CREATE TABLE IF NOT EXISTS users (
        user_id INTEGER PRIMARY KEY AUTOINCREMENT,
        username TEXT NOT NULL UNIQUE,
        email TEXT UNIQUE,
        registration_date TEXT DEFAULT CURRENT_TIMESTAMP
    );
    """
    cursor.execute(create_table_sql)
    conn.commit()
    print(f"Table 'users' created successfully with constraints in {database_file}.")

Explanation

This example demonstrates adding common SQL table constraints to your schema. UNIQUE ensures that no two rows have the same value for username or email, which is vital for preventing duplicate user entries. DEFAULT CURRENT_TIMESTAMP automatically populates the registration_date with the current date and time if no value is provided during insertion, simplifying data entry. This shows good practices for defining robust database schemas.

Example 3: Creating Multiple Tables with a Foreign Key Relationship

import sqlite3

database_file = 'ecommerce.db'

with sqlite3.connect(database_file) as conn:
    cursor = conn.cursor()

    # Enable foreign key support (important for integrity)
    cursor.execute("PRAGMA foreign_keys = ON;")

    # Create 'customers' table
    create_customers_table = """
    CREATE TABLE IF NOT EXISTS customers (
        customer_id INTEGER PRIMARY KEY AUTOINCREMENT,
        first_name TEXT NOT NULL,
        last_name TEXT NOT NULL,
        email TEXT UNIQUE
    );
    """
    cursor.execute(create_customers_table)

    # Create 'orders' table with a FOREIGN KEY linking to 'customers'
    create_orders_table = """
    CREATE TABLE IF NOT EXISTS orders (
        order_id INTEGER PRIMARY KEY AUTOINCREMENT,
        customer_id INTEGER NOT NULL,
        order_date TEXT DEFAULT CURRENT_TIMESTAMP,
        total_amount REAL,
        FOREIGN KEY (customer_id) REFERENCES customers (customer_id) ON DELETE CASCADE
    );
    """
    cursor.execute(create_orders_table)
    conn.commit()
    print(f"Tables 'customers' and 'orders' created with foreign key relationship in {database_file}.")

Explanation

This is a more advanced example showcasing relational database design in SQLite. We create two tables, customers and orders, and establish a foreign key constraint between them. FOREIGN KEY (customer_id) REFERENCES customers (customer_id) ensures referential integrity, meaning an order cannot exist without a corresponding customer. ON DELETE CASCADE is an advanced option that means if a customer is deleted, all their associated orders will also be automatically deleted. Crucially, PRAGMA foreign_keys = ON; must be executed to enable foreign key enforcement in SQLite.

Example 4: Checking if a Table Exists Before Creating

import sqlite3

database_file = 'system_config.db'

with sqlite3.connect(database_file) as conn:
    cursor = conn.cursor()

    table_name = "settings"

    # Check if table exists using sqlite_master
    cursor.execute(f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table_name}';")
    existing_table = cursor.fetchone()

    if existing_table:
        print(f"Table '{table_name}' already exists.")
    else:
        create_table_sql = f"""
        CREATE TABLE {table_name} (
            setting_key TEXT PRIMARY KEY,
            setting_value TEXT
        );
        """
        cursor.execute(create_table_sql)
        conn.commit()
        print(f"Table '{table_name}' created in {database_file}.")

Explanation

While IF NOT EXISTS is convenient, this example demonstrates a more explicit way to check for table existence in SQLite. We query the sqlite_master table, which is a system table containing schema information. This approach gives you more granular control and is useful for conditional database setup or when you need to perform different actions based on a table's presence.

Example 5: Altering an Existing Table (Adding a Column)

import sqlite3

database_file = 'product_catalog.db'

with sqlite3.connect(database_file) as conn:
    cursor = conn.cursor()

    # First, ensure the 'products' table exists for alteration
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS products (
        product_id INTEGER PRIMARY KEY AUTOINCREMENT,
        product_name TEXT NOT NULL,
        category TEXT
    );
    """)
    conn.commit()
    print("Ensured 'products' table exists.")

    # Check if the 'description' column already exists before adding
    cursor.execute("PRAGMA table_info(products);")
    columns = cursor.fetchall()
    column_names = [col[1] for col in columns] # col[1] is the column name

    if 'description' not in column_names:
        # SQL command to add a new column to an existing table
        alter_table_sql = """
        ALTER TABLE products
        ADD COLUMN description TEXT;
        """
        cursor.execute(alter_table_sql)
        conn.commit()
        print("Column 'description' added to 'products' table.")
    else:
        print("Column 'description' already exists in 'products' table.")

    # You can also add a column with a default value
    if 'stock_count' not in column_names: # Re-check if column exists after previous alteration
        cursor.execute("PRAGMA table_info(products);")
        columns_after_desc = cursor.fetchall()
        column_names_after_desc = [col[1] for col in columns_after_desc]

        if 'stock_count' not in column_names_after_desc:
            alter_table_add_stock = """
            ALTER TABLE products
            ADD COLUMN stock_count INTEGER DEFAULT 0;
            """
            cursor.execute(alter_table_add_stock)
            conn.commit()
            print("Column 'stock_count' added to 'products' table with a default value.")
        else:
            print("Column 'stock_count' already exists.")

Explanation

This advanced example demonstrates modifying an existing SQLite table using the ALTER TABLE ADD COLUMN statement. It's common in database schema evolution to add new fields. The PRAGMA table_info(table_name) query is used to inspect the table's structure and check for the existence of a column before attempting to add it, preventing errors. This example also shows how to add a column with a DEFAULT value, useful for database migrations.

Inserting, Querying, Updating, Deleting Data

Once your tables are set up, you'll want to manipulate the data within them. This involves the fundamental CRUD (Create, Read, Update, Delete) operations, which correspond to SQL commands INSERT INTO, SELECT, UPDATE, and DELETE FROM. Python's sqlite3 module makes these operations straightforward using the cursor's execute() and executemany() methods.

Example 1: Inserting Single and Multiple Rows (Parameterization)

import sqlite3

database_file = 'crm_data.db'

with sqlite3.connect(database_file) as conn:
    cursor = conn.cursor()

    cursor.execute("""
    CREATE TABLE IF NOT EXISTS contacts (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE,
        phone TEXT
    );
    """)
    conn.commit()
    print("Ensured 'contacts' table exists.")

    # Insert a single row using parameterization (best practice to prevent SQL injection)
    name1 = "Alice Wonderland"
    email1 = "alice@example.com"
    phone1 = "111-222-3333"
    cursor.execute("INSERT INTO contacts (name, email, phone) VALUES (?, ?, ?)",
                   (name1, email1, phone1))
    print(f"Inserted single contact: {name1}")

    # Insert multiple rows using executemany for efficiency
    contacts_to_add = [
        ("Bob The Builder", "bob@example.com", "444-555-6666"),
        ("Charlie Chaplin", "charlie@example.com", None), # Phone can be NULL
        ("Diana Prince", "diana@example.com", "777-888-9999")
    ]
    cursor.executemany("INSERT INTO contacts (name, email, phone) VALUES (?, ?, ?)",
                       contacts_to_add)
    print(f"Inserted {len(contacts_to_add)} contacts using executemany.")

    conn.commit() # Commit all insertions
    print("All insertions committed.")

Explanation

This example covers inserting data into SQLite tables. It highlights two critical methods: execute() for single row insertions and executemany() for efficiently inserting multiple rows. Crucially, it demonstrates parameterized queries using ? placeholders. This is the most important security practice to prevent SQL injection vulnerabilities, a common attack vector in web applications. executemany() is highly optimized for bulk data insertion in Python.

Example 2: Querying Data (SELECT statements)

import sqlite3

database_file = 'crm_data.db' # Using the same database from previous insert example

with sqlite3.connect(database_file) as conn:
    cursor = conn.cursor()

    # Query all data from the 'contacts' table
    cursor.execute("SELECT id, name, email, phone FROM contacts;")
    all_contacts = cursor.fetchall()
    print("\n--- All Contacts ---")
    for contact in all_contacts:
        print(f"ID: {contact[0]}, Name: {contact[1]}, Email: {contact[2]}, Phone: {contact[3]}")

    # Query with a WHERE clause (filtering data)
    search_name = "Alice Wonderland"
    cursor.execute("SELECT * FROM contacts WHERE name = ?;", (search_name,))
    alice = cursor.fetchone()
    print(f"\n--- Contact found for '{search_name}' ---")
    if alice:
        print(f"ID: {alice[0]}, Name: {alice[1]}, Email: {alice[2]}, Phone: {alice[3]}")
    else:
        print("Contact not found.")

    # Query with ORDER BY and LIMIT (sorting and pagination)
    cursor.execute("SELECT name, email FROM contacts ORDER BY name DESC LIMIT 2;")
    limited_sorted_contacts = cursor.fetchall()
    print("\n--- Top 2 Contacts (alphabetical, descending) ---")
    for contact in limited_sorted_contacts:
        print(f"Name: {contact[0]}, Email: {contact[1]}")

    # Query with COUNT aggregate function
    cursor.execute("SELECT COUNT(*) FROM contacts;")
    count = cursor.fetchone()[0]
    print(f"\nTotal number of contacts: {count}")

Explanation

This example focuses on retrieving data from SQLite using SELECT statements. It covers various common querying patterns:

fetchall(): Retrieves all matching rows as a list of tuples.

fetchone(): Retrieves a single row as a tuple (or None if no match).

Filtering data with the WHERE clause and parameterized queries.

Sorting data using ORDER BY (ascending ASC or descending DESC).

Limiting results with LIMIT for pagination.

Using SQL aggregate functions like COUNT(*), essential for database reporting.

Example 3: Updating Data (UPDATE statements)

import sqlite3

database_file = 'crm_data.db'

with sqlite3.connect(database_file) as conn:
    cursor = conn.cursor()

    # Update a single contact's email
    old_email = "bob@example.com"
    new_email = "bob.updated@example.com"
    cursor.execute("UPDATE contacts SET email = ? WHERE email = ?", (new_email, old_email))
    print(f"\nUpdated email for {old_email} to {new_email}. Rows affected: {cursor.rowcount}")

    # Update multiple contacts (e.g., set phone to NULL for those with no phone)
    cursor.execute("UPDATE contacts SET phone = NULL WHERE phone IS NOT NULL AND phone = '';")
    print(f"Updated contacts with empty phone to NULL. Rows affected: {cursor.rowcount}")

    # Update a record based on ID
    contact_id_to_update = 1
    new_name = "Alice L. Wonderland"
    cursor.execute("UPDATE contacts SET name = ? WHERE id = ?", (new_name, contact_id_to_update))
    print(f"Updated name for ID {contact_id_to_update} to '{new_name}'. Rows affected: {cursor.rowcount}")

    conn.commit() # Commit all updates

    # Verify updates
    cursor.execute("SELECT id, name, email, phone FROM contacts WHERE id IN (?, ?)", (contact_id_to_update, 2))
    updated_contacts = cursor.fetchall()
    print("\n--- Verified Updated Contacts ---")
    for contact in updated_contacts:
        print(f"ID: {contact[0]}, Name: {contact[1]}, Email: {contact[2]}, Phone: {contact[3]}")

Explanation

This example demonstrates modifying data in SQLite using the UPDATE SQL statement. It shows how to update single rows based on specific conditions (WHERE clause) and how to affect multiple rows. The cursor.rowcount attribute is useful for checking how many rows were affected by the last execute() call, providing feedback on database operations. Again, parameterized queries are used for safety and best practice in updating database records.

Example 4: Deleting Data (DELETE statements)

import sqlite3

database_file = 'crm_data.db'

with sqlite3.connect(database_file) as conn:
    cursor = conn.cursor()

    # Delete a specific contact by email
    email_to_delete = "charlie@example.com"
    cursor.execute("DELETE FROM contacts WHERE email = ?", (email_to_delete,))
    print(f"\nDeleted contact with email '{email_to_delete}'. Rows affected: {cursor.rowcount}")

    # Delete records based on ID range
    # Assuming IDs 4 and 5 might exist after previous operations
    cursor.execute("DELETE FROM contacts WHERE id >= 4;")
    print(f"Deleted contacts with ID >= 4. Rows affected: {cursor.rowcount}")

    conn.commit() # Commit all deletions

    # Verify deletions
    cursor.execute("SELECT * FROM contacts;")
    remaining_contacts = cursor.fetchall()
    print("\n--- Remaining Contacts After Deletion ---")
    if remaining_contacts:
        for contact in remaining_contacts:
            print(f"ID: {contact[0]}, Name: {contact[1]}, Email: {contact[2]}, Phone: {contact[3]}")
    else:
        print("No contacts remaining.")

Explanation

This example covers removing data from SQLite tables using the DELETE FROM SQL statement. It illustrates how to delete specific rows based on conditions and how to delete multiple rows based on a range or other criteria. The WHERE clause is crucial here; without it, DELETE FROM table_name; would delete all rows from the table, so use it with caution! This is a core part of managing database records.

Example 5: Full CRUD Cycle (Combined Operations)

import sqlite3
import os

database_file = 'full_crud_example.db'

# Clean up previous database for fresh start
if os.path.exists(database_file):
    os.remove(database_file)
    print(f"Cleaned up existing {database_file}.")

with sqlite3.connect(database_file) as conn:
    cursor = conn.cursor()

    # C - Create Table
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS products (
        product_id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        price REAL NOT NULL
    );
    """)
    print("\n--- CRUD Cycle: Products Table Created ---")

    # C - Insert Initial Data
    products_data = [
        ("Laptop", 1200.00),
        ("Monitor", 300.00),
        ("Keyboard", 75.00),
        ("Mouse", 25.00)
    ]
    cursor.executemany("INSERT INTO products (name, price) VALUES (?, ?)", products_data)
    conn.commit()
    print(f"Inserted {len(products_data)} initial products.")

    # R - Read All Data
    print("\n--- All Products (Initial) ---")
    cursor.execute("SELECT product_id, name, price FROM products;")
    for row in cursor.fetchall():
        print(f"ID: {row[0]}, Name: {row[1]}, Price: ${row[2]:.2f}")

    # U - Update Data
    product_to_update_id = 2 # Monitor
    new_price_monitor = 320.50
    cursor.execute("UPDATE products SET price = ? WHERE product_id = ?",
                   (new_price_monitor, product_to_update_id))
    conn.commit()
    print(f"\nUpdated product ID {product_to_update_id} price to ${new_price_monitor:.2f}.")

    # R - Read Updated Data
    print("\n--- Product ID 2 (After Update) ---")
    cursor.execute("SELECT product_id, name, price FROM products WHERE product_id = ?",
                   (product_to_update_id,))
    updated_product = cursor.fetchone()
    if updated_product:
        print(f"ID: {updated_product[0]}, Name: {updated_product[1]}, Price: ${updated_product[2]:.2f}")

    # D - Delete Data
    product_to_delete_name = "Mouse"
    cursor.execute("DELETE FROM products WHERE name = ?", (product_to_delete_name,))
    conn.commit()
    print(f"\nDeleted product '{product_to_delete_name}'.")

    # R - Read Remaining Data
    print("\n--- Remaining Products ---")
    cursor.execute("SELECT product_id, name, price FROM products;")
    remaining_products = cursor.fetchall()
    if remaining_products:
        for row in remaining_products:
            print(f"ID: {row[0]}, Name: {row[1]}, Price: ${row[2]:.2f}")
    else:
        print("No products remaining.")

    print("\nFull CRUD cycle completed successfully.")

Explanation

This comprehensive example demonstrates a complete CRUD cycle in SQLite using Python. It combines creating a table, inserting initial data, reading all records, updating a specific record, reading the updated record, deleting a record, and finally reading the remaining data. This example serves as a full-picture demonstration of database manipulation in Python, incorporating all previously discussed concepts from basic SQL operations to data persistence.

Committing and Rolling Back Transactions

Transactions are crucial for maintaining database integrity and consistency, especially when performing multiple related database operations. In SQLite, by default, each execute() call often acts as its own transaction (auto-commit behavior can vary, but explicit control is best). The connection.commit() method saves all changes made within the current transaction to the database, making them permanent. Conversely, connection.rollback() reverts all changes made since the last commit() or the start of the transaction, effectively undoing them.

Example 1: Basic Commit Operation

import sqlite3
import os

database_file = 'transaction_commit.db'
if os.path.exists(database_file):
    os.remove(database_file)

with sqlite3.connect(database_file) as conn:
    cursor = conn.cursor()

    cursor.execute("""
    CREATE TABLE IF NOT EXISTS accounts (
        account_id INTEGER PRIMARY KEY,
        balance REAL NOT NULL
    );
    """)
    print("Table 'accounts' created.")

    # Insert initial data
    cursor.execute("INSERT INTO accounts (account_id, balance) VALUES (1, 100.00), (2, 200.00);")
    print("Initial account balances inserted.")

    # Perform a transaction: transfer money
    amount = 50.00
    cursor.execute("UPDATE accounts SET balance = balance - ? WHERE account_id = ?", (amount, 1))
    cursor.execute("UPDATE accounts SET balance = balance + ? WHERE account_id = ?", (amount, 2))
    print(f"Attempting to transfer ${amount:.2f} from account 1 to account 2...")

    # Commit the changes, making them permanent
    conn.commit()
    print("Transaction committed successfully.")

    # Verify balances
    cursor.execute("SELECT account_id, balance FROM accounts;")
    balances = cursor.fetchall()
    print("\n--- Account Balances After Commit ---")
    for acc_id, balance in balances:
        print(f"Account ID: {acc_id}, Balance: ${balance:.2f}")

Explanation

This example demonstrates a fundamental database transaction involving a money transfer between two accounts. After performing two UPDATE operations, conn.commit() is called. This persists all changes within the transaction to the database file. If commit() were not called, the changes would be lost when the connection closes (unless auto-commit is enabled, which isn't recommended for multi-step operations). This showcases how to save changes to SQLite.

Example 2: Basic Rollback Operation

import sqlite3
import os

database_file = 'transaction_rollback.db'
if os.path.exists(database_file):
    os.remove(database_file)

with sqlite3.connect(database_file) as conn:
    cursor = conn.cursor()

    cursor.execute("""
    CREATE TABLE IF NOT EXISTS logs (
        log_id INTEGER PRIMARY KEY AUTOINCREMENT,
        message TEXT NOT NULL
    );
    """)
    print("Table 'logs' created.")

    # Insert some data that we intend to roll back
    cursor.execute("INSERT INTO logs (message) VALUES ('Attempt 1: Log entry 1');")
    cursor.execute("INSERT INTO logs (message) VALUES ('Attempt 1: Log entry 2');")
    print("Two log entries inserted (pending commit).")

    # Rollback the changes – these insertions will be undone
    conn.rollback()
    print("Transaction rolled back. Changes undone.")

    # Verify that no logs were actually added
    cursor.execute("SELECT * FROM logs;")
    remaining_logs = cursor.fetchall()
    print("\n--- Logs After Rollback ---")
    if not remaining_logs:
        print("No log entries found. Rollback successful.")
    else:
        print("Rollback failed. Logs found:", remaining_logs)

Explanation

This example illustrates the database rollback mechanism. After inserting two log entries, conn.rollback() is explicitly called. This effectively reverts all uncommitted changes, leaving the database in its state before these insertions. This is crucial for maintaining data consistency if a series of operations cannot be fully completed or encounters an error. It's a key feature for undoing database changes.

Example 3: Error Handling with Rollback (ACID Properties)

import sqlite3
import os

database_file = 'transaction_error_handling.db'
if os.path.exists(database_file):
    os.remove(database_file)

with sqlite3.connect(database_file) as conn:
    cursor = conn.cursor()

    cursor.execute("""
    CREATE TABLE IF NOT EXISTS inventory (
        product_id INTEGER PRIMARY KEY,
        stock INTEGER NOT NULL
    );
    """)
    cursor.execute("INSERT INTO inventory (product_id, stock) VALUES (101, 50), (102, 30);")
    conn.commit()
    print("Initial inventory set up.")

    try:
        # Simulate an order transaction: decrease stock of two products
        product_a_id = 101
        product_b_id = 102
        order_quantity_a = 10
        order_quantity_b = 40 # This will cause an insufficient stock error for product_b

        # Deduct stock for product A
        cursor.execute("UPDATE inventory SET stock = stock - ? WHERE product_id = ?",
                       (order_quantity_a, product_a_id))
        print(f"Deducted {order_quantity_a} from product {product_a_id}.")

        # Check stock for product B before deducting
        cursor.execute("SELECT stock FROM inventory WHERE product_id = ?", (product_b_id,))
        current_stock_b = cursor.fetchone()[0]

        if current_stock_b < order_quantity_b:
            raise ValueError("Insufficient stock for product B!")

        # Deduct stock for product B (this line won't be reached if ValueError raised)
        cursor.execute("UPDATE inventory SET stock = stock - ? WHERE product_id = ?",
                       (order_quantity_b, product_b_id))
        print(f"Deducted {order_quantity_b} from product {product_b_id}.")

        # If all operations succeed, commit
        conn.commit()
        print("Order transaction completed and committed.")

    except ValueError as e:
        print(f"Transaction failed: {e}")
        conn.rollback() # Rollback all changes if an error occurs
        print("Transaction rolled back due to error.")
    except sqlite3.Error as e:
        print(f"Database error during transaction: {e}")
        conn.rollback()
        print("Transaction rolled back due to database error.")

    # Verify final inventory balances
    cursor.execute("SELECT product_id, stock FROM inventory;")
    final_inventory = cursor.fetchall()
    print("\n--- Final Inventory Balances ---")
    for prod_id, stock in final_inventory:
        print(f"Product ID: {prod_id}, Stock: {stock}")

Explanation

This advanced example demonstrates transactional integrity using try-except blocks and rollback(). We simulate an order process that attempts to deduct stock from two products. If any part of the transaction fails (e.g., due to insufficient stock, simulated by a ValueError), the except block catches the error, and conn.rollback() is called. This ensures that either all operations within the transaction succeed and are committed, or none of them are, maintaining the ACID properties of transactions (Atomicity, Consistency, Isolation, Durability). This is crucial for reliable database operations.

Example 4: Transaction with Context Manager (Simplified Transaction Management)

import sqlite3
import os

database_file = 'transaction_context_manager.db'
if os.path.exists(database_file):
    os.remove(database_file)

# Using the connection as a context manager simplifies transaction handling.
# By default, exiting the 'with' block commits if no exception, otherwise rolls back.
try:
    with sqlite3.connect(database_file) as conn:
        cursor = conn.cursor()

        cursor.execute("""
        CREATE TABLE IF NOT EXISTS events (
            event_id INTEGER PRIMARY KEY AUTOINCREMENT,
            description TEXT NOT NULL,
            timestamp TEXT DEFAULT CURRENT_TIMESTAMP
        );
        """)
        print("Table 'events' created.")

        # Insert some events
        cursor.execute("INSERT INTO events (description) VALUES ('User login');")
        cursor.execute("INSERT INTO events (description) VALUES ('Data updated');")
        print("Two events inserted (implicitly part of a transaction).")

        # Simulate an error to trigger an automatic rollback
        # raise sqlite3.Error("Simulating database error during transaction.")

        # If no error, the context manager will implicitly commit here
        print("Exiting 'with' block... (Implicit commit if no error, else rollback)")

except sqlite3.Error as e:
    print(f"Caught an error: {e}")
    print("The transaction was automatically rolled back by the context manager.")

# Verify the state of the database after the 'with' block
with sqlite3.connect(database_file) as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM events;")
    events = cursor.fetchall()
    print("\n--- Events after transaction attempt ---")
    if events:
        for event in events:
            print(event)
    else:
        print("No events found (rollback likely occurred or table empty).")

Explanation

This example demonstrates the power of using the sqlite3.connect() object as a context manager for transactions. When used with with, conn.commit() is automatically called upon successful exit of the with block, and conn.rollback() is automatically called if an unhandled exception occurs within the block. This makes transaction management in Python cleaner and less prone to errors compared to manual try-finally blocks for commit/rollback. This is a recommended Pythonic way for database transactions.

Example 5: Manual Transaction Control (BEGIN, COMMIT, ROLLBACK statements)

import sqlite3
import os

database_file = 'manual_transactions.db'
if os.path.exists(database_file):
    os.remove(database_file)

conn = None # Initialize conn

try:
    conn = sqlite3.connect(database_file)
    cursor = conn.cursor()

    cursor.execute("PRAGMA journal_mode=WAL;") # Use WAL for better concurrency and recovery
    conn.execute('BEGIN TRANSACTION;') # Explicitly start a transaction

    cursor.execute("""
    CREATE TABLE IF NOT EXISTS sales (
        sale_id INTEGER PRIMARY KEY AUTOINCREMENT,
        item TEXT NOT NULL,
        amount REAL NOT NULL
    );
    """)
    print("Table 'sales' created.")

    # Insert a sale
    cursor.execute("INSERT INTO sales (item, amount) VALUES ('Product X', 150.00);")
    print("Inserted 'Product X'.")

    # This is a point where you might decide to commit or rollback
    # For demonstration, let's insert another item
    cursor.execute("INSERT INTO sales (item, amount) VALUES ('Service Y', 50.00);")
    print("Inserted 'Service Y'.")

    # You can explicitly commit here, or defer it.
    # For this example, let's demonstrate a deliberate rollback after a potential issue.

    # Simulate an external check or validation that fails
    if True: # Change to False to see commit path
        raise Exception("External validation failed! Rolling back.")

    conn.commit() # This line will only be reached if no exception
    print("Manual transaction committed successfully.")

except Exception as e:
    print(f"Error: {e}. Performing manual rollback.")
    if conn:
        conn.rollback() # Explicitly rollback on error
    print("Manual transaction rolled back.")

finally:
    if conn:
        conn.close()
        print("Connection closed.")

# Verify data after the manual transaction attempt
with sqlite3.connect(database_file) as conn_verify:
    cursor_verify = conn_verify.cursor()
    cursor_verify.execute("SELECT * FROM sales;")
    sales_data = cursor_verify.fetchall()
    print("\n--- Sales Data After Manual Transaction Attempt ---")
    if sales_data:
        for sale in sales_data:
            print(sale)
    else:
        print("No sales data found (rollback successful).")

Explanation

This advanced example demonstrates explicit transaction control using BEGIN TRANSACTION;, COMMIT;, and ROLLBACK; SQL statements (though sqlite3's default behavior often makes BEGIN implicit). By setting conn.isolation_level = None (which is often done implicitly or not needed depending on the operation), you take full control of transactions. This approach is useful when you need very precise control over transaction boundaries, potentially across multiple functions or complex logic. It also briefly introduces PRAGMA journal_mode=WAL; which can improve SQLite concurrency and crash recovery. This example solidifies understanding of database transaction management.