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) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 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(fID: {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}, 25i%10, random.choice([技术部,市场部,人事部]), random.randint(5000,15000), (datetime.now() - timedelta(daysrandom.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:ibatch_size] cursor.executemany(insert_sql, batch) conn.commit() print(f已插入{ibatch_size if ibatch_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_namemypool, pool_size5, # 连接池大小 **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_keyTrue) name Column(String(50)) age Column(Integer) department Column(String(50)) salary Column(Float) join_date Column(Date) # 创建引擎 engine create_engine(mysqlmysqlconnector://username:passwordlocalhost/database) # 创建表 Base.metadata.create_all(engine) # 创建会话 Session sessionmaker(bindengine) session Session() # 添加新员工 new_employee Employee( name钱七, age29, department财务部, salary8800.00, join_date2022-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, paramsNone, max_retries3, delay1): 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 head5. 性能监控与优化当数据量增大时数据库性能变得尤为重要。以下是一些监控和优化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(maxsize100) 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 fSELECT * 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_userlocalhost;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 fbackup_{timestamp}.sql command [ mysqldump, --useryour_username, --passwordyour_password, --hostlocalhost, your_database, , backup_file ] try: subprocess.run( .join(command), shellTrue, checkTrue) 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_size5): self.pool pooling.MySQLConnectionPool( pool_namemypool, pool_sizepool_size, hosthost, useruser, passwordpassword, databasedatabase, autocommitTrue ) contextmanager def get_cursor(self): conn self.pool.get_connection() try: cursor conn.cursor(dictionaryTrue) # 返回字典形式的结果 yield cursor finally: cursor.close() conn.close() def execute_query(self, sql, paramsNone): with self.get_cursor() as cursor: cursor.execute(sql, params or ()) return cursor.fetchall() def execute_update(self, sql, paramsNone): 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)中使用MySQLfrom 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_namewebpool, pool_size5, **dbconfig ) app.before_request def before_request(): g.db connection_pool.get_connection() g.cursor g.db.cursor(dictionaryTrue) 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驱动如aiomysqlimport asyncio import aiomysql async def fetch_employees(): conn await aiomysql.connect( hostlocalhost, useruser, passwordpassword, dbcompany_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 连接问题如果遇到连接问题可以按照以下步骤排查检查MySQL服务是否运行检查连接参数是否正确检查用户权限检查防火墙设置import socket def check_connection(host, port): try: with socket.create_connection((host, port), timeout5): 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( hostlocalhost, useruser, passwordpassword, databasedb, charsetutf8mb4, collationutf8mb4_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( hostlocalhost, usertest_user, passwordtest_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}, detachTrue ) # 等待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-cmdmysqladmin ping --health-interval10s --health-timeout5s --health-retries3 steps: - uses: actions/checkoutv2 - name: Set up Python uses: actions/setup-pythonv2 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 --covyour_module tests/