Python MySQL - Best Practices

What You'll Learn Here

Security practices — protecting your database
Performance optimization — making queries faster
Code organization — writing maintainable code
Connection management — handling connections properly
Query best practices — writing efficient SQL
Error handling — robust error management

Why Best Practices Matter

Following best practices when working with MySQL in Python is essential for building secure, performant, and maintainable applications. Poor practices can lead to security breaches, slow performance, and difficult-to-maintain code.

💡 Key concept: Best practices are not just rules — they're lessons learned from years of experience. Following them saves you from common mistakes.

Security

Protect against SQL injection, data breaches, and unauthorized access.

Performance

Write efficient queries and manage connections properly for speed.

Maintainability

Organize code so it's easy to understand, modify, and debug.

The Impact of Following Best Practices

# ============================================================
# WITHOUT BEST PRACTICES
# ============================================================

# ❌ Insecure, slow, hard to maintain
def get_user(name):
    cursor.execute(f"SELECT * FROM users WHERE name = '{name}'")  # SQL Injection!
    return cursor.fetchone()

# Every request creates a new connection
def get_data():
    connection = mysql.connector.connect(...)  # Slow!
    # ... operations
    return result

# No error handling
def update_user(id, data):
    cursor.execute("UPDATE users SET ...")
    # If this fails, app crashes!

# ============================================================
# WITH BEST PRACTICES
# ============================================================

# ✅ Secure, fast, maintainable
def get_user(name):
    cursor.execute("SELECT * FROM users WHERE name = %s", (name,))  # Safe!
    return cursor.fetchone()

# Connection pooling for efficiency
pool = create_connection_pool()

# Error handling with try/except
def update_user(id, data):
    try:
        cursor.execute("UPDATE users SET ...")
        connection.commit()
    except Exception as e:
        connection.rollback()
        logger.error(f"Update failed: {e}")
        raise

# The difference is huge!

Key point: Best practices make your application secure, fast, and reliable.

Quick Check: Why are best practices important? (Answer: They ensure security, performance, and maintainability)

Security Best Practices

Protecting Your Database

# ============================================================
# SECURITY BEST PRACTICES
# ============================================================

print("1. ALWAYS USE PARAMETERIZED QUERIES")
print("   - Use %s placeholders")
print("   - Never use string concatenation or f-strings")
print("   - Protects against SQL injection")

print("\n2. NEVER STORE PASSWORDS IN PLAIN TEXT")
print("   - Hash passwords using bcrypt or argon2")
print("   - Never store actual passwords")

print("\n3. USE ENVIRONMENT VARIABLES")
print("   - Store credentials in environment variables")
print("   - Never hardcode passwords in code")

print("\n4. USE THE LEAST PRIVILEGE PRINCIPLE")
print("   - Database user should have minimal permissions")
print("   - Only what the application needs")

print("\n5. VALIDATE USER INPUT")
print("   - Validate all input before using in queries")
print("   - Check types, lengths, and format")

print("\n6. USE HTTPS FOR CONNECTIONS")
print("   - Encrypt data in transit")
print("   - Use SSL/TLS for database connections")

print("\n7. KEEP DRIVERS UPDATED")
print("   - Regular updates for security patches")
print("   - Update mysql-connector-python")

print("\n8. LOG SUSPICIOUS ACTIVITY")
print("   - Monitor for SQL injection attempts")
print("   - Log and alert on suspicious queries")

# ============================================================
# IMPLEMENTATION EXAMPLE
# ============================================================

import os
import bcrypt
import mysql.connector

class SecureDatabase:
    """Database with security best practices"""
    
    def __init__(self):
        # Credentials from environment variables
        self.config = {
            'host': os.getenv('DB_HOST', 'localhost'),
            'user': os.getenv('DB_USER', 'root'),
            'password': os.getenv('DB_PASSWORD'),
            'database': os.getenv('DB_NAME', 'myapp_db')
        }
        
        if not self.config['password']:
            raise ValueError("DB_PASSWORD environment variable is required")
    
    def hash_password(self, password):
        """Hash password using bcrypt"""
        salt = bcrypt.gensalt()
        return bcrypt.hashpw(password.encode(), salt)
    
    def verify_password(self, password, hashed):
        """Verify password against hash"""
        return bcrypt.checkpw(password.encode(), hashed)
    
    def create_user(self, username, password, email):
        """Create user with hashed password"""
        hashed = self.hash_password(password)
        
        query = "INSERT INTO users (username, password_hash, email) VALUES (%s, %s, %s)"
        # Parameterized query - safe from injection
        self.cursor.execute(query, (username, hashed, email))
        self.connection.commit()

