PyMySQL 常用连接方式

适用于 Python 3.11 及以上版本。先安装依赖：python -m pip install "PyMySQL[rsa]" DBUtils。

复制下方代码，保存为 pymysql_client.py。设置 MYSQL_HOST、MYSQL_USER、MYSQL_PASSWORD、MYSQL_DATABASE，再运行 python pymysql_client.py；MYSQL_PORT 默认是 3306。

入口使用正文的 intl_sixin_data 示例表，筛选 session_id = 'session_a'，最多返回 5 条记录的 id、date_num、session_id；没有匹配记录时返回 []。建表语句见正文 1.2。连接自己的表时，修改 main() 中的查询。

在其他脚本中可通过 from pymysql_client import MySQLClient 导入。事务和连接池的调用方式见正文。

pymysql_client.py Python

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
"""PyMySQL 同步客户端，适用于 Python 3.11+。

安装依赖：python -m pip install "PyMySQL[rsa]" DBUtils
运行前设置 MYSQL_HOST、MYSQL_USER、MYSQL_PASSWORD、MYSQL_DATABASE；MYSQL_PORT 默认 3306。
直接运行本文件只查询 intl_sixin_data 中的演示会话 session_a；无匹配记录时返回 []。
"""

import os
from contextlib import contextmanager

import pymysql
from dbutils.pooled_db import PooledDB


class MySQLClient:
    """在进程启动时创建一个实例，所有使用者退出后再关闭连接池。"""

    def __init__(self, host, user, password, database, port=3306, maxconnections=5,
                 connect_timeout=10, read_timeout=30):
        self._pool = PooledDB(
            creator=pymysql, mincached=1, maxcached=maxconnections, maxconnections=maxconnections,
            maxshared=0, blocking=True, ping=1,
            host=host, port=port, user=user, password=password, database=database, charset="utf8mb4",
            cursorclass=pymysql.cursors.DictCursor, autocommit=True,
            connect_timeout=connect_timeout, read_timeout=read_timeout,
        )

    def query(self, sql: str, params=None) -> list[dict]:
        """执行返回结果集的语句；保留日期、Decimal 等原始 Python 类型。"""
        with self._pool.connection(shareable=False) as conn:
            with conn.cursor() as cursor:
                cursor.execute(sql, params)
                return list(cursor.fetchall())

    def execute(self, sql: str, params=None) -> int:
        """在独立事务中执行一条写入语句，返回受影响行数。"""
        with self.transaction() as cursor:
            return cursor.execute(sql, params)

    def executemany(self, sql: str, params) -> int:
        """params 为参数列表；空列表返回 0，非空列表在同一事务中执行。"""
        if not params:
            return 0
        with self.transaction() as cursor:
            return cursor.executemany(sql, params)

    @contextmanager
    def transaction(self):
        """多条语句共用同一游标；正常退出提交，异常退出回滚。"""
        with self._pool.connection(shareable=False) as conn:
            try:
                # 显式 begin 让 DBUtils 暂停自动重连、重试，避免断线后重放写语句。
                conn.begin()
                with conn.cursor() as cursor:
                    yield cursor
                conn.commit()
            except BaseException:
                conn.rollback()
                raise
        # 退出连接上下文时，DBUtils 将连接归还池中，供后续调用复用。

    def close(self):
        """先等待所有工作线程完成，再关闭池中连接；此后不再使用该客户端。"""
        self._pool.close()


def main():
    client = MySQLClient(
        host=os.environ["MYSQL_HOST"], port=int(os.environ.get("MYSQL_PORT", "3306")),
        user=os.environ["MYSQL_USER"], password=os.environ["MYSQL_PASSWORD"],
        database=os.environ["MYSQL_DATABASE"],
    )
    try:
        sql = ("SELECT id, date_num, session_id FROM intl_sixin_data "
               "WHERE session_id = %s ORDER BY date_num DESC, id DESC LIMIT 5")
        print(client.query(sql, ("session_a",)))
    finally:
        client.close()


if __name__ == "__main__":
    main()