Master Python Database Access: From DB‑API to Connection Pools

1. Introduction to Python Database Operations

Python's standard database interface is the Python DB‑API, which provides a uniform API for developers to work with many different databases such as GadFly, mSQL, MySQL, PostgreSQL, Microsoft SQL Server, Informix, Interbase, Oracle, Sybase, etc. Different databases require installing their respective DB‑API modules (e.g., Oracle or MySQL drivers).

The DB‑API defines a set of required objects and access methods, allowing consistent interaction with various underlying database systems.

Typical DB‑API workflow:

Import the API module.

Establish a connection to the database.

Execute SQL statements or stored procedures.

Close the connection.

2. Python MySQL Modules

Python can work with MySQL using two approaches:

Native DB modules (raw SQL)

PyMySQL (supports Python 2.x/3.x)

MySQLdb (Python 2.x only)

ORM frameworks

SQLAlchemy

2.1 PyMySQL Module

PyMySQL is a pure‑Python MySQL driver that allows Python code to interact with MySQL databases.

2.1.1 Installing PyMySQL

pip install PyMySQL

2.2 Basic Usage

#! /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 results as dictionaries
cursor = conn.cursor(pymysql.cursors.DictCursor)

# Execute a SELECT 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 changes for INSERT/UPDATE/DELETE
conn.commit()

# Clean up
cursor.close()
conn.close()

print(result)

2.3 Getting the Auto‑Increment ID of Newly Inserted Data

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

# Insert a new row and retrieve its auto‑increment ID
effect_row = cursor.executemany("insert into USER (NAME) values(%s)", [("eric")])
conn.commit()

new_id = cursor.lastrowid
print(new_id)

cursor.close()
conn.close()

2.4 Query Operations

#! /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()          # first row
row_2 = cursor.fetchmany(3)        # next 3 rows
row_3 = cursor.fetchall()          # all remaining rows

print(row_1)
print(row_2)
print(row_3)

cursor.close()
conn.close()

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

2.5 Preventing SQL Injection

#! /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 string formatting (vulnerable to injection)
sql = "insert into USER (NAME) values('%s')" % ('zhangsan',)
cursor.execute(sql)

# Safe parameterized execution (preferred)
sql = "insert into USER (NAME) values(%s)"
cursor.execute(sql, ['wang6'])
cursor.execute(sql, ('wang7',))

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

# Insert multiple rows safely
cursor.executemany("insert into USER (NAME) values(%s)", [('ermazi'), ('dianxiaoer')])

conn.commit()
cursor.close()
conn.close()

Use parameterized queries to avoid injection; note that different drivers may use different placeholder styles (e.g., '?' for sqlite3).

3. Database Connection Pools

Repeatedly opening and closing connections in multithreaded programs can cause performance issues; connection pools solve this by reusing connections.

3.1 DBUtils Module

DBUtils provides two pooling modes:

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

Shared pool of connections for all threads (recommended).

3.2 Mode One (PersistentDB)

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

3.3 Mode Two (PooledDB)

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

import time, threading
import pymysql
from DBUtils.PooledDB import PooledDB

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)

Because pymysql and MySQLdb have a threadsafety value of 1, connections obtained from the pool are safely shared among threads. Without a pool, multithreaded programs would need explicit locking, which can become a bottleneck.

3.4 Locking

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

3.5 No Lock (Error)

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

You can monitor active connections in MySQL with

show status like 'Threads%';

4. Using Database Connection Pool with PyMySQL

# sql_helper.py
import pymysql, threading
from DBUtils.PooledDB import PooledDB

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

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

These helper functions can be wrapped into a class for convenient reuse.