Python操作MySQL数据库：从基础到高级实践

1. Python与MySQL交互基础准备

在开始使用Python操作MySQL之前，我们需要完成一些基础准备工作。首先确保你的系统已经安装了Python（建议3.6以上版本）和MySQL数据库服务器。如果你还没有安装MySQL，可以从MySQL官网下载社区版进行安装。

1.1 安装MySQL驱动

Python需要通过特定的驱动来连接MySQL数据库。目前最常用的驱动是mysql-connector-python，这是MySQL官方提供的纯Python驱动。安装方法如下：

pip install mysql-connector-python

如果你遇到网络问题，可以使用国内镜像源加速安装：

pip install mysql-connector-python -i https://pypi.tuna.tsinghua.edu.cn/simple

安装完成后，可以通过以下代码验证是否安装成功：

import mysql.connector print(mysql.connector.__version__)

如果没有报错并输出版本号，说明安装成功。

1.2 创建数据库连接

建立与MySQL数据库的连接是操作的第一步。我们需要提供数据库的主机地址、用户名、密码等信息：

import mysql.connector config = { 'host': 'localhost', 'user': 'your_username', 'password': 'your_password', 'database': 'your_database', # 可选，也可以在连接后选择数据库 'port': 3306, # 默认端口 'charset': 'utf8mb4', # 推荐使用utf8mb4以支持完整的Unicode字符 'autocommit': True # 自动提交事务 } try: conn = mysql.connector.connect(**config) print("数据库连接成功！") cursor = conn.cursor() # 执行SQL语句... except mysql.connector.Error as err: print(f"连接失败: {err}") finally: if 'conn' in locals() and conn.is_connected(): cursor.close() conn.close()

提示：在实际项目中，不要将数据库凭据直接写在代码中，应该使用环境变量或配置文件来管理敏感信息。

2. 基本CRUD操作

掌握了数据库连接后，我们来看如何使用Python执行基本的增删改查(CRUD)操作。

2.1 创建表

首先创建一个示例表来演示后续操作：

create_table_sql = """ CREATE TABLE IF NOT EXISTS employees ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(50) NOT NULL, age INT, department VARCHAR(50), salary DECIMAL(10,2), join_date DATE, INDEX idx_department (department) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ cursor.execute(create_table_sql) print("表创建成功")

2.2 插入数据

插入数据可以使用INSERT语句。为了防止SQL注入，应该使用参数化查询：

insert_sql = "INSERT INTO employees (name, age, department, salary, join_date) VALUES (%s, %s, %s, %s, %s)" employees = [ ('张三', 28, '技术部', 8500.00, '2020-05-15'), ('李四', 32, '市场部', 9200.00, '2019-08-22'), ('王五', 25, '技术部', 7800.00, '2021-03-10'), ('赵六', 35, '人事部', 9500.00, '2018-11-05') ] cursor.executemany(insert_sql, employees) print(f"插入了{cursor.rowcount}条记录")

2.3 查询数据

查询数据是最常用的操作，我们可以使用SELECT语句：

# 查询所有记录 cursor.execute("SELECT * FROM employees") rows = cursor.fetchall() print("所有员工信息:") for row in rows: print(f"ID: {row[0]}, 姓名: {row[1]}, 年龄: {row[2]}, 部门: {row[3]}, 薪资: {row[4]}, 入职日期: {row[5]}") # 带条件的查询 query = "SELECT name, salary FROM employees WHERE department = %s AND salary > %s" params = ('技术部', 8000) cursor.execute(query, params) print("\n技术部薪资高于8000的员工:") for (name, salary) in cursor: print(f"{name}: {salary}")

2.4 更新数据

更新数据使用UPDATE语句：

update_sql = "UPDATE employees SET salary = salary * %s WHERE department = %s" params = (1.1, '技术部') # 技术部员工加薪10% cursor.execute(update_sql, params) print(f"更新了{cursor.rowcount}条记录")

2.5 删除数据

删除数据使用DELETE语句：

delete_sql = "DELETE FROM employees WHERE name = %s" cursor.execute(delete_sql, ('赵六',)) print(f"删除了{cursor.rowcount}条记录")

3. 高级操作与性能优化

掌握了基本操作后，我们来看一些更高级的用法和性能优化技巧。

3.1 事务处理

MySQL支持事务，可以确保一组操作要么全部成功，要么全部失败：

try: conn.start_transaction() # 转账操作示例 cursor.execute("UPDATE accounts SET balance = balance - 500 WHERE account_id = 1") cursor.execute("UPDATE accounts SET balance = balance + 500 WHERE account_id = 2") conn.commit() print("事务执行成功") except Exception as e: conn.rollback() print(f"事务执行失败，已回滚: {e}")

3.2 批量操作

对于大量数据操作，使用批量处理可以显著提高性能：

# 批量插入 data = [(f"员工{i}", 25+i%10, random.choice(['技术部','市场部','人事部']), random.randint(5000,15000), (datetime.now() - timedelta(days=random.randint(1,1000))).strftime('%Y-%m-%d')) for i in range(1000)] insert_sql = "INSERT INTO employees (name, age, department, salary, join_date) VALUES (%s, %s, %s, %s, %s)" # 每次插入100条 batch_size = 100 for i in range(0, len(data), batch_size): batch = data[i:i+batch_size] cursor.executemany(insert_sql, batch) conn.commit() print(f"已插入{i+batch_size if i+batch_size < len(data) else len(data)}条记录")

3.3 使用连接池

对于Web应用等高频访问数据库的场景，使用连接池可以提高性能：

from mysql.connector import pooling # 创建连接池 dbconfig = { "host": "localhost", "user": "your_username", "password": "your_password", "database": "your_database" } connection_pool = pooling.MySQLConnectionPool( pool_name="mypool", pool_size=5, # 连接池大小 **dbconfig ) # 从连接池获取连接 conn = connection_pool.get_connection() try: cursor = conn.cursor() cursor.execute("SELECT * FROM employees LIMIT 5") for row in cursor: print(row) finally: if conn.is_connected(): cursor.close() conn.close() # 实际是返回到连接池

4. 实际应用中的最佳实践

在实际项目中使用Python操作MySQL时，有一些最佳实践值得注意。

4.1 使用ORM框架

对于复杂项目，使用ORM(Object-Relational Mapping)框架可以简化数据库操作。Python中常用的ORM有SQLAlchemy和Django ORM等。以下是使用SQLAlchemy的示例：

from sqlalchemy import create_engine, Column, Integer, String, Float, Date from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker Base = declarative_base() class Employee(Base): __tablename__ = 'employees' id = Column(Integer, primary_key=True) name = Column(String(50)) age = Column(Integer) department = Column(String(50)) salary = Column(Float) join_date = Column(Date) # 创建引擎 engine = create_engine('mysql+mysqlconnector://username:password@localhost/database') # 创建表 Base.metadata.create_all(engine) # 创建会话 Session = sessionmaker(bind=engine) session = Session() # 添加新员工 new_employee = Employee( name='钱七', age=29, department='财务部', salary=8800.00, join_date='2022-01-15' ) session.add(new_employee) session.commit() # 查询 employees = session.query(Employee).filter(Employee.department=='财务部').all() for emp in employees: print(emp.name, emp.salary)

4.2 错误处理与重试机制

数据库操作可能会因各种原因失败，实现健壮的错误处理和重试机制很重要：

import time from mysql.connector import Error def execute_with_retry(cursor, sql, params=None, max_retries=3, delay=1): attempts = 0 while attempts < max_retries: try: if params: cursor.execute(sql, params) else: cursor.execute(sql) return True except Error as err: attempts += 1 print(f"操作失败 (尝试 {attempts}/{max_retries}): {err}") if attempts == max_retries: return False time.sleep(delay) # 使用示例 success = execute_with_retry(cursor, "UPDATE employees SET salary = salary * 1.05 WHERE department = %s", ('技术部',)) if success: print("更新成功") else: print("更新失败，请检查")

4.3 数据库迁移与版本控制

对于长期维护的项目，数据库结构会不断演进。使用迁移工具可以更好地管理这些变更。Alembic是一个常用的数据库迁移工具：

pip install alembic alembic init migrations

然后配置alembic.ini和migrations/env.py文件，之后可以通过以下命令创建和应用迁移：

# 创建新的迁移 alembic revision --autogenerate -m "add email column to employees" # 应用迁移 alembic upgrade head

5. 性能监控与优化

当数据量增大时，数据库性能变得尤为重要。以下是一些监控和优化MySQL性能的方法。

5.1 使用EXPLAIN分析查询

对于慢查询，可以使用EXPLAIN来分析执行计划：

query = "EXPLAIN SELECT * FROM employees WHERE department = %s AND salary > %s" cursor.execute(query, ('技术部', 8000)) print("执行计划分析:") for row in cursor: print(row)

5.2 添加适当的索引

合理的索引可以大幅提高查询性能。例如，如果我们经常按部门和薪资查询：

cursor.execute("ALTER TABLE employees ADD INDEX idx_dept_salary (department, salary)")

5.3 监控慢查询

在MySQL配置中开启慢查询日志：

# 查询当前慢查询设置 cursor.execute("SHOW VARIABLES LIKE 'slow_query%'") cursor.execute("SHOW VARIABLES LIKE 'long_query_time'") # 可以临时设置(重启后失效) cursor.execute("SET GLOBAL slow_query_log = 'ON'") cursor.execute("SET GLOBAL long_query_time = 1") # 超过1秒的查询记录

5.4 使用缓存

对于不经常变化的数据，可以使用缓存减少数据库访问：

from functools import lru_cache import time @lru_cache(maxsize=100) def get_department_employees(department): time.sleep(1) # 模拟耗时查询 cursor.execute("SELECT name FROM employees WHERE department = %s", (department,)) return [row[0] for row in cursor] # 第一次调用会查询数据库 print(get_department_employees('技术部')) # 后续调用会从缓存获取 print(get_department_employees('技术部'))

6. 安全注意事项

数据库安全至关重要，以下是一些关键的安全实践。

6.1 防止SQL注入

始终使用参数化查询，不要拼接SQL字符串：

# 错误做法 - 容易导致SQL注入 user_input = "技术部'; DROP TABLE employees; --" sql = f"SELECT * FROM employees WHERE department = '{user_input}'" # 正确做法 sql = "SELECT * FROM employees WHERE department = %s" cursor.execute(sql, (user_input,))

6.2 最小权限原则

为应用数据库用户分配最小必要权限：

-- 只授予必要的权限 GRANT SELECT, INSERT, UPDATE ON database.employees TO 'app_user'@'localhost';

6.3 敏感数据加密

对敏感数据如密码应该加密存储：

import hashlib def hash_password(password): return hashlib.sha256(password.encode()).hexdigest() # 存储加密后的密码 hashed_pwd = hash_password('user_password') cursor.execute("INSERT INTO users (username, password) VALUES (%s, %s)", ('user1', hashed_pwd))

6.4 定期备份

实现自动化的数据库备份策略：

import subprocess from datetime import datetime def backup_database(): timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') backup_file = f"backup_{timestamp}.sql" command = [ 'mysqldump', '--user=your_username', '--password=your_password', '--host=localhost', 'your_database', '>', backup_file ] try: subprocess.run(' '.join(command), shell=True, check=True) print(f"备份成功: {backup_file}") except subprocess.CalledProcessError as e: print(f"备份失败: {e}") # 定期执行备份 backup_database()

7. 实际项目集成

在实际项目中，我们通常会将数据库操作封装成更易用的模块或类。

7.1 封装数据库操作类

import mysql.connector from mysql.connector import pooling from contextlib import contextmanager class DatabaseManager: def __init__(self, host, user, password, database, pool_size=5): self.pool = pooling.MySQLConnectionPool( pool_name="mypool", pool_size=pool_size, host=host, user=user, password=password, database=database, autocommit=True ) @contextmanager def get_cursor(self): conn = self.pool.get_connection() try: cursor = conn.cursor(dictionary=True) # 返回字典形式的结果 yield cursor finally: cursor.close() conn.close() def execute_query(self, sql, params=None): with self.get_cursor() as cursor: cursor.execute(sql, params or ()) return cursor.fetchall() def execute_update(self, sql, params=None): with self.get_cursor() as cursor: cursor.execute(sql, params or ()) return cursor.rowcount # 使用示例 db = DatabaseManager('localhost', 'user', 'password', 'company_db') # 查询 employees = db.execute_query("SELECT * FROM employees WHERE department = %s", ('技术部',)) for emp in employees: print(emp['name'], emp['salary']) # 更新 affected = db.execute_update("UPDATE employees SET salary = salary * 1.05 WHERE department = %s", ('技术部',)) print(f"更新了{affected}条记录")

7.2 与Web框架集成

在Web应用(如Flask)中使用MySQL：

from flask import Flask, g import mysql.connector from mysql.connector import pooling app = Flask(__name__) # 配置数据库连接池 dbconfig = { "host": "localhost", "user": "web_user", "password": "secure_password", "database": "myapp_db" } connection_pool = pooling.MySQLConnectionPool( pool_name="webpool", pool_size=5, **dbconfig ) @app.before_request def before_request(): g.db = connection_pool.get_connection() g.cursor = g.db.cursor(dictionary=True) @app.teardown_request def teardown_request(exception): if hasattr(g, 'cursor'): g.cursor.close() if hasattr(g, 'db'): g.db.close() @app.route('/employees') def list_employees(): g.cursor.execute("SELECT * FROM employees") employees = g.cursor.fetchall() return {'employees': employees} if __name__ == '__main__': app.run()

7.3 异步操作

对于高性能应用，可以使用异步MySQL驱动如aiomysql：

import asyncio import aiomysql async def fetch_employees(): conn = await aiomysql.connect( host='localhost', user='user', password='password', db='company_db' ) async with conn.cursor(aiomysql.DictCursor) as cursor: await cursor.execute("SELECT * FROM employees WHERE department=%s", ('技术部',)) result = await cursor.fetchall() print(result) conn.close() # 运行 loop = asyncio.get_event_loop() loop.run_until_complete(fetch_employees())

8. 调试与问题排查

在开发过程中，可能会遇到各种数据库相关问题，以下是一些常见问题的解决方法。

8.1 连接问题

如果遇到连接问题，可以按照以下步骤排查：

1. 检查MySQL服务是否运行
2. 检查连接参数是否正确
3. 检查用户权限
4. 检查防火墙设置

import socket def check_connection(host, port): try: with socket.create_connection((host, port), timeout=5): return True except (socket.timeout, ConnectionRefusedError) as e: print(f"无法连接到 {host}:{port} - {e}") return False if check_connection('localhost', 3306): print("MySQL端口可访问") else: print("无法访问MySQL端口")

8.2 性能问题

对于慢查询，可以使用MySQL的慢查询日志或performance_schema来识别瓶颈：

# 启用性能监控 cursor.execute("SET GLOBAL performance_schema = ON") cursor.execute("SET GLOBAL performance_schema_consumer_events_statements_history_long = ON") # 查询最耗时的SQL cursor.execute(""" SELECT sql_text, timer_wait/1000000000 as seconds FROM performance_schema.events_statements_history_long ORDER BY timer_wait DESC LIMIT 5 """) for row in cursor: print(f"{row[1]:.2f}s: {row[0]}")

8.3 字符编码问题

确保数据库、连接和表都使用一致的字符编码（推荐utf8mb4）：

# 检查当前编码设置 cursor.execute("SHOW VARIABLES LIKE 'character_set%'") for row in cursor: print(row) # 创建连接时指定编码 conn = mysql.connector.connect( host='localhost', user='user', password='password', database='db', charset='utf8mb4', collation='utf8mb4_unicode_ci' )

8.4 连接池问题

如果使用连接池，可能会遇到连接泄漏或连接数不足的问题：

# 检查当前连接数 cursor.execute("SHOW STATUS LIKE 'Threads_connected'") print(cursor.fetchone()) # 检查连接池状态 print(f"连接池大小: {connection_pool.pool_size}") print(f"当前使用连接数: {connection_pool._cnx_queue.qsize()}")

9. 扩展知识与进阶主题

掌握了基本操作后，可以进一步学习以下进阶主题。

9.1 存储过程调用

MySQL支持存储过程，可以在Python中调用：

# 创建存储过程 cursor.execute(""" CREATE PROCEDURE get_employees_by_dept(IN dept VARCHAR(50)) BEGIN SELECT * FROM employees WHERE department = dept; END """) # 调用存储过程 cursor.callproc('get_employees_by_dept', ('技术部',)) for result in cursor.stored_results(): for row in result: print(row)

9.2 使用MySQL JSON类型

MySQL 5.7+支持JSON数据类型：

# 创建包含JSON列的表 cursor.execute(""" CREATE TABLE IF NOT EXISTS products ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100), attributes JSON, INDEX idx_attributes ((CAST(attributes->'$.color' AS CHAR(20)))) ) """) # 插入JSON数据 insert_sql = "INSERT INTO products (name, attributes) VALUES (%s, %s)" products = [ ('手机', '{"color": "black", "memory": "128GB", "price": 3999}'), ('笔记本', '{"color": "silver", "memory": "16GB", "price": 8999}') ] cursor.executemany(insert_sql, products) # 查询JSON数据 cursor.execute("SELECT name, attributes->'$.price' AS price FROM products WHERE attributes->'$.color' = 'black'") for row in cursor: print(row)

9.3 使用MySQL窗口函数

MySQL 8.0+支持窗口函数，可以进行复杂分析：

cursor.execute(""" SELECT name, department, salary, RANK() OVER (PARTITION BY department ORDER BY salary DESC) as dept_rank, salary - AVG(salary) OVER (PARTITION BY department) as diff_from_avg FROM employees """) print("员工薪资分析:") for row in cursor: print(f"{row[0]}({row[1]}): 薪资 {row[2]}, 部门排名 {row[3]}, 与部门平均差 {row[4]:.2f}")

9.4 使用MySQL CTE

公用表表达式(CTE)可以简化复杂查询：

cursor.execute(""" WITH dept_stats AS ( SELECT department, AVG(salary) as avg_salary, COUNT(*) as emp_count FROM employees GROUP BY department ) SELECT e.name, e.department, e.salary, ds.avg_salary, e.salary - ds.avg_salary as diff FROM employees e JOIN dept_stats ds ON e.department = ds.department ORDER BY diff DESC """) print("员工薪资与部门平均对比:") for row in cursor: print(f"{row[0]}({row[1]}): 薪资 {row[2]}, 部门平均 {row[3]:.2f}, 差异 {row[4]:.2f}")

10. 测试与持续集成

在实际项目中，我们需要为数据库相关代码编写测试，并集成到CI/CD流程中。

10.1 单元测试

使用unittest或pytest测试数据库操作：

import unittest from unittest.mock import MagicMock class TestEmployeeQueries(unittest.TestCase): def setUp(self): self.mock_cursor = MagicMock() # 模拟查询结果 self.mock_cursor.fetchall.return_value = [ (1, '张三', 28, '技术部', 8500.00), (2, '李四', 32, '市场部', 9200.00) ] def test_get_all_employees(self): from your_module import get_all_employees # 调用被测函数 result = get_all_employees(self.mock_cursor) # 验证 self.assertEqual(len(result), 2) self.mock_cursor.execute.assert_called_once_with("SELECT * FROM employees") if __name__ == '__main__': unittest.main()

10.2 集成测试

使用测试数据库进行集成测试：

import pytest import mysql.connector from your_module import DatabaseManager @pytest.fixture def test_db(): # 创建测试数据库 conn = mysql.connector.connect( host="localhost", user="test_user", password="test_password" ) cursor = conn.cursor() cursor.execute("CREATE DATABASE IF NOT EXISTS test_db") cursor.execute("USE test_db") cursor.execute(""" CREATE TABLE IF NOT EXISTS employees ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(50), department VARCHAR(50) ) """) yield DatabaseManager('localhost', 'test_user', 'test_password', 'test_db') # 清理 cursor.execute("DROP DATABASE test_db") cursor.close() conn.close() def test_add_employee(test_db): test_db.execute_update("INSERT INTO employees (name, department) VALUES (%s, %s)", ('测试员工', '测试部门')) result = test_db.execute_query("SELECT * FROM employees") assert len(result) == 1 assert result[0]['name'] == '测试员工'

10.3 使用Docker进行测试

可以使用Docker快速启动测试用的MySQL实例：

import docker import time def start_test_mysql(): client = docker.from_env() container = client.containers.run( "mysql:8.0", environment={ "MYSQL_ROOT_PASSWORD": "test", "MYSQL_DATABASE": "test_db", "MYSQL_USER": "test_user", "MYSQL_PASSWORD": "test_password" }, ports={'3306/tcp': 3307}, detach=True ) # 等待MySQL启动 time.sleep(20) return container def stop_test_mysql(container): container.stop() container.remove()

10.4 持续集成配置

在GitHub Actions等CI系统中配置MySQL测试：

# .github/workflows/test.yml name: Test on: [push, pull_request] jobs: test: runs-on: ubuntu-latest services: mysql: image: mysql:8.0 env: MYSQL_ROOT_PASSWORD: root MYSQL_DATABASE: test_db MYSQL_USER: test_user MYSQL_PASSWORD: test_password ports: - 3306:3306 options: --health-cmd="mysqladmin ping" --health-interval=10s --health-timeout=5s --health-retries=3 steps: - uses: actions/checkout@v2 - name: Set up Python uses: actions/setup-python@v2 with: python-version: '3.9' - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt pip install pytest pytest-cov - name: Run tests env: DB_HOST: localhost DB_USER: test_user DB_PASSWORD: test_password DB_NAME: test_db run: | pytest --cov=your_module tests/