PyMySQL Security Best Practices

1. Preventing SQL Injection

SQL injection is the most critical database vulnerability. PyMySQL provides safe parameterization via its parameter substitution — never bypass it.

Use Parameterized Queries

import pymysql

# Safe — parameter is bound, never interpolated into SQL
with connection.cursor() as cursor:
    cursor.execute(
        "SELECT id, username, email, role FROM users WHERE email = %s",
        (user_input,)
    )
    result = cursor.fetchall()

# Also safe — named parameters with a dict
with connection.cursor() as cursor:
    cursor.execute(
        "SELECT * FROM users WHERE email = %(email)s",
        {"email": user_input}
    )

Never Interpolate User Input

# DANGEROUS — direct string formatting
cursor.execute(f"SELECT * FROM users WHERE email = '{user_input}'")

# DANGEROUS — string concatenation
query = "SELECT * FROM users WHERE name = '" + name + "'"
cursor.execute(query)

# DANGEROUS — %-formatting the string yourself
cursor.execute("SELECT * FROM users WHERE name = '%s'" % name)

Never Use Raw String Formatting

Even though PyMySQL uses %s placeholders, you must pass the values as the second argument to execute(), never format them into the query string yourself. PyMySQL's driver handles escaping — manual formatting bypasses all protections.

2. Securing the Database Connection

Use Environment Variables for Credentials

Never hardcode database credentials in source code.

Install python-dotenv

pip install python-dotenv

Create a .env file

Store credentials in a .env file at the root of your project. This file is for local development only — use a proper secrets manager (Vault, AWS Secrets Manager, etc.) in production.

# .env
DB_HOST=localhost
DB_USER=appuser
DB_PASSWORD=s3cur3pass
DB_NAME=mydb
DB_PORT=3306

Keep .env out of version control

Add it to .gitignore immediately. Committing credentials is one of the most common causes of secret leaks.

# .gitignore
.env

Provide a .env.example with placeholder values so collaborators know which variables are required:

# .env.example — safe to commit
DB_HOST=
DB_USER=
DB_PASSWORD=
DB_NAME=
DB_PORT=3306

Load the .env file

Call load_dotenv() once at application startup, before any code that reads environment variables. It populates os.environ from the .env file, but does not override variables that are already set — so real environment variables in production take precedence automatically.

from dotenv import load_dotenv
import os
import pymysql

load_dotenv()  # Call once, at startup

connection = pymysql.connect(
    host=os.environ["DB_HOST"],
    user=os.environ["DB_USER"],
    password=os.environ["DB_PASSWORD"],
    database=os.environ["DB_NAME"],
    port=int(os.environ.get("DB_PORT", 3306)),
)

os.environ[key] vs os.environ.get(key)

Use os.environ["KEY"] (bracket syntax) for required variables. It raises a KeyError immediately at startup if the variable is missing, making misconfigurations obvious. Use os.environ.get("KEY", default) only for truly optional variables with safe defaults — such as DB_PORT.

Use SSL/TLS for Remote Connections

import ssl

ssl_context = ssl.create_default_context(cafile="/path/to/ca-cert.pem")
ssl_context.load_cert_chain(
    certfile="/path/to/client-cert.pem",
    keyfile="/path/to/client-key.pem"
)

connection = pymysql.connect(
    host=os.environ["DB_HOST"],
    user=os.environ["DB_USER"],
    password=os.environ["DB_PASSWORD"],
    database=os.environ["DB_NAME"],
    ssl=ssl_context,
)

Verify Server Certificates

Always verify the server's SSL certificate in production. Use ssl.create_default_context() (which verifies by default) rather than ssl_verify_cert=False, to prevent man-in-the-middle attacks.

3. Principle of Least Privilege

Create dedicated database users with only the permissions your application actually needs.

-- Create a restricted application user
CREATE USER 'appuser'@'%' IDENTIFIED BY 'strong_password';

-- Grant only what is needed
GRANT SELECT, INSERT, UPDATE, DELETE ON mydb.* TO 'appuser'@'%';

-- Never grant these to the application user:
-- GRANT ALL PRIVILEGES ...
-- GRANT DROP, ALTER, CREATE ...
FLUSH PRIVILEGES;

Role	Permissions
App (read/write)	SELECT, INSERT, UPDATE, DELETE
App (read-only)	SELECT
Migrations only	ALTER, CREATE, DROP
Admin	All (never used by app code)

4. Connection Management

Always Close Connections and Cursors

Use context managers to guarantee cursors and connections are closed.

import pymysql

# Cursor closes automatically on __exit__
with connection.cursor() as cursor:
    cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
    user = cursor.fetchone()

# Full connection lifecycle with context manager
connection = pymysql.connect(**db_config)
try:
    with connection.cursor() as cursor:
        cursor.execute("INSERT INTO events (name) VALUES (%s)", ("login",))
    connection.commit()
except pymysql.MySQLError as e:
    connection.rollback()
    raise
finally:
    connection.close()

Use a Connection Pool for Long-Running Applications

PyMySQL itself is not thread-safe and does not include a pool. Use a pooling library such as DBUtils for multi-threaded applications.

from dbutils.pooled_db import PooledDB
import pymysql

