Python Database Access: DB‑API Overview, MySQL Operations, and Connection Pooling

Python provides a standard database interface known as DB‑API, which defines a consistent set of objects and methods for interacting with various relational databases.

GadFly

mSQL

MySQL

PostgreSQL

Microsoft SQL Server 2000

Informix

Interbase

Oracle

Sybase …

The typical usage flow of Python DB‑API is:

Import the API module.

Obtain a connection to the database.

Execute SQL statements or stored procedures.

Close the connection.

For MySQL, two approaches are common:

Native DB modules (raw SQL): PyMySQL (supports Python 2.x/3.x) and MySQLdb (Python 2.x only).

ORM frameworks: SQLAlchemy .

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 changes
conn.commit()

# Close resources
cursor.close()
conn.close()

print(result)
"""
[{'id': 6, 'name': 'boom'}, {'id': 5, 'name': 'jack'}, {'id': 7, 'name': 'lucy'}, {'id': 4, 'name': 'tome'}, {'id': 3, 'name': 'zff'}, {'id': 1, 'name': 'zhaofengfeng'}, {'id': 2, 'name': 'zhaofengfeng02'}]
"""
</code>

Getting the Last Inserted Auto‑Increment 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()

# Insert a row
effect_row = cursor.executemany("insert into USER (NAME) values(%s)", [("eric")])
conn.commit()

cursor.close()
conn.close()

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

cursor.close()
conn.close()

print(row_1)
print(row_2)
print(row_3)
</code>

To move the cursor manually, use cursor.scroll(num, mode) where mode='relative' moves relative to the current position and mode='absolute' moves to an absolute offset.

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 (do NOT use string formatting)
sql = "insert into USER (NAME) values('%s')" % ('zhangsan',)
cursor.execute(sql)

# Safe method 1: pass parameters as a sequence
sql = "insert into USER (NAME) values(%s)"
cursor.execute(sql, ['wang6'])
cursor.execute(sql, ('wang7',))

# Safe method 2: 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()
</code>

Using parameterized queries eliminates the risk of SQL injection; note that different drivers may use different placeholder styles (e.g., ? for sqlite3 ).

Database Connection Pooling with DBUtils

DBUtils provides two pooling modes:

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

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, threading
import pymysql
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>

Because pymysql and MySQLdb have a threadsafety value of 1, the pooled connections are thread‑safe and can be shared across threads without additional locking.

Locking vs. No‑Locking in Multithreaded 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>

Without a lock, concurrent threads share the same connection, which can lead to race conditions; using a pool or explicit locking avoids these issues.

Utility Functions Using a Pooled Connection

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

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

These helper functions encapsulate common CRUD operations while automatically handling connection acquisition and release via the pool.