CRUD Operations & Executing SQL Queries in Python

Building on our introduction to database connectivity, this session covers standard CRUD (Create, Read, Update, Delete) operations in Python. You will learn how to write robust database logic that modifies records safely, handles unexpected errors gracefully, and manages transactions.

1. Overview of CRUD Operations in Python

Every backend application centers on managing persistent state. Here is how core HTTP/Application verbs align with SQL commands and Python DB-API methods:

Operation	SQL Command	Python Method	Requires Commit?
Create	`INSERT INTO ...`	`cursor.execute()` / `executemany()`	Yes (Modifies state)
Read	`SELECT ... FROM ...`	`cursor.fetchone()` / `fetchall()`	No (Read-only)
Update	`UPDATE ... SET ...`	`cursor.execute()`	Yes (Modifies state)
Delete	`DELETE FROM ...`	`cursor.execute()`	Yes (Modifies state)

2. Implementing Full CRUD Workflow

The example below demonstrates a full, practical implementation of creating a table, inserting records, reading results, updating data, and deleting records safely with parameterization.

import sqlite3 # 1. Establish Connection conn = sqlite3.connect(':memory:') cursor = conn.cursor() # CREATE TABLE cursor.execute(''' CREATE TABLE accounts ( account_id INTEGER PRIMARY KEY, owner_name TEXT NOT NULL, balance REAL NOT NULL ) ''') # CREATE (Insert) new_accounts = [ (101, 'Alice', 1500.00), (102, 'Bob', 2300.50), (103, 'Charlie', 450.75) ] cursor.executemany("INSERT INTO accounts VALUES (?, ?, ?)", new_accounts) conn.commit() # READ (Select) cursor.execute("SELECT owner_name, balance FROM accounts WHERE balance > ?", (1000.00,)) print("Accounts with > $1000:", cursor.fetchall()) # UPDATE (Modify) cursor.execute("UPDATE accounts SET balance = balance + ? WHERE account_id = ?", (200.00, 103)) conn.commit() print("Updated rows count:", cursor.rowcount) # DELETE (Remove) cursor.execute("DELETE FROM accounts WHERE account_id = ?", (101,)) conn.commit() print("Deleted rows count:", cursor.rowcount) conn.close()

3. Robust Transaction Management & Rollbacks

When executing multiple related SQL updates (e.g., transferring funds between two bank accounts), either all updates must succeed or none should take effect. Wrapping database logic in `try...except` blocks with `conn.rollback()` prevents partial state changes when errors occur.

import sqlite3 def transfer_funds(conn, sender_id, receiver_id, amount): cursor = conn.cursor() try: # Check sender balance cursor.execute("SELECT balance FROM accounts WHERE account_id = ?", (sender_id,)) res = cursor.fetchone() if not res or res[0] < amount: raise ValueError("Insufficient balance or sender does not exist.") # Deduct from sender cursor.execute("UPDATE accounts SET balance = balance - ? WHERE account_id = ?", (amount, sender_id)) # Credit to receiver cursor.execute("UPDATE accounts SET balance = balance + ? WHERE account_id = ?", (amount, receiver_id)) # Commit transaction as a single unit conn.commit() print(f"Successfully transferred ${amount:.2f} from #{sender_id} to #{receiver_id}") except (sqlite3.Error, ValueError) as err: # Roll back changes on error conn.rollback() print(f"Transaction failed and rolled back! Error: {err}")

Transaction Safety: Failing to call conn.rollback() after a failed statement leaves uncommitted changes in memory, which can lead to invalid states or file locks.

🏋️ Try It Yourself: Practice Challenges

Challenge 1: Conditional Update

Write an UPDATE query that grants a 10% bonus to all employees in the 'Engineering' department.

Challenge 2: Safe Deletion with Row Verification

Write a function that deletes a record by ID and prints a warning if no matching row was found (using cursor.rowcount).

Challenge 3: Transaction Rollback Guard

Simulate a failed insert (e.g. inserting a duplicate PRIMARY KEY) inside a transaction block, catch sqlite3.IntegrityError, and execute a rollback.

⚡ Interactive Sandbox (CRUD & Rollback)

import sqlite3

# Initialize Database
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()

cursor.execute('''
    CREATE TABLE employees (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        salary REAL NOT NULL
    )
''')

# 1. CREATE (Insert)
employees = [(1, 'Sarah', 65000), (2, 'David', 52000), (3, 'Elena', 72000)]
cursor.executemany("INSERT INTO employees VALUES (?, ?, ?)", employees)
conn.commit()

# 2. UPDATE (Raise salaries)
cursor.execute("UPDATE employees SET salary = salary * 1.05 WHERE salary < ?", (60000,))
print(f"Updated {cursor.rowcount} employee salary record(s).")

# 3. TRANSACTION WITH ROLLBACK DEMO
try:
    print("\nAttempting invalid transaction (Duplicate ID)...")
    cursor.execute("INSERT INTO employees VALUES (1, 'Duplicate User', 90000)")
    conn.commit()
except sqlite3.IntegrityError as err:
    conn.rollback()
    print("Caught Exception:", err)
    print("Transaction rolled back cleanly!")

# 4. READ Final State
cursor.execute("SELECT * FROM employees")
print("\nCurrent Roster:", cursor.fetchall())

conn.close()

Console Output:

Click "Run Code" above to execute CRUD operations script...

📝 Knowledge Check Quiz

1. Which method must be explicitly called on the connection object to permanently save data changes (INSERT, UPDATE, DELETE)?

cursor.save() conn.commit() conn.flush()

2. How do you check how many records were affected by the most recent UPDATE or DELETE query?

len(cursor.fetchall()) cursor.rowcount conn.affected_rows

3. What is the purpose of executing `conn.rollback()` inside an exception handler block?

It reverses pending uncommitted changes, returning the database to its last consistent state It deletes the database file from disk It closes the database connection automatically

Horizontal Ad Banner Slot (728x90 / Responsive)