Tutorials
Python
Python SQLite Database

Python SQLite Database

SQLite is a lightweight, file-based database that comes built-in with Python. The sqlite3 module provides a simple yet powerful interface for database operations without requiring a separate database server.

SQLite Basics

SQLite stores the entire database in a single file.

Connecting to Database 

import sqlite3

# Connect to database (creates file if it doesn't exist)
conn = sqlite3.connect('example.db')

# Create cursor object
cursor = conn.cursor()

print("Connected to SQLite database")
print(f"Database file: example.db")

# Close connection
conn.close()

sqlite3.connect() creates or opens a database file. Always close connections when done.

Basic Operations 

import sqlite3

# Connect and create table
conn = sqlite3.connect('users.db')
cursor = conn.cursor()

# Create table
cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        email TEXT UNIQUE,
        age INTEGER,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
''')

# Insert data
cursor.execute("INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
               ("Alice", "[email protected]", 30))

cursor.execute("INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
               ("Bob", "[email protected]", 25))

# Save changes
conn.commit()

print("Table created and data inserted")

conn.close()

Use ? placeholders to prevent SQL injection. conn.commit() saves changes.

CRUD Operations

Create, Read, Update, Delete operations.

Creating Records 

import sqlite3

conn = sqlite3.connect('users.db')
cursor = conn.cursor()

# Single insert
cursor.execute("""
    INSERT INTO users (name, email, age) 
    VALUES (?, ?, ?)
""", ("Charlie", "[email protected]", 35))

# Multiple inserts
users_data = [
    ("David", "[email protected]", 28),
    ("Eve", "[email protected]", 32),
    ("Frank", "[email protected]", 45)
]

cursor.executemany("""
    INSERT INTO users (name, email, age) 
    VALUES (?, ?, ?)
""", users_data)

conn.commit()
print(f"Inserted {cursor.rowcount} rows")

conn.close()

executemany() efficiently inserts multiple rows.

Reading Records 

import sqlite3

conn = sqlite3.connect('users.db')
cursor = conn.cursor()

# Select all users
cursor.execute("SELECT id, name, email, age FROM users")
rows = cursor.fetchall()

print("All users:")
for row in rows:
    user_id, name, email, age = row
    print(f"ID: {user_id}, Name: {name}, Email: {email}, Age: {age}")

print(f"\nTotal users: {len(rows)}")

# Select with conditions
cursor.execute("SELECT name, age FROM users WHERE age > ?", (30,))
older_users = cursor.fetchall()

print(f"\nUsers older than 30: {older_users}")

conn.close()

fetchall() returns all results as a list of tuples.

Updating Records 

import sqlite3

conn = sqlite3.connect('users.db')
cursor = conn.cursor()

# Update single record
cursor.execute("""
    UPDATE users 
    SET age = ?, email = ? 
    WHERE name = ?
""", (31, "[email protected]", "Alice"))

print(f"Rows updated: {cursor.rowcount}")

# Update multiple records
cursor.execute("""
    UPDATE users 
    SET age = age + 1 
    WHERE age < 30
""")

print(f"Young users aged: {cursor.rowcount}")

conn.commit()
conn.close()

Use WHERE clauses to target specific records.

Deleting Records 

import sqlite3

conn = sqlite3.connect('users.db')
cursor = conn.cursor()

# Delete specific record
cursor.execute("DELETE FROM users WHERE name = ?", ("Frank",))
print(f"Deleted: {cursor.rowcount} rows")

# Delete with conditions
cursor.execute("DELETE FROM users WHERE age < ?", (25,))
print(f"Deleted young users: {cursor.rowcount} rows")

# Delete all records
cursor.execute("DELETE FROM users")
print(f"Cleared table: {cursor.rowcount} rows deleted")

conn.commit()
conn.close()

Be careful with DELETE operations - they permanently remove data.

Advanced Queries

Complex SQL operations with SQLite.

Joins and Relationships 

import sqlite3

conn = sqlite3.connect('company.db')
cursor = conn.cursor()

# Create tables
cursor.execute('''
    CREATE TABLE IF NOT EXISTS departments (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        location TEXT
    )
''')

cursor.execute('''
    CREATE TABLE IF NOT EXISTS employees (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        department_id INTEGER,
        salary REAL,
        FOREIGN KEY (department_id) REFERENCES departments (id)
    )
