MySQL #

MySQL is one of the most popular relational databases in the world, used widely from simple web apps to large enterprise systems. Python provides several libraries for interacting with MySQL, and understanding how to use them properly — including parameterized queries to prevent SQL injection, transaction management, and connection pooling — is an essential foundation for building reliable, secure applications.

Installation #

The officially recommended library for connecting to MySQL from Python is mysql-connector-python, created by Oracle:

pip install mysql-connector-python

Another popular alternative is PyMySQL (pure Python, no C dependency):

pip install pymysql

mysql-connector-python is the more stable choice for production because it’s directly supported by Oracle. PyMySQL is easier to install in environments that can’t compile C extensions. This article uses mysql-connector-python, but the core API is compatible with the DB-API 2.0 standard that both follow.

Creating a Connection #

Connecting to MySQL requires several required parameters: host, user, password, and database name. Never store these values directly in code — use environment variables.

import mysql.connector
import os

# ANTI-PATTERN: hardcoding credentials in code
conn = mysql.connector.connect(
    host="localhost",
    user="root",
    password="password123",  # ✗ -- don't do this
    database="myapp"
)

# CORRECT: read from environment variables
conn = mysql.connector.connect(
    host=os.getenv("DB_HOST", "localhost"),
    port=int(os.getenv("DB_PORT", "3306")),
    user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD"),
    database=os.getenv("DB_NAME"),
    charset="utf8mb4",          # support emoji and full Unicode characters
    use_unicode=True,
    autocommit=False            # manage transactions explicitly
)

print("Connection successful:", conn.is_connected())
conn.close()

Connection with a Context Manager #

A cleaner way is to wrap the connection in a function and use try/finally to make sure the connection is always closed:

import mysql.connector
from contextlib import contextmanager
import os

def get_connection():
    return mysql.connector.connect(
        host=os.getenv("DB_HOST", "localhost"),
        user=os.getenv("DB_USER", "root"),
        password=os.getenv("DB_PASSWORD", ""),
        database=os.getenv("DB_NAME", "myapp"),
        charset="utf8mb4",
        autocommit=False
    )

@contextmanager
def db_connection():
    conn = get_connection()
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# Usage
with db_connection() as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT VERSION()")
    version = cursor.fetchone()
    print("MySQL version:", version[0])

Creating Tables #

Before data operations, create the tables you need. Use IF NOT EXISTS so the script is safe to run repeatedly.

import mysql.connector

def create_tables(conn):
    cursor = conn.cursor()
    
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS pengguna (
            id          INT AUTO_INCREMENT PRIMARY KEY,
            nama        VARCHAR(100) NOT NULL,
            email       VARCHAR(150) NOT NULL UNIQUE,
            usia        INT,
            aktif       BOOLEAN DEFAULT TRUE,
            dibuat_pada TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """)
    
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS produk (
            id       INT AUTO_INCREMENT PRIMARY KEY,
            nama     VARCHAR(200) NOT NULL,
            harga    DECIMAL(15, 2) NOT NULL,
            stok     INT DEFAULT 0,
            kategori VARCHAR(100)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """)
    
    conn.commit()
    cursor.close()
    print("Tables created successfully.")

with db_connection() as conn:
    create_tables(conn)

CRUD Operations #

Inserting Data #

import mysql.connector

def tambah_pengguna(conn, nama: str, email: str, usia: int) -> int:
    cursor = conn.cursor()
    
    # ANTI-PATTERN: direct string formatting -- vulnerable to SQL Injection
    query = f"INSERT INTO pengguna (nama, email) VALUES ('{nama}', '{email}')"  # ✗
    
    # CORRECT: use a parameterized query with %s placeholders
    query = "INSERT INTO pengguna (nama, email, usia) VALUES (%s, %s, %s)"
    cursor.execute(query, (nama, email, usia))
    
    new_id = cursor.lastrowid
    cursor.close()
    return new_id