Security summary:

Parameterized queries — prevent SQL injection
Environment variables — keep credentials safe
Least privilege — minimize access
Input validation — verify all data
Password hashing — never store plain text

Quick Check: What is the #1 way to prevent SQL injection? (Answer: Use parameterized queries)

Performance Best Practices

Making Your Application Fast

# ============================================================
# PERFORMANCE BEST PRACTICES
# ============================================================

print("1. USE CONNECTION POOLING")
print("   - Reuse connections instead of creating new ones")
print("   - Significant speed improvement")

print("\n2. SELECT ONLY NEEDED COLUMNS")
print("   - Use specific columns instead of SELECT *")
print("   - Reduces data transfer")

print("\n3. USE INDEXES WISELY")
print("   - Index columns used in WHERE, JOIN, ORDER BY")
print("   - But don't over-index (slows INSERT/UPDATE)")

print("\n4. USE BATCH OPERATIONS")
print("   - Use executemany() for multiple inserts")
print("   - Reduces round trips")

print("\n5. USE LIMIT FOR LARGE RESULTS")
print("   - Always use LIMIT for large datasets")
print("   - Implement pagination")

print("\n6. AVOID SELECT * IN JOINS")
print("   - Specify only needed columns")
print("   - Reduces data transfer")

print("\n7. USE EXPLAIN TO ANALYZE QUERIES")
print("   - Check query execution plans")
print("   - Identify slow queries")

print("\n8. CACHE FREQUENTLY ACCESSED DATA")
print("   - Use Redis or Memcached")
print("   - Reduce database load")

# ============================================================
# PERFORMANCE COMPARISON
# ============================================================

print("""
┌─────────────────────────────┬─────────────────────────────────────┐
│ Practice                    │ Impact                            │
├─────────────────────────────┼─────────────────────────────────────┤
│ Connection Pooling          │ 5-10x faster                      │
│ Specific Columns            │ 2-3x faster                      │
│ Indexes                     │ 10-100x faster for searches       │
│ Batch Operations            │ 5-10x faster for inserts          │
│ LIMIT                       │ Prevents timeouts                 │
│ Caching                     │ 100x faster for repeated queries  │
└─────────────────────────────┴─────────────────────────────────────┘
""")

Performance summary:

Connection pooling — reuse connections
Specific columns — reduce data transfer
Indexes — speed up searches
Batch operations — reduce round trips
LIMIT — prevent large result sets
Caching — avoid repeated queries

Quick Check: What is the most important performance practice? (Answer: Use connection pooling and indexes)

Code Organization

Writing Maintainable Database Code

# ============================================================
# CODE ORGANIZATION BEST PRACTICES
# ============================================================

print("1. USE A DATABASE LAYER")
print("   - Separate database code from business logic")
print("   - Use a dedicated Database class")

print("\n2. USE CONTEXT MANAGERS")
print("   - Use 'with' statements for connections")
print("   - Automatic cleanup")

print("\n3. USE CONSTANTS FOR QUERIES")
print("   - Store queries as constants")
print("   - Easier to maintain")

print("\n4. USE TYPES AND DOCSTRINGS")
print("   - Add type hints")
print("   - Document functions and queries")

print("\n5. SEPARATE READ AND WRITE OPERATIONS")
print("   - Different functions for different operations")
print("   - Clear separation of concerns")

print("\n6. USE REPOSITORY PATTERN")
print("   - One repository per table/model")
print("   - Encapsulates database operations")

# ============================================================
# EXAMPLE: CLEAN DATABASE LAYER
# ============================================================

from contextlib import contextmanager