''')

# Insert data
cursor.execute("INSERT INTO departments (name, location) VALUES (?, ?)",
               ("Engineering", "Building A"))
cursor.execute("INSERT INTO departments (name, location) VALUES (?, ?)",
               ("Sales", "Building B"))

cursor.execute("INSERT INTO employees (name, department_id, salary) VALUES (?, ?, ?)",
               ("Alice", 1, 75000))
cursor.execute("INSERT INTO employees (name, department_id, salary) VALUES (?, ?, ?)",
               ("Bob", 1, 70000))
cursor.execute("INSERT INTO employees (name, department_id, salary) VALUES (?, ?, ?)",
               ("Charlie", 2, 65000))

conn.commit()

# Join query
cursor.execute('''
    SELECT e.name, d.name, e.salary
    FROM employees e
    JOIN departments d ON e.department_id = d.id
    ORDER BY e.salary DESC
''')

print("Employee Department Report:")
for row in cursor.fetchall():
    name, dept, salary = row
    print(f"{name} - {dept} - ${salary}")

conn.close()

SQLite supports JOIN operations for relational queries.

Aggregations and Grouping 

import sqlite3

conn = sqlite3.connect('company.db')
cursor = conn.cursor()

# Aggregation queries
cursor.execute('''
    SELECT 
        d.name,
        COUNT(e.id) as employee_count,
        AVG(e.salary) as avg_salary,
        MAX(e.salary) as max_salary
    FROM departments d
    LEFT JOIN employees e ON d.id = e.department_id
    GROUP BY d.id, d.name
''')

print("Department Statistics:")
for row in cursor.fetchall():
    dept_name, count, avg_salary, max_salary = row
    print(f"{dept_name}: {count} employees, avg salary ${avg_salary:.0f}, max ${max_salary}")

# Having clause
cursor.execute('''
    SELECT department_id, AVG(salary), COUNT(*)
    FROM employees
    GROUP BY department_id
    HAVING COUNT(*) > 1
''')

print("\nDepartments with multiple employees:")
for row in cursor.fetchall():
    dept_id, avg_salary, count = row
    print(f"Dept {dept_id}: {count} employees, avg salary ${avg_salary:.0f}")

conn.close()

Use aggregate functions like COUNT, AVG, MAX with GROUP BY and HAVING.

Data Types and Constraints

SQLite data types and table constraints.

SQLite Data Types 

import sqlite3

conn = sqlite3.connect('datatypes.db')
cursor = conn.cursor()

# Create table with various data types
cursor.execute('''
    CREATE TABLE IF NOT EXISTS examples (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        age INTEGER,
        height REAL,
        is_active BOOLEAN,
        birth_date DATE,
        notes BLOB
    )
''')

# Insert sample data
cursor.execute('''
    INSERT INTO examples (name, age, height, is_active, birth_date, notes)
    VALUES (?, ?, ?, ?, ?, ?)
