Querying Mysql Database

YAML
---
name: querying-mysql-database
description: Connects to MySQL databases through SSH tunnels and executes queries automatically. Use when you need to retrieve data from MySQL databases with SSH jump host access.
---

# MySQL Database Query Skill

Python
import pymysql
import sshtunnel
from contextlib import contextmanager

@contextmanager
def mysql_connection(ssh_config, db_config):
    """Create MySQL connection through SSH tunnel"""
    with sshtunnel.SSHTunnelForwarder(
        (ssh_config['host'], ssh_config['port']),
        ssh_username=ssh_config['username'],
        ssh_private_key=ssh_config['private_key'],
        remote_bind_address=(db_config['host'], db_config['port'])
    ) as tunnel:
        connection = pymysql.connect(
            host='127.0.0.1',
            port=tunnel.local_bind_port,
            user=db_config['username'],
            password=db_config['password'],
            database=db_config['database'],
            charset='utf8mb4'
        )
        yield connection
        connection.close()

# Usage
ssh_config = {
    'host': 'jump.example.com',
    'port': 22,
    'username': 'user',
    'private_key': '/path/to/key.pem'
}

db_config = {
    'host': '10.0.1.100',
    'port': 3306,
    'username': 'dbuser',
    'password': 'dbpass',
    'database': 'mydb'
}

with mysql_connection(ssh_config, db_config) as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users LIMIT 10")
    results = cursor.fetchall()

Progress:

• Set up SSH tunnel configuration
• Configure MySQL connection parameters
• Establish SSH tunnel
• Create MySQL connection through tunnel
• Execute query
• Process results
• Clean up connections

Step-by-step Process

1. Install dependencies

   Bash
   pip install pymysql sshtunnel pandas

2. Configure connection parameters

   Python
   config = {
       'ssh': {
           'host': 'jumphost.com',
           'port': 22,
           'username': 'ssh_user',
           'private_key': '/path/to/key.pem',  # or use password
           'password': None
       },
       'mysql': {
           'host': 'internal-db-host',
           'port': 3306,
           'username': 'db_user',
           'password': 'db_password',
           'database': 'target_db'
       }
   }

3. Create query executor

   Python
   class MySQLQueryExecutor:
       def __init__(self, config):
           self.ssh_config = config['ssh']
           self.db_config = config['mysql']
       
       def execute_query(self, query, fetch_type='all'):
           with self.mysql_connection() as conn:
               cursor = conn.cursor(pymysql.cursors.DictCursor)
               cursor.execute(query)
               
               if fetch_type == 'one':
                   return cursor.fetchone()
               elif fetch_type == 'many':
                   return cursor.fetchmany(100)
               else:
                   return cursor.fetchall()
       
       @contextmanager
       def mysql_connection(self):
           # Implementation from Quick Start

Example 1: Simple SELECT query Input:

Python
executor = MySQLQueryExecutor(config)
query = "SELECT id, name, email FROM users WHERE status = 'active'"
results = executor.execute_query(query)

Output:

Python
[
    {'id': 1, 'name': 'John', 'email': 'john@example.com'},
    {'id': 2, 'name': 'Jane', 'email': 'jane@example.com'}
]

Example 2: Parameterized query with pandas Input:

Python
import pandas as pd

def get_user_orders(user_id):
    query = """
    SELECT o.id, o.order_date, o.total, p.name as product_name
    FROM orders o
    JOIN order_items oi ON o.id = oi.order_id
    JOIN products p ON oi.product_id = p.id
    WHERE o.user_id = %s
    ORDER BY o.order_date DESC
    """
    
    with executor.mysql_connection() as conn:
        return pd.read_sql(query, conn, params=[user_id])

df = get_user_orders(123)

Output: DataFrame with user's order history

Example 3: Environment-based configuration Input:

Python
import os
from dotenv import load_dotenv

load_dotenv()

config = {
    'ssh': {
        'host': os.getenv('SSH_HOST'),
        'port': int(os.getenv('SSH_PORT', 22)),
        'username': os.getenv('SSH_USER'),
        'private_key': os.getenv('SSH_KEY_PATH'),
        'password': os.getenv('SSH_PASSWORD')
    },
    'mysql': {
        'host': os.getenv('DB_HOST'),
        'port': int(os.getenv('DB_PORT', 3306)),
        'username': os.getenv('DB_USER'),
        'password': os.getenv('DB_PASSWORD'),
        'database': os.getenv('DB_NAME')
    }
}

• Use connection pooling for multiple queries
• Always use parameterized queries to prevent SQL injection
• Set connection timeouts to avoid hanging connections
• Use environment variables for sensitive configuration
• Implement retry logic for network issues
• Log query execution time for performance monitoring
• Use pandas.read_sql() for analytical queries
• Close connections explicitly in finally blocks

Python
# Connection with timeout and retry
@retry(stop_max_attempt_number=3, wait_fixed=2000)
def create_connection_with_timeout():
    return pymysql.connect(
        host='127.0.0.1',
        port=tunnel.local_bind_port,
        user=db_config['username'],
        password=db_config['password'],
        database=db_config['database'],
        charset='utf8mb4',
        connect_timeout=10,
        read_timeout=30,
        write_timeout=30
    )

• Don't hardcode credentials in source code
• Don't forget to close SSH tunnels - use context managers
• Don't use string formatting for SQL queries - use parameterized queries
• Don't ignore connection errors - implement proper error handling
• Don't keep connections open indefinitely - close after use
• **Don't use SELECT *** on large tables without LIMIT
• Don't forget to handle MySQL time zones for datetime fields
• Don't use the same SSH tunnel for concurrent connections without proper synchronization

Python
# Bad
query = f"SELECT * FROM users WHERE id = {user_id}"  # SQL injection risk

# Good
query = "SELECT * FROM users WHERE id = %s"
cursor.execute(query, [user_id])