Building db_utils.py

Practical Python Scripting — Building db_utils.py

Practical Python Scripting

Course 1 · Chapter 4 · Building db_utils.py

🏗️ Building db_utils.py

This chapter is a complete line-by-line breakdown of the actual db_utils.py module used in production to populate the learning_blog database with course chapters. You'll see how everything from chapters 1-3 comes together: paths, configuration, database connections, error handling, and real-world SQL operations.

📋 Overview: What db_utils.py Does

The db_utils.py module provides database utility functions for:

DatabaseConnection class — Connects to MySQL/MariaDB
get_or_create_subtopic() — Find or create a course
insert_chapter_with_content() — Insert a chapter and its HTML content
rebuild_course() — Rebuild an entire course from chapters

It's used by rebuild_typescript_course.py to populate the database with generated course chapters.

📁 File Location

debserver/claude/generated/scripts/python/db_utils.py

📦 Imports & Setup

import mysql.connector
from mysql.connector import Error
import os
from dotenv import load_dotenv
from typing import Optional, Dict, List, Tuple
from urllib.parse import quote

# Load environment variables from .env file
load_dotenv()
        

✅ WHAT EACH IMPORT DOES:

mysql.connector — Database connection library
Error — Database error exception class
dotenv — Load .env configuration
typing — Type hints for function parameters
urllib.parse.quote — URL-encode strings for slug generation

🔌 DatabaseConnection Class

The core class that manages all database operations.

The init Method

class DatabaseConnection:
    """Manage MySQL/MariaDB database connections."""

    def __init__(self):
        """Initialize connection from environment variables."""
        self.connection = None

        # Get credentials from environment
        self.config = {
            'host': os.environ.get('DB_HOST'),
            'user': os.environ.get('DB_USER'),
            'password': os.environ.get('DB_PASS'),
            'database': os.environ.get('DB_NAME'),
            'port': int(os.environ.get('DB_PORT', '3306')),
            'charset': os.environ.get('DB_CHARSET', 'utf8mb4')
        }
        

🤔 WHY THIS STRUCTURE:

Loads credentials from .env (not hard-coded)
Sets sensible defaults (port=3306, charset=utf8mb4)
Stores config for reuse across method calls
Converts port string to integer

The connect() Method

def connect(self) -> bool:
    """Establish database connection."""
    try:
        self.connection = mysql.connector.connect(**self.config)
        print("✅ Database connection established")
        return True

    except Error as e:
        print(f"❌ Connection failed: {e}")
        return False
        

The disconnect() Method

def disconnect(self) -> None:
    """Close database connection."""
    if self.connection and self.connection.is_connected():
        self.connection.close()
        print("✅ Database connection closed")
        

🔧 Helper Functions

generate_slug() Function

Convert a title into a URL-safe slug:

def generate_slug(title: str) -> str:
    """
    Convert title to URL-safe slug.

    Example: "Why TypeScript & Setup" → "why-typescript-setup"
    """
    # Convert to lowercase
    slug = title.lower()

    # Replace spaces with hyphens
    slug = slug.replace(' ', '-')

    # Remove special characters
    slug = quote(slug, safe='-')

    # Remove multiple consecutive hyphens
    while '--' in slug:
        slug = slug.replace('--', '-')

    return slug
        

✅ WHAT IT DOES:

Converts "Why TypeScript & Setup" → "why-typescript-setup"
.lower() makes it lowercase
.replace() converts spaces to hyphens
quote() removes special characters
Final loop removes double hyphens

💾 Core Database Functions

get_or_create_subtopic()

Find a course (subtopic) or create it if it doesn't exist:

def get_or_create_subtopic(
    db: DatabaseConnection,
    subject_id: int,
    subtopic_name: str,
    description: str = ""
) -> Optional[int]:
    """
    Get or create a subtopic (course).
    Returns the subtopic ID, or None if error.
    """
    cursor = db.connection.cursor()

    try:
        # Check if subtopic already exists
        query = "SELECT id FROM subtopics WHERE name = %s AND subject_id = %s"
        cursor.execute(query, (subtopic_name, subject_id))
        result = cursor.fetchone()

        if result:
            return result[0]  # Found existing subtopic

        # Create new subtopic
        insert_query = "INSERT INTO subtopics (subject_id, name, description) VALUES (%s, %s, %s)"
        cursor.execute(insert_query, (subject_id, subtopic_name, description))
        db.connection.commit()

        subtopic_id = cursor.lastrowid
        print(f"✅ Created subtopic '{subtopic_name}' with ID {subtopic_id}")

        return subtopic_id

    except Error as e:
        print(f"❌ Error: {e}")
        db.connection.rollback()
        return None

    finally:
        cursor.close()
        

✅ WHAT IT DOES:

