python-pymysql

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

PyMySQL Best Practices

PyMySQL 最佳实践

PyMySQL is a pure-Python MySQL client library implementing the DB-API 2.0 specification (PEP 249). It supports MySQL >= 5.7 and MariaDB >= 10.3 on Python >= 3.7.

PyMySQL是一个纯Python编写的MySQL客户端库，实现了DB-API 2.0规范（PEP 249）。它支持Python >= 3.7环境下的MySQL >= 5.7和MariaDB >= 10.3。

Installation

安装

bash

pip install PyMySQL

bash

pip install PyMySQL

For SHA-256 / caching_sha2_password authentication:

如需SHA-256 / caching_sha2_password认证：

pip install "PyMySQL[rsa]"

undefined

pip install "PyMySQL[rsa]"

undefined

Establishing a Connection

建立连接

Use

pymysql.connect()

as the entry point. Always specify

charset='utf8mb4'

and use context managers to guarantee cleanup.

python

import pymysql
import pymysql.cursors

connection = pymysql.connect(
    host="localhost",
    user="app_user",
    password="secret",
    database="mydb",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor,
    connect_timeout=10,
    read_timeout=30,
    write_timeout=30,
    autocommit=False,  # explicit transaction control (recommended)
)

使用

pymysql.connect()

作为入口点。请始终指定

charset='utf8mb4'

，并使用上下文管理器确保资源清理。

python

import pymysql
import pymysql.cursors

connection = pymysql.connect(
    host="localhost",
    user="app_user",
    password="secret",
    database="mydb",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor,
    connect_timeout=10,
    read_timeout=30,
    write_timeout=30,
    autocommit=False,  # 显式事务控制（推荐）
)

Key Connection Parameters

关键连接参数

Parameter	Default	Notes
`charset`	`''`	Always set `utf8mb4` for full Unicode support
`cursorclass`	`Cursor`	Use `DictCursor` for dict rows; `SSCursor` for large result sets
`autocommit`	`False`	Keep `False` ; commit/rollback explicitly
`connect_timeout`	`10`	Seconds before connection attempt fails
`read_timeout`	`None`	Set to prevent hung reads
`write_timeout`	`None`	Set to prevent hung writes
`ssl_ca`	`None`	Path to CA cert for TLS connections

参数	默认值	说明
`charset`	`''`	始终设置为 `utf8mb4` 以获得完整Unicode支持
`cursorclass`	`Cursor`	使用 `DictCursor` 返回字典格式行；使用 `SSCursor` 处理大型结果集
`autocommit`	`False`	保持为 `False` ；显式执行提交/回滚操作
`connect_timeout`	`10`	连接尝试失败前的等待秒数
`read_timeout`	`None`	设置该值以防止读取操作挂起
`write_timeout`	`None`	设置该值以防止写入操作挂起
`ssl_ca`	`None`	TLS连接使用的CA证书路径

Context Manager Usage

上下文管理器用法

Connections and cursors implement the context manager protocol. Use

with

blocks to ensure proper resource cleanup.

python

undefined

连接和游标实现了上下文管理器协议。使用

with

语句块确保资源被正确清理。

python

undefined

Connection as context manager — commits on success, rolls back on exception

连接作为上下文管理器——成功时提交，异常时回滚

with pymysql.connect(**db_config) as connection: with connection.cursor() as cursor: cursor.execute("SELECT id, email FROM users WHERE active = %s", (1,)) rows = cursor.fetchall()


> **Note**: Using `with connection:` handles transaction commit/rollback but does **not** close the connection. Call `connection.close()` explicitly or manage it via a pool.

with pymysql.connect(**db_config) as connection: with connection.cursor() as cursor: cursor.execute("SELECT id, email FROM users WHERE active = %s", (1,)) rows = cursor.fetchall()


> **注意**：使用`with connection:`会处理事务的提交/回滚，但**不会**关闭连接。请显式调用`connection.close()`或通过连接池管理连接。

