Quick Start: Connect MySQL with Python in 5 Minutes
First, install both libraries:
pip install PyMySQL mysql-connector-pythonHere is the most basic connection example using PyMySQL:
import pymysql
conn = pymysql.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydb',
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
try:
with conn.cursor() as cursor:
cursor.execute("SELECT VERSION()")
result = cursor.fetchone()
print(f"MySQL version: {result['VERSION()']}")
finally:
conn.close()Got it running? Great. But in a real environment, code like this will cause all kinds of problems under peak load.
The Real Problems with MySQL Connections in Production
My production database runs MySQL 8.0 with around 50GB of data. Previously, each request created a new connection and closed it when done — the system kept slowing down during the day. After profiling, I discovered that most of the response time was being consumed by TCP connection setup and MySQL authentication handshakes, not slow queries.
The three most common problems:
- Connection overhead: Creating a new connection per request costs 20–100ms. Multiply that by 1,000 requests per minute and you have a serious problem.
- SQL Injection: Directly concatenating strings into queries is an extremely dangerous security vulnerability.
- Careless transactions: Data can become corrupted if an error occurs mid-operation without a rollback.
Solution 1: Connection Pooling — Reuse Connections Instead of Creating New Ones
Instead of constantly opening and closing connections, a Connection Pool maintains a set of ready-to-use connections for reuse. With mysql-connector-python, setting up a pool takes just a few lines:
from mysql.connector import pooling
db_pool = pooling.MySQLConnectionPool(
pool_name="mypool",
pool_size=5, # Adjust based on your workload
host='localhost',
user='myuser',
password='mypassword',
database='mydb',
charset='utf8mb4',
use_unicode=True
)
def get_connection():
return db_pool.get_connection()
# Usage
conn = get_connection()
try:
cursor = conn.cursor(dictionary=True)
cursor.execute("SELECT id, name FROM users WHERE active = %s", (1,))
rows = cursor.fetchall()
print(rows)
finally:
conn.close() # Returns the connection to the pool, does not actually close itNote: pool_size=5 is suitable for small to medium applications. For higher traffic, increase it to 10–20, but also check max_connections in your MySQL config.
PyMySQL with SQLAlchemy Pool
If you are using PyMySQL and want connection pooling, combining it with SQLAlchemy is my go-to approach:
from sqlalchemy import create_engine, text
engine = create_engine(
"mysql+pymysql://myuser:mypassword@localhost/mydb?charset=utf8mb4",
pool_size=5,
max_overflow=10, # Allow up to 15 connections when needed
pool_timeout=30, # Wait up to 30s to get a connection from the pool
pool_recycle=1800, # Recycle connections after 30 minutes (prevents "MySQL has gone away")
pool_pre_ping=True # Check if a connection is alive before using it
)
with engine.connect() as conn:
result = conn.execute(
text("SELECT id, name FROM users WHERE active = :active"),
{"active": 1}
)
for row in result:
print(row)pool_pre_ping=True is a parameter I always enable — it automatically checks and reconnects if a connection has died after a MySQL server restart or extended idle period.
Solution 2: Prepared Statements — Fast and Secure
Prepared Statements solve two problems at once: they prevent SQL Injection and improve performance when running the same query multiple times.
How it works: The query is sent to MySQL first (without data), and MySQL parses and compiles it once. Data is then passed in separately via parameters — MySQL never interprets that data as SQL code.
import pymysql
conn = pymysql.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydb',
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
# CORRECT — Use a parameterized query
def find_user(email):
with conn.cursor() as cursor:
sql = "SELECT id, name, email FROM users WHERE email = %s AND active = %s"
cursor.execute(sql, (email, 1)) # %s is a placeholder, not string formatting!
return cursor.fetchone()
# WRONG — Direct string concatenation (SQL Injection!)
def find_user_unsafe(email):
with conn.cursor() as cursor:
sql = f"SELECT * FROM users WHERE email = '{email}'" # DANGEROUS!
cursor.execute(sql)
return cursor.fetchone()The find_user_unsafe function with input email = "' OR '1'='1" will return every user in the database. This is the classic SQL Injection vulnerability that I still encounter in real-world codebases.
True Prepared Statements with mysql-connector-python
import mysql.connector
conn = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydb'
)
cursor = conn.cursor(dictionary=True, prepared=True) # prepared=True enables server-side caching
sql = "SELECT id, name FROM orders WHERE user_id = %s AND status = %s"
cursor.execute(sql, (123, "completed"))
orders = cursor.fetchall()
cursor.close()
conn.close()The prepared=True parameter enables true prepared statements at the MySQL server level — the server caches the execution plan, making subsequent queries noticeably faster on large databases.
Solution 3: Safe Transactions with Context Managers
Transactions guarantee data integrity: either everything succeeds, or everything rolls back to the original state. I have personally witnessed a system fail mid-operation after deducting a customer’s balance but before creating the order — not using transactions correctly is a real risk.
import pymysql
from contextlib import contextmanager
@contextmanager
def db_transaction(conn):
"""Automatically commits or rolls back."""
try:
yield conn
conn.commit()
except Exception as e:
conn.rollback()
raise e
conn = pymysql.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydb',
charset='utf8mb4',
autocommit=False # Disable autocommit to manually control transactions
)
def create_order(user_id, product_id, quantity, price):
with db_transaction(conn):
with conn.cursor() as cursor:
# Step 1: Deduct stock
cursor.execute(
"UPDATE products SET stock = stock - %s WHERE id = %s AND stock >= %s",
(quantity, product_id, quantity)
)
if cursor.rowcount == 0:
raise ValueError("Insufficient stock")
# Step 2: Create the order
cursor.execute(
"INSERT INTO orders (user_id, product_id, quantity, total_price) VALUES (%s, %s, %s, %s)",
(user_id, product_id, quantity, price * quantity)
)
order_id = cursor.lastrowid
# Step 3: Write to the log
cursor.execute(
"INSERT INTO transaction_logs (order_id, action, created_at) VALUES (%s, %s, NOW())",
(order_id, "ORDER_CREATED")
)
return order_id # Only returns after a successful commitIf any step raises an exception — whether a network error or a constraint violation — db_transaction automatically rolls back everything. There is no scenario where stock gets deducted without a corresponding order being created.
Advanced: Batch Inserts with executemany()
When you need to insert thousands of rows, calling execute() one row at a time is very slow. Use executemany() for batch inserts:
data = [
("[email protected]", "Alice", 1),
("[email protected]", "Bob", 1),
("[email protected]", "Charlie", 0),
# ... thousands of rows
]
try:
with conn.cursor() as cursor:
sql = "INSERT INTO users (email, name, active) VALUES (%s, %s, %s)"
cursor.executemany(sql, data)
conn.commit()
print(f"Inserted {cursor.rowcount} rows")
except Exception as e:
conn.rollback()
print(f"Error: {e}")
finally:
conn.close()On a production database with 50GB of data, executemany() inserts 10,000 rows roughly 15–20 times faster than looping with individual execute() calls — the difference is most apparent when running data migrations.
Practical Tips from Operational Experience
1. Always Set Timeouts to Prevent Hanging Connections
conn = pymysql.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydb',
charset='utf8mb4',
connect_timeout=5, # Connection timeout: 5 seconds
read_timeout=30, # Read timeout: 30 seconds
write_timeout=30 # Write timeout: 30 seconds
)2. Handle Errors by Exception Type
import pymysql
try:
with conn.cursor() as cursor:
cursor.execute("INSERT INTO users (email) VALUES (%s)", ("[email protected]",))
conn.commit()
except pymysql.err.IntegrityError as e:
conn.rollback()
if e.args[0] == 1062: # Duplicate entry
print("Email already exists in the system")
else:
raise
except pymysql.err.OperationalError as e:
conn.rollback()
print(f"Database connection error: {e}")
raise3. Quick Comparison: PyMySQL vs mysql-connector-python
- PyMySQL: Pure Python, no C extension required, easy to deploy, suitable for most use cases. Downside: slightly slower when queries return large amounts of data.
- mysql-connector-python: Official driver from Oracle, optional C extension, supports true server-side prepared statements, built-in connection pool. Better suited for production environments with high traffic.
I use PyMySQL + SQLAlchemy for Flask web apps, and mysql-connector-python for ETL scripts that handle large data volumes due to its better performance with prepared statements. Choose based on your use case — both are solid options when used correctly.
