Python Database Access: DB‑API, PyMySQL, and Connection Pooling

Python's standard database interface, DB‑API, defines a common set of objects and methods for interacting with many relational databases such as MySQL, PostgreSQL, Oracle, and others. By importing the appropriate DB‑API module, developers can write database‑agnostic code.

The article lists several supported databases and outlines the typical DB‑API workflow: import the module, establish a connection, execute SQL statements, and close the connection.

PyMySQL Installation

<code>pip install PyMySQL</code>

Basic PyMySQL Usage

<code>#!/usr/bin/env python
# -*- coding: utf-8 -*-
# __author__ = "shuke"
# Date: 2018/5/13

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 query
effect_row1 = cursor.execute("select * from USER")

# Insert multiple rows
effect_row2 = cursor.executemany("insert into USER (NAME) values(%s)", [("jack"), ("boom"), ("lucy")])

# Fetch all results
result = cursor.fetchall()

# Commit and close
conn.commit()
cursor.close()
conn.close()

print(result)
"""
[{'id': 6, 'name': 'boom'}, {'id': 5, 'name': 'jack'}, {'id': 7, 'name': 'lucy'}, ...]
"""
</code>

Retrieving the Last Inserted ID

<code>#!/usr/bin/env python
# -*- coding: utf-8 -*-
# __author__ = "shuke"
# Date: 2018/5/13

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)
"""
8
"""
</code>

Query Operations

<code>#!/usr/bin/env python
# -*- coding: utf-8 -*-
# __author__ = "shuke"
# Date: 2018/5/13

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>

To move the cursor within a result set, cursor.scroll(num, mode) can be used, where mode='relative' moves relative to the current position and mode='absolute' moves to an absolute index.

Preventing SQL Injection

<code>#!/usr/bin/env python
# -*- coding: utf-8 -*-
# __author__ = "shuke"
# Date: 2018/5/13

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',))

sql = "insert into USER (NAME) values(%(name)s)"
cursor.execute(sql, {'name': 'wudalang'})

cursor.executemany("insert into USER (NAME) values(%s)", [('ermazi'), ('dianxiaoer')])
conn.commit()
cursor.close()
conn.close()
</code>

Because different DB‑API modules use different placeholder styles (e.g., %s for PyMySQL, ? for sqlite3), always consult the module's documentation.

Database Connection Pooling with DBUtils

Two pooling strategies are described:

Per‑thread connections that are returned to the pool when close() is called.

A shared pool of pre‑created connections (recommended).

Mode 1 – PersistentDB

<code>#!/usr/bin/env python
# -*- coding: utf-8 -*-
# __author__ = "shuke"
# Date: 2018/5/13

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>

Mode 2 – PooledDB

<code>#!/usr/bin/env python
# -*- coding: utf-8 -*-
# __author__ = "shuke"
# Date: 2018/5/13

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>

Since pymysql and MySQLdb have a threadsafety value of 1, the pool's connections are thread‑safe, but without a pool each thread would need its own connection or explicit locking.

Lock‑Based Access

<code>#!/usr/bin/env python
# -*- coding: utf-8 -*-
# __author__ = "shuke"
# Date: 2018/5/13

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()
</code>

Lock‑Free (Error‑Prone) Example

<code>#!/usr/bin/env python
# -*- coding: utf-8 -*-
# __author__ = "shuke"
# Date: 2018/5/13

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()
</code>

Running the lock‑free version can cause race conditions and may be observed in MySQL with show status like 'Threads%'; .

Helper Module Combining DBUtils and PyMySQL

<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()
    effect_row = 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()
    effect_row = cursor.execute(sql, args)
    conn.commit()
    close(conn, cursor)
    return effect_row

def update(sql, args):
    conn, cursor = connect()
    effect_row = cursor.execute(sql, args)
    conn.commit()
    close(conn, cursor)
    return effect_row
</code>

The article concludes with suggestions to encapsulate these utilities in a class for easier reuse and provides several recommended reading links.