Master Python Database Access: From DB‑API Basics to Connection Pooling with PyMySQL

Python Database Interface Overview

Python's standard database interface is the DB‑API, which provides a uniform API for many databases such as GadFly, mSQL, MySQL, PostgreSQL, Microsoft SQL Server, Informix, Oracle, and others. Different databases require their own DB‑API modules (e.g., PyMySQL for MySQL).

Using Python DB‑API

Import the API module.

Establish a connection.

Execute SQL statements and stored procedures.

Close the connection.

PyMySQL Module

Installation

pip install PyMySQL

Basic Usage

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pymysql
conn = pymysql.connect(host="127.0.0.1", port=3306, user='zff', passwd='zff123', db='zff', charset='utf8mb4')
cursor = conn.cursor(pymysql.cursors.DictCursor)
# Execute a query
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)

Retrieve Last Inserted ID

#!/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("insert into USER (NAME) values('er')")
new_id = cursor.lastrowid
conn.commit()
cursor.close()
conn.close()
print(new_id)

Query Operations

#!/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)

When fetching data, you can move the cursor with cursor.scroll(num, mode) (e.g., cursor.scroll(1, mode='relative') or cursor.scroll(2, mode='absolute')).

Prevent SQL Injection

#!/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()
# Vulnerable example (do NOT use)
sql = "insert into USER (NAME) values('%s')" % ('zhangsan')
cursor.execute(sql)
# Safe example using parameters
sql = "insert into USER (NAME) values(%s)"
cursor.execute(sql, ['wang6'])
# Named parameters
sql = "insert into USER (NAME) values(%(name)s)"
cursor.execute(sql, {'name': 'wudalang'})
conn.commit()
cursor.close()
conn.close()

Note: Different DB‑API modules use different placeholder styles (e.g., %s for PyMySQL, ? for sqlite3).

Database Connection Pooling with DBUtils

DBUtils provides two pooling modes:

Per‑thread connections (connections are returned to the pool when closed).

Shared pool of connections (recommended).

Mode 1 – PersistentDB

#!/usr/bin/env python
from DBUtils.PersistentDB import PersistentDB
import pymysql
POOL = PersistentDB(creator=pymysql, maxusage=None, setsession=[], ping=0, closeable=False, 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)

Mode 2 – PooledDB

#!/usr/bin/env python
from DBUtils.PooledDB import PooledDB
import pymysql
POOL = PooledDB(creator=pymysql, maxconnections=6, mincached=2, maxcached=5, 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)

Locking Example (Thread‑Safe)

#!/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', database='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()

No‑Lock Example (May Fail)

#!/usr/bin/env python
import pymysql, threading
CONN = pymysql.connect(host='127.0.0.1', port=3306, user='zff', passwd='zff123', database='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()

Use show status like 'Threads%'; in MySQL to monitor thread usage.