Parameterized Queries (SQL Injection Prevention)

参数化查询（防止SQL注入）

Always pass values as the second argument to

execute()

. Never use string formatting or concatenation to build SQL.

python

undefined

请始终将值作为

execute()

的第二个参数传入。切勿使用字符串格式化或拼接来构建SQL语句。

python

undefined

Correct — parameterized

正确方式——参数化查询

cursor.execute( "INSERT INTO orders (user_id, amount) VALUES (%s, %s)", (user_id, amount), )

Named placeholders with dict

使用字典的命名占位符

cursor.execute( "SELECT * FROM products WHERE category = %(cat)s AND price < %(max_price)s", {"cat": "electronics", "max_price": 500}, )

WRONG — never do this

错误方式——切勿这样做

cursor.execute(f"SELECT * FROM users WHERE name = '{name}'") # SQL injection risk


Use `cursor.mogrify(query, args)` to preview the interpolated query string during debugging.

cursor.execute(f"SELECT * FROM users WHERE name = '{name}'") # 存在SQL注入风险


调试期间，可使用`cursor.mogrify(query, args)`预览插值后的查询字符串。

Cursor Types

游标类型

Cursor Class	Returns	Buffered	Use Case
`Cursor`	tuple	Yes	Default; small-to-medium result sets
`DictCursor`	dict	Yes	When column-name access is needed
`SSCursor`	tuple	No	Large result sets; memory-constrained
`SSDictCursor`	dict	No	Large result sets with dict access

Pass

cursorclass

at connection time (global default) or per-cursor:

python

undefined

游标类	返回格式	是否缓冲	使用场景
`Cursor`	元组	是	默认类型；适用于中小型结果集
`DictCursor`	字典	是	需要按列名访问数据时使用
`SSCursor`	元组	否	大型结果集；内存受限场景
`SSDictCursor`	字典	否	需要字典格式的大型结果集场景

可在连接时指定

cursorclass

（全局默认），或为每个游标单独设置：

python

undefined

Per-cursor override

为单个游标覆盖全局设置

with connection.cursor(pymysql.cursors.DictCursor) as cursor: cursor.execute("SELECT * FROM users") rows = cursor.fetchall() # list of dicts


For large tables, prefer `SSCursor` and iterate without calling `fetchall()`:

```python
with connection.cursor(pymysql.cursors.SSCursor) as cursor:
    cursor.execute("SELECT * FROM large_table")
    for row in cursor:  # streams row-by-row
        process(row)

with connection.cursor(pymysql.cursors.DictCursor) as cursor: cursor.execute("SELECT * FROM users") rows = cursor.fetchall() # 字典列表


处理大表时，优先使用`SSCursor`并逐行迭代，避免调用`fetchall()`：

```python
with connection.cursor(pymysql.cursors.SSCursor) as cursor:
    cursor.execute("SELECT * FROM large_table")
    for row in cursor:  # 逐行流式处理
        process(row)

Transaction Management

事务管理

PyMySQL defaults to

autocommit=False

. Explicitly commit successful work and roll back on errors.

python

try:
    with connection.cursor() as cursor:
        cursor.execute(
            "UPDATE accounts SET balance = balance - %s WHERE id = %s",
            (amount, from_id),
        )
        cursor.execute(
            "UPDATE accounts SET balance = balance + %s WHERE id = %s",
            (amount, to_id),
        )
    connection.commit()
except Exception:
    connection.rollback()
    raise

Call

connection.begin()

to start a transaction block explicitly when needed.

PyMySQL默认

autocommit=False

。请为成功的操作显式提交，出错时回滚。

python

try:
    with connection.cursor() as cursor:
        cursor.execute(
            "UPDATE accounts SET balance = balance - %s WHERE id = %s",
            (amount, from_id),
        )
        cursor.execute(
            "UPDATE accounts SET balance = balance + %s WHERE id = %s",
            (amount, to_id),
        )
    connection.commit()
