Using Python DB‑API to Operate MySQL: PyMySQL Basics, Connection Pooling, and Safe SQL Practices

Python Database Operations Overview

Python's standard database interface is the Python DB‑API, which provides a uniform API for many relational databases such as MySQL, PostgreSQL, Oracle, and others. Different databases require their own DB‑API driver modules.

GadFly

mSQL

MySQL

PostgreSQL

Microsoft SQL Server 2000

Informix

Interbase

Oracle

Sybase …

The DB‑API defines a set of required objects and methods so that code can interact with different databases in a consistent way.

1. Python DB‑API Usage Flow

Import the API module.

Obtain a connection to the database.

Execute SQL statements or stored procedures.

Close the connection.

2. Python MySQL Modules

Two main approaches are used for MySQL:

Native DB modules (raw SQL): PyMySQL (supports Python 2.x/3.x) and MySQLdb (Python 2.x only).

ORM frameworks: SQLAlchemy .

2.1 Installing PyMySQL

<code>pip install PyMySQL</code>

2.2 Basic Usage

<code>#! /usr/bin/env python
# -*- coding: utf-8 -*-
import pymysql
# Create connection
conn = pymysql.connect(host="127.0.0.1", port=3306, user='zff', passwd='zff123', db='zff', charset='utf8mb4')
# Create cursor returning dicts
cursor = conn.cursor(pymysql.cursors.DictCursor)
# Execute a SELECT
effect_row1 = cursor.execute("select * from USER")
# Insert multiple rows
effect_row2 = cursor.executemany("insert into USER (NAME) values(%s)", [("jack"), ("boom"), ("lucy")])
result = cursor.fetchall()
conn.commit()
cursor.close()
conn.close()
print(result)</code>

2.3 Getting the Last Inserted ID

<code>#! /usr/bin/env python
import pymysql
conn = pymysql.connect(host="127.0.0.1", port=3306, user='zff', passwd='zff123', db='zff', charset='utf8mb4')
cursor = conn.cursor()
cursor.executemany("insert into USER (NAME) values(%s)", [("eric")])
conn.commit()
new_id = cursor.lastrowid
print(new_id)</code>

2.4 Query Operations

<code>#! /usr/bin/env python
import pymysql
conn = pymysql.connect(host="127.0.0.1", port=3306, user='zff', passwd='zff123', db='zff', charset='utf8mb4')
cursor = conn.cursor()
cursor.execute("select * from USER")
row_1 = cursor.fetchone()
row_2 = cursor.fetchmany(3)
row_3 = cursor.fetchall()
cursor.close()
conn.close()
print(row_1)
print(row_2)
print(row_3)</code>

When fetching data you can move the cursor with cursor.scroll(num, mode) (relative or absolute).

2.5 Preventing SQL Injection

<code>#! /usr/bin/env python
import pymysql
conn = pymysql.connect(host="127.0.0.1", port=3306, user='zff', passwd='zff123', db='zff', charset='utf8mb4')
cursor = conn.cursor()
# Unsafe example (do NOT use)
sql = "insert into USER (NAME) values('%s')" % ('zhangsan',)
cursor.execute(sql)
# Safe examples
sql = "insert into USER (NAME) values(%s)"
cursor.execute(sql, ['wang6'])
cursor.execute(sql, ('wang7',))
# Named parameters
sql = "insert into USER (NAME) values(%(name)s)"
cursor.execute(sql, {'name': 'wudalang'})
# Insert many rows safely
cursor.executemany("insert into USER (NAME) values(%s)", [('ermazi'), ('dianxiaoer')])
conn.commit()
cursor.close()
conn.close()</code>

3. Database Connection Pooling

In multithreaded programs creating and closing connections repeatedly can become a bottleneck, so a connection pool is recommended.

3.1 DBUtils Module

DBUtils provides two pooling models:

One connection per thread (closed only when the thread ends).

A shared pool of connections (recommended).

3.2 Model 1 – PersistentDB