# Insert many rows at once (more efficient)
def tambah_banyak_pengguna(conn, daftar_pengguna: list[tuple]) -> int:
    cursor = conn.cursor()
    query = "INSERT INTO pengguna (nama, email, usia) VALUES (%s, %s, %s)"
    cursor.executemany(query, daftar_pengguna)
    jumlah = cursor.rowcount
    cursor.close()
    return jumlah

with db_connection() as conn:
    id1 = tambah_pengguna(conn, "Budi Santoso", "[email protected]", 28)
    print(f"New user ID: {id1}")
    
    data_baru = [
        ("Sari Dewi", "[email protected]", 25),
        ("Andi Prasetyo", "[email protected]", 32),
        ("Rina Marlina", "[email protected]", 29),
    ]
    jumlah = tambah_banyak_pengguna(conn, data_baru)
    print(f"{jumlah} users added.")

SQL Injection is one of the most common and dangerous security vulnerabilities. Never build SQL queries with string formatting or concatenation using user input. Always use parameterized queries with %s placeholders and a separate tuple of values.

Reading Data #

def ambil_semua_pengguna(conn) -> list[dict]:
    # dictionary=True -- rows returned as dicts, not tuples
    cursor = conn.cursor(dictionary=True)
    cursor.execute("SELECT * FROM pengguna WHERE aktif = TRUE ORDER BY dibuat_pada DESC")
    hasil = cursor.fetchall()
    cursor.close()
    return hasil

def ambil_pengguna_by_id(conn, pengguna_id: int) -> dict | None:
    cursor = conn.cursor(dictionary=True)
    cursor.execute("SELECT * FROM pengguna WHERE id = %s", (pengguna_id,))
    hasil = cursor.fetchone()
    cursor.close()
    return hasil

def cari_pengguna(conn, kata_kunci: str) -> list[dict]:
    cursor = conn.cursor(dictionary=True)
    # LIKE with a parameterized query -- add % in the values, not the query
    cursor.execute(
        "SELECT * FROM pengguna WHERE nama LIKE %s OR email LIKE %s",
        (f"%{kata_kunci}%", f"%{kata_kunci}%")
    )
    hasil = cursor.fetchall()
    cursor.close()
    return hasil

with db_connection() as conn:
    semua = ambil_semua_pengguna(conn)
    for p in semua:
        print(f"[{p['id']}] {p['nama']} — {p['email']}")
    
    satu = ambil_pengguna_by_id(conn, 1)
    if satu:
        print(f"Found: {satu['nama']}")
    
    hasil_cari = cari_pengguna(conn, "budi")
    print(f"Search results: {len(hasil_cari)} users")

Updating Data #

def update_pengguna(conn, pengguna_id: int, nama: str, usia: int) -> bool:
    cursor = conn.cursor()
    query = "UPDATE pengguna SET nama = %s, usia = %s WHERE id = %s"
    cursor.execute(query, (nama, usia, pengguna_id))
    berhasil = cursor.rowcount > 0
    cursor.close()
    return berhasil

def nonaktifkan_pengguna(conn, pengguna_id: int) -> bool:
    cursor = conn.cursor()
    cursor.execute("UPDATE pengguna SET aktif = FALSE WHERE id = %s", (pengguna_id,))
    berhasil = cursor.rowcount > 0
    cursor.close()
    return berhasil

with db_connection() as conn:
    ok = update_pengguna(conn, 1, "Budi Santoso Wijaya", 29)
    print("Update successful:", ok)

Deleting Data #

def hapus_pengguna(conn, pengguna_id: int) -> bool:
    cursor = conn.cursor()
    cursor.execute("DELETE FROM pengguna WHERE id = %s", (pengguna_id,))
    berhasil = cursor.rowcount > 0
    cursor.close()
    return berhasil

with db_connection() as conn:
    ok = hapus_pengguna(conn, 99)
    print("Delete successful:", ok)

Transactions #

