PyMySQL Best Practices

PyMySQL is a pure-Python MySQL client library implementing the DB-API 2.0 specification (PEP 249). It supports MySQL >= 5.7 and MariaDB >= 10.3 on Python >= 3.7.

Installation

pip install PyMySQL

# For SHA-256 / caching_sha2_password authentication:
pip install "PyMySQL[rsa]"

Establishing a Connection

Use pymysql.connect() as the entry point. Always specify charset='utf8mb4' and use context managers to guarantee cleanup.

import pymysql
import pymysql.cursors

connection = pymysql.connect(
    host="localhost",
    user="app_user",
    password="secret",
    database="mydb",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor,
    connect_timeout=10,
    read_timeout=30,
    write_timeout=30,
    autocommit=False,  # explicit transaction control (recommended)
)

Key Connection Parameters

Parameter	Default	Notes
charset	''	Always set utf8mb4 for full Unicode support
cursorclass	Cursor	Use DictCursor for dict rows; SSCursor for large result sets
autocommit	False	Keep False; commit/rollback explicitly
connect_timeout	10	Seconds before connection attempt fails
read_timeout	None	Set to prevent hung reads
write_timeout	None	Set to prevent hung writes
ssl_ca	None	Path to CA cert for TLS connections

Context Manager Usage

Connections and cursors implement the context manager protocol. Use with blocks to ensure proper resource cleanup.

# Connection as context manager — commits on success, rolls back on exception
with pymysql.connect(**db_config) as connection:
    with connection.cursor() as cursor:
        cursor.execute("SELECT id, email FROM users WHERE active = %s", (1,))
        rows = cursor.fetchall()

Note: Using with connection: handles transaction commit/rollback but does not close the connection. Call connection.close() explicitly or manage it via a pool.

Parameterized Queries (SQL Injection Prevention)

Always pass values as the second argument to execute(). Never use string formatting or concatenation to build SQL.

# Correct — parameterized
cursor.execute(
    "INSERT INTO orders (user_id, amount) VALUES (%s, %s)",
    (user_id, amount),
)

# Named placeholders with dict
cursor.execute(
    "SELECT * FROM products WHERE category = %(cat)s AND price < %(max_price)s",
    {"cat": "electronics", "max_price": 500},
)

# WRONG — never do this
cursor.execute(f"SELECT * FROM users WHERE name = '{name}'")  # SQL injection risk

Use cursor.mogrify(query, args) to preview the interpolated query string during debugging.

Cursor Types

Cursor Class	Returns	Buffered	Use Case
Cursor	tuple	Yes	Default; small-to-medium result sets
DictCursor	dict	Yes	When column-name access is needed
SSCursor	tuple	No	Large result sets; memory-constrained
SSDictCursor	dict	No	Large result sets with dict access

Pass cursorclass at connection time (global default) or per-cursor:

# Per-cursor override
with connection.cursor(pymysql.cursors.DictCursor) as cursor:
    cursor.execute("SELECT * FROM users")
    rows = cursor.fetchall()  # list of dicts

For large tables, prefer SSCursor and iterate without calling fetchall():

with connection.cursor(pymysql.cursors.SSCursor) as cursor:
    cursor.execute("SELECT * FROM large_table")
    for row in cursor:  # streams row-by-row
        process(row)

Transaction Management

PyMySQL defaults to autocommit=False. Explicitly commit successful work and roll back on errors.

try:
    with connection.cursor() as cursor:
        cursor.execute(
            "UPDATE accounts SET balance = balance - %s WHERE id = %s",
            (amount, from_id),
        )
        cursor.execute(
            "UPDATE accounts SET balance = balance + %s WHERE id = %s",
            (amount, to_id),
        )
    connection.commit()
except Exception:
    connection.rollback()
    raise

Call connection.begin() to start a transaction block explicitly when needed.

Bulk Inserts with executemany()

Use executemany() for inserting multiple rows. PyMySQL batches the statements up to Cursor.max_stmt_length (1 MB) for efficiency.

records = [
    ("alice@example.com", "hash1"),
    ("bob@example.com", "hash2"),
]

with connection.cursor() as cursor:
    cursor.execute("TRUNCATE TABLE staging_users")
    cursor.executemany(
        "INSERT INTO users (email, password_hash) VALUES (%s, %s)",
        records,
    )
connection.commit()

Fetching Rows

Method	Returns	Notes
fetchone()	single row or None	Efficient for single-row lookups
fetchmany(size)	list of rows	Page through results
fetchall()	list of all rows	Avoid on large result sets

cursor.execute("SELECT id, name FROM users WHERE id = %s", (user_id,))
row = cursor.fetchone()
if row is None:
    raise ValueError(f"User {user_id} not found")

Error Handling

Catch PyMySQL exceptions from pymysql.err:

import pymysql.err

try:
    with connection.cursor() as cursor:
        cursor.execute(sql, args)
    connection.commit()
except pymysql.err.IntegrityError as exc:
    connection.rollback()
    # duplicate key, foreign key violation, etc.
    raise
except pymysql.err.OperationalError as exc:
    # connection dropped, timeout, server gone away
    raise
except pymysql.err.ProgrammingError as exc:
    # bad SQL syntax, wrong number of params
    raise

Key exception classes:

• pymysql.err.IntegrityError — constraint violations (duplicate key, FK)
• pymysql.err.OperationalError — connection / server errors
• pymysql.err.ProgrammingError — SQL syntax errors, wrong param count
• pymysql.err.DataError — invalid data for column type
• pymysql.err.DatabaseError — base class for all DB errors

Connection Health Check

Use connection.ping(reconnect=True) to check liveness before executing queries in long-lived connections (e.g., background workers):

connection.ping(reconnect=True)
with connection.cursor() as cursor:
    cursor.execute(query)

Reading Config from my.cnf

Avoid hardcoding credentials. Use read_default_file to read from a MySQL option file:

connection = pymysql.connect(
    read_default_file="~/.my.cnf",
    read_default_group="client",
    database="mydb",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor,
)

~/.my.cnf example:

[client]
host     = db.internal
user     = app_user
password = secret

Quick Reference

import pymysql
import pymysql.cursors

# Connect
conn = pymysql.connect(
    host="localhost", user="u", password="p", database="db",
    charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor,
)

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

# Write + commit
with conn.cursor() as cur:
    cur.execute("INSERT INTO t (col) VALUES (%s)", ("val",))
conn.commit()

# Bulk insert
with conn.cursor() as cur:
    cur.executemany("INSERT INTO t (a, b) VALUES (%s, %s)", rows)
conn.commit()

conn.close()

Additional Resources

For detailed patterns and advanced usage, consult:

• references/connection-patterns.md — Connection pooling, SSL/TLS configuration, timeout tuning, my.cnf patterns
• references/cursor-guide.md — Cursor type selection, streaming large result sets, stored procedures, mogrify debugging