python
examples
examples.py🐍python
"""
15 - Database: Examples
Run this file to see SQLite database operations in action!
"""
print("=" * 60)
print("DATABASE - EXAMPLES")
print("=" * 60)
import sqlite3
import os
import tempfile
from contextlib import contextmanager
# Use temporary database for examples
DB_PATH = os.path.join(tempfile.gettempdir(), 'example_db.sqlite')
# Clean up any existing database
if os.path.exists(DB_PATH):
os.remove(DB_PATH)
# =============================================================================
# 1. BASIC CONNECTION
# =============================================================================
print("\n--- 1. Basic Connection ---\n")
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("SELECT sqlite_version()")
version = cursor.fetchone()
print(f"SQLite version: {version[0]}")
conn.close()
# =============================================================================
# 2. CREATE TABLES
# =============================================================================
print("\n--- 2. Create Tables ---\n")
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()
# Users table
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE,
age INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Posts table with foreign key
cursor.execute('''
CREATE TABLE IF NOT EXISTS posts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
title TEXT NOT NULL,
content TEXT,
FOREIGN KEY (user_id) REFERENCES users (id)
)
''')
conn.commit()
print("Tables created successfully!")
# =============================================================================
# 3. INSERT DATA
# =============================================================================
print("\n--- 3. Insert Data ---\n")
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()
# Insert single user
cursor.execute('''
INSERT INTO users (name, email, age)
VALUES (?, ?, ?)
''', ('Alice', 'alice@example.com', 25))
alice_id = cursor.lastrowid
print(f"Inserted Alice with ID: {alice_id}")
# Insert multiple users
users = [
('Bob', 'bob@example.com', 30),
('Charlie', 'charlie@example.com', 35),
('Diana', 'diana@example.com', 28),
]
cursor.executemany('''
INSERT INTO users (name, email, age)
VALUES (?, ?, ?)
''', users)
print(f"Inserted {cursor.rowcount} more users")
# Insert posts
posts = [
(alice_id, 'Hello World', 'My first post!'),
(alice_id, 'Python Tips', 'Some useful Python tips...'),
(2, 'Database Guide', 'How to use SQLite...'),
]
cursor.executemany('''
INSERT INTO posts (user_id, title, content)
VALUES (?, ?, ?)
''', posts)
print(f"Inserted {cursor.rowcount} posts")
conn.commit()
# =============================================================================
# 4. SELECT DATA
# =============================================================================
print("\n--- 4. Select Data ---\n")
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()
# Fetch one
cursor.execute('SELECT * FROM users WHERE id = ?', (1,))
user = cursor.fetchone()
print(f"Single user: {user}")
# Fetch all
cursor.execute('SELECT id, name, email FROM users')
all_users = cursor.fetchall()
print(f"\nAll users:")
for user in all_users:
print(f" ID: {user[0]}, Name: {user[1]}, Email: {user[2]}")
# With WHERE clause
cursor.execute('SELECT name, age FROM users WHERE age > ?', (27,))
older_users = cursor.fetchall()
print(f"\nUsers older than 27: {older_users}")
# =============================================================================
# 5. ROW FACTORY (DICT RESULTS)
# =============================================================================
print("\n--- 5. Row Factory ---\n")
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute('SELECT * FROM users WHERE id = 1')
user = cursor.fetchone()
# Access by column name
print(f"Name: {user['name']}")
print(f"Email: {user['email']}")
print(f"Age: {user['age']}")
# Convert to dict
user_dict = dict(user)
print(f"As dict: {user_dict}")
# =============================================================================
# 6. UPDATE DATA
# =============================================================================
print("\n--- 6. Update Data ---\n")
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()
# Show before
cursor.execute('SELECT name, age FROM users WHERE id = 1')
print(f"Before: {cursor.fetchone()}")
# Update
cursor.execute('''
UPDATE users
SET age = ?
WHERE id = ?
''', (26, 1))
print(f"Rows updated: {cursor.rowcount}")
# Show after
cursor.execute('SELECT name, age FROM users WHERE id = 1')
print(f"After: {cursor.fetchone()}")
conn.commit()
# =============================================================================
# 7. DELETE DATA
# =============================================================================
print("\n--- 7. Delete Data ---\n")
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()
# Count before
cursor.execute('SELECT COUNT(*) FROM users')
print(f"Users before: {cursor.fetchone()[0]}")
# Delete
cursor.execute('DELETE FROM users WHERE id = ?', (4,))
print(f"Deleted {cursor.rowcount} user(s)")
# Count after
cursor.execute('SELECT COUNT(*) FROM users')
print(f"Users after: {cursor.fetchone()[0]}")
conn.commit()
# =============================================================================
# 8. JOIN QUERIES
# =============================================================================
print("\n--- 8. Join Queries ---\n")
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute('''
SELECT users.name, posts.title, posts.content
FROM posts
INNER JOIN users ON posts.user_id = users.id
''')
print("Posts with authors:")
for row in cursor.fetchall():
print(f" {row['name']}: {row['title']}")
# =============================================================================
# 9. AGGREGATE FUNCTIONS
# =============================================================================
print("\n--- 9. Aggregate Functions ---\n")
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()
# Count
cursor.execute('SELECT COUNT(*) FROM users')
print(f"Total users: {cursor.fetchone()[0]}")
# Average
cursor.execute('SELECT AVG(age) FROM users')
print(f"Average age: {cursor.fetchone()[0]:.1f}")
# Min/Max
cursor.execute('SELECT MIN(age), MAX(age) FROM users')
min_age, max_age = cursor.fetchone()
print(f"Age range: {min_age} - {max_age}")
# Group by
cursor.execute('''
SELECT user_id, COUNT(*) as post_count
FROM posts
GROUP BY user_id
''')
print("Posts per user:")
for row in cursor.fetchall():
print(f" User {row[0]}: {row[1]} posts")
# =============================================================================
# 10. TRANSACTIONS
# =============================================================================
print("\n--- 10. Transactions ---\n")
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
try:
# Start transaction
cursor.execute("INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
('Eve', 'eve@example.com', 22))
# This would fail (duplicate email) - uncomment to test rollback
# cursor.execute("INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
# ('Eve2', 'eve@example.com', 23))
conn.commit()
print("Transaction committed successfully!")
except sqlite3.IntegrityError as e:
conn.rollback()
print(f"Transaction rolled back: {e}")
finally:
conn.close()
# =============================================================================
# 11. DATABASE HELPER CLASS
# =============================================================================
print("\n--- 11. Database Helper Class ---\n")
class Database:
def __init__(self, db_path):
self.db_path = db_path
@contextmanager
def get_cursor(self):
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
try:
yield cursor
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
def execute(self, query, params=()):
with self.get_cursor() as cursor:
cursor.execute(query, params)
return [dict(row) for row in cursor.fetchall()]
def insert(self, table, data):
columns = ', '.join(data.keys())
placeholders = ', '.join('?' * len(data))
query = f"INSERT INTO {table} ({columns}) VALUES ({placeholders})"
with self.get_cursor() as cursor:
cursor.execute(query, tuple(data.values()))
return cursor.lastrowid
# Use the helper
db = Database(DB_PATH)
# Query
users = db.execute("SELECT name, email FROM users LIMIT 3")
print("Users via helper:")
for user in users:
print(f" {user}")
# Insert
new_id = db.insert('users', {'name': 'Frank', 'email': 'frank@example.com', 'age': 40})
print(f"Inserted new user with ID: {new_id}")
# =============================================================================
# 12. PRACTICAL EXAMPLE - TODO APP
# =============================================================================
print("\n--- 12. Practical: Todo App ---\n")
class TodoDB:
def __init__(self, db_path):
self.db_path = db_path
self._init_db()
def _init_db(self):
with sqlite3.connect(self.db_path) as conn:
conn.execute('''
CREATE TABLE IF NOT EXISTS todos (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
completed INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
def add(self, title):
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute('INSERT INTO todos (title) VALUES (?)', (title,))
return cursor.lastrowid
def complete(self, todo_id):
with sqlite3.connect(self.db_path) as conn:
conn.execute('UPDATE todos SET completed = 1 WHERE id = ?', (todo_id,))
def list_all(self):
with sqlite3.connect(self.db_path) as conn:
conn.row_factory = sqlite3.Row
return [dict(row) for row in conn.execute('SELECT * FROM todos')]
def delete(self, todo_id):
with sqlite3.connect(self.db_path) as conn:
conn.execute('DELETE FROM todos WHERE id = ?', (todo_id,))
# Use Todo app
todo_db_path = os.path.join(tempfile.gettempdir(), 'todos.sqlite')
todo = TodoDB(todo_db_path)
# Add todos
todo.add("Learn Python")
todo.add("Practice SQLite")
todo.add("Build a project")
# Complete one
todo.complete(1)
# List all
print("All todos:")
for t in todo.list_all():
status = "✓" if t['completed'] else " "
print(f" [{status}] {t['id']}: {t['title']}")
# Cleanup
os.remove(todo_db_path)
# =============================================================================
# CLEANUP
# =============================================================================
print("\n--- Cleanup ---\n")
if os.path.exists(DB_PATH):
os.remove(DB_PATH)
print(f"Removed temporary database: {DB_PATH}")
print("\n" + "=" * 60)
print("END OF EXAMPLES")
print("=" * 60)