Checks if subtopic (course) already exists
If exists: returns its ID
If not: creates it and returns new ID
Uses cursor.lastrowid to get the new ID
Commits transaction if successful
Rolls back if error occurs

insert_chapter_with_content()

Insert a chapter and its HTML content into the database:

def insert_chapter_with_content(
    db: DatabaseConnection,
    subtopic_id: int,
    chapter_title: str,
    html_content: str,
    sort_order: int = 10
) -> Optional[int]:
    """
    Insert a chapter with HTML content.
    Returns page_id, or None if error.
    """
    cursor = db.connection.cursor()

    try:
        # Generate slug from title
        slug = generate_slug(chapter_title)

        # Insert into pages table
        page_query = """
            INSERT INTO pages (subtopic_id, title, slug, sort_order)
            VALUES (%s, %s, %s, %s)
        """
        cursor.execute(page_query, (subtopic_id, chapter_title, slug, sort_order))
        page_id = cursor.lastrowid

        # Insert HTML content into page_content table
        content_query = "INSERT INTO page_content (page_id, body) VALUES (%s, %s)"
        cursor.execute(content_query, (page_id, html_content))

        db.connection.commit()

        return page_id

    except Error as e:
        print(f"❌ Error inserting chapter: {e}")
        db.connection.rollback()
        return None

    finally:
        cursor.close()
        

🎯 Why This Pattern Works

💡 Key Design Decisions

Separation of Concerns: DatabaseConnection handles connections, helper functions handle operations
Parameterized Queries: All queries use %s placeholders (prevents SQL injection)
Error Handling: Every database operation has try/except
Transactions: Commit on success, rollback on error
Resource Cleanup: finally blocks ensure cursors always close
Type Hints: Return types are clear (int, bool, None, etc.)
Configuration: All credentials come from .env, never hard-coded

📌 Common Patterns Used

Pattern	Purpose	Example
Parameterized Query	Prevent SQL injection	`cursor.execute("SELECT * FROM t WHERE id = %s", (id,))`
Check Before Insert	Avoid duplicates	SELECT first, then INSERT if not found
lastrowid	Get inserted row ID	`page_id = cursor.lastrowid`
Commit/Rollback	Transaction control	commit() on success, rollback() on error
try/except/finally	Error handling + cleanup	finally block always closes cursor

🌍 Real-World Usage Example

from db_utils import DatabaseConnection, get_or_create_subtopic, insert_chapter_with_content

# Connect to database
db = DatabaseConnection()
if not db.connect():
    exit(1)

try:
    # Get or create TypeScript course
    subtopic_id = get_or_create_subtopic(
        db,
        subject_id=3,  # Programming
        subtopic_name="TypeScript Fundamentals",
        description="Learn TypeScript from basics to generics"
    )

    # Insert a chapter
    page_id = insert_chapter_with_content(
        db,
        subtopic_id=subtopic_id,
        chapter_title="Why TypeScript & Setup",
        html_content="<h1>Chapter 1</h1>...",
        sort_order=10
    )

    print(f"✅ Chapter inserted with page_id: {page_id}")

finally:
    db.disconnect()
        

💻 Coding Challenges

Challenge 1: Test db_utils.py Connection

Create a script that:

Imports DatabaseConnection from db_utils
Attempts to connect using credentials from .env
Shows connection status and database info
Safely closes the connection

Goal: Verify db_utils works with your database.

→ Solution

Challenge 2: Insert a Test Subtopic

Create a script that:

Connects using db_utils
Uses get_or_create_subtopic() to create a test course
Shows the returned subtopic_id
Calls it again to verify it returns the same ID (no duplicate)

Goal: Practice using db_utils helper functions.

→ Solution

Challenge 3: Complete Course Integration

Create a script that:

Creates a test subtopic
Inserts 3 test chapters with HTML content
Verifies chapters are in database by querying them back
Reports success with all IDs

Goal: Full database integration workflow.

→ Solution

🎯 What's Next

Chapter 5 shows how rebuild_typescript_course.py uses db_utils to automatically populate the database by reading generated HTML chapter files and inserting them with all the functions you've learned here.

Building db_utils.py

🏗️ Building db_utils.py

📋 Overview: What db_utils.py Does

📁 File Location

📦 Imports & Setup

🔌 DatabaseConnection Class

The __init__ Method

The connect() Method

The disconnect() Method

🔧 Helper Functions

generate_slug() Function

💾 Core Database Functions

get_or_create_subtopic()

insert_chapter_with_content()

🎯 Why This Pattern Works

📌 Common Patterns Used

🌍 Real-World Usage Example

💻 Coding Challenges

Challenge 1: Test db_utils.py Connection

Challenge 2: Insert a Test Subtopic

Challenge 3: Complete Course Integration

🎯 What's Next

The init Method