''', ("Alice", 30, 5.6, True, "1993-05-15", b"Some binary data"))

conn.commit()

# Query with type checking
cursor.execute("SELECT name, age, height, is_active FROM examples")
row = cursor.fetchone()

name, age, height, is_active = row
print(f"Name: {name} (type: {type(name)})")
print(f"Age: {age} (type: {type(age)})")
print(f"Height: {height} (type: {type(height)})")
print(f"Active: {is_active} (type: {type(is_active)})")

conn.close()

SQLite uses dynamic typing - data types are per-value, not per-column.

Constraints and Indexes 

import sqlite3

conn = sqlite3.connect('constraints.db')
cursor = conn.cursor()

# Create table with constraints
cursor.execute('''
    CREATE TABLE IF NOT EXISTS products (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL UNIQUE,
        price REAL NOT NULL CHECK(price > 0),
        category TEXT DEFAULT 'General',
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
''')

# Create index for faster queries
cursor.execute("CREATE INDEX IF NOT EXISTS idx_category ON products(category)")

# Insert valid data
try:
    cursor.execute("""
        INSERT INTO products (name, price, category) 
        VALUES (?, ?, ?)
    """, ("Laptop", 999.99, "Electronics"))
    print("Product inserted successfully")
except sqlite3.IntegrityError as e:
    print(f"Constraint violation: {e}")

# Try invalid data
try:
    cursor.execute("""
        INSERT INTO products (name, price) 
        VALUES (?, ?)
    """, ("Laptop", -100))  # Negative price violates CHECK
except sqlite3.IntegrityError as e:
    print(f"Check constraint failed: {e}")

conn.commit()
conn.close()

Use constraints to maintain data integrity and indexes for performance.

Context Managers

Use context managers for automatic connection handling.

Custom Context Manager 

import sqlite3
from contextlib import contextmanager

@contextmanager
def get_db_connection(db_name):
    """Context manager for database connections."""
    conn = sqlite3.connect(db_name)
    try:
        yield conn
    except Exception as e:
        conn.rollback()
        raise e
    else:
        conn.commit()
    finally:
        conn.close()

# Usage
with get_db_connection('test.db') as conn:
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE IF NOT EXISTS test (id INTEGER, name TEXT)")
    cursor.execute("INSERT INTO test VALUES (1, 'Test')")
    print("Operation completed successfully")

# Connection automatically closed and committed

Context managers ensure proper resource cleanup.

Error Handling

Handle database errors gracefully.

Common Exceptions 

import sqlite3

def safe_database_operation():
    try:
        conn = sqlite3.connect('example.db')
        cursor = conn.cursor()
        
        # This might fail
        cursor.execute("INSERT INTO nonexistent_table VALUES (?)", (1,))
        
        conn.commit()
        
    except sqlite3.OperationalError as e:
        print(f"SQL error: {e}")
        if conn:
            conn.rollback()
    
    except sqlite3.IntegrityError as e:
        print(f"Constraint violation: {e}")
        if conn:
            conn.rollback()
    
    except Exception as e:
        print(f"Unexpected error: {e}")
        if conn:
            conn.rollback()
    
    finally:
        if conn:
            conn.close()

safe_database_operation()

Handle specific SQLite exceptions appropriately.

Performance Optimization

Techniques for faster database operations.

Prepared Statements 

import sqlite3
import time

# Create test data
conn = sqlite3.connect('performance.db')
cursor = conn.cursor()

cursor.execute('''
    CREATE TABLE IF NOT EXISTS logs (
        id INTEGER PRIMARY KEY,
        timestamp TEXT,
        level TEXT,
        message TEXT
    )
''')

# Method 1: Multiple executes (slower)
start = time.time()
for i in range(1000):
    cursor.execute("""
        INSERT INTO logs (timestamp, level, message) 
        VALUES (?, ?, ?)
    """, (f"2023-01-{i%30:02d}", "INFO", f"Log message {i}"))
method1_time = time.time() - start

# Method 2: Execute many (faster)
cursor.execute("DELETE FROM logs")  # Clear table
start = time.time()
data = [(f"2023-01-{i%30:02d}", "INFO", f"Log message {i}") for i in range(1000)]
cursor.executemany("""
    INSERT INTO logs (timestamp, level, message) 
    VALUES (?, ?, ?)
""", data)
method2_time = time.time() - start

print(f"Multiple executes: {method1_time:.4f} seconds")
print(f"Execute many: {method2_time:.4f} seconds")
print(f"Speedup: {method1_time/method2_time:.1f}x")

conn.commit()
conn.close()

executemany() is much faster for bulk inserts.

Connection Pooling 

import sqlite3
import threading
from queue import Queue

class ConnectionPool:
    def __init__(self, db_name, pool_size=5):
        self.db_name = db_name
        self.pool_size = pool_size
        self.pool = Queue(maxsize=pool_size)
        
        # Initialize pool
        for _ in range(pool_size):
            conn = sqlite3.connect(db_name, check_same_thread=False)
            self.pool.put(conn)
    
    def get_connection(self):
        return self.pool.get()
    
    def return_connection(self, conn):
        self.pool.put(conn)
    
    def close_all(self):
        while not self.pool.empty():
            conn = self.pool.get()
            conn.close()

# Usage
pool = ConnectionPool('shared.db')

def worker_thread(thread_id):
    conn = pool.get_connection()
    cursor = conn.cursor()
    
    cursor.execute("INSERT INTO test (thread_id) VALUES (?)", (thread_id,))
    conn.commit()
    
    pool.return_connection(conn)
    print(f"Thread {thread_id} completed")

# Create test table
conn = pool.get_connection()
conn.execute("CREATE TABLE IF NOT EXISTS test (thread_id INTEGER)")
conn.commit()
pool.return_connection(conn)

# Run threads
threads = []
for i in range(10):
    t = threading.Thread(target=worker_thread, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

pool.close_all()
print("All operations completed")

Connection pooling improves performance in multi-threaded applications.

Best Practices

  1. Use context managers for connections
  2. Always commit or rollback transactions
  3. Use parameterized queries to prevent SQL injection
  4. Create indexes for frequently queried columns
  5. Use foreign keys for data integrity
  6. Handle exceptions appropriately
  7. Close connections when done
  8. Consider connection pooling for multi-threaded apps

External Resources:

Related Tutorials:

Last updated on
Python CSV HandlingPython Threading