Python and MySQL — Connect, Query, and Build Database-Backed Apps

Introduction

Why This Matters

MySQL is one of the world's most popular relational databases, powering billions of applications from WordPress blogs to large-scale e-commerce platforms. Knowing how to interact with MySQL from Python is an essential skill for backend developers, data engineers, and automation scripters.

Python provides two main approaches: the mysql-connector-python library for raw SQL, and SQLAlchemy for an ORM layer that abstracts the database behind Python objects. Both have their place — raw SQL is ideal for complex queries and reporting, while ORMs shine for CRUD-heavy application code.

Security is paramount when working with databases. This guide emphasizes parameterized queries to prevent SQL injection, proper connection management, and transaction handling.

Installation

pip install mysql-connector-python sqlalchemy pymysql

Connecting with mysql-connector-python

import mysql.connector
from mysql.connector import Error
 
def get_connection():
    return mysql.connector.connect(
        host="localhost",
        port=3306,
        user="root",
        password="yourpassword",
        database="mydb",
        autocommit=False,
    )
 
try:
    conn = get_connection()
    print("Connected to MySQL")
except Error as e:
    print(f"Connection failed: {e}")
finally:
    if conn.is_connected():
        conn.close()

Executing Queries

import mysql.connector
 
conn = mysql.connector.connect(
    host="localhost", user="root", password="pass", database="mydb"
)
cursor = conn.cursor(dictionary=True)  # returns rows as dicts
 
# Simple SELECT
cursor.execute("SELECT id, name, email FROM users LIMIT 10")
users = cursor.fetchall()
for user in users:
    print(user["name"], user["email"])
 
cursor.close()
conn.close()

Parameterized Queries (Prevent SQL Injection)

def get_user_by_email(email: str) -> dict | None:
    conn = mysql.connector.connect(
        host="localhost", user="root", password="pass", database="mydb"
    )
    cursor = conn.cursor(dictionary=True)
 
    # ALWAYS use %s placeholders, never f-strings for user input
    cursor.execute(
        "SELECT id, name, email FROM users WHERE email = %s",
        (email,)   # tuple is required even for single value
    )
    user = cursor.fetchone()
 
    cursor.close()
    conn.close()
    return user

INSERT, UPDATE, DELETE with Transactions

def create_user(name: str, email: str, age: int) -> int:
    conn = mysql.connector.connect(
        host="localhost", user="root", password="pass", database="mydb"
    )
    cursor = conn.cursor()
 
    try:
        cursor.execute(
            "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)",
            (name, email, age)
        )
        conn.commit()
        user_id = cursor.lastrowid
        print(f"Created user with ID: {user_id}")
        return user_id
    except Exception as e:
        conn.rollback()
        print(f"Transaction rolled back: {e}")
        raise
    finally:
        cursor.close()
        conn.close()

Batch INSERT

def bulk_insert_users(users: list[tuple]) -> None:
    conn = mysql.connector.connect(
        host="localhost", user="root", password="pass", database="mydb"
    )
    cursor = conn.cursor()
 
    sql = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
    cursor.executemany(sql, users)
    conn.commit()
 
    print(f"Inserted {cursor.rowcount} rows")
    cursor.close()
    conn.close()
 
bulk_insert_users([
    ("Alice", "alice@example.com", 30),
    ("Bob", "bob@example.com", 25),
    ("Charlie", "charlie@example.com", 35),
])

SQLAlchemy ORM (Modern Approach)

from sqlalchemy import create_engine, String, Integer, select
from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped, Session
 
engine = create_engine("mysql+pymysql://root:pass@localhost/mydb", echo=False)
 
class Base(DeclarativeBase):
    pass
 
class User(Base):
    __tablename__ = "users"
 
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(255), unique=True)
    age: Mapped[int] = mapped_column(Integer)
 
Base.metadata.create_all(engine)
 
# Create
with Session(engine) as session:
    user = User(name="Alice", email="alice@example.com", age=30)
    session.add(user)
    session.commit()
 
# Query
with Session(engine) as session:
    stmt = select(User).where(User.age > 25).order_by(User.name)
    users = session.scalars(stmt).all()
    for u in users:
        print(u.name, u.email)

Connection Pooling

from sqlalchemy import create_engine
 
# Pool keeps connections open and reuses them
engine = create_engine(
    "mysql+pymysql://root:pass@localhost/mydb",
    pool_size=10,        # connections kept open
    max_overflow=20,     # extra connections allowed under load
    pool_recycle=3600,   # recycle connections after 1 hour
)

Common Mistakes

Using f-strings to build queries with user input — SQL injection vulnerability
Not committing transactions — changes are rolled back when connection closes
Not closing cursors and connections — causes connection pool exhaustion
Using fetchall() for large result sets — load in batches with fetchmany(n)
Storing database credentials in source code — use environment variables

Best Practices

Always use parameterized queries with %s placeholders
Use a connection pool (SQLAlchemy or mysql.connector.pooling) in web applications
Wrap mutating operations in explicit transactions with rollback on failure
Store credentials in environment variables or a secrets manager
Use SQLAlchemy ORM for application CRUD; raw SQL for complex analytical queries

Key Takeaways

mysql-connector-python provides raw SQL access; SQLAlchemy adds an ORM layer
Parameterized queries with %s are the only safe way to include user data in SQL
Always call conn.commit() after INSERT/UPDATE/DELETE, or set autocommit=True
cursor.executemany() is the efficient way to bulk-insert rows
SQLAlchemy's Session context manager handles commit and rollback automatically
Connection pools are essential in web apps — avoid creating a new connection per request
Never store database passwords in source code — use os.environ or .env files

Introduction#

Why This Matters#

Installation#

Connecting with mysql-connector-python#

Executing Queries#

Parameterized Queries (Prevent SQL Injection)#

INSERT, UPDATE, DELETE with Transactions#

Batch INSERT#

SQLAlchemy ORM (Modern Approach)#

Connection Pooling#

Common Mistakes#

Best Practices#

Key Takeaways#

Related reading