Python 操作MySQL数据库

作用：预先创建并管理多个数据库连接，避免频繁创建/关闭连接的开销。优势：提升性能、控制并发连接数、防止资源耗尽。READ UNCOMMITTED：允许脏读，最低隔离级别，性能最好，但容易出现数据不一致的情况。READ COMMITTED：解决了脏读问题，但可能出现不可重复读。REPEATABLE READ：解决了脏读和不可重复读问题，但可能出现幻读。（MySQL事务默认级别）SERIALIZABL

伤不起bb

1320人浏览 · 2025-05-24 12:07:21

伤不起bb · 2025-05-24 12:07:21 发布

Python连接MySQL

1. 概述

2. 核心作用

3. 核心流程（以 PyMySQL 为例）

4. 重点总结

5. 典型应用场景

一、安装 Python MySQL 连接库

1. 安装mysql-connector-python

2. 安装PyMySQL

二、 Python 连接MySQL 数据库

（1）READ UNCOMMITTED (未提交读)

（2）READ COMMITTED（提交读）

（3）REPEATABLE READ（可重复读）

Python连接MySQL

1. 概述

Python 通过数据库驱动或 ORM 工具与 MySQL 交互，支持数据存储、查询和管理。常用库包括：

原生驱动：mysql-connector-python（官方）、PyMySQL（纯Python）。
ORM 框架：SQLAlchemy（通用）、Django ORM（Django 专属）。

2. 核心作用

数据持久化：存储用户信息、日志、交易记录等。
动态查询：支持条件检索、聚合统计、多表关联等操作。
事务管理：确保数据操作的原子性（如转账操作）。
应用开发：为 Web、数据分析、自动化工具提供数据支持。

3. 核心流程（以 PyMySQL 为例）

连接数据库：配置主机、用户、密码、数据库名。
创建游标：通过游标执行 SQL 命令。
执行 SQL：查询（SELECT）、增删改（INSERT/UPDATE/DELETE）。
处理结果：获取数据或提交事务。
关闭连接：释放资源，避免泄漏。

4. 重点总结

关键点	说明
驱动选择	根据需求选库：快速开发用 ORM，高性能用原生驱动。
参数化查询	使用 `%s` 占位符，避免 SQL 注入攻击。
事务管理	写操作后需 `commit()`，异常时 `rollback()` 保证数据一致性。
资源释放	使用 `with` 语句或手动关闭连接，防止资源泄漏。
ORM 优势	用类操作表（如 `User.query.filter_by(name='Alice')`），简化 SQL 编写。
连接池优化	高并发场景使用连接池（如 `DBUtils`），提升性能。

5. 典型应用场景

Web 后端：Django/Flask 集成 ORM 处理用户数据。
数据分析：pandas 从 MySQL 读取数据进行分析。
自动化脚本：定时备份、数据迁移（如 mysqldump + Python 处理）。

一、安装 Python MySQL 连接库

1. 安装mysql-connector-python

pip3 install mysql-connector-python

2. 安装PyMySQL

pip3 install pymysql

二、 Python 连接MySQL 数据库

1. 导入连接库

创建文件 aaa.py（导入模块）
import pymysql

2. 创建数据库连接

#连接名字+值()
db=pymysql.connect(
    host="192.168.10.101",    #MySQL数据库的地址，通常是localhsot或IP地址
    user="root",        #数据库的用户名，通常是root
    password="pwd123",    #数据库的密码
    database="testdb"    #要连接的数据库名称
)	

##要注意远程必须可以连接到MySQL服务器的，在服务器上创建用于远程用户并授权
例如:create user 'root'@'%' identified by 'pwd123';
    grant all on *.* to 'root'@'%';
    flush privileges;

3. 创建游标对象

#建立连接（游标：cursor，动态变化位置）
cursor=db.cursor()			#针对上面的值 建立游标用于连接

4. 执行SQL语句

#执行SQL语句,可以使用%s占位符来避免SQL注入攻击
cursor.execute("select * from users")

5. 获取查询结果

#定义执行结果为它赋值
results=cursor.fetchall()
for row in results:
	print(row)

6. 关闭连接

#关闭连接（执行完成后要关闭游标和数据库连接）
cursor.close()
db.close()

三、常见的MySQL操作

1. 插入数据（INSERT）

cursor.execute("insert into users (name,age) values(%s,%s)",("zhangsan",20))
db.commit()    --提交事务

2. 更新数据（UPDATE）

cursor.execute("UPDATE users SET age = %s WHERE name = %s , (26,"ALice")")
db.commit()  #提交

3.删除数据（DELETE）

cursor.execute("DELETE FROM users WHERE name = %s",("Alice",))
db.commit()      #提交事务

4. 查询数据（SELECT）

cursor.execute("SELECT * FROM users")
results = cursor.fetchall()
for row in results:
    print(row)

5. 执行多条SQL语句

cursor.executemany(
    "INSERT INTO users (name,age) VALUES (%s,%s)",
    [("Bob",30),("Charlie",35),("David",28)]
)
db.commit()

6. 使用LIKE进行模糊查询

作用：根据模式匹配检索数据，支持通配符 %（任意字符）和 _（单个字符）。

cursor.execute("select * from users where name like %s",("zhang%"))
results = cursor.fetchall()
for row in results:
    print(row)

7. 使用JOIN 进行联合查询

作用：联合多表数据，常用类型：

