เชื่อมต่อ MySQL ด้วย Python (PyMySQL)

PyMySQL คืออะไร

PyMySQL เป็น Pure-Python MySQL Client Library ใช้เชื่อมต่อ MySQL/MariaDB จาก Python ไม่ต้องพึ่ง C Extension (ต่างจาก mysqlclient/MySQLdb)

ติดตั้ง PyMySQL

# pip (มาตรฐาน)
pip install pymysql

# Conda (Anaconda/Miniconda)
conda install -c anaconda pymysql

# หรือผ่าน IDE (PyCharm, VS Code, ฯลฯ) ค้นหา pymysql ใน Package Manager

เชื่อมต่อ MySQL และตรวจสอบ Version

import pymysql

# เปิดการเชื่อมต่อ
db = pymysql.connect(
    host="localhost",
    user="testuser",
    password="test123",
    database="TESTDB"
)

# สร้าง Cursor
cursor = db.cursor()

# รัน Query
cursor.execute("SELECT VERSION()")

# ดึงผลลัพธ์ 1 แถว
data = cursor.fetchone()
print("Database version : %s" % data)

# ปิดการเชื่อมต่อ
db.close()

ผลลัพธ์ตัวอย่าง

Database version : 8.0.19

รัน Query ดึงข้อมูล (Parameterized Query)

import pymysql

db = pymysql.connect("localhost", "username", "paswd", "DBname")
cursor = db.cursor()

# ใช้ Parameterized Query ป้องกัน SQL Injection
sql = "SELECT * FROM EMPLOYEE WHERE INCOME > %s"
threshold = 1000

try:
    cursor.execute(sql, (threshold,))  # ส่ง Parameter เป็น Tuple
    results = cursor.fetchall()
    
    for row in results:
        fname, lname, age, sex, income = row
        print("fname=%s, lname=%s, age=%d, sex=%s, income=%d" % 
              (fname, lname, age, sex, income))
except Exception as e:
    print("Error: unable to fetch data -", e)
finally:
    db.close()

ผลลัพธ์ตัวอย่าง

fname=Jack, lname=Ma, age=31, sex=M, income=12000

Best Practices สำคัญ

• Parameterized Queries: ใช้ %s Placeholder + Tuple ใน execute() ห้าม ใช้ String Formatting (%s ใน String)
• Context Manager (with): ปิด Connection อัตโนมัติ แม้ Error
• Connection Pool: Production ควรใช้ pymysql.pooling หรือ DBUtils
• DictCursor: คืนค่าเป็น Dictionary แทน Tuple

ตัวอย่างแบบ Modern (Python 3.7+)

import pymysql
from pymysql.cursors import DictCursor

with pymysql.connect(
    host="localhost",
    user="username",
    password="paswd",
    database="DBname",
    cursorclass=DictCursor,
    autocommit=True
) as db:
    with db.cursor() as cursor:
        cursor.execute("SELECT * FROM EMPLOYEE WHERE INCOME > %s", (1000,))
        for row in cursor.fetchall():
            print(f"{row['fname']} {row['lname']}: ${row['income']}")

Troubleshooting

• ModuleNotFoundError: No module named 'pymysql' → pip install pymysql
• Access denied for user → ตรวจ User/Host/Password ใน MySQL (CREATE USER 'user'@'%' IDENTIFIED BY 'pass'; GRANT ALL ON db.* TO 'user'@'%';)
• Can't connect to MySQL server → ตรวจ Host/Port/Firewall/MySQL Service Running
• Authentication plugin 'caching_sha2_password' not supported → ใช้ MySQL 8.0+ ต้อง Upgrade PyMySQL หรือตั้ง User ใช้ mysql_native_password