class Database:
    """Database layer with clean organization"""
    
    def __init__(self, config):
        self.config = config
        self.pool = self._create_pool()
    
    def _create_pool(self):
        """Create connection pool"""
        from mysql.connector import pooling
        return pooling.MySQLConnectionPool(**self.config)
    
    @contextmanager
    def get_connection(self):
        """Context manager for connections"""
        connection = self.pool.get_connection()
        try:
            yield connection
        finally:
            connection.close()
    
    def execute_query(self, query, params=None, fetch_all=True):
        """Execute query with automatic connection management"""
        with self.get_connection() as conn:
            cursor = conn.cursor()
            try:
                if params:
                    cursor.execute(query, params)
                else:
                    cursor.execute(query)
                
                if query.strip().upper().startswith("SELECT"):
                    return cursor.fetchall() if fetch_all else cursor.fetchone()
                else:
                    conn.commit()
                    return cursor.rowcount
            except Exception as e:
                conn.rollback()
                raise

# Repository pattern
class UserRepository:
    """User repository for database operations"""
    
    def __init__(self, db):
        self.db = db
    
    def find_by_id(self, user_id):
        query = "SELECT * FROM users WHERE id = %s"
        return self.db.execute_query(query, (user_id,), fetch_all=False)
    
    def find_by_email(self, email):
        query = "SELECT * FROM users WHERE email = %s"
        return self.db.execute_query(query, (email,), fetch_all=False)
    
    def create(self, username, email, password_hash):
        query = "INSERT INTO users (username, email, password_hash) VALUES (%s, %s, %s)"
        return self.db.execute_query(query, (username, email, password_hash))

Code organization summary:

Database layer — separate from business logic
Context managers — automatic cleanup
Repository pattern — organized data access
Constants — maintainable queries
Type hints — better code quality

Quick Check: What pattern is recommended for organizing database code? (Answer: Repository pattern with a database layer)

Connection Management

Handling Database Connections Properly

# ============================================================
# CONNECTION MANAGEMENT BEST PRACTICES
# ============================================================

print("1. USE CONNECTION POOLING")
print("   - Create a pool once, reuse connections")
print("   - Prevents connection overhead")

print("\n2. ALWAYS CLOSE CONNECTIONS")
print("   - Use context managers (with statement)")
print("   - Never leave connections open")

print("\n3. SET CONNECTION TIMEOUTS")
print("   - Prevent hanging on dead connections")
print("   - Set reasonable timeout values")

print("\n4. HANDLE CONNECTION LOSS")
print("   - Check if connection is alive")
print("   - Reconnect if needed")

print("\n5. USE SINGLE CONNECTION PER REQUEST")
print("   - In web apps, one connection per request")
print("   - Don't create multiple connections")

print("\n6. MONITOR CONNECTION USAGE")
print("   - Track active connections")
print("   - Detect connection leaks")

# ============================================================
# IMPLEMENTATION EXAMPLE
# ============================================================

import time
from contextlib import contextmanager

class ConnectionManager:
    """Connection management with retry and pooling"""
    
    def __init__(self, config, pool_size=5):
        self.config = config
        self.pool_size = pool_size
        self.pool = None
        self._create_pool()
    
    def _create_pool(self):
        """Create connection pool"""
        config = self.config.copy()
        config.update({
            'pool_name': 'app_pool',
            'pool_size': self.pool_size,
            'pool_reset_session': True
        })
        from mysql.connector import pooling
        self.pool = pooling.MySQLConnectionPool(**config)
    
    @contextmanager
    def get_connection(self, retries=3, delay=1):
        """Get connection with retry logic"""
        for attempt in range(retries):
            try:
                connection = self.pool.get_connection()
                try:
                    yield connection
                finally:
                    connection.close()  # Returns to pool
                return
            except Exception as e:
                if attempt < retries - 1:
                    time.sleep(delay * (attempt + 1))
                else:
                    raise
    
    def check_connection(self):
        """Check if pool has available connections"""
        # Implementation depends on connector version
        return True

# Usage
manager = ConnectionManager(config)

with manager.get_connection() as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT 1")
    result = cursor.fetchone()
    # Connection automatically returned to pool

Connection management summary:

Connection pooling — reuse connections
Context managers — automatic cleanup
Timeouts — prevent hanging
Reconnection — handle connection loss
Monitoring — track usage

