15 - Database

📌 What You'll Learn

• •SQLite basics
• •CRUD operations
• •Parameterized queries
• •Context managers for connections
• •SQLAlchemy ORM basics
• •Database best practices

---

💾 SQLite Basics

SQLite is a lightweight, file-based database built into Python.

Connection and Cursor

import sqlite3

# Connect to database (creates if not exists)
conn = sqlite3.connect('example.db')

# Create cursor for executing SQL
cursor = conn.cursor()

# Execute SQL
cursor.execute("SELECT sqlite_version()")
print(cursor.fetchone())

# Close connection
conn.close()

Context Manager (Recommended)

import sqlite3

with sqlite3.connect('example.db') as conn:
    cursor = conn.cursor()
    # Database operations here
    # Auto-commits on success, rollbacks on error

---

📊 Creating Tables

import sqlite3

with sqlite3.connect('example.db') as conn:
    cursor = conn.cursor()

    # Create 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
        )
    ''')

    # Create another 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()

---

✏️ CRUD Operations

Create (INSERT)

import sqlite3

with sqlite3.connect('example.db') as conn:
    cursor = conn.cursor()

    # Insert single row
    cursor.execute('''
        INSERT INTO users (name, email, age)
        VALUES (?, ?, ?)
    ''', ('Alice', 'alice@example.com', 25))

    # Get inserted row id
    print(f"Inserted ID: {cursor.lastrowid}")

    # Insert multiple rows
    users = [
        ('Bob', 'bob@example.com', 30),
        ('Charlie', 'charlie@example.com', 35),
    ]
    cursor.executemany('''
        INSERT INTO users (name, email, age)
        VALUES (?, ?, ?)
    ''', users)

    conn.commit()

Read (SELECT)

import sqlite3

with sqlite3.connect('example.db') as conn:
    cursor = conn.cursor()

    # Fetch one
    cursor.execute('SELECT * FROM users WHERE id = ?', (1,))
    user = cursor.fetchone()
    print(f"One user: {user}")

    # Fetch all
    cursor.execute('SELECT * FROM users')
    all_users = cursor.fetchall()
    print(f"All users: {all_users}")

    # Iterate with cursor
    cursor.execute('SELECT name, email FROM users')
    for row in cursor:
        print(f"Name: {row[0]}, Email: {row[1]}")

    # Fetch with limit
    cursor.execute('SELECT * FROM users LIMIT ?', (2,))
    limited = cursor.fetchall()

Update

import sqlite3

with sqlite3.connect('example.db') as conn:
    cursor = conn.cursor()

    cursor.execute('''
        UPDATE users
        SET age = ?
        WHERE name = ?
    ''', (26, 'Alice'))

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

Delete

import sqlite3

with sqlite3.connect('example.db') as conn:
    cursor = conn.cursor()

    cursor.execute('DELETE FROM users WHERE id = ?', (3,))

    print(f"Rows deleted: {cursor.rowcount}")
    conn.commit()

---

🔒 Parameterized Queries (SQL Injection Prevention)

# NEVER do this (SQL injection vulnerable!)
name = "Alice'; DROP TABLE users;--"
# cursor.execute(f"SELECT * FROM users WHERE name = '{name}'")  # DANGEROUS!

# ALWAYS use parameterized queries
cursor.execute("SELECT * FROM users WHERE name = ?", (name,))

# Named parameters
cursor.execute('''
    SELECT * FROM users
    WHERE name = :name AND age > :min_age
''', {'name': 'Alice', 'min_age': 18})

---

📦 Row Factory (Dict Results)

import sqlite3

with sqlite3.connect('example.db') as conn:
    # Return rows as dictionaries
    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']}")

    # Convert to dict
    user_dict = dict(user)
    print(user_dict)

---

🔄 Transactions

import sqlite3

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

try:
    # Start transaction (implicit)
    cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)",
                   ('Dave', 'dave@example.com'))
    cursor.execute("INSERT INTO posts (user_id, title) VALUES (?, ?)",
                   (cursor.lastrowid, 'First Post'))

    # Commit if all successful
    conn.commit()
except Exception as e:
    # Rollback on error
    conn.rollback()
    print(f"Error: {e}")
finally:
    conn.close()

---

🗂️ Database Helper Class

import sqlite3
from contextlib import contextmanager

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 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

# Usage
db = Database('example.db')
users = db.execute("SELECT * FROM users")
new_id = db.insert('users', {'name': 'Eve', 'email': 'eve@example.com'})

---

🔮 SQLAlchemy ORM

SQLAlchemy provides an Object-Relational Mapping (ORM) for Python.

Setup

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

# Create engine
engine = create_engine('sqlite:///example.db', echo=True)

# Create base class
Base = declarative_base()

# Define models
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True)

    posts = relationship('Post', back_populates='author')

    def __repr__(self):
        return f"<User(name='{self.name}')>"

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String(100), nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))

    author = relationship('User', back_populates='posts')

# Create tables
Base.metadata.create_all(engine)

# Create session
Session = sessionmaker(bind=engine)
session = Session()

CRUD with SQLAlchemy

# Create
user = User(name='Alice', email='alice@example.com')
session.add(user)
session.commit()

# Read
user = session.query(User).filter_by(name='Alice').first()
all_users = session.query(User).all()

# Update
user.email = 'newemail@example.com'
session.commit()

# Delete
session.delete(user)
session.commit()

# Query with filter
adults = session.query(User).filter(User.age >= 18).all()

---

📋 Summary

---

🎯 Best Practices

1. •Always use parameterized queries to prevent SQL injection
2. •Use context managers for connection handling
3. •Handle transactions properly with commit/rollback
4. •Use row_factory for easier data access
5. •Index frequently queried columns
6. •Close connections when done

---

🎯 Next Steps

After mastering databases, proceed to 16_networking to learn about HTTP requests, APIs, and socket programming!