INNER JOIN：返回两表匹配的行。
LEFT JOIN：返回左表所有行，右表无匹配则为 NULL。

cursor.execute("""
    SELECT users.name, orders.amount
    FROM users
    INNER JOIN orders ON users.id = orders.user_id
""")
results = cursor.fetchall()
for row in results:
    print(row)

代码解析：

SQL 查询：使用 INNER JOIN 连接 users 和 orders 表，通过 user_id 关联，查询用户姓名（users.name）和订单金额（orders.amount）。
Python 操作：
- cursor.execute() 执行 SQL 语句。
- fetchall() 获取所有查询结果。
- 遍历结果并打印每行数据（用户姓名与对应订单金额的组合）。

四、使用连接池

1. 连接池简介

作用：预先创建并管理多个数据库连接，避免频繁创建/关闭连接的开销。
优势：提升性能、控制并发连接数、防止资源耗尽。

2. 创建连接池

先安装库用来建立来连接池

pip install dbutils

from dbutils.pooled_db import PooledDB
import pymysql

# 数据库连接配置
dbconfig = {
    "host": "localhost",
    "user": "root",
    "password": "password",
    "database": "testdb"
}

# 创建连接池
connection_pool = PooledDB(
    creator=pymysql,  # 使用PyMySQL作为数据库连接库
    maxconnections=5,  # 连接池中最大连接数
    **dbconfig
)

3. 获取连接

从连接池中获取连接时，可以使用connection()方法。每次获取到的连接都可以直接执行数据库操作。

db_connection = connection_pool.connection()
cursor = db_connection.cursor()

cursor.execute("SELECT * FROM users")
results = cursor.fetchall()

for row in results:
    print(row)

cursor.close()
db_connection.close()  # 连接会自动归还给连接池

4. 连接池的优势

资源复用：避免重复创建连接，减少 TCP 握手和认证开销。
流量控制：限制最大连接数，防止数据库过载。
健康检查：自动检测失效连接并重建。

五、事务管理

1. 开始事务

cursor.execute("START TRANSACTION")

2. 提交事务

db.commit()

3. 回滚事务

db.rollback()

4. 事务的隔离级别

（1）READ UNCOMMITTED (未提交读)

描述：事务可读取其他事务未提交的数据，可能导致 脏读（Dirty Read），事务间隔离性最差。
应用场景：通常不推荐使用，除非对数据一致性要求极低。

使用方法：

cursor.execute("SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED")

（2）READ COMMITTED（提交读）

描述：事务仅读取其他事务已提交的数据，避免脏读，但可能出现 不可重复读（Non-repeatable Read）（同一事务内多次读取同一数据，值因其他事务提交而变化）。
应用场景：适用于多数常见场景，平衡隔离性与性能（如普通业务查询）。

使用方法：

cursor.execute("SET TRANSACTION ISOLATION LEVEL READ COMMITTED")

（3）REPEATABLE READ（可重复读）

- 描述：事务内多次读取同一数据，值保持不变（避免不可重复读），但可能出现 幻读（Phantom Read）（同一事务内查询结果集因其他事务插入 / 删除而变化）。
- 应用场景：需保证事务内数据一致性（如金融系统余额操作、订单状态变更）。
- 使用方法：
```
cursor.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
```

（4）SERIAIZABLE （串行化）

描述：最严格隔离级别，事务串行执行（完全避免脏读、不可重复读、幻读），但 性能开销大（锁竞争剧烈）。
应用场景：对数据一致性要求极高（如库存管理、银行转账、关键业务交易）。

使用方法：

cursor.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")

核心对比（隔离级别与问题）：

隔离级别	脏读	不可重复读	幻读	隔离性	性能
READ UNCOMMITTED	✔️	✔️	✔️	最低	最高
READ COMMITTED	❌	✔️	✔️	中等	中等
REPEATABLE READ	❌	❌	✔️	较高	中等
SERIALIZABLE	❌	❌	❌	最高	最低

5. 事务隔离级别总结

READ UNCOMMITTED：允许脏读，最低隔离级别，性能最好，但容易出现数据不一致的情况。
READ COMMITTED：解决了脏读问题，但可能出现不可重复读。
REPEATABLE READ：解决了脏读和不可重复读问题，但可能出现幻读。（MySQL事务默认级别）
SERIALIZABLE：解决了所有问题，但性能最差，可能导致事务时间等待。

注意：在选择隔离级别是，需根据具体需求平衡数据一致性和性能，如果事务数据不频繁冲突，可以选择较低的隔离级别以提高性能，而对数据一致性要求高的场景，则选择更高的隔离级别。

6. 案例

import pymysql

# 连接到数据库
conn = pymysql.connect(
    host='localhost',
    user='root',
    password='password',
    database='testdb'
)

# 创建一个游标对象
cursor = conn.cursor()

# 关闭自动提交模式
conn.autocommit = False

try:
    # 开始事务

    # 执行插入操作
    cursor.execute("INSERT INTO users (name, age) VALUES ('Eve', 22)")
    cursor.execute("INSERT INTO orders (user_id, amount) VALUES ((SELECT id FROM users WHERE name = 'Eve'), 120.50)")

    # 提交事务
    conn.commit()
    print("事务已提交。")

except pymysql.MySQLError as err:
    # 如果发生错误，回滚事务
    print(f"错误: {err}")
    conn.rollback()
    print("事务已回滚。")

finally:
    # 关闭游标和连接
    cursor.close()
    conn.close()