<code>#! /usr/bin/env python
from DBUtils.PersistentDB import PersistentDB
import pymysql
POOL = PersistentDB(
    creator=pymysql,
    maxusage=None,
    setsession=[],
    ping=0,
    closeable=False,
    threadlocal=None,
    host='127.0.0.1',
    port=3306,
    user='zff',
    password='zff123',
    database='zff',
    charset='utf8'
)
def func():
    conn = POOL.connection(shareable=False)
    cursor = conn.cursor()
    cursor.execute('select * from USER')
    result = cursor.fetchall()
    cursor.close()
    conn.close()
    return result
result = func()
print(result)</code>

3.3 Model 2 – PooledDB

<code>#! /usr/bin/env python
import time, pymysql, threading
from DBUtils.PooledDB import PooledDB, SharedDBConnection
POOL = PooledDB(
    creator=pymysql,
    maxconnections=6,
    mincached=2,
    maxcached=5,
    maxshared=3,
    blocking=True,
    maxusage=None,
    setsession=[],
    ping=0,
    host='127.0.0.1',
    port=3306,
    user='zff',
    password='zff123',
    database='zff',
    charset='utf8'
)
def func():
    conn = POOL.connection()
    cursor = conn.cursor()
    cursor.execute('select * from USER')
    result = cursor.fetchall()
    conn.close()
    return result
result = func()
print(result)</code>

Because pymysql and MySQLdb have a threadsafety value of 1, the pooled connections are thread‑safe; without a pool you would need explicit locks, which reduces performance.

3.4 Lock‑Based Access (Example)

<code>#! /usr/bin/env python
import pymysql, threading
from threading import RLock
LOCK = RLock()
CONN = pymysql.connect(host='127.0.0.1', port=3306, user='zff', passwd='zff123', db='zff', charset='utf8')
def task(arg):
    with LOCK:
        cursor = CONN.cursor()
        cursor.execute('select * from USER')
        result = cursor.fetchall()
        cursor.close()
        print(result)
for i in range(10):
    t = threading.Thread(target=task, args=(i,))
    t.start()</code>

3.5 No‑Lock Access (May Fail)

<code>#! /usr/bin/env python
import pymysql, threading
CONN = pymysql.connect(host='127.0.0.1', port=3306, user='zff', passwd='zff123', db='zff', charset='utf8')
def task(arg):
    cursor = CONN.cursor()
    cursor.execute('select * from USER')
    result = cursor.fetchall()
    cursor.close()
    print(result)
for i in range(10):
    t = threading.Thread(target=task, args=(i,))
    t.start()</code>

You can monitor active threads in MySQL with show status like 'Threads%';

4. Helper Module Combining PyMySQL with a Pool

<code># sql_helper.py
import pymysql, threading
from DBUtils.PooledDB import PooledDB, SharedDBConnection
POOL = PooledDB(
    creator=pymysql,
    maxconnections=20,
    mincached=2,
    maxcached=5,
    blocking=True,
    maxusage=None,
    setsession=[],
    ping=0,
    host='192.168.11.38',
    port=3306,
    user='root',
    passwd='apNXgF6RDitFtDQx',
    db='m2day03db',
    charset='utf8'
)
def connect():
    conn = POOL.connection()
    cursor = conn.cursor(pymysql.cursors.DictCursor)
    return conn, cursor
def close(conn, cursor):
    cursor.close()
    conn.close()
def fetch_one(sql, args):
    conn, cursor = connect()
    cursor.execute(sql, args)
    result = cursor.fetchone()
    close(conn, cursor)
    return result
def fetch_all(sql, args):
    conn, cursor = connect()
    cursor.execute(sql, args)
    result = cursor.fetchall()
    close(conn, cursor)
    return result
def insert(sql, args):
    conn, cursor = connect()
    cursor.execute(sql, args)
    conn.commit()
    close(conn, cursor)
def delete(sql, args):
    conn, cursor = connect()
    cursor.execute(sql, args)
    conn.commit()
    close(conn, cursor)
    return cursor.rowcount
def update(sql, args):
    conn, cursor = connect()
    cursor.execute(sql, args)
    conn.commit()
    close(conn, cursor)
    return cursor.rowcount</code>

At the end of the article a QR code is presented for a free Python public‑course resource, which is promotional in nature.