Working with SQL proficiency forms the basis for using Python modules to interact with MySQL in production tasks.

Installing and Importing the Driver

Install the MySQL driver via package manager:

pip install PyMySQL

Import the module in code:

import pymysql

Connection Object

Establishes a link to the database. Instantiate it using pymysql.connect() with parameters:

• host: database server address ('localhost' for local)
• port: server port (default 3306)
• database: target schema name
• user: authentication username
• password: authentication password
• charset: communication encoding (should match DB collation, e.g., 'utf8mb4')

Example:

db_conn = pymysql.connect(
    host='127.0.0.1',
    port=3306,
    database='school',
    user='admin',
    password='secret',
    charset='utf8mb4'
)

Methods:

• close(): temrinate connection
• commit(): persist transaction changes
• rollback(): discard transaction changes
• cursor(): obtain a Cursor for executing statements

Cursor Object

Used to run queries and process results. Create via db_conn.cursor().

Methods:

• close(): release cursor resources
• execute(stmt, args=None): run a statement; returns affected row count
• fetchone(): retrieve the first matching row as a tuple
• fetchall(): retrieve all rows, each as a tuple within a containing tuple
• scroll(offset, mode='relative'): reposition pointer; relative moves from current position, absolute from start (index 0)

Attributes:

• rowcount: number of rows impacted by last execute()
• connection: reference to originating Connection

Insert Operation

Example inserting a student record:

import pymysql

try:
    conn = pymysql.connect(host='127.0.0.1', port=3306, database='school', user='admin', password='secret', charset='utf8mb4')
    cur = conn.cursor()
    affected = cur.execute("INSERT INTO pupil(name) VALUES ('Li Wei')")
    print(affected)
    conn.commit()
    cur.close()
    conn.close()
except Exception as err:
    print(err)

Update Operation

Modify an existing entry:

import pymysql

try:
    conn = pymysql.connect(host='127.0.0.1', port=3306, database='school', user='admin', password='secret', charset='utf8mb4')
    cur = conn.cursor()
    affected = cur.execute("UPDATE pupil SET name='Liu Bang' WHERE uid=6")
    print(affected)
    conn.commit()
    cur.close()
    conn.close()
except Exception as err:
    print(err)

Delete Operation

Remove a record:

import pymysql

try:
    conn = pymysql.connect(host='127.0.0.1', port=3306, database='school', user='admin', password='secret', charset='utf8mb4')
    cur = conn.cursor()
    affected = cur.execute("DELETE FROM pupil WHERE uid=6")
    print(affected)
    conn.commit()
    cur.close()
    conn.close()
except Exception as err:
    print(err)

Parameterized Queries

Prevent injection and handle dynamic input:

import pymysql

try:
    conn = pymysql.connect(host='127.0.0.1', port=3306, database='school', user='admin', password='secret', charset='utf8mb4')
    cur = conn.cursor()
    name = input("Enter student name: ")
    affected = cur.execute("INSERT INTO pupil(name) VALUES (%s)", (name,))
    print(affected)
    conn.commit()
    cur.close()
    conn.close()
except Exception as err:
    print(err)

Other Statements

execute() can run DDL such as CREATE TABLE. Defining schema should occur before runtime execution.

Fetch Single Row

Retrieve one record:

import pymysql

try:
    conn = pymysql.connect(host='127.0.0.1', port=3306, database='school', user='admin', password='secret', charset='utf8mb4')
    cur = conn.cursor()
    cur.execute("SELECT * FROM pupil WHERE uid=7")
    row = cur.fetchone()
    print(row)
    cur.close()
    conn.close()
except Exception as err:
    print(err)

Fetch Multiple Rows

Retrieve all records:

import pymysql

try:
    conn = pymysql.connect(host='127.0.0.1', port=3306, database='school', user='admin', password='secret', charset='utf8mb4')
    cur = conn.cursor()
    cur.execute("SELECT * FROM pupil")
    rows = cur.fetchall()
    print(rows)
    cur.close()
    conn.close()
except Exception as err:
    print(err)

Helper Class for Reuse

Common connection and CRUD logic can be abstracted:

import pymysql

class DbClient:
    def __init__(self, host, port, db, user, pwd, charset='utf8mb4'):
        self.host = host
        self.port = port
        self.db = db
        self.user = user
        self.pwd = pwd
        self.charset = charset

    def _open(self):
        self.link = pymysql.connect(
            host=self.host,
            port=self.port,
            database=self.db,
            user=self.user,
            password=self.pwd,
            charset=self.charset
        )
        self.handle = self.link.cursor()

    def _shutdown(self):
        self.handle.close()
        self.link.close()

    def fetch_one(self, query, vals=()):
        self._open()
        self.handle.execute(query, vals)
        rec = self.handle.fetchone()
        self._shutdown()
        return rec

    def fetch_all(self, query, vals=()):
        self._open()
        self.handle.execute(query, vals)
        recs = self.handle.fetchall()
        self._shutdown()
        return recs

    def modify(self, query, vals=()):
        self._open()
        cnt = self.handle.execute(query, vals)
        self.link.commit()
        self._shutdown()
        return cnt

Using the helper for insertion:

from DbClient import DbClient

query = "INSERT INTO pupil(name, sex) VALUES (%s, %s)"
name = input("Enter name: ")
sex = input("Enter gender (True/False): ")
client = DbClient('127.0.0.1', 3306, 'school', 'admin', 'secret')
res = client.modify(query, (name, sex == 'True'))
print('OK' if res == 1 else 'Error')

Fetch most recent entry:

from DbClient import DbClient

query = "SELECT name, sex FROM pupil ORDER BY uid DESC"
client = DbClient('127.0.0.1', 3306, 'school', 'admin', 'secret')
entry = client.fetch_one(query)
print(entry)

User Login Example

Schema definition with SHA-1 hashed passwords:

CREATE TABLE user_account (
    uid INT AUTO_INCREMENT PRIMARY KEY,
    username VARCHAR(20),
    pass_hash CHAR(40),
    disabled BIT DEFAULT 0
);

Seed data (SHA-1 of 123):

INSERT INTO user_account VALUES (NULL, '123', '40bd001563085fc35165329ea1ff5c5ecbdbbeef', 0);

Verification script:

from DbClient import DbClient
import hashlib

usr = input("Username: ")
pwd = input("Password: ")
hasher = hashlib.sha1(pwd.encode())
pwd_hash = hasher.hexdigest()

query = "SELECT pass_hash FROM user_account WHERE username=%s"
client = DbClient('127.0.0.1', 3306, 'school', 'admin', 'secret')
record = client.fetch_one(query, (usr,))

if not record:
    print('Invalid username')
elif record[0] == pwd_hash:
    print('Login successful')
else:
    print('Incorrect password')