数据库操作是 Python 开发的必备核心技能，下面以 MySQL 数据库为例展开讲解。
可以从 MySQL 官网 下载安装，或使用包管理器。
点击学习 MySQL 相关知识。
这里我们直接使用官方提供的 mysql-connnector-python 驱动，因为该驱动包含 C 扩展加速：
C:\Users\hxstr> pip install mysql-connector-python
Defaulting to user installation because normal site-packages is not writeable
Collecting mysql-connector-python
Downloading mysql_connector_python-9.7.0-cp313-cp313-win_amd64.whl.metadata (11 kB)
Downloading mysql_connector_python-9.7.0-cp313-cp313-win_amd64.whl (17.7 MB)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 17.7/17.7 MB 930.5 kB/s 0:00:19
Installing collected packages: mysql-connector-python
Successfully installed mysql-connector-python-9.7.0
[notice] A new release of pip is available: 26.0.1 -> 26.1.2
[notice] To update, run: python.exe -m pip install --upgrade pip注意，除了官方驱动外，还存在一下 mysql 驱动，但是推荐使用 mysql 官方提供的驱动：
# 纯 Python 实现（无需编译，兼容性好）
pip install pymysql
# ORM 框架（大型项目推荐）
pip install sqlalchemy pymysql
# 异步驱动（高并发场景）
pip install aiomysql为了后续更好的操作，我们提前准备如下数据库和数据表，SQL 如下：
-- 创建数据库
CREATE DATABASE python_mysql_demo
CHARACTER SET utf8mb4
COLLATE utf8mb4_unicode_ci;
-- 切换到新创建的数据库
USE python_mysql_demo;
-- 在新数据库中创建数据表
CREATE TABLE students (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(50) NOT NULL,
age INT,
email VARCHAR(100) UNIQUE,
enrolled_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 模拟数据
INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('张三', 18, 'zhangsan@demo.com', '2026-01-10 08:20:15');
INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('李四', 19, 'lisi@demo.com', '2026-01-12 09:10:33');
INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('王五', 17, 'wangwu@demo.com', '2026-01-15 14:05:22');
INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('赵六', 20, 'zhaoliu@demo.com', '2026-02-03 10:44:18');
INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('钱小七', 18, 'qianxiaoqi@demo.com', '2026-02-07 11:22:56');
INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('孙小八', 19, 'sunxiaoba@demo.com', '2026-02-11 15:30:09');
INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('周九', 17, 'zhoujiu@demo.com', '2026-03-01 08:55:41');
INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('吴十', 20, 'wushi@demo.com', '2026-03-05 16:12:37');
INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('郑十一', 18, 'zheng11@demo.com', '2026-03-09 13:28:44');
INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('冯十二', 19, 'feng12@demo.com', '2026-03-14 09:47:10');在 mysql 客户端命令中运行上面的 SQL 语句，如下：
（1）使用 “mysql -h 主机地址 -u root -p” 命令连接到 mysql 服务，如下：
C:\Users\hxstr> mysql -h 192.168.116.70 -u root -p
Enter password: ******
Welcome to the MySQL monitor. Commands end with ; or \g.
Your MySQL connection id is 47
Server version: 8.0.46 MySQL Community Server - GPL
Copyright (c) 2000, 2021, Oracle and/or its affiliates.
Oracle is a registered trademark of Oracle Corporation and/or its
affiliates. Other names may be trademarks of their respective
owners.
Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.
mysql>（2）开始执行创建数据库和数据表的 SQL 语句：
-- 创建数据库
mysql> CREATE DATABASE python_mysql_demo
-> CHARACTER SET utf8mb4
-> COLLATE utf8mb4_unicode_ci;
Query OK, 1 row affected (0.00 sec)
-- 切换到数据库
mysql> USE python_mysql_demo;
Database changed
-- 创建数据表
mysql> CREATE TABLE students (
-> id INT AUTO_INCREMENT PRIMARY KEY,
-> name VARCHAR(50) NOT NULL,
-> age INT,
-> email VARCHAR(100) UNIQUE,
-> enrolled_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-> );
Query OK, 0 rows affected (0.01 sec)
-- 查看数据表结构
mysql> desc students;
+-------------+--------------+------+-----+-------------------+-------------------+
| Field | Type | Null | Key | Default | Extra |
+-------------+--------------+------+-----+-------------------+-------------------+
| id | int | NO | PRI | NULL | auto_increment |
| name | varchar(50) | NO | | NULL | |
| age | int | YES | | NULL | |
| email | varchar(100) | YES | UNI | NULL | |
| enrolled_at | timestamp | YES | | CURRENT_TIMESTAMP | DEFAULT_GENERATED |
+-------------+--------------+------+-----+-------------------+-------------------+
5 rows in set (0.00 sec)
-- 插入数据
mysql> INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('张三', 18, 'zhangsan@demo.com', '2026-01-10 08:20:15');
Query OK, 1 row affected (0.00 sec)
mysql> INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('李四', 19, 'lisi@demo.com', '2026-01-12 09:10:33');
Query OK, 1 row affected (0.00 sec)
mysql> INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('王五', 17, 'wangwu@demo.com', '2026-01-15 14:05:22');
Query OK, 1 row affected (0.00 sec)
mysql> INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('赵六', 20, 'zhaoliu@demo.com', '2026-02-03 10:44:18');
Query OK, 1 row affected (0.00 sec)
mysql> INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('钱小七', 18, 'qianxiaoqi@demo.com', '2026-02-07 11:22:56');
Query OK, 1 row affected (0.00 sec)
mysql> INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('孙小八', 19, 'sunxiaoba@demo.com', '2026-02-11 15:30:09');
Query OK, 1 row affected (0.00 sec)
mysql> INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('周九', 17, 'zhoujiu@demo.com', '2026-03-01 08:55:41');
Query OK, 1 row affected (0.00 sec)
mysql> INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('吴十', 20, 'wushi@demo.com', '2026-03-05 16:12:37');
Query OK, 1 row affected (0.00 sec)
mysql> INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('郑十一', 18, 'zheng11@demo.com', '2026-03-09 13:28:44');
Query OK, 1 row affected (0.00 sec)
mysql> INSERT INTO `students` (`name`, `age`, `email`, `enrolled_at`) VALUES ('冯十二', 19, 'feng12@demo.com', '2026-03-14 09:47:10');
Query OK, 1 row affected (0.00 sec)什么是连接？简单说就是让 python 程序和 mysql 数据库之间铺设一条管道，后续通过该管道实现数据传输，如发送 SQL 命令到 mysql，或者从 mysql 数据库获取结果。
示例：演示使用 python 程序建立到 mysql 的连接，然后打印 mysql 的版本信息。
import mysql.connector
from mysql.connector import Error
def create_connection():
"""建立与 MySQL 数据库的连接"""
try:
connection = mysql.connector.connect(
host='192.168.116.70',
port=3306,
database='python_mysql_demo',
user='root',
password='aaaaaa',
charset='utf8mb4', # 支持 emoji 和中文
autocommit=False, # 手动控制事务
raise_on_warnings=True # 警告转为异常
)
if connection.is_connected():
print("✓ 连接成功！MySQL 版本:", connection.server_info)
return connection
except Error as e:
print("✗ 连接失败:", e)
return None
if __name__ == "__main__":
conn = create_connection()
if conn:
conn.close()
print("连接已关闭")运行示例，输出如下：
✓ 连接成功！MySQL 版本: 8.0.46
连接已关闭上面虽然成功连接到 mysql 中，但在实际开发中，不推荐每次都这样去直接连接数据库、手动释放资源，如果忘记释放了，怎么办？
为了解决资源释放的问题，可以使用上下文管理器自动管理连接生命周期，避免资源泄漏，例如：
from contextlib import contextmanager
import mysql.connector
from mysql.connector import Error
class DatabaseConfig:
"""数据库配置类，集中管理连接参数"""
HOST = '192.168.116.70' # 数据库服务器地址
PORT = 3306 # 数据库端口号
DATABASE = 'python_mysql_demo' # 数据库名
USER = 'root' # 数据库用户名
PASSWORD = 'aaaaaa' # 数据库密码
CHARSET = 'utf8mb4' # 字符集，该字符集可支持emoji表情
class Database:
"""数据库操作封装"""
def __init__(self, config=DatabaseConfig):
self.config = config
self.connection = None
def connect(self):
"""建立连接"""
self.connection = mysql.connector.connect(
host=self.config.HOST,
port=self.config.PORT,
database=self.config.DATABASE,
user=self.config.USER,
password=self.config.PASSWORD,
charset=self.config.CHARSET,
autocommit=False
)
return self.connection
def close(self):
"""关闭连接"""
if self.connection and self.connection.is_connected():
self.connection.close()
def __enter__(self):
"""
进入上下文管理器，建立数据库连接
在 with 语句入口处被调用，建立连接并返回实例本身
"""
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""
退出上下文管理器，关闭数据库连接
"""
self.close()
if __name__ == "__main__":
# 使用上下文管理器, 自动关闭连接
with Database() as db:
cursor = db.connection.cursor()
cursor.execute("SELECT VERSION()")
version = cursor.fetchone()
print("MySQL 版本:", version[0])运行示例，输出如下：
MySQL 版本: 8.0.46注意：推荐将数据库配置存储在环境变量或 .env 文件中，不要硬编码在源码里。生产环境中务必使用加密方式管理密码。
在测试或者编写 Demo 时，使用上面方式建立连接完全没有问题。但在高并发场景下，频繁创建/销毁连接代价高昂。
此时，我们可以使用连接池技术，如果你有其他编程语言的经验，应该对它体会很深。连接池指程序启动 / 连接池实例化时，会根据配置的 minconnections（最小空闲连接数），一次性向 MySQL 建立对应数量的物理 TCP 连接，存入空闲队列。这部分长连接会持续保持，不会随单次查询销毁，相当于提前备好 “通道”。
当我们执行业务 SQL 时，从空闲队列取出一条连接交给代码使用。操作完成（无异常 / 异常）不执行底层 close() 销毁 TCP，而是重置会话状态（事务回滚、清空临时变量），放回空闲队列等待下一次取用。
在并发高峰时，空闲连接用完后，自动新建连接，上限受 maxconnections 限制。
如果某些连接长期闲置，一旦超过闲置超时时间的多余空闲连接会自动释放，保留最小连接数避免频繁重建。
这和原生 connect() 本质区别：原生每次 pymysql.connect() = 新建 TCP → 用完断开 TCP，反复握手、认证，高并发损耗极高；连接池只在扩容 / 初始化时建立 TCP，绝大多数场景只是取出、归还，复用已有链路。
例如：
from mysql.connector.pooling import MySQLConnectionPool
# 创建连接池（全局单例）
pool = MySQLConnectionPool(
pool_name="mypool",
pool_size=10, # 池中连接数
pool_reset_session=True, # 归还时重置会话状态
host='192.168.116.70',
database='python_mysql_demo',
user='root',
password='aaaaaa',
charset='utf8mb4'
)
def execute_with_pool(sql, params=None):
"""从连接池获取连接执行 SQL"""
try:
conn = pool.get_connection() # 从池中借用
cursor = conn.cursor()
cursor.execute(sql, params or ())
result = cursor.fetchall()
conn.commit()
return result
finally:
if conn.is_connected():
cursor.close()
conn.close() # 归还到池中
# 使用示例
rows = execute_with_pool("SELECT * FROM students WHERE age > %s", (18,))
for row in rows:
print(row)运行示例，输出如下：
(22, '李四', 19, 'lisi@demo.com', datetime.datetime(2026, 1, 12, 9, 10, 33))
(24, '赵六', 20, 'zhaoliu@demo.com', datetime.datetime(2026, 2, 3, 10, 44, 18))
(26, '孙小八', 19, 'sunxiaoba@demo.com', datetime.datetime(2026, 2, 11, 15, 30, 9))
(28, '吴十', 20, 'wushi@demo.com', datetime.datetime(2026, 3, 5, 16, 12, 37))
(30, '冯十二', 19, 'feng12@demo.com', datetime.datetime(2026, 3, 14, 9, 47, 10))到这里，我们已经学会了如何连接到数据库，下面将介绍如何使用 Python 对数据库表进行 CRUD 操作。
注意了，对数据库操作主要是将数据存入数据库、修改、读取和删除，下面将逐一介绍：
创建数据即将数据写入到指定的数据库表，将数据持久化到磁盘，下次重启程序数据不会丢失。
示例：插入一条数据和批量插入三条数据到数据库
# ...省略数据库封装操作...
def insert_student(db, name, age, email):
"""插入单条记录 —— 使用参数化查询防止 SQL 注入"""
sql = """
INSERT INTO students (name, age, email)
VALUES (%s, %s, %s)
"""
cursor = db.connection.cursor()
cursor.execute(sql, (name, age, email))
db.connection.commit()
return cursor.lastrowid # 返回自增 ID
def batch_insert_students(db, records):
"""批量插入 —— 使用 executemany 提升性能"""
sql = "INSERT INTO students (name, age, email) VALUES (%s, %s, %s)"
cursor = db.connection.cursor()
cursor.executemany(sql, records) # 批量执行
db.connection.commit()
return cursor.rowcount # 受影响行数
# 使用示例
with Database() as db:
# 单条插入
new_id = insert_student(db, '张三', 21, 'zhangsan@example.com')
print("新记录 ID:", new_id)
# 批量插入
data = [
('李四', 22, 'lisi@example.com'),
('王五', 23, 'wangwu@example.com'),
('赵六', 20, 'zhaoliu@example.com'),
]
count = batch_insert_students(db, data)
print("批量插入了", count, "条记录")运行示例，输出如下：
新记录 ID: 31
批量插入了 3 条记录我们将数据存储到数据库的最终目的是进行各种查询，以满足业务需要。如按照年龄范围查询、按照名称模糊查询等等。
示例：提供根据范围、模糊、分页和聚合查询示例
# ...省略数据库封装操作...
def get_all_students(db):
"""查询所有学生"""
sql = "SELECT id, name, age, email, enrolled_at FROM students"
cursor = db.connection.cursor(dictionary=True) # 返回字典格式
cursor.execute(sql)
return cursor.fetchall()
def get_students_by_age(db, min_age, max_age):
"""按年龄范围查询，参数化查询"""
sql = "SELECT * FROM students WHERE age BETWEEN %s AND %s"
cursor = db.connection.cursor(dictionary=True)
cursor.execute(sql, (min_age, max_age))
return cursor.fetchall()
def search_students(db, keyword):
"""模糊搜索，LIKE 查询"""
sql = "SELECT * FROM students WHERE name LIKE %s"
cursor = db.connection.cursor(dictionary=True)
cursor.execute(sql, ('%' + keyword + '%',))
return cursor.fetchall()
def get_student_count(db):
"""聚合查询，COUNT"""
sql = "SELECT COUNT(*) as total, AVG(age) as avg_age FROM students"
cursor = db.connection.cursor(dictionary=True)
cursor.execute(sql)
return cursor.fetchone()
# 分页查询
def get_students_paginated(db, page=1, page_size=10):
"""分页查询，LIMIT / OFFSET"""
offset = (page - 1) * page_size
sql = "SELECT * FROM students ORDER BY id LIMIT %s OFFSET %s"
cursor = db.connection.cursor(dictionary=True)
cursor.execute(sql, (page_size, offset))
return cursor.fetchall()
if __name__ == '__main__':
with Database() as db:
# 查询所有学生
students = get_all_students(db)
print("所有学生：")
for student in students:
print(student)
# 按年龄范围查询
students_in_age_range = get_students_by_age(db, 19, 20)
print("\n年龄在19-20之间的学生：")
for student in students_in_age_range:
print(student)
# 模糊搜索
search_result = search_students(db, '张')
print("\n查询'张'的结果：")
for student in search_result:
print(student)
# 聚合查询
student_count = get_student_count(db)
print("\n学生总数和平均年龄：")
print(student_count)
# 分页查询
paginated_students = get_students_paginated(db, page=1, page_size=3)
print("\n分页学生数据（第1页，每页3条）：")
for student in paginated_students:
print(student)运行示例，输出如下：
所有学生：
{'id': 21, 'name': '张三', 'age': 18, 'email': 'zhangsan@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 10, 8, 20, 15)}
{'id': 22, 'name': '李四', 'age': 19, 'email': 'lisi@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 12, 9, 10, 33)}
{'id': 23, 'name': '王五', 'age': 17, 'email': 'wangwu@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 15, 14, 5, 22)}
{'id': 24, 'name': '赵六', 'age': 20, 'email': 'zhaoliu@demo.com', 'enrolled_at': datetime.datetime(2026, 2, 3, 10, 44, 18)}
{'id': 25, 'name': '钱小七', 'age': 18, 'email': 'qianxiaoqi@demo.com', 'enrolled_at': datetime.datetime(2026, 2, 7, 11, 22, 56)}
{'id': 26, 'name': '孙小八', 'age': 19, 'email': 'sunxiaoba@demo.com', 'enrolled_at': datetime.datetime(2026, 2, 11, 15, 30, 9)}
{'id': 27, 'name': '周九', 'age': 17, 'email': 'zhoujiu@demo.com', 'enrolled_at': datetime.datetime(2026, 3, 1, 8, 55, 41)}
{'id': 28, 'name': '吴十', 'age': 20, 'email': 'wushi@demo.com', 'enrolled_at': datetime.datetime(2026, 3, 5, 16, 12, 37)}
{'id': 29, 'name': '郑十一', 'age': 18, 'email': 'zheng11@demo.com', 'enrolled_at': datetime.datetime(2026, 3, 9, 13, 28, 44)}
{'id': 30, 'name': '冯十二', 'age': 19, 'email': 'feng12@demo.com', 'enrolled_at': datetime.datetime(2026, 3, 14, 9, 47, 10)}
{'id': 31, 'name': '张三', 'age': 21, 'email': 'zhangsan@example.com', 'enrolled_at': datetime.datetime(2026, 7, 12, 13, 50, 27)}
{'id': 32, 'name': '李四', 'age': 22, 'email': 'lisi@example.com', 'enrolled_at': datetime.datetime(2026, 7, 12, 13, 50, 27)}
{'id': 33, 'name': '王五', 'age': 23, 'email': 'wangwu@example.com', 'enrolled_at': datetime.datetime(2026, 7, 12, 13, 50, 27)}
{'id': 34, 'name': '赵六', 'age': 20, 'email': 'zhaoliu@example.com', 'enrolled_at': datetime.datetime(2026, 7, 12, 13, 50, 27)}
年龄在19-20之间的学生：
{'id': 22, 'name': '李四', 'age': 19, 'email': 'lisi@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 12, 9, 10, 33)}
{'id': 24, 'name': '赵六', 'age': 20, 'email': 'zhaoliu@demo.com', 'enrolled_at': datetime.datetime(2026, 2, 3, 10, 44, 18)}
{'id': 26, 'name': '孙小八', 'age': 19, 'email': 'sunxiaoba@demo.com', 'enrolled_at': datetime.datetime(2026, 2, 11, 15, 30, 9)}
{'id': 28, 'name': '吴十', 'age': 20, 'email': 'wushi@demo.com', 'enrolled_at': datetime.datetime(2026, 3, 5, 16, 12, 37)}
{'id': 30, 'name': '冯十二', 'age': 19, 'email': 'feng12@demo.com', 'enrolled_at': datetime.datetime(2026, 3, 14, 9, 47, 10)}
{'id': 34, 'name': '赵六', 'age': 20, 'email': 'zhaoliu@example.com', 'enrolled_at': datetime.datetime(2026, 7, 12, 13, 50, 27)}
查询'张'的结果：
{'id': 21, 'name': '张三', 'age': 18, 'email': 'zhangsan@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 10, 8, 20, 15)}
{'id': 31, 'name': '张三', 'age': 21, 'email': 'zhangsan@example.com', 'enrolled_at': datetime.datetime(2026, 7, 12, 13, 50, 27)}
学生总数和平均年龄：
{'total': 14, 'avg_age': Decimal('19.3571')}
分页学生数据（第1页，每页3条）：
{'id': 21, 'name': '张三', 'age': 18, 'email': 'zhangsan@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 10, 8, 20, 15)}
{'id': 22, 'name': '李四', 'age': 19, 'email': 'lisi@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 12, 9, 10, 33)}
{'id': 23, 'name': '王五', 'age': 17, 'email': 'wangwu@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 15, 14, 5, 22)}查询技巧： 使用 dictionary=True 让结果以字典形式返回，代码可读性更好。使用 fetchone() 取单行、fetchmany(n) 分批取、fetchall() 取全部。
当我们业务数据发生变化时，我们需要通过更新操作来更新数据库中已经保存的数据。
例如：更新学生的邮箱地址、更新年龄
# ...省略数据库封装操作...
def update_student_email(db, student_id, new_email):
"""更新学生邮箱"""
sql = "UPDATE students SET email = %s WHERE id = %s"
cursor = db.connection.cursor()
cursor.execute(sql, (new_email, student_id))
db.connection.commit()
return cursor.rowcount # 返回受影响行数
def increment_age(db, student_id):
"""原子递增年龄，避免并发问题"""
sql = "UPDATE students SET age = age + 1 WHERE id = %s"
cursor = db.connection.cursor()
cursor.execute(sql, (student_id,))
db.connection.commit()
return cursor.rowcount
if __name__ == "__main__":
# 使用上下文管理器进行数据库操作
with Database() as db:
# 更新学生邮箱
rows_updated = update_student_email(db, student_id=23, new_email="newemail@example.com")
print(f"更新了 {rows_updated} 行学生的邮箱")
# 原子递增年龄
rows_incremented = increment_age(db, student_id=23)
print(f"增加了 {rows_incremented} 行学生的年龄")运行示例，输出如下：
更新了 1 行学生的邮箱
增加了 1 行学生的年龄当业务数据不在需要时，我们可以通过删除操作将数据从数据库表中删除。
示例：根据学生 ID 删除和根据年龄批量删除
def delete_student(db, student_id):
"""删除指定学生"""
sql = "DELETE FROM students WHERE id = %s"
cursor = db.connection.cursor()
cursor.execute(sql, (student_id,))
db.connection.commit()
return cursor.rowcount
def delete_students_below_age(db, age):
"""批量删除 —— 条件删除"""
sql = "DELETE FROM students WHERE age < %s"
cursor = db.connection.cursor()
cursor.execute(sql, (age,))
db.connection.commit()
return cursor.rowcount运行示例，输出如下：
删除ID为 23 的学生 1 个
删除了 1 个年龄小于 18 的学生事务是一组不可分割的 SQL 操作单元。
执行规则如下：
单元内所有 SQL全部执行成功 → 提交 commit，数据永久生效；
任意一条 SQL 失败 / 程序异常 / 手动回滚 → 全部撤销 rollback，数据库回到执行前原始状态；
简单来说，就是同生共死，不存在中间半完成状态。
如银行转账，A 扣钱、B 加钱
若 A 扣款成功，B 加钱时报错，不加事务会导致资金丢失；
开启事务后，两条 SQL 捆绑，出错则全部撤销，资金平衡不被破坏；
事务保证一组操作要么全部成功，要么全部回滚（即 ACID 原则）：
原子性 Atomicity（事务的最小不可分割单位）：要么全成，要么全废，没有中间状态
一致性 Consistency（数据合法，业务约束不被破坏）：事务执行前后，数据库完整性约束始终有效
隔离性 Isolation（多事务并发互不干扰）：多个事务并发执行时，彼此操作互相隔离，看不到对方未提交的脏数据。
持久性 Durability（提交后永久生效）：事务一旦执行 commit 提交成功，修改永久保存在磁盘，数据库宕机、断电也不会丢失
示例：
# ...省略代码...
def get_students_by_id(db, student_id):
"""根据ID查询学生信息"""
sql = "SELECT * FROM students WHERE id = %s"
cursor = db.connection.cursor(dictionary=True)
cursor.execute(sql, (student_id,))
return cursor.fetchone()
def transfer_age(db, from_id, to_id, age):
try:
cursor = db.connection.cursor()
# 操作 1: 扣减年龄
cursor.execute(
"UPDATE students SET age = age - %s WHERE id = %s AND age >= %s",
(age, from_id, age)
)
if cursor.rowcount == 0:
raise ValueError("年龄不足或用户不存在")
# 操作 2: 增加年龄
cursor.execute(
"UPDATE students SET age = age + %s WHERE id = %s",
(age, to_id)
)
if cursor.rowcount == 0:
raise ValueError("目标用户不存在")
# 提交事务 —— 所有操作生效
db.connection.commit()
print("年龄扣减成功")
except Exception as e:
# 回滚事务 —— 撤销所有操作
db.connection.rollback()
print("年龄扣减失败，已回滚:", e)
# raise # 抛出异常，让上层调用者知道事务失败
if __name__ == "__main__":
# 通过年龄字段来模拟数据
with Database() as db:
try:
# 扣减前
student1 = get_students_by_id(db, 21)
print("扣减前学生1信息:\n", student1)
student2 = get_students_by_id(db, 22)
print("扣减前学生2信息:\n", student2)
# 扣减
transfer_age(db, from_id=21, to_id=22, age=10)
student1 = get_students_by_id(db, 21)
print("扣减后学生1信息:\n", student1)
student2 = get_students_by_id(db, 22)
print("扣减后学生2信息:\n", student2)
except Exception as e:
print("操作失败:", e)运行示例，输出如下：
扣减前学生1信息:
{'id': 21, 'name': '张三', 'age': 18, 'email': 'zhangsan@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 10, 8, 20, 15)}
扣减前学生2信息:
{'id': 22, 'name': '李四', 'age': 19, 'email': 'lisi@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 12, 9, 10, 33)}
年龄扣减成功
扣减后学生1信息:
{'id': 21, 'name': '张三', 'age': 8, 'email': 'zhangsan@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 10, 8, 20, 15)}
扣减后学生2信息:
{'id': 22, 'name': '李四', 'age': 29, 'email': 'lisi@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 12, 9, 10, 33)}注意查看 age 字段的信息。如果在执行第二条语句错误（故意模拟错误，误将 = 号写成 ==）：
cursor.execute(
"UPDATE students SET age = age + %s WHERE id == %s",
(age, to_id)
)输出如下：
扣减前学生1信息:
{'id': 21, 'name': '张三', 'age': 8, 'email': 'zhangsan@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 10, 8, 20, 15)}
扣减前学生2信息:
{'id': 22, 'name': '李四', 'age': 29, 'email': 'lisi@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 12, 9, 10, 33)}
年龄扣减失败，已回滚: 年龄不足或用户不存在
扣减后学生1信息:
{'id': 21, 'name': '张三', 'age': 8, 'email': 'zhangsan@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 10, 8, 20, 15)}
扣减后学生2信息:
{'id': 22, 'name': '李四', 'age': 29, 'email': 'lisi@demo.com', 'enrolled_at': datetime.datetime(2026, 1, 12, 9, 10, 33)}记住了，永远不要用字符串拼接构建 SQL，必须使用参数化查询。因为使用字符串拼接非常容易出现 SQL 注入漏洞，例如：
# 非常危险！永远不要这样做
# 用户输入的信息，可能是前端输入的查询条件，用户恶意输入 SQL 语句
user_input = "'; DROP TABLE students; --"
sql = "SELECT * FROM students WHERE name = '" + user_input + "'"
# 拼接结果:
# SELECT * FROM students WHERE name = ''; DROP TABLE students; --'
# 上面 SQL 会直接将表的数据清空，非常危险
# 推荐做法使用参数化查询，安全
sql = "SELECT * FROM students WHERE name = %s"
cursor.execute(sql, (user_input,))
# 驱动自动转义，输入被当作普通字符串处理安全警示： SQL 注入是 OWASP Top 10 安全风险之一。参数化查询是防御 SQL 注入的最基本也是最重要的手段。同时建议使用最小权限原则配置数据库用户。如用户仅仅能查询数据，那么他的数据库权限就只能查询，即使出现 SQL 注入，最多导致数据暴露，不会导致数据被删。
到这里，python 数据库编程就介绍完了，更多内容请阅读后续章节。
川公网安备51010802032098