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

Python's standard database interface, DB‑API, provides a uniform way to interact with many databases such as MySQL, PostgreSQL, Oracle, and SQLite.

The typical DB‑API workflow includes importing the module, establishing a connection, executing SQL statements, and finally closing the connection.

To work with MySQL, install the PyMySQL driver:

pip install PyMySQL

Basic usage involves creating a connection and a cursor (optionally a DictCursor for dictionary results), executing queries, committing changes, and closing resources. Example:

#! /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)
cursor.execute("select * from USER")
result = cursor.fetchall()
conn.commit()
cursor.close()
conn.close()
print(result)

To obtain the auto‑increment ID of a newly inserted row, use cursor.lastrowid after an INSERT operation.

effect_row = cursor.executemany("insert into USER (NAME) values(%s)", [("eric",)])
conn.commit()
new_id = cursor.lastrowid
print(new_id)

Querying data can be done with fetchone() , fetchmany(n) , or fetchall() . Cursor positioning can be adjusted with cursor.scroll() if needed.

To prevent SQL injection, avoid string formatting and use parameterized queries, either passing a tuple/list or a dictionary of parameters.

sql = "insert into USER (NAME) values(%s)"
cursor.execute(sql, ['wang6'])
# or using named placeholders
sql = "insert into USER (NAME) values(%(name)s)"
cursor.execute(sql, {'name': 'wudalang'})
conn.commit()

For multithreaded applications, a database connection pool is recommended. The DBUtils library offers two patterns:

PersistentDB creates a dedicated connection per thread.

PooledDB shares a pool of connections among threads.

Example of PersistentDB:

from DBUtils.PersistentDB import PersistentDB
import pymysql
POOL = PersistentDB(creator=pymysql, maxusage=None, 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

print(func())

Example of PooledDB:

from DBUtils.PooledDB import PooledDB
import pymysql
POOL = PooledDB(creator=pymysql, maxconnections=6, mincached=2, maxcached=5,
                blocking=True, maxusage=None, 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

print(func())

When threads share a single connection, a re‑entrant lock can serialize access:

import pymysql, threading
LOCK = threading.RLock()
CONN = pymysql.connect(host='127.0.0.1', port=3306, user='zff', password='zff123', database='zff', charset='utf8')

def task(i):
    with LOCK:
        cur = CONN.cursor()
        cur.execute('select * from USER')
        print(cur.fetchall())
        cur.close()

for i in range(10):
    threading.Thread(target=task, args=(i,)).start()

A helper module can encapsulate pool creation and common CRUD functions, returning connections and cursors as needed and ensuring they are closed after each operation.