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

python-pymysql