Quick Check: What happens when you close a connection from a pool? (Answer: It returns to the pool, doesn't close)

Query Best Practices

Writing Better SQL Queries

# ============================================================
# QUERY BEST PRACTICES
# ============================================================

print("1. USE EXPLICIT COLUMN NAMES")
print("   - Don't use SELECT *")
print("   - List only needed columns")

print("\n2. USE WHERE CLAUSE EFFECTIVELY")
print("   - Filter at the database level")
print("   - Don't filter in Python")

print("\n3. USE LIMIT FOR PAGINATION")
print("   - Use LIMIT and OFFSET")
print("   - Avoid fetching all data")

print("\n4. AVOID N+1 QUERIES")
print("   - Use JOINs to fetch related data")
print("   - Don't query in loops")

print("\n5. USE EXPLAIN TO ANALYZE")
print("   - Check query execution plans")
print("   - Identify slow queries")

print("\n6. USE PARAMETERIZED QUERIES")
print("   - Always use %s placeholders")
print("   - Security and performance")

print("\n7. USE TRANSACTIONS FOR RELATED OPERATIONS")
print("   - Group related operations")
print("   - Commit or rollback as a unit")

print("\n8. AVOID FUNCTIONS ON INDEXED COLUMNS")
print("   - WHERE YEAR(date) = 2024 is slow")
print("   - Use date BETWEEN '2024-01-01' AND '2024-12-31'")

# ============================================================
# QUERY COMPARISON
# ============================================================

# ❌ BAD: SELECT *
cursor.execute("SELECT * FROM users")
# Fetches all columns, even unused ones

# ✅ GOOD: Specific columns
cursor.execute("SELECT id, username, email FROM users")
# Only what you need

# ❌ BAD: Filtering in Python
cursor.execute("SELECT * FROM users")
for user in cursor.fetchall():
    if user[2] == 'admin':  # Filter in Python
        process(user)

# ✅ GOOD: Filter in SQL
cursor.execute("SELECT * FROM users WHERE role = 'admin'")
for user in cursor.fetchall():
    process(user)

# ❌ BAD: N+1 queries
for user in users:
    cursor.execute("SELECT * FROM orders WHERE user_id = %s", (user['id'],))

# ✅ GOOD: Single query with JOIN
cursor.execute("""
    SELECT u.*, o.* 
    FROM users u
    LEFT JOIN orders o ON u.id = o.user_id
""")

Query best practices summary:

Explicit columns — avoid SELECT *
Filter at database — use WHERE effectively
LIMIT — for pagination
Avoid N+1 — use JOINs
EXPLAIN — analyze queries
Parameterized — always for security

Quick Check: What is the N+1 query problem? (Answer: Querying in a loop instead of using a JOIN)

Error Handling

Handling Errors Gracefully

# ============================================================
# ERROR HANDLING BEST PRACTICES
# ============================================================

print("1. USE TRY/EXCEPT BLOCKS")
print("   - Always wrap database operations in try/except")
print("   - Never let errors crash your program")

print("\n2. HANDLE SPECIFIC ERROR TYPES")
print("   - Different errors need different handling")
print("   - Use specific exception classes")

print("\n3. USE ROLLBACK ON ERRORS")
print("   - Rollback on any database error")
print("   - Keep data consistent")

print("\n4. LOG ERRORS FOR DEBUGGING")
print("   - Log all database errors")
print("   - Include context (query, params)")

print("\n5. PROVIDE USER-FRIENDLY MESSAGES")
print("   - Don't show raw database errors")
print("   - Show helpful messages")

print("\n6. RETRY TRANSIENT ERRORS")
print("   - Retry on timeouts and network issues")
print("   - Use exponential backoff")

print("\n7. NEVER IGNORE ERRORS")
print("   - Don't use empty except blocks")
print("   - Always handle or log errors")

# ============================================================
# IMPLEMENTATION EXAMPLE
# ============================================================

import logging
from mysql.connector import Error, IntegrityError, OperationalError

logger = logging.getLogger(__name__)

def execute_safely(query, params=None):
    """Execute query with comprehensive error handling"""
    try:
        if params:
            cursor.execute(query, params)
        else:
            cursor.execute(query)
        connection.commit()
        return cursor.fetchall() if query.strip().upper().startswith("SELECT") else cursor.rowcount
        
    except IntegrityError as e:
        # Data integrity errors (duplicate, foreign key)
        connection.rollback()
        if e.errno == 1062:
            logger.warning(f"Duplicate entry: {e}")
            return {"error": "duplicate", "message": "Record already exists"}
        elif e.errno == 1452:
            logger.warning(f"Foreign key violation: {e}")
            return {"error": "foreign_key", "message": "Referenced record not found"}
        else:
            logger.error(f"Integrity error: {e}")
            return {"error": "integrity", "message": "Data integrity error"}
            
    except OperationalError as e:
        # Connection issues, timeouts
        connection.rollback()
        logger.error(f"Operational error: {e}")
        # Could retry here
        return {"error": "operational", "message": "Database operation failed"}
        
    except Error as e:
        # Other database errors
        connection.rollback()
        logger.error(f"Database error: {e}")
        return {"error": "database", "message": "Database error occurred"}
        
    except Exception as e:
        # Unexpected errors
        connection.rollback()
        logger.error(f"Unexpected error: {e}")
        return {"error": "unknown", "message": "An unexpected error occurred"}

Error handling summary:

Try/except — always handle errors
Specific errors — different handling for different errors
Rollback — on any error
Logging — for debugging
User messages — friendly and helpful
Retry — for transient errors

Quick Check: What should you do on any database error? (Answer: Rollback, log, and handle appropriately)

Real-World Example: Production-Ready Service

Building a Production-Ready Database Service

# ============================================================
# PRODUCTION-READY DATABASE SERVICE
# ============================================================

import os
import logging
import time
from contextlib import contextmanager
from mysql.connector import pooling, Error

class ProductionDatabase:
    """Production-ready database service with best practices"""
    
    def __init__(self):
        self.config = self._load_config()
        self.pool = self._create_pool()
        self.logger = self._setup_logger()
        self.max_retries = 3
        self.retry_delay = 1
    
    def _load_config(self):
        """Load configuration from environment"""
        return {
            'host': os.getenv('DB_HOST', 'localhost'),
            'user': os.getenv('DB_USER', 'root'),
            'password': os.getenv('DB_PASSWORD'),
            'database': os.getenv('DB_NAME', 'myapp_db'),
            'pool_name': os.getenv('DB_POOL_NAME', 'app_pool'),
            'pool_size': int(os.getenv('DB_POOL_SIZE', '5')),
            'charset': 'utf8mb4',
            'use_unicode': True,
            'autocommit': False
        }
    
    def _setup_logger(self):
        """Set up logging"""
        logger = logging.getLogger(__name__)
        if not logger.handlers:
            handler = logging.StreamHandler()
            formatter = logging.Formatter(
                '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
            )
            handler.setFormatter(formatter)
            logger.addHandler(handler)
            logger.setLevel(logging.INFO)
        return logger
    
    def _create_pool(self):
        """Create connection pool"""
        try:
            return pooling.MySQLConnectionPool(**self.config)
        except Exception as e:
            self.logger.error(f"Failed to create connection pool: {e}")
            raise
    
    @contextmanager
    def get_connection(self):
        """Get connection with automatic cleanup"""
        connection = None
        try:
            connection = self.pool.get_connection()
            yield connection
        except Exception as e:
            if connection:
                connection.rollback()
            self.logger.error(f"Connection error: {e}")
            raise
        finally:
            if connection:
                connection.close()
    
    def execute(self, query, params=None, retry=True):
        """Execute query with retry and error handling"""
        def _execute():
            with self.get_connection() as conn:
                cursor = conn.cursor()
                try:
                    if params:
                        cursor.execute(query, params)
                    else:
                        cursor.execute(query)
                    
                    if query.strip().upper().startswith('SELECT'):
                        return cursor.fetchall()
                    else:
                        conn.commit()
                        return cursor.rowcount
                except Exception as e:
                    conn.rollback()
                    raise
        
        if not retry:
            return _execute()
        
        for attempt in range(self.max_retries):
            try:
                return _execute()
            except Error as e:
                self.logger.warning(
                    f"Query attempt {attempt + 1} failed: {e}"
                )
                if attempt < self.max_retries - 1:
                    time.sleep(self.retry_delay * (attempt + 1))
                else:
                    self.logger.error(f"Query failed after {self.max_retries} attempts")
                    raise
    
    def get_stats(self):
        """Get database statistics"""
        try:
            result = self.execute("SELECT COUNT(*) FROM information_schema.tables")
            return {
                'status': 'healthy',
                'table_count': result[0][0] if result else 0
            }
        except:
            return {'status': 'unhealthy'}
    
    def close(self):
        """Close all connections"""
        try:
            self.pool.close()
            self.logger.info("Connection pool closed")
        except Exception as e:
            self.logger.error(f"Error closing pool: {e}")

# ============================================================
# APPLICATION ENTRY POINT
# ============================================================

def main():
    """Main application with production database"""
    # Initialize database
    db = ProductionDatabase()
    
    try:
        # Health check
        stats = db.get_stats()
        print(f"Database status: {stats['status']}")
        
        # Example operations
        result = db.execute("SELECT * FROM students LIMIT 5")
        print(f"Students: {len(result)}")
        
    except Exception as e:
        print(f"Application error: {e}")
    finally:
        db.close()

if __name__ == "__main__":
    main()

This production service includes:

Connection pooling with configuration
Context managers for cleanup
Retry logic with exponential backoff
Comprehensive logging
Error handling and rollback
Health monitoring
Environment-based configuration

Quick Check: What makes a database service production-ready? (Answer: Connection pooling, retry logic, logging, error handling, and proper configuration)

Complete Best Practices Checklist

Quick Reference for All Practices

# ============================================================
# BEST PRACTICES CHECKLIST
# ============================================================

print("SECURITY:")
print("  ☐ Use parameterized queries for all user input")
print("  ☐ Store credentials in environment variables")
print("  ☐ Hash passwords using bcrypt or argon2")
print("  ☐ Use least privilege database users")
print("  ☐ Validate all user input")
print("  ☐ Keep drivers updated")
print("  ☐ Use SSL/TLS for connections")

print("\nPERFORMANCE:")
print("  ☐ Use connection pooling")
print("  ☐ Select only needed columns")
print("  ☐ Use indexes on WHERE/JOIN columns")
print("  ☐ Use batch operations (executemany)")
print("  ☐ Use LIMIT for pagination")
print("  ☐ Avoid N+1 queries")
print("  ☐ Cache frequently accessed data")
print("  ☐ Use EXPLAIN to analyze queries")

print("\nCODE ORGANIZATION:")
print("  ☐ Use a database layer")
print("  ☐ Use repository pattern")
print("  ☐ Use context managers")
print("  ☐ Store queries as constants")
print("  ☐ Use type hints")
print("  ☐ Write docstrings")
print("  ☐ Separate read and write operations")

print("\nCONNECTION MANAGEMENT:")
print("  ☐ Use connection pooling")
print("  ☐ Always close connections")
print("  ☐ Set connection timeouts")
print("  ☐ Handle connection loss")
print("  ☐ Monitor connection usage")

print("\nQUERY BEST PRACTICES:")
print("  ☐ Use explicit column names")
print("  ☐ Filter at the database level")
print("  ☐ Use LIMIT for pagination")
print("  ☐ Avoid N+1 queries")
print("  ☐ Use EXPLAIN to analyze")
print("  ☐ Use parameterized queries")
print("  ☐ Use transactions")
print("  ☐ Avoid functions on indexed columns")

print("\nERROR HANDLING:")
print("  ☐ Use try/except blocks")
print("  ☐ Handle specific error types")
print("  ☐ Use rollback on errors")
print("  ☐ Log errors with context")
print("  ☐ Provide user-friendly messages")
print("  ☐ Retry transient errors")
print("  ☐ Never ignore errors")

Use this checklist when building any MySQL application with Python.

Quick Check: What should you do before deploying any database application? (Answer: Review this checklist and ensure all points are covered)

Try It Yourself

See how best practices compare to bad practices in the editor below.

Loading Pyodide... 0%

Python Code Editor

# Python MySQL Best Practices - Practice

print("=" * 40)
print("BEST PRACTICES - PRACTICE")
print("=" * 40)

# Simulated comparison of good vs bad practices

class SimulatedDatabase:
    """Simulate database operations with best practices"""
    
    @staticmethod
    def good_execute(query, params):
        """Good practice: parameterized query"""
        print(f"  [Good] Executing: {query[:30]}... with params: {params}")
        print("  [Good] SQL injection prevented")
        return "Result"
    
    @staticmethod
    def bad_execute(query):
        """Bad practice: string concatenation"""
        print(f"  [Bad] Executing: {query[:30]}...")
        print("  [Bad] SQL injection possible")
        return "Result"

print("\n1. SECURITY COMPARISON")
print("-" * 40)

print("\nGood Practice - Parameterized Query:")
SimulatedDatabase.good_execute(
    "SELECT * FROM users WHERE name = %s",
    ("Robert'; DROP TABLE users; --",)
)

print("\nBad Practice - String Concatenation:")
SimulatedDatabase.bad_execute(
    "SELECT * FROM users WHERE name = 'Robert'; DROP TABLE users; --'"
)

print("\n2. PERFORMANCE COMPARISON")
print("-" * 40)

print("\nGood Practice - Specific Columns:")
print("  SELECT id, username, email FROM users")
print("  → Only what you need, faster")

print("\nBad Practice - SELECT *:")
print("  SELECT * FROM users")
print("  → All columns, slower")

print("\n3. CONNECTION MANAGEMENT")
print("-" * 40)

print("\nGood Practice - Connection Pool:")
print("  pool.get_connection()  # Reuses connections")
print("  → 100 requests = 5 connections created")

print("\nBad Practice - New Connection:")
print("  mysql.connector.connect(...)  # Creates new connection")
print("  → 100 requests = 100 connections created")

print("\n4. CODE ORGANIZATION")
print("-" * 40)

print("\nGood Practice - Repository Pattern:")
print("  user_repo.find_by_id(1)")
print("  user_repo.create(data)")
print("  → Clean, organized, maintainable")

print("\nBad Practice - Scattered Queries:")
print("  db.execute('SELECT * FROM users WHERE id = 1')")
print("  db.execute('INSERT INTO users ...')")
print("  → Hard to maintain")

print("\n5. ERROR HANDLING")
print("-" * 40)

print("\nGood Practice - Try/Except:")
print("  try:")
print("      cursor.execute(query)")
print("  except Exception as e:")
print("      logger.error(e)")
print("      connection.rollback()")
print("  → Graceful error handling")

print("\nBad Practice - No Error Handling:")
print("  cursor.execute(query)")
print("  → Crashes on error")

print("\n📋 Best Practices Summary:")
print("""
┌─────────────────┬─────────────────────────────────────────┐
│ Category        │ Key Practice                          │
├─────────────────┼─────────────────────────────────────────┤
│ Security        │ Parameterized queries                  │
│ Performance     │ Connection pooling + indexes           │
│ Code            │ Repository pattern                     │
│ Connections     │ Use pools, always close               │
│ Queries         │ Specific columns, JOIN over N+1       │
│ Errors          │ Try/except + rollback + logging       │
└─────────────────┴─────────────────────────────────────────┘
""")

print("Following best practices makes your application secure and reliable!")

========================================
BEST PRACTICES - PRACTICE
========================================

1. SECURITY COMPARISON
----------------------------------------

Good Practice - Parameterized Query:
[Good] Executing: SELECT * FROM users WHERE name ... with params: ("Robert'; DROP TABLE users; --",)
[Good] SQL injection prevented

Bad Practice - String Concatenation:
[Bad] Executing: SELECT * FROM users WHERE name ...
[Bad] SQL injection possible

2. PERFORMANCE COMPARISON
----------------------------------------

Good Practice - Specific Columns:
SELECT id, username, email FROM users
→ Only what you need, faster

Bad Practice - SELECT *:
SELECT * FROM users
→ All columns, slower

3. CONNECTION MANAGEMENT
----------------------------------------

Good Practice - Connection Pool:
pool.get_connection() # Reuses connections
→ 100 requests = 5 connections created

Bad Practice - New Connection:
mysql.connector.connect(...) # Creates new connection
→ 100 requests = 100 connections created

4. CODE ORGANIZATION
----------------------------------------

Good Practice - Repository Pattern:
user_repo.find_by_id(1)
user_repo.create(data)
→ Clean, organized, maintainable

Bad Practice - Scattered Queries:
db.execute('SELECT * FROM users WHERE id = 1')
db.execute('INSERT INTO users ...')
→ Hard to maintain

5. ERROR HANDLING
----------------------------------------

Good Practice - Try/Except:
try:
cursor.execute(query)
except Exception as e:
logger.error(e)
connection.rollback()
→ Graceful error handling

Bad Practice - No Error Handling:
cursor.execute(query)
→ Crashes on error

📋 Best Practices Summary:
┌─────────────────┬─────────────────────────────────────────┐
│ Category │ Key Practice │
├─────────────────┼─────────────────────────────────────────┤
│ Security │ Parameterized queries │
│ Performance │ Connection pooling + indexes │
│ Code │ Repository pattern │
│ Connections │ Use pools, always close │
│ Queries │ Specific columns, JOIN over N+1 │
│ Errors │ Try/except + rollback + logging │
└─────────────────┴─────────────────────────────────────────┘

Following best practices makes your application secure and reliable!

🏆

You've Got It!

You now know the best practices for using MySQL with Python. You understand security, performance, code organization, and error handling.

Quick Quiz

Test what you've learned:

1. What is the #1 way to prevent SQL injection?

A) Input validation B) Parameterized queries C) Using a firewall D) SSL encryption

