🐍 Python Day47：Python 操作 MySQL — 生产级数据库
🕐 预计用时：2-3 小时 ｜ 🎯 目标：掌握 pymysql 连接配置、增删改查、防 SQL 注入
📖 今日目录
1. MySQL 简介与安装
# MySQL 是最流行的开源关系型数据库
# 安装方式：
# Ubuntu/Debian
# sudo apt update && sudo apt install mysql-server
# CentOS/RHEL
# sudo yum install mysql-server
# macOS
# brew install mysql
# Docker（推荐学习用）
# docker run -d --name mysql -e MYSQL_ROOT_PASSWORD=123456 -p 3306:3306 mysql:8
# 安装 Python 驱动
# pip install pymysql
2. pymysql 安装与连接
# pip install pymysql
import pymysql
# 连接 MySQL
conn = pymysql.connect(
host="localhost", # 主机地址
port=3306, # 端口（默认3306）
user="root", # 用户名
password="123456", # 密码
database="school", # 数据库名
charset="utf8mb4", # 字符集（支持中文和 emoji）
cursorclass=pymysql.cursors.DictCursor # 返回字典格式
)
cursor = conn.cursor()
# 测试连接
cursor.execute("SELECT VERSION()")
print(f"MySQL 版本: {cursor.fetchone()['VERSION()']}")
conn.close()
💡 cursorclass=pymysql.cursors.DictCursor 让查询结果返回字典格式（如 {'name': '张三', 'age': 20}），比元组格式更易用。
3. 创建数据库和表
import pymysql
conn = pymysql.connect(
host="localhost",
user="root",
password="123456",
charset="utf8mb4"
)
cursor = conn.cursor()
# 创建数据库
cursor.execute("CREATE DATABASE IF NOT EXISTS school CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")
cursor.execute("USE school")
# 创建表
cursor.execute("""
CREATE TABLE IF NOT EXISTS students (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
age INT CHECK(age > 0),
gender ENUM('男', '女', '未知') DEFAULT '未知',
class VARCHAR(50) NOT NULL,
email VARCHAR(200) UNIQUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""")
conn.commit()
conn.close()
print("✅ 数据库和表创建成功")
4. 增删改查 (CRUD)
import pymysql
conn = pymysql.connect(
host="localhost", user="root", password="123456",
database="school", charset="utf8mb4",
cursorclass=pymysql.cursors.DictCursor
)
cursor = conn.cursor()
# INSERT — 插入
cursor.execute(
"INSERT INTO students (name, age, gender, class) VALUES (%s, %s, %s, %s)",
("张三", 20, "男", "一班")
)
print(f"插入 ID: {cursor.lastrowid}")
# 批量插入
students = [
("李四", 22, "女", "二班"),
("王五", 21, "男", "一班"),
("赵六", 23, "男", "二班"),
]
cursor.executemany(
"INSERT INTO students (name, age, gender, class) VALUES (%s, %s, %s, %s)",
students
)
conn.commit()
# SELECT — 查询
cursor.execute("SELECT * FROM students")
for row in cursor.fetchall():
print(f" {row['name']} | {row['age']}岁 | {row['class']}")
# UPDATE — 更新
cursor.execute("UPDATE students SET age = %s WHERE name = %s", (21, "张三"))
print(f"更新了 {cursor.rowcount} 行")
# DELETE — 删除
cursor.execute("DELETE FROM students WHERE name = %s", ("赵六",))
print(f"删除了 {cursor.rowcount} 行")
conn.commit()
conn.close()
⚠️ MySQL 用 %s 占位符，SQLite 用 ? 占位符！
5. 参数化查询 — 防 SQL 注入
import pymysql
conn = pymysql.connect(host="localhost", user="root", password="123456",
database="school", charset="utf8mb4",
cursorclass=pymysql.cursors.DictCursor)
cursor = conn.cursor()
# ❌ 危险！字符串格式化
# cursor.execute(f"SELECT * FROM students WHERE name = '{user_input}'")
# ✅ 安全！参数化查询
user_input = "张三' OR '1'='1"
cursor.execute("SELECT * FROM students WHERE name = %s", (user_input,))
print(f"安全查询: {cursor.fetchall()}") # []（空结果）
# ✅ 也支持字典参数
cursor.execute(
"SELECT * FROM students WHERE name = %(name)s AND age > %(min_age)s",
{"name": "张三", "min_age": 18}
)
conn.close()
6. 事务管理
import pymysql
conn = pymysql.connect(host="localhost", user="root", password="123456",
database="school", charset="utf8mb4",
cursorclass=pymysql.cursors.DictCursor)
cursor = conn.cursor()
try:
# 开始事务（pymysql 默认自动开启）
cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE name = '张三'")
cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE name = '李四'")
# 提交事务
conn.commit()
print("✅ 转账成功")
except Exception as e:
# 回滚事务
conn.rollback()
print(f"❌ 转账失败: {e}")
finally:
conn.close()
7. 连接池
# DBUtils 提供连接池功能
# pip install dbutils
from dbutils.pooled_db import PooledDB
import pymysql
# 创建连接池
pool = PooledDB(
creator=pymysql, # 数据库驱动
maxconnections=10, # 最大连接数
mincached=2, # 最小空闲连接
maxcached=5, # 最大空闲连接
host="localhost",
user="root",
password="123456",
database="school",
charset="utf8mb4",
cursorclass=pymysql.cursors.DictCursor
)
# 从连接池获取连接
conn = pool.connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM students")
print(cursor.fetchall())
conn.close() # 归还到连接池（不是真正关闭）
# 连接池的好处：
# 1. 避免频繁创建/销毁连接（开销大）
# 2. 控制最大连接数（防止数据库过载）
# 3. 自动管理连接生命周期
8. 错误处理
import pymysql
def safe_connect():
"""安全的数据库连接"""
try:
conn = pymysql.connect(
host="localhost",
user="root",
password="wrong_password",
database="school",
charset="utf8mb4"
)
return conn
except pymysql.err.OperationalError as e:
print(f"❌ 连接失败: {e}")
return None
def safe_execute(conn, sql, params=None):
"""安全执行 SQL"""
try:
cursor = conn.cursor()
cursor.execute(sql, params)
conn.commit()
return cursor
except pymysql.err.ProgrammingError as e:
print(f"❌ SQL 语法错误: {e}")
conn.rollback()
return None
except pymysql.err.IntegrityError as e:
print(f"❌ 数据完整性错误（如主键重复）: {e}")
conn.rollback()
return None
except pymysql.err.OperationalError as e:
print(f"❌ 操作错误: {e}")
conn.rollback()
return None
9. 实战：用户管理系统
import pymysql
from datetime import datetime
class UserDB:
"""用户管理系统"""
def __init__(self, host="localhost", user="root", password="123456", database="school"):
self.conn = pymysql.connect(
host=host, user=user, password=password,
database=database, charset="utf8mb4",
cursorclass=pymysql.cursors.DictCursor
)
self._create_table()
def _create_table(self):
self.conn.cursor().execute("""
CREATE TABLE IF NOT EXISTS users (
id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(50) NOT NULL UNIQUE,
email VARCHAR(200) NOT NULL,
age INT,
status TINYINT DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""")
self.conn.commit()
def create(self, username, email, age=None):
"""创建用户"""
try:
cursor = self.conn.cursor()
cursor.execute(
"INSERT INTO users (username, email, age) VALUES (%s, %s, %s)",
(username, email, age)
)
self.conn.commit()
return cursor.lastrowid
except pymysql.err.IntegrityError:
print(f"❌ 用户名 '{username}' 已存在")
return None
def get(self, user_id):
"""获取单个用户"""
cursor = self.conn.cursor()
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
return cursor.fetchone()
def list_all(self, page=1, page_size=10):
"""分页列出用户"""
cursor = self.conn.cursor()
offset = (page - 1) * page_size
cursor.execute("SELECT * FROM users ORDER BY id LIMIT %s OFFSET %s", (page_size, offset))
return cursor.fetchall()
def update(self, user_id, **kwargs):
"""更新用户"""
allowed = {"username", "email", "age", "status"}
updates = {k: v for k, v in kwargs.items() if k in allowed}
if not updates:
return 0
set_clause = ", ".join(f"{k} = %s" for k in updates)
values = list(updates.values()) + [user_id]
cursor = self.conn.cursor()
cursor.execute(f"UPDATE users SET {set_clause} WHERE id = %s", values)
self.conn.commit()
return cursor.rowcount
def delete(self, user_id):
"""删除用户"""
cursor = self.conn.cursor()
cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
self.conn.commit()
return cursor.rowcount
def search(self, keyword):
"""搜索用户"""
cursor = self.conn.cursor()
cursor.execute(
"SELECT * FROM users WHERE username LIKE %s OR email LIKE %s",
(f"%{keyword}%", f"%{keyword}%")
)
return cursor.fetchall()
def stats(self):
"""统计信息"""
cursor = self.conn.cursor()
cursor.execute("""
SELECT
COUNT(*) as total,
AVG(age) as avg_age,
SUM(CASE WHEN status = 1 THEN 1 ELSE 0 END) as active
FROM users
""")
return cursor.fetchone()
def close(self):
self.conn.close()
# 使用
db = UserDB()
db.create("alice", "alice@example.com", 25)
db.create("bob", "bob@example.com", 30)
db.create("charlie", "charlie@example.com", 22)
print("📋 所有用户:")
for u in db.list_all():
print(f" [{u['id']}] {u['username']} | {u['email']} | {u['age']}岁")
print(f"\n📊 统计: {db.stats()}")
db.close()
10. SQLite vs MySQL 对比
11. 今日小结
- ✅
pymysql.connect() 连接 MySQL - ✅ MySQL 用
%s 占位符（SQLite 用 ?） - ✅
cursorclass=DictCursor 返回字典格式 - ✅
conn.commit() 提交，conn.rollback() 回滚
🎯 练习建议：
1. 给用户管理系统添加"批量导入"功能（从 CSV 导入）
2. 实现一个"数据库备份"工具（导出为 SQL 文件）
3. 对比 pymysql 和 sqlite3 的代码差异
📚 Day47 完成！明天学习 ORM — SQLAlchemy 入门
10个月宝宝每天需要喝多少奶粉？