except Exception:
    connection.rollback()
    raise

必要时，可调用

connection.begin()

显式启动事务块。

Bulk Inserts with

executemany()

使用

executemany()

批量插入

Use

executemany()

for inserting multiple rows. PyMySQL batches the statements up to

Cursor.max_stmt_length

(1 MB) for efficiency.

python

records = [
    ("alice@example.com", "hash1"),
    ("bob@example.com", "hash2"),
]

with connection.cursor() as cursor:
    cursor.execute("TRUNCATE TABLE staging_users")
    cursor.executemany(
        "INSERT INTO users (email, password_hash) VALUES (%s, %s)",
        records,
    )
connection.commit()

使用

executemany()

插入多行数据。PyMySQL会将语句分批处理，最大批次大小为

Cursor.max_stmt_length

（1 MB）以提升效率。

python

records = [
    ("alice@example.com", "hash1"),
    ("bob@example.com", "hash2"),
]

with connection.cursor() as cursor:
    cursor.execute("TRUNCATE TABLE staging_users")
    cursor.executemany(
        "INSERT INTO users (email, password_hash) VALUES (%s, %s)",
        records,
    )
connection.commit()

Fetching Rows

行数据获取

Method	Returns	Notes
`fetchone()`	single row or `None`	Efficient for single-row lookups
`fetchmany(size)`	list of rows	Page through results
`fetchall()`	list of all rows	Avoid on large result sets

python

cursor.execute("SELECT id, name FROM users WHERE id = %s", (user_id,))
row = cursor.fetchone()
if row is None:
    raise ValueError(f"User {user_id} not found")

方法	返回值	说明
`fetchone()`	单行数据或 `None`	高效查找单行数据
`fetchmany(size)`	行数据列表	分页获取结果
`fetchall()`	所有行数据列表	大型结果集请避免使用

python

cursor.execute("SELECT id, name FROM users WHERE id = %s", (user_id,))
row = cursor.fetchone()
if row is None:
    raise ValueError(f"未找到用户 {user_id}")

Error Handling

错误处理

Catch PyMySQL exceptions from

pymysql.err

python

import pymysql.err

try:
    with connection.cursor() as cursor:
        cursor.execute(sql, args)
    connection.commit()
except pymysql.err.IntegrityError as exc:
    connection.rollback()
    # duplicate key, foreign key violation, etc.
    raise
except pymysql.err.OperationalError as exc:
    # connection dropped, timeout, server gone away
    raise
except pymysql.err.ProgrammingError as exc:
    # bad SQL syntax, wrong number of params
    raise

Key exception classes:

```
pymysql.err.IntegrityError
```
— constraint violations (duplicate key, FK)
```
pymysql.err.OperationalError
```
— connection / server errors
```
pymysql.err.ProgrammingError
```
— SQL syntax errors, wrong param count
```
pymysql.err.DataError
```
— invalid data for column type
```
pymysql.err.DatabaseError
```
— base class for all DB errors

捕获

pymysql.err

中的PyMySQL异常：

python

import pymysql.err

try:
    with connection.cursor() as cursor:
        cursor.execute(sql, args)
    connection.commit()
except pymysql.err.IntegrityError as exc:
    connection.rollback()
    # 重复键、外键约束 violation 等
    raise
except pymysql.err.OperationalError as exc:
    # 连接断开、超时、服务器不可用
    raise
except pymysql.err.ProgrammingError as exc:
    # SQL语法错误、参数数量错误
    raise

关键异常类：

```
pymysql.err.IntegrityError
```
— 约束 violation（重复键、外键）
```
pymysql.err.OperationalError
```
— 连接/服务器错误
```
pymysql.err.ProgrammingError
```
— SQL语法错误、参数数量错误
```
pymysql.err.DataError
```
— 列类型数据无效
```
pymysql.err.DatabaseError
```
— 所有数据库错误的基类

Connection Health Check

连接健康检查

Use

connection.ping(reconnect=True)

