💾 Working with Databases

Databases store and organize data for efficient retrieval and manipulation. Python provides built-in support for SQLite databases and can connect to other database systems like PostgreSQL, MySQL, and more.

import sqlite3

# Create in-memory database for demo
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()

# Create a table
cursor.execute('''
    CREATE TABLE users (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        email TEXT UNIQUE,
        age INTEGER
    )
''')

# Insert some data
users_data = [
    ('Alice Johnson', 'alice@example.com', 30),
    ('Bob Smith', 'bob@example.com', 25),
    ('Charlie Brown', 'charlie@example.com', 35)
]

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

# Query the data
cursor.execute('SELECT * FROM users WHERE age > 25')
results = cursor.fetchall()

print("Users older than 25:")
for user in results:
    print(f"  {user[1]} ({user[3]} years old)")

conn.close()

🎯 Understanding Database Operations

Databases use SQL (Structured Query Language) for creating, reading, updating, and deleting data.

Creating and Managing Tables

import sqlite3

def create_database():
    """Create database with multiple tables"""
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()
    
    # Create users table
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            email TEXT UNIQUE NOT NULL,
            created_date TEXT DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    
    # Create posts table
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS posts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            content TEXT,
            user_id INTEGER,
            created_date TEXT DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (user_id) REFERENCES users (id)
        )
    ''')
    
    conn.commit()
    return conn

def add_sample_data(conn):
    """Add sample data to tables"""
    cursor = conn.cursor()
    
    # Add users
    users = [
        ('alice123', 'alice@example.com'),
        ('bob456', 'bob@example.com'),
        ('charlie789', 'charlie@example.com')
    ]
    
    cursor.executemany('INSERT OR IGNORE INTO users (username, email) VALUES (?, ?)', users)
    
    # Add posts
    posts = [
        ('My First Post', 'This is the content of my first post', 1),
        ('Python Tips', 'Here are some useful Python tips', 1),
        ('Database Tutorial', 'Learning about databases', 2),
        ('Web Development', 'Building web applications', 3)
    ]
    
    cursor.executemany('INSERT INTO posts (title, content, user_id) VALUES (?, ?, ?)', posts)
    
    conn.commit()
    print("Sample data added successfully")

# Create database and add data
conn = create_database()
add_sample_data(conn)
conn.close()

Querying Data

import sqlite3

def connect_database():
    """Connect to database with row factory for named access"""
    conn = sqlite3.connect('example.db')
    # Enable row factory for named column access
    conn.row_factory = sqlite3.Row
    return conn

def get_all_users():
    """Retrieve all users"""
    conn = connect_database()
    cursor = conn.cursor()
    
    cursor.execute('SELECT * FROM users ORDER BY username')
    users = cursor.fetchall()
    
    conn.close()
    return users

def get_user_posts(username):
    """Get posts by a specific user"""
    conn = connect_database()
    cursor = conn.cursor()
    
    cursor.execute('''
        SELECT p.title, p.content, p.created_date, u.username
        FROM posts p
        JOIN users u ON p.user_id = u.id
        WHERE u.username = ?
        ORDER BY p.created_date DESC
    ''', (username,))
    
    posts = cursor.fetchall()
    conn.close()
    return posts

def search_posts(keyword):
    """Search posts by keyword in title or content"""
    conn = connect_database()
    cursor = conn.cursor()
    
    cursor.execute('''
        SELECT p.title, p.content, u.username
        FROM posts p
        JOIN users u ON p.user_id = u.id
        WHERE p.title LIKE ? OR p.content LIKE ?
        ORDER BY p.title
    ''', (f'%{keyword}%', f'%{keyword}%'))
    
    results = cursor.fetchall()
    conn.close()
    return results

# Test queries
print("All users:")
users = get_all_users()
for user in users:
    print(f"  {user['username']} - {user['email']}")

print("\nPosts by alice123:")
alice_posts = get_user_posts('alice123')
for post in alice_posts:
    print(f"  {post['title']}")

print("\nPosts containing 'Python':")
python_posts = search_posts('Python')
for post in python_posts:
    print(f"  {post['title']} by {post['username']}")

📋 SQL Commands Reference

Operation	SQL Example	Purpose
Select	`SELECT * FROM table`	Retrieve data
Insert	`INSERT INTO table VALUES (...)`	Add new record
Update	`UPDATE table SET col=val WHERE ...`	Modify record
Delete	`DELETE FROM table WHERE ...`	Remove record
Join	`SELECT * FROM a JOIN b ON ...`	Combine tables
Filter	`WHERE condition`	Filter results
Sort	`ORDER BY column`	Sort results
Group	`GROUP BY column`	Aggregate data

🔧 Database Operations

Inserting and Updating Data

import sqlite3

class DatabaseManager:
    def __init__(self, db_path='example.db'):
        self.db_path = db_path
    
    def get_connection(self):
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        return conn
    
    def create_user(self, username, email):
        """Create a new user"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            cursor.execute(
                'INSERT INTO users (username, email) VALUES (?, ?)',
                (username, email)
            )
            user_id = cursor.lastrowid
            conn.commit()
            print(f"User created with ID: {user_id}")
            return user_id
        except sqlite3.IntegrityError as e:
            print(f"Error creating user: {e}")
            return None
        finally:
            conn.close()
    
    def update_user_email(self, username, new_email):
        """Update user's email address"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        cursor.execute(
            'UPDATE users SET email = ? WHERE username = ?',
            (new_email, username)
        )
        
        rows_affected = cursor.rowcount
        conn.commit()
        conn.close()
        
        if rows_affected > 0:
            print(f"Updated email for {username}")
            return True
        else:
            print(f"User {username} not found")
            return False
    
    def create_post(self, title, content, username):
        """Create a new post"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        # First, get user ID
        cursor.execute('SELECT id FROM users WHERE username = ?', (username,))
        user_row = cursor.fetchone()
        
        if not user_row:
            print(f"User {username} not found")
            conn.close()
            return None
        
        user_id = user_row['id']
        
        # Create post
        cursor.execute(
            'INSERT INTO posts (title, content, user_id) VALUES (?, ?, ?)',
            (title, content, user_id)
        )
        
        post_id = cursor.lastrowid
        conn.commit()
        conn.close()
        
        print(f"Post created with ID: {post_id}")
        return post_id
    
    def delete_post(self, post_id):
        """Delete a post by ID"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        cursor.execute('DELETE FROM posts WHERE id = ?', (post_id,))
        rows_affected = cursor.rowcount
        conn.commit()
        conn.close()
        
        if rows_affected > 0:
            print(f"Post {post_id} deleted")
            return True
        else:
            print(f"Post {post_id} not found")
            return False

# Test database manager
db = DatabaseManager()

# Create new user
new_user_id = db.create_user('diana456', 'diana@example.com')

# Update user email
db.update_user_email('diana456', 'diana.new@example.com')

# Create post
if new_user_id:
    post_id = db.create_post('New User Post', 'This is my first post!', 'diana456')
    
    # Delete post
    if post_id:
        db.delete_post(post_id)

Data Analysis and Aggregation

import sqlite3

def get_user_statistics():
    """Get statistics about users and posts"""
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()
    
    # Count total users
    cursor.execute('SELECT COUNT(*) FROM users')
    total_users = cursor.fetchone()[0]
    
    # Count total posts
    cursor.execute('SELECT COUNT(*) FROM posts')
    total_posts = cursor.fetchone()[0]
    
    # Get posts per user
    cursor.execute('''
        SELECT u.username, COUNT(p.id) as post_count
        FROM users u
        LEFT JOIN posts p ON u.id = p.user_id
        GROUP BY u.id, u.username
        ORDER BY post_count DESC
    ''')
    user_post_counts = cursor.fetchall()
    
    # Get most recent posts
    cursor.execute('''
        SELECT p.title, u.username, p.created_date
        FROM posts p
        JOIN users u ON p.user_id = u.id
        ORDER BY p.created_date DESC
        LIMIT 3
    ''')
    recent_posts = cursor.fetchall()
    
    conn.close()
    
    return {
        'total_users': total_users,
        'total_posts': total_posts,
        'user_post_counts': user_post_counts,
        'recent_posts': recent_posts
    }

def search_database(search_term):
    """Search across users and posts"""
    conn = sqlite3.connect('example.db')
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    
    # Search in usernames and emails
    cursor.execute('''
        SELECT 'user' as type, username as title, email as content
        FROM users
        WHERE username LIKE ? OR email LIKE ?
    ''', (f'%{search_term}%', f'%{search_term}%'))
    
    user_results = cursor.fetchall()
    
    # Search in post titles and content
    cursor.execute('''
        SELECT 'post' as type, p.title, p.content, u.username as author
        FROM posts p
        JOIN users u ON p.user_id = u.id
        WHERE p.title LIKE ? OR p.content LIKE ?
    ''', (f'%{search_term}%', f'%{search_term}%'))
    
    post_results = cursor.fetchall()
    
    conn.close()
    
    return {
        'users': user_results,
        'posts': post_results
    }

# Get statistics
stats = get_user_statistics()
print(f"Database Statistics:")
print(f"  Total users: {stats['total_users']}")
print(f"  Total posts: {stats['total_posts']}")

print("\nPosts per user:")
for username, count in stats['user_post_counts']:
    print(f"  {username}: {count} posts")

print("\nRecent posts:")
for title, username, date in stats['recent_posts']:
    print(f"  '{title}' by {username}")

# Search database
search_results = search_database('Python')
print(f"\nSearch results for 'Python':")

if search_results['users']:
    print("Users:")
    for user in search_results['users']:
        print(f"  {user['title']} - {user['content']}")

if search_results['posts']:
    print("Posts:")
    for post in search_results['posts']:
        print(f"  '{post['title']}' by {post['author']}")

🎯 Key Takeaways

🚀 What's Next?

Learn techniques for processing large files efficiently without overwhelming system memory.

Continue to: Process Large Files

Online Python

💾 Working with Databases

Track Your Learning Progress