Connect to MySQL

mysql-connector-python is the official Oracle driver with better performance. PyMySQL is a pure-Python alternative useful when C extensions cannot be installed.

Set connectTimeout, acquireTimeout, and timeout in pool options. Also configure enableKeepAlive for long-running connections.

query() sends the SQL as a plain string. execute() sends a prepared statement with bound parameters, which is safe against SQL injection and can improve performance for repeated queries.

import mysql.connector
from mysql.connector import pooling
from contextlib import contextmanager

# Create a connection pool
mysql_pool = pooling.MySQLConnectionPool(
    pool_name="mypool",
    pool_size=10,
    host="localhost",
    database="mydb",
    user="user",
    password="pass",
    ssl_ca="ca.pem",
    ssl_verify_cert=True
)

@contextmanager
def get_db_cursor():
    conn = mysql_pool.get_connection()
    try:
        cursor = conn.cursor(dictionary=True)
        yield cursor
        conn.commit()
        cursor.close()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# Usage
with get_db_cursor() as cur:
    cur.execute("SELECT * FROM users WHERE active = %s", (True,))
    rows = cur.fetchall()
    for row in rows:
        print(row)

const mysql = require('mysql2/promise');

const pool = mysql.createPool({
    host: 'localhost',
    database: 'mydb',
    user: 'user',
    password: 'pass',
    ssl: { ca: require('fs').readFileSync('ca.pem') },
    waitForConnections: true,
    connectionLimit: 10,
    queueLimit: 0,
    enableKeepAlive: true,
    keepAliveInitialDelay: 10000
});

async function transferBalance(fromId, toId, amount) {
    const conn = await pool.getConnection();
    try {
        await conn.beginTransaction();

        await conn.execute(
            'UPDATE accounts SET balance = balance - ? WHERE id = ?',
            [amount, fromId]
        );

        await conn.execute(
            'UPDATE accounts SET balance = balance + ? WHERE id = ?',
            [amount, toId]
        );

        await conn.commit();
    } catch (err) {
        await conn.rollback();
        throw err;
    } finally {
        conn.release();
    }
}

import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import java.sql.*;

public class MySQLPool {
    private static final HikariDataSource ds;

    static {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:mysql://localhost:3306/mydb?sslMode=VERIFY_CA&serverTimezone=UTC");
        config.setUsername("user");
        config.setPassword("pass");
        config.setMaximumPoolSize(20);
        config.setMinimumIdle(5);
        config.setIdleTimeout(30000);
        config.setConnectionTimeout(5000);
        ds = new HikariDataSource(config);
    }

    public static Connection getConnection() throws SQLException {
        return ds.getConnection();
    }

    public static void batchInsert(List<String> emails) throws SQLException {
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(
                 "INSERT INTO users (email) VALUES (?)")) {
            for (String email : emails) {
                stmt.setString(1, email);
                stmt.addBatch();
            }
            stmt.executeBatch();
        }
    }
}

import pymysql
from contextlib import contextmanager

@contextmanager
def get_db():
    conn = pymysql.connect(
        host="localhost",
        database="mydb",
        user="user",
        password="pass",
        ssl={"ca": "ca.pem"},
        cursorclass=pymysql.cursors.DictCursor
    )
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# Usage
with get_db() as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT * FROM users WHERE id = %s", (1,))
        row = cur.fetchone()
        print(row)