to check liveness before executing queries in long-lived connections (e.g., background workers):

python

connection.ping(reconnect=True)
with connection.cursor() as cursor:
    cursor.execute(query)

在长连接场景（如后台工作进程）中，执行查询前可使用

connection.ping(reconnect=True)

检查连接活性：

python

connection.ping(reconnect=True)
with connection.cursor() as cursor:
    cursor.execute(query)

Reading Config from

my.cnf

从

my.cnf

读取配置

Avoid hardcoding credentials. Use

read_default_file

to read from a MySQL option file:

python

connection = pymysql.connect(
    read_default_file="~/.my.cnf",
    read_default_group="client",
    database="mydb",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor,
)

~/.my.cnf

example:

ini

[client]
host     = db.internal
user     = app_user
password = secret

避免硬编码凭据。使用

read_default_file

从MySQL配置文件中读取：

python

connection = pymysql.connect(
    read_default_file="~/.my.cnf",
    read_default_group="client",
    database="mydb",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor,
)

~/.my.cnf

示例：

ini

[client]
host     = db.internal
user     = app_user
password = secret

Quick Reference

快速参考

python

import pymysql
import pymysql.cursors

python

import pymysql
import pymysql.cursors

Connect

连接

conn = pymysql.connect( host="localhost", user="u", password="p", database="db", charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, )

Query

查询

with conn.cursor() as cur: cur.execute("SELECT * FROM t WHERE id = %s", (1,)) row = cur.fetchone()

Write + commit

写入并提交

with conn.cursor() as cur: cur.execute("INSERT INTO t (col) VALUES (%s)", ("val",)) conn.commit()

Bulk insert

批量插入

with conn.cursor() as cur: cur.executemany("INSERT INTO t (a, b) VALUES (%s, %s)", rows) conn.commit()

conn.close()

undefined

with conn.cursor() as cur: cur.executemany("INSERT INTO t (a, b) VALUES (%s, %s)", rows) conn.commit()

conn.close()

undefined

Additional Resources

额外资源

For detailed patterns and advanced usage, consult:

references/connection-patterns.md
— Connection pooling, SSL/TLS configuration, timeout tuning,
```
my.cnf
```
patterns
references/cursor-guide.md
— Cursor type selection, streaming large result sets, stored procedures,
```
mogrify
```
debugging

如需了解详细模式和高级用法，请参考：

references/connection-patterns.md
— 连接池、SSL/TLS配置、超时调优、
```
my.cnf
```
模式
references/cursor-guide.md
— 游标类型选择、大型结果集流式处理、存储过程、
```
mogrify
```
调试

python-pymysql

Original

Translation

PyMySQL Best Practices

PyMySQL 最佳实践

Installation

安装

For SHA-256 / caching_sha2_password authentication:

如需SHA-256 / caching_sha2_password认证：

Establishing a Connection

建立连接

Key Connection Parameters

关键连接参数

Context Manager Usage

上下文管理器用法

Connection as context manager — commits on success, rolls back on exception

连接作为上下文管理器——成功时提交，异常时回滚

Parameterized Queries (SQL Injection Prevention)

参数化查询（防止SQL注入）

Correct — parameterized

正确方式——参数化查询

Named placeholders with dict

使用字典的命名占位符

WRONG — never do this

错误方式——切勿这样做

Cursor Types

游标类型

Per-cursor override

为单个游标覆盖全局设置

Transaction Management

事务管理

Bulk Inserts with executemany()

使用executemany()批量插入

Fetching Rows

行数据获取

Error Handling

错误处理

Connection Health Check

连接健康检查

Reading Config from my.cnf

从my.cnf读取配置

Quick Reference

快速参考

Connect

连接

Query

查询

Write + commit

写入并提交

Bulk insert

批量插入

Additional Resources

额外资源

Bulk Inserts with
`executemany()`

使用
`executemany()`
批量插入

Reading Config from
`my.cnf`

从
`my.cnf`
读取配置