Master MySQL Transactions, Locks, and Connection Pools with Python Examples

1. Transactions

InnoDB supports transactions while MyISAM does not.

CREATE TABLE `users` (
  `id` int(11) NOT NULL AUTO_INCREMENT PRIMARY KEY,
  `name` varchar(32) DEFAULT NULL,
  `amount` int(11) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

Example: Li Jie transfers 100 to Wu Peiqi, which requires two steps—decreasing Li Jie's balance and increasing Wu Peiqi's balance. Both steps must succeed together or be rolled back, embodying the ACID principle: all‑or‑nothing.

Atomicity – all operations in a transaction are indivisible.

Consistency – data integrity is preserved before and after the transaction.

Isolation – a transaction runs without interference from others.

Durability – once committed, changes persist permanently.

MySQL client workflow:

mysql> SELECT * FROM users;
+----+---------+--------+
| id | name    | amount |
+----+---------+--------+
| 1  | wupeiqi | 5      |
| 2  | alex    | 6      |
+----+---------+--------+

mysql> BEGIN;   -- start transaction
mysql> UPDATE users SET amount = amount-2 WHERE id=1;
mysql> UPDATE users SET amount = amount+2 WHERE id=2;
mysql> COMMIT;
mysql> SELECT * FROM users;

1.1 MySQL Client

mysql> SELECT * FROM users;
... (output omitted for brevity) ...
mysql> BEGIN;
mysql> UPDATE users SET amount=1 WHERE id=1;
... (error occurs) ...
mysql> ROLLBACK;
mysql> SELECT * FROM users;

2. Locks

MySQL provides locking to ensure data consistency under concurrent updates, inserts, or deletes.

Table‑level lock – the whole table is locked.

Row‑level lock – only the targeted rows are locked.

MyISAM only supports table locks; InnoDB supports both row and table locks. Therefore InnoDB is usually preferred, especially when indexes are used.

CREATE TABLE `L1` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `name` varchar(255) DEFAULT NULL,
  `count` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

InnoDB automatically acquires exclusive locks for UPDATE/INSERT/DELETE operations before execution and releases them afterward. SELECT statements do not lock by default.

2.1 Exclusive Lock (FOR UPDATE)

Exclusive locks prevent other transactions from reading or writing the locked rows.

Scenario: selling a product with limited stock. Use FOR UPDATE to lock the row, check the remaining quantity, and then decrement it only if the quantity is greater than zero.

BEGIN;
SELECT count FROM goods WHERE id=3 FOR UPDATE;
-- if count > 0 then
UPDATE goods SET count = count-1 WHERE id=3;
COMMIT;

Python example:

import pymysql, threading

def task():
    conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='root123', db='userdb', charset='utf8')
    cursor = conn.cursor(pymysql.cursors.DictCursor)
    conn.begin()
    cursor.execute("SELECT id, age FROM tran WHERE id=2 FOR UPDATE")
    row = cursor.fetchone()
    if row['age'] > 0:
        cursor.execute("UPDATE tran SET age=age-1 WHERE id=2")
    else:
        print('Sold out')
    conn.commit()
    cursor.close()
    conn.close()

def run():
    for _ in range(5):
        threading.Thread(target=task).start()

if __name__ == '__main__':
    run()

2.2 Shared Lock (LOCK IN SHARE MODE)

Shared locks allow other transactions to read but not write the locked rows.

SELECT * FROM parent WHERE name='Jones' LOCK IN SHARE MODE;

After acquiring a shared lock, you can safely insert a child row that references the parent, because any concurrent transaction attempting an exclusive lock on the same row will wait until the shared lock is released.

3. Database Connection Pool

Using a connection pool improves performance when many requests need database access.

pip3 install pymysql
pip3 install dbutils

import pymysql
from dbutils.pooled_db import PooledDB

MYSQL_DB_POOL = PooledDB(
    creator=pymysql,
    maxconnections=5,
    mincached=2,
    maxcached=3,
    blocking=True,
    setsession=[],
    ping=0,
    host='127.0.0.1',
    port=3306,
    user='root',
    password='root123',
    database='userdb',
    charset='utf8'
)

def task():
    conn = MYSQL_DB_POOL.connection()
    cursor = conn.cursor(pymysql.cursors.DictCursor)
    cursor.execute('SELECT SLEEP(2)')
    print(cursor.fetchall())
    cursor.close()
    conn.close()

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

4. SQL Utility Class

A reusable helper class abstracts connection‑pool handling and common CRUD operations.

4.1 Singleton and Methods

# db.py
import pymysql
from dbutils.pooled_db import PooledDB

class DBHelper(object):
    def __init__(self):
        self.pool = PooledDB(
            creator=pymysql,
            maxconnections=5,
            mincached=2,
            maxcached=3,
            blocking=True,
            setsession=[],
            ping=0,
            host='127.0.0.1',
            port=3306,
            user='root',
            password='root123',
            database='userdb',
            charset='utf8'
        )
    def get_conn_cursor(self):
        conn = self.pool.connection()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        return conn, cursor
    def close_conn_cursor(self, *args):
        for item in args:
            item.close()
    def exec(self, sql, **kwargs):
        conn, cursor = self.get_conn_cursor()
        cursor.execute(sql, kwargs)
        conn.commit()
        self.close_conn_cursor(conn, cursor)
    def fetch_one(self, sql, **kwargs):
        conn, cursor = self.get_conn_cursor()
        cursor.execute(sql, kwargs)
        result = cursor.fetchone()
        self.close_conn_cursor(conn, cursor)
        return result
    def fetch_all(self, sql, **kwargs):
        conn, cursor = self.get_conn_cursor()
        cursor.execute(sql, kwargs)
        result = cursor.fetchall()
        self.close_conn_cursor(conn, cursor)
        return result

db = DBHelper()

from db import db

db.exec("INSERT INTO d1(name) VALUES(%(name)s)", name='Wu Peiqi666')
print(db.fetch_one('SELECT * FROM d1'))
print(db.fetch_all('SELECT * FROM d1'))

4.2 Context Manager

Support the with statement to automatically return connections to the pool.

# db_context.py
import pymysql
from dbutils.pooled_db import PooledDB

POOL = PooledDB(
    creator=pymysql,
    maxconnections=5,
    mincached=2,
    maxcached=3,
    blocking=True,
    setsession=[],
    ping=0,
    host='127.0.0.1',
    port=3306,
    user='root',
    password='root123',
    database='userdb',
    charset='utf8'
)

class Connect(object):
    def __init__(self):
        self.conn = POOL.connection()
        self.cursor = self.conn.cursor(pymysql.cursors.DictCursor)
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.cursor.close()
        self.conn.close()
    def exec(self, sql, **kwargs):
        self.cursor.execute(sql, kwargs)
        self.conn.commit()
    def fetch_one(self, sql, **kwargs):
        self.cursor.execute(sql, kwargs)
        return self.cursor.fetchone()
    def fetch_all(self, sql, **kwargs):
        self.cursor.execute(sql, kwargs)
        return self.cursor.fetchall()

from db_context import Connect

with Connect() as db:
    print(db.fetch_one('SELECT * FROM d1'))
    print(db.fetch_one('SELECT * FROM d1 WHERE id=%(id)s', id=3))

Summary

Transactions ensure that a batch of operations either all succeed or all fail.

Locks (exclusive and shared) handle concurrency control.

Connection pools manage multiple simultaneous database connections efficiently.

Reusable SQL utility classes reduce repetitive connection code.

Tools like Navicat can assist with visual database management.