Python Database Operations: Using DB-API, PyMySQL, and Connection Pools

Python provides a standard DB‑API that offers a uniform interface for many relational databases such as MySQL, PostgreSQL, Oracle, and others.

The article first lists the databases supported by Python's DB‑API and outlines the typical workflow: import the driver, obtain a connection, execute SQL, and close the connection.

PyMySQL installation

pip install PyMySQL

Basic usage example

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

Getting the last inserted auto‑increment ID

#!/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()
cursor.executemany("insert into USER (NAME) values(%s)", [("eric")])
conn.commit()
new_id = cursor.lastrowid
print(new_id)

Query operations

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

Preventing SQL injection

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

# unsafe concatenation (do not use)
sql = "insert into USER (NAME) values('%s')" % ('zhangsan',)
cursor.execute(sql)

# safe parameterized execution
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'})
conn.commit()
cursor.close()
conn.close()

Database connection pool with DBUtils (PersistentDB mode)

#!/usr/bin/env python
# -*- coding: utf-8 -*-
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)

Database connection pool with DBUtils (PooledDB mode)

#!/usr/bin/env python
# -*- coding: utf-8 -*-
from DBUtils.PooledDB import PooledDB
import pymysql

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)

Thread‑safe access using a lock

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pymysql, threading
from threading import RLock

LOCK = RLock()
CONN = pymysql.connect(host='127.0.0.1', port=3306, user='zff', password='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()

Without a lock (may raise errors in multithreaded use)

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pymysql, threading

CONN = pymysql.connect(host='127.0.0.1', port=3306, user='zff', password='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()

The article concludes with a reusable helper module (sql_helper.py) that encapsulates connection‑pool creation and common CRUD functions (fetch_one, fetch_all, insert, delete, update) using DBUtils, providing a clean API for database operations in Python projects.