================================================================================ PRACTICAL PYTHON SCRIPTING - CHALLENGE 2 SOLUTION Chapter 3: Database Connections Challenge: Query and Display Results ================================================================================ PROBLEM: Create a script that: 1. Connects to the database 2. Executes a SELECT query 3. Displays results in a formatted table 4. Handles empty results gracefully SOLUTION: ================================================================================ from dotenv import load_dotenv import os import sys import mysql.connector from mysql.connector import Error # Load configuration load_dotenv() db_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': 'utf8mb4' } def display_table(columns, rows): """Display results in a formatted table.""" if not rows: print("📭 No results found") return # Calculate column widths col_widths = [] for i, col_name in enumerate(columns): # Start with column name width max_width = len(str(col_name)) # Check all rows for this column for row in rows: max_width = max(max_width, len(str(row[i]))) col_widths.append(max_width) # Print header print() header = " | ".join( str(col).ljust(col_widths[i]) for i, col in enumerate(columns) ) print(header) print("-" * len(header)) # Print rows for row in rows: row_str = " | ".join( str(val).ljust(col_widths[i]) for i, val in enumerate(row) ) print(row_str) print() print(f"📊 Total rows: {len(rows)}\n") def execute_query(connection, query, description="Query Results"): """Execute a query and display results.""" try: cursor = connection.cursor() cursor.execute(query) # Get column names columns = [desc[0] for desc in cursor.description] # Get all rows rows = cursor.fetchall() cursor.close() # Display print(f"\n{'=' * 60}") print(f"📋 {description}") print(f"{'=' * 60}") display_table(columns, rows) return rows except Error as e: print(f"❌ Query failed: {e}") return None # Main execution try: connection = mysql.connector.connect(**db_config) if not connection.is_connected(): print("❌ Failed to connect to database") sys.exit(1) print("✅ Connected to database\n") # Example Query 1: Show database information execute_query( connection, "SELECT TABLE_NAME, TABLE_TYPE FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE()", "Tables in Database" ) # Example Query 2: Show column count execute_query( connection, """ SELECT TABLE_NAME, COUNT(*) as COLUMN_COUNT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() GROUP BY TABLE_NAME ORDER BY TABLE_NAME """, "Columns per Table" ) # Example Query 3: Show a sample query (if pages table exists) try: execute_query( connection, "SELECT id, title, slug FROM pages LIMIT 5", "Sample Pages" ) except: print("(pages table not found or doesn't have expected columns)\n") except Error as e: print(f"❌ Connection error: {e.errno} - {e.msg}") sys.exit(1) finally: if 'connection' in locals() and connection.is_connected(): connection.close() print("✅ Connection closed") ================================================================================ HOW IT WORKS: ================================================================================ 1. DISPLAY_TABLE FUNCTION def display_table(columns, rows): - Takes column names and result rows - Calculates width for each column - Formats and prints a nice table 2. CALCULATE COLUMN WIDTHS for i, col_name in enumerate(columns): max_width = len(str(col_name)) for row in rows: max_width = max(max_width, len(str(row[i]))) - Start with column name length - Check each row for that column - Width is the maximum of all values - This ensures nothing gets truncated 3. PRINT HEADER header = " | ".join( str(col).ljust(col_widths[i]) ... ) - ljust(width) pads strings to the left - " | " separates columns - Creates nice aligned columns 4. PRINT ROWS for row in rows: row_str = " | ".join(...) print(row_str) - Same formatting as header - Columns line up nicely - Shows count at the end 5. EXECUTE_QUERY FUNCTION cursor.description - Returns tuple of (name, type_code, ..., ...) - desc[0] extracts just the column name cursor.fetchall() - Gets all rows at once - Returns list of tuples 6. ERROR HANDLING except Error as e: print(f"Query failed: {e}") return None - Catches query errors - Returns None if query fails - Allows script to continue safely ================================================================================ UNDERSTANDING CURSOR.DESCRIPTION: ================================================================================ cursor.description returns metadata about columns: cursor.execute("SELECT id, title, created_at FROM pages") print(cursor.description) # Output: # ( # ('id', 3, None, None, None, None, 0, 0), # ('title', 253, None, None, None, None, 0, 0), # ('created_at', 7, None, None, None, None, 0, 0) # ) Element 0: Column name ('id', 'title', 'created_at') Element 1: Type code (3=int, 253=varchar, 7=datetime) Elements 2-7: Other metadata You typically just need element 0 (the name): columns = [desc[0] for desc in cursor.description] ================================================================================ TESTING THE SOLUTION: ================================================================================ Test 1: Query a table that exists python display_results.py Output: ✅ Connected to database ============================================================ 📋 Tables in Database ============================================================ TABLE_NAME | TABLE_TYPE -----------------|----------- pages | BASE TABLE subtopics | BASE TABLE subjects | BASE TABLE page_content | BASE TABLE 📊 Total rows: 4 ============================================================ 📋 Columns per Table ============================================================ TABLE_NAME | COLUMN_COUNT -----------------|----------- page_content | 3 pages | 8 subtopics | 4 subjects | 2 📊 Total rows: 3 ... ✅ Connection closed --- Test 2: Query with no results (modify the LIMIT to 0) execute_query(connection, "SELECT * FROM pages WHERE id = -1", "Non-existent Page") Output: ============================================================ 📋 Non-existent Page ============================================================ 📭 No results found --- Test 3: Query with error (wrong table name) execute_query(connection, "SELECT * FROM nonexistent_table", "Error Test") Output: ❌ Query failed: 1146 (42S02): Table 'learning_blog.nonexistent_table' doesn't exist (Script continues without crashing) ================================================================================ ADVANCED: FORMATTED OUTPUT OPTIONS: ================================================================================ Option 1: JSON Output import json rows = cursor.fetchall() columns = [desc[0] for desc in cursor.description] # Convert to list of dictionaries results = [dict(zip(columns, row)) for row in rows] print(json.dumps(results, indent=2)) # Output: # [ # { # "id": 1, # "title": "Chapter 1" # }, # { # "id": 2, # "title": "Chapter 2" # } # ] Option 2: CSV Output import csv with open('results.csv', 'w', newline='') as f: writer = csv.writer(f) writer.writerow(columns) # Header writer.writerows(rows) # Data rows Option 3: HTML Table def to_html_table(columns, rows): html = "\n" for col in columns: html += f"" html += "\n" for row in rows: html += "" for val in row: html += f"" html += "\n" html += "
{col}
{val}
" return html ================================================================================ REAL-WORLD USAGE: ================================================================================ In production code, you'd parameterize queries: def query_pages(connection, limit=10): """Get pages from database.""" query = "SELECT id, title, slug FROM pages LIMIT %s" try: cursor = connection.cursor() cursor.execute(query, (limit,)) # ← Parameter is in tuple! rows = cursor.fetchall() cursor.close() return rows except Error as e: print(f"Query failed: {e}") return [] # Usage: pages = query_pages(connection, limit=5) # Never do this (SQL injection risk): # query = f"SELECT * FROM pages LIMIT {limit}" # ❌ DANGEROUS! # Always use parameters: # cursor.execute(query, (limit,)) # ✅ SAFE! ================================================================================ KEY TAKEAWAYS: ================================================================================ ✅ Use cursor.description to get column names ✅ Use cursor.fetchall() to get all rows ✅ Use cursor.fetchone() for single row ✅ Calculate column widths for alignment ✅ Use ljust() and rjust() for padding ✅ Handle empty results gracefully ✅ Always use try/except for queries ✅ Always close cursors ✅ Use parameterized queries (with %s) to prevent SQL injection ✅ Never format SQL with f-strings (security risk!) Query Checklist: - Get column names from cursor.description ✅ - Fetch all/one rows with fetchall/fetchone ✅ - Format data for display ✅ - Handle empty results ✅ - Handle query errors ✅ - Close cursor ✅ - Use parameterized queries ✅ - Never use string formatting for SQL ✅ Display Checklist: - Calculate column widths ✅ - Align columns nicely ✅ - Show row count ✅ - Handle different data types ✅ - Format dates/times nicely (optional) ✅ ================================================================================