Transactions ensure a series of database operations execute atomically — all succeed or all are rolled back. This is crucial for operations involving multiple tables or rows that must stay consistent.

def transfer_stok(conn, dari_produk_id: int, ke_produk_id: int, jumlah: int) -> None:
    cursor = conn.cursor()
    
    try:
        # Check available stock
        cursor.execute("SELECT stok FROM produk WHERE id = %s FOR UPDATE", (dari_produk_id,))
        baris = cursor.fetchone()
        
        if not baris or baris[0] < jumlah:
            raise ValueError(f"Insufficient stock. Available: {baris[0] if baris else 0}")
        
        # Decrease source stock
        cursor.execute(
            "UPDATE produk SET stok = stok - %s WHERE id = %s",
            (jumlah, dari_produk_id)
        )
        
        # Increase destination stock
        cursor.execute(
            "UPDATE produk SET stok = stok + %s WHERE id = %s",
            (jumlah, ke_produk_id)
        )
        
        conn.commit()
        print(f"Transferred {jumlah} units successfully.")
        
    except Exception as e:
        conn.rollback()
        print(f"Transaction rolled back: {e}")
        raise
    finally:
        cursor.close()

Connection Pooling #

Creating a new database connection for every request is an expensive operation. A connection pool maintains a set of ready-to-use connections and recycles them.

import mysql.connector
from mysql.connector import pooling
import os

# Create the pool when the app starts
connection_pool = pooling.MySQLConnectionPool(
    pool_name="myapp_pool",
    pool_size=5,                    # number of connections maintained
    pool_reset_session=True,
    host=os.getenv("DB_HOST", "localhost"),
    user=os.getenv("DB_USER", "root"),
    password=os.getenv("DB_PASSWORD", ""),
    database=os.getenv("DB_NAME", "myapp"),
    charset="utf8mb4"
)

def get_pooled_connection():
    return connection_pool.get_connection()

# Usage -- the connection is automatically returned to the pool on close()
def ambil_pengguna_pool(pengguna_id: int) -> dict | None:
    conn = get_pooled_connection()
    try:
        cursor = conn.cursor(dictionary=True)
        cursor.execute("SELECT * FROM pengguna WHERE id = %s", (pengguna_id,))
        return cursor.fetchone()
    finally:
        conn.close()  # returned to the pool, not actually closed

Error Handling #

import mysql.connector
from mysql.connector import Error, errorcode

def tambah_pengguna_aman(conn, nama: str, email: str) -> int | None:
    cursor = conn.cursor()
    try:
        cursor.execute(
            "INSERT INTO pengguna (nama, email) VALUES (%s, %s)",
            (nama, email)
        )
        conn.commit()
        return cursor.lastrowid
    
    except Error as e:
        conn.rollback()
        
        if e.errno == errorcode.ER_DUP_ENTRY:
            print(f"Email '{email}' is already registered.")
        elif e.errno == errorcode.ER_BAD_NULL_ERROR:
            print("A required field is empty.")
        else:
            print(f"Database error [{e.errno}]: {e.msg}")
        
        return None
    finally:
        cursor.close()

Summary #

Parameterized queries are mandatory — always use %s placeholders with a separate tuple of values; never format SQL strings directly from user input.
cursor(dictionary=True) — use it so query result rows come back as dicts (access by column name) instead of tuples (access by index).
charset="utf8mb4" — use this charset (not utf8) to properly support emoji and all Unicode characters.
Don’t hardcode credentials — read host, user, password from environment variables, not directly in code.
Explicit commit() and rollback() — set autocommit=False and manage transactions manually for full control over atomicity.
executemany() — use it for inserting many rows at once; far more efficient than a loop of individual execute() calls.
Connection pooling — use MySQLConnectionPool in web/server apps so you don’t create a new connection per request.
cursor.lastrowid — get the ID of the row just inserted without an extra query.
cursor.rowcount — check whether an UPDATE or DELETE actually changed rows, rather than just running with no effect.

← Previous: YAML Next: MSSQL →