Building db_utils.py

🏗️ Building db_utils.py

📋 Overview: What db_utils.py Does

The db_utils.py module provides database utility functions for:

1. DatabaseConnection class — Connects to MySQL/MariaDB
2. get_or_create_subtopic() — Find or create a course
3. insert_chapter_with_content() — Insert a chapter and its HTML content
4. 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

📦 Imports & Setup

Copyimport 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()

🔌 DatabaseConnection Class

The core class that manages all database operations.

The __init__ Method

Copyclass 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')
        }

The connect() Method

Copydef 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

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

Copydef 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

💾 Core Database Functions

get_or_create_subtopic()

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

Copydef 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()

insert_chapter_with_content()

Insert a chapter and its HTML content into the database:

Copydef 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

📌 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

Copyfrom 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

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