pool = PooledDB(
    creator=pymysql,
    maxconnections=10,   # Hard cap on total connections
    mincached=2,         # Connections kept open when idle
    maxcached=5,         # Max idle connections in pool
    blocking=True,       # Block instead of raising when pool is full
    host=os.environ["DB_HOST"],
    user=os.environ["DB_USER"],
    password=os.environ["DB_PASSWORD"],
    database=os.environ["DB_NAME"],
    charset="utf8mb4",
)

# Borrow a connection from the pool
connection = pool.connection()
try:
    with connection.cursor() as cursor:
        cursor.execute("SELECT 1")
finally:
    connection.close()  # Returns connection to pool, does not disconnect

Connection Leaks

Unreturned connections exhaust the pool and cause application downtime. Always use try/finally or a context manager to guarantee the connection is returned.

5. Input Validation & Sanitization

PyMySQL parameterization prevents injection, but you still need to validate inputs before they reach the database.

from pydantic import BaseModel, EmailStr, constr

class UserCreate(BaseModel):
    username: constr(min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_]+$")
    email: EmailStr
    age: int

def create_user(data: UserCreate, connection):
    with connection.cursor() as cursor:
        cursor.execute(
            "INSERT INTO users (username, email, age) VALUES (%s, %s, %s)",
            (data.username, data.email, data.age),
        )
    connection.commit()

Validate before constructing any database operation:

• Enforce expected types and lengths
• Reject unexpected characters using allow-lists
• Validate business rules (e.g. age > 0) at the model level

6. Password Storage

Never store plaintext passwords. Use a strong adaptive hashing algorithm.

import bcrypt

def hash_password(plain: str) -> str:
    return bcrypt.hashpw(plain.encode(), bcrypt.gensalt()).decode()

def verify_password(plain: str, hashed: str) -> bool:
    return bcrypt.checkpw(plain.encode(), hashed.encode())

# Storing
with connection.cursor() as cursor:
    cursor.execute(
        "INSERT INTO users (username, password_hash) VALUES (%s, %s)",
        (username, hash_password(raw_password)),
    )
connection.commit()

# Verifying
with connection.cursor() as cursor:
    cursor.execute(
        "SELECT password_hash FROM users WHERE username = %s", (username,)
    )
    row = cursor.fetchone()

if not row or not verify_password(submitted_password, row["password_hash"]):
    raise ValueError("Invalid credentials")

Recommended Algorithms

Use bcrypt, Argon2, or scrypt. Never use MD5, SHA-1, or unsalted SHA-256 for passwords.

7. Protecting Against Mass Assignment

When mapping request payloads to database writes, be explicit about which fields are allowed.

# DANGEROUS — attacker could inject is_admin=True
def update_user(user_id: int, payload: dict, connection):
    columns = ", ".join(f"{k} = %s" for k in payload)
    values = list(payload.values()) + [user_id]
    with connection.cursor() as cursor:
        cursor.execute(f"UPDATE users SET {columns} WHERE id = %s", values)
    connection.commit()  # Never do this — columns come from user input

# Safe — only update allowed fields
ALLOWED_FIELDS = {"username", "email", "bio"}

def update_user(user_id: int, payload: dict, connection):
    updates = {k: v for k, v in payload.items() if k in ALLOWED_FIELDS}
    if not updates:
        return
    columns = ", ".join(f"{k} = %s" for k in updates)
    values = list(updates.values()) + [user_id]
    with connection.cursor() as cursor:
        cursor.execute(f"UPDATE users SET {columns} WHERE id = %s", values)
    connection.commit()

8. Error Handling & Information Disclosure

Never expose raw database errors to users — they can leak schema details, table names, or stack traces useful to an attacker.

import logging
import pymysql

logger = logging.getLogger(__name__)

def get_user(user_id: int, connection):
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                "SELECT id, username, email FROM users WHERE id = %s", (user_id,)
            )
            return cursor.fetchone()
    except pymysql.MySQLError as e:
        logger.error("Database error fetching user %s: %s", user_id, e)
        raise RuntimeError("An internal error occurred") from None

Log Internally, Surface Generically

Log the full exception server-side for debugging. Return only a generic message to the client. Never include table names, column names, or SQL in API responses.

9. Auditing & Logging

Track sensitive operations for security auditing. With PyMySQL you can wrap your cursor calls in a helper that logs before executing.

import logging

logger = logging.getLogger(__name__)

def audited_execute(cursor, query, params=None, *, label="query"):
    logger.debug("[SQL:%s] %s | params=%s", label, query, params)
    cursor.execute(query, params)

# Usage
with connection.cursor() as cursor:
    audited_execute(
        cursor,
        "DELETE FROM sessions WHERE user_id = %s",
        (user_id,),
        label="logout",
    )
connection.commit()

Consider logging: login attempts, privilege changes, bulk deletes/updates, and access to sensitive tables.

Quick Reference Checklist

• Using %s / %(name)s bound parameters — no raw string interpolation
• Credentials stored in environment variables; .env in .gitignore, .env.example committed
• SSL/TLS enabled for remote connections
• Database user has least-privilege permissions
• Connections always closed via try/finally or context managers
• Connection pool configured for multi-threaded applications
• All inputs validated before hitting the database
• Passwords hashed with bcrypt/Argon2
• Mass assignment prevented via field allowlists
• Database errors caught and not exposed to users