2. What is the N+1 query problem?

A) Running N queries then 1 more B) Querying in a loop instead of using JOIN C) Using too many JOINs D) Not using LIMIT

3. What is the benefit of connection pooling?

A) It creates more connections B) It reuses connections for better performance C) It encrypts connections D) It closes all connections

4. What should you do when a database error occurs?

A) Ignore it B) Show the error to the user C) Rollback, log, and handle appropriately D) Restart the application

5. Why should you store credentials in environment variables?

A) To make code run faster B) To keep credentials out of source code C) To use more memory D) To make it work on all platforms

Frequently Asked Questions

What is the most important best practice?

Always use parameterized queries. This single practice prevents SQL injection, which is the most common and dangerous security vulnerability. Everything else is secondary to security.

How do I know if my code follows best practices?

Review the checklist in this tutorial. Also, consider using code review processes, static analysis tools, and having experienced developers review your database code.

What is a common interview question about MySQL best practices?

Common questions include: "How do you prevent SQL injection?" "Explain the N+1 query problem" "What is connection pooling and why use it?" and "How would you organize database code in a large application?"

Should I use an ORM or write raw SQL?

Both approaches have their place. ORMs (like SQLAlchemy) are good for complex applications with many models. Raw SQL gives you more control and is better for complex queries. The best practice is to use the right tool for the job.

How often should I review my database code?

Regularly review your database code, especially when adding new features. Also review when you upgrade MySQL versions, as new features and optimizations may be available.

Where to Go From Here

Now that you know the best practices, here are some next steps:

Previous: Backup MySQL Database Next: Assignments →

Python

Python: MySQL Best Practices

Why Best Practices Matter

Security

Performance

Maintainability

The Impact of Following Best Practices

Security Best Practices

Protecting Your Database

Performance Best Practices

Making Your Application Fast

Code Organization

Writing Maintainable Database Code

Connection Management

Handling Database Connections Properly

Query Best Practices

Writing Better SQL Queries

Error Handling

Handling Errors Gracefully

Real-World Example: Production-Ready Service

Building a Production-Ready Database Service

Complete Best Practices Checklist

Quick Reference for All Practices

Try It Yourself

You've Got It!

Quick Quiz

Frequently Asked Questions

What is the most important best practice?

How do I know if my code follows best practices?

What is a common interview question about MySQL best practices?

Should I use an ORM or write raw SQL?

How often should I review my database code?

Where to Go From Here

Practice Assignments

Connection Management

Parameterized Queries

Python

Python: MySQL Best Practices

Why Best Practices Matter

Security

Performance

Maintainability

The Impact of Following Best Practices

Security Best Practices

Protecting Your Database

Performance Best Practices

Making Your Application Fast

Code Organization

Writing Maintainable Database Code

Connection Management

Handling Database Connections Properly

Query Best Practices

Writing Better SQL Queries

Error Handling

Handling Errors Gracefully

Real-World Example: Production-Ready Service

Building a Production-Ready Database Service

Complete Best Practices Checklist

Quick Reference for All Practices

Try It Yourself

You've Got It!

Quick Quiz

Frequently Asked Questions

What is the most important best practice? ▼

How do I know if my code follows best practices? ▼

What is a common interview question about MySQL best practices? ▼

Should I use an ORM or write raw SQL? ▼

How often should I review my database code? ▼

Where to Go From Here

Practice Assignments

Connection Management

Parameterized Queries

What is the most important best practice?

How do I know if my code follows best practices?

What is a common interview question about MySQL best practices?

Should I use an ORM or write raw SQL?

How often should I review my database code?