1. Python连接MySQL的几种主流方式在Python生态中连接MySQL数据库主要有三种主流方式每种方式都有其适用场景和特点1.1 MySQLdb (MySQL-python)这是Python连接MySQL最经典的方式实现了Python DB-API 2.0规范。它的特点是基于MySQL C API开发性能较好只支持Python 2.x安装需要编译环境最新版本停留在1.2.52014年安装命令pip install MySQL-python1.2 mysql-connector-pythonMySQL官方推出的纯Python驱动特点包括由Oracle官方维护纯Python实现无需编译支持Python 2.7和3.x完全兼容MySQL协议支持MySQL 8.0的新特性安装命令pip install mysql-connector-python1.3 PyMySQL纯Python实现的MySQL客户端主要特点兼容MySQLdb的API支持Python 3.x无需编译安装简单性能略低于MySQLdb支持MariaDB安装命令pip install PyMySQL2. 环境准备与基础连接2.1 MySQL服务端安装在开始Python连接前需要确保MySQL服务端已正确安装。推荐使用MySQL Community Server 8.0版本Windows平台下载MySQL InstallerLinux平台以Ubuntu为例sudo apt update sudo apt install mysql-server sudo mysql_secure_installation2.2 Python连接MySQL基础示例使用mysql-connector-python的基本连接代码import mysql.connector # 创建连接配置 config { user: username, password: password, host: 127.0.0.1, database: testdb, raise_on_warnings: True } try: # 建立连接 conn mysql.connector.connect(**config) # 创建游标 cursor conn.cursor() # 执行SQL查询 cursor.execute(SELECT VERSION()) # 获取结果 version cursor.fetchone() print(fMySQL Server version: {version[0]}) except mysql.connector.Error as err: print(fError: {err}) finally: # 关闭连接 if conn in locals() and conn.is_connected(): cursor.close() conn.close()3. 数据库CRUD操作详解3.1 创建表def create_table(): try: conn mysql.connector.connect(**config) cursor conn.cursor() create_table_sql CREATE TABLE IF NOT EXISTS employees ( id INT AUTO_INCREMENT PRIMARY KEY, first_name VARCHAR(50) NOT NULL, last_name VARCHAR(50), age INT, department VARCHAR(50), salary DECIMAL(10,2), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) cursor.execute(create_table_sql) print(Table created successfully) except mysql.connector.Error as err: print(fFailed to create table: {err}) finally: if conn in locals() and conn.is_connected(): cursor.close() conn.close()3.2 插入数据def insert_employee(data): try: conn mysql.connector.connect(**config) cursor conn.cursor() insert_sql INSERT INTO employees (first_name, last_name, age, department, salary) VALUES (%s, %s, %s, %s, %s) cursor.execute(insert_sql, data) conn.commit() print(fInserted {cursor.rowcount} row(s)) except mysql.connector.Error as err: conn.rollback() print(fFailed to insert data: {err}) finally: if conn in locals() and conn.is_connected(): cursor.close() conn.close() # 使用示例 employee_data (John, Doe, 30, IT, 75000.00) insert_employee(employee_data)3.3 查询数据def query_employees(departmentNone): try: conn mysql.connector.connect(**config) cursor conn.cursor(dictionaryTrue) # 返回字典形式的结果 query_sql SELECT * FROM employees params () if department: query_sql WHERE department %s params (department,) cursor.execute(query_sql, params) results cursor.fetchall() for row in results: print(fID: {row[id]}, Name: {row[first_name]} {row[last_name]}, fDepartment: {row[department]}, Salary: {row[salary]}) except mysql.connector.Error as err: print(fFailed to query data: {err}) finally: if conn in locals() and conn.is_connected(): cursor.close() conn.close()3.4 更新数据def update_salary(employee_id, new_salary): try: conn mysql.connector.connect(**config) cursor conn.cursor() update_sql UPDATE employees SET salary %s WHERE id %s cursor.execute(update_sql, (new_salary, employee_id)) conn.commit() print(fUpdated {cursor.rowcount} row(s)) except mysql.connector.Error as err: conn.rollback() print(fFailed to update data: {err}) finally: if conn in locals() and conn.is_connected(): cursor.close() conn.close()3.5 删除数据def delete_employee(employee_id): try: conn mysql.connector.connect(**config) cursor conn.cursor() delete_sql DELETE FROM employees WHERE id %s cursor.execute(delete_sql, (employee_id,)) conn.commit() print(fDeleted {cursor.rowcount} row(s)) except mysql.connector.Error as err: conn.rollback() print(fFailed to delete data: {err}) finally: if conn in locals() and conn.is_connected(): cursor.close() conn.close()4. 高级特性与最佳实践4.1 连接池管理对于Web应用等需要频繁连接数据库的场景使用连接池可以显著提高性能from mysql.connector import pooling # 创建连接池 dbconfig { host: localhost, user: username, password: password, database: testdb } connection_pool pooling.MySQLConnectionPool( pool_namemypool, pool_size5, **dbconfig ) # 使用连接池 def get_employees(): try: # 从连接池获取连接 conn connection_pool.get_connection() cursor conn.cursor(dictionaryTrue) cursor.execute(SELECT * FROM employees) return cursor.fetchall() except mysql.connector.Error as err: print(fError: {err}) return None finally: if conn in locals() and conn.is_connected(): cursor.close() conn.close() # 连接返回到连接池4.2 事务处理def transfer_funds(from_id, to_id, amount): try: conn mysql.connector.connect(**config) cursor conn.cursor() # 开始事务 conn.start_transaction() # 扣除转出账户金额 cursor.execute( UPDATE accounts SET balance balance - %s WHERE id %s, (amount, from_id) ) # 增加转入账户金额 cursor.execute( UPDATE accounts SET balance balance %s WHERE id %s, (amount, to_id) ) # 验证转出账户余额 cursor.execute( SELECT balance FROM accounts WHERE id %s FOR UPDATE, (from_id,) ) balance cursor.fetchone()[0] if balance 0: raise ValueError(Insufficient funds) # 提交事务 conn.commit() print(Transfer completed successfully) except Exception as err: conn.rollback() print(fTransfer failed: {err}) finally: if conn in locals() and conn.is_connected(): cursor.close() conn.close()4.3 批量操作def batch_insert_employees(employee_list): try: conn mysql.connector.connect(**config) cursor conn.cursor() insert_sql INSERT INTO employees (first_name, last_name, age, department, salary) VALUES (%s, %s, %s, %s, %s) cursor.executemany(insert_sql, employee_list) conn.commit() print(fInserted {cursor.rowcount} rows) except mysql.connector.Error as err: conn.rollback() print(fBatch insert failed: {err}) finally: if conn in locals() and conn.is_connected(): cursor.close() conn.close() # 使用示例 employees [ (Alice, Smith, 28, HR, 65000), (Bob, Johnson, 35, Finance, 82000), (Charlie, Brown, 42, IT, 95000) ] batch_insert_employees(employees)4.4 使用上下文管理器from contextlib import contextmanager contextmanager def get_db_connection(): conn None try: conn mysql.connector.connect(**config) yield conn except mysql.connector.Error as err: print(fDatabase error: {err}) if conn: conn.rollback() raise finally: if conn and conn.is_connected(): conn.close() # 使用示例 with get_db_connection() as conn: cursor conn.cursor(dictionaryTrue) cursor.execute(SELECT * FROM employees LIMIT 5) for row in cursor: print(row)5. 常见问题与解决方案5.1 连接超时问题MySQL默认连接超时为8小时可以通过以下方式解决在连接配置中添加连接超时参数config { user: username, password: password, host: 127.0.0.1, database: testdb, pool_name: mypool, pool_size: 5, pool_reset_session: True, connection_timeout: 30 # 超时时间(秒) }使用连接池并设置自动重连connection_pool pooling.MySQLConnectionPool( pool_namemypool, pool_size5, pool_reset_sessionTrue, **dbconfig )5.2 字符编码问题确保连接使用UTF-8编码conn mysql.connector.connect( hostlocalhost, userusername, passwordpassword, databasetestdb, charsetutf8mb4, # 支持完整的Unicode字符 collationutf8mb4_unicode_ci )5.3 安全注意事项永远不要直接拼接SQL语句# 错误做法SQL注入风险 sql fSELECT * FROM users WHERE username {username} AND password {password} # 正确做法使用参数化查询 sql SELECT * FROM users WHERE username %s AND password %s cursor.execute(sql, (username, password))使用环境变量存储数据库凭证import os from dotenv import load_dotenv load_dotenv() # 从.env文件加载环境变量 config { user: os.getenv(DB_USER), password: os.getenv(DB_PASSWORD), host: os.getenv(DB_HOST), database: os.getenv(DB_NAME) }5.4 性能优化建议使用索引加速查询# 创建索引 cursor.execute( CREATE INDEX idx_employee_department ON employees(department) )批量操作代替循环单条操作# 低效做法 for employee in employees: cursor.execute(insert_sql, employee) # 高效做法 cursor.executemany(insert_sql, employees)使用服务器端游标处理大量数据# 创建服务器端游标 cursor conn.cursor(bufferedFalse) # 流式处理结果 cursor.execute(SELECT * FROM large_table) while True: batch cursor.fetchmany(1000) # 每次获取1000条 if not batch: break process_batch(batch)6. ORM框架简介虽然直接使用数据库驱动提供了最大的灵活性但对于复杂应用可以考虑使用ORM框架6.1 SQLAlchemyfrom sqlalchemy import create_engine, Column, Integer, String, Float from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # 创建引擎 engine create_engine(mysqlmysqlconnector://username:passwordlocalhost/testdb) # 声明基类 Base declarative_base() # 定义模型 class Employee(Base): __tablename__ employees id Column(Integer, primary_keyTrue) first_name Column(String(50)) last_name Column(String(50)) age Column(Integer) department Column(String(50)) salary Column(Float) # 创建表 Base.metadata.create_all(engine) # 创建会话 Session sessionmaker(bindengine) session Session() # 使用示例 new_employee Employee( first_nameJane, last_nameDoe, age32, departmentMarketing, salary68000.00 ) session.add(new_employee) session.commit() # 查询 employees session.query(Employee).filter_by(departmentIT).all() for emp in employees: print(emp.first_name, emp.last_name, emp.salary)6.2 Django ORM如果你使用Django框架它内置了强大的ORM# models.py from django.db import models class Employee(models.Model): first_name models.CharField(max_length50) last_name models.CharField(max_length50) age models.IntegerField() department models.CharField(max_length50) salary models.DecimalField(max_digits10, decimal_places2) created_at models.DateTimeField(auto_now_addTrue) # 使用示例 from yourapp.models import Employee # 创建 Employee.objects.create( first_nameMike, last_nameJohnson, age45, departmentFinance, salary92000.00 ) # 查询 it_employees Employee.objects.filter(departmentIT) for emp in it_employees: print(emp.first_name, emp.last_name)7. 实际项目中的应用模式7.1 数据访问层设计# db.py - 数据库访问层 import mysql.connector from mysql.connector import pooling from contextlib import contextmanager class Database: _pool None classmethod def initialize(cls, **config): cls._pool pooling.MySQLConnectionPool( pool_namemypool, pool_size5, **config ) classmethod contextmanager def get_connection(cls): conn cls._pool.get_connection() try: yield conn except Exception: conn.rollback() raise finally: conn.close() # employee_dao.py - 数据访问对象 from db import Database class EmployeeDAO: staticmethod def get_all(): with Database.get_connection() as conn: cursor conn.cursor(dictionaryTrue) cursor.execute(SELECT * FROM employees) return cursor.fetchall() staticmethod def get_by_id(employee_id): with Database.get_connection() as conn: cursor conn.cursor(dictionaryTrue) cursor.execute( SELECT * FROM employees WHERE id %s, (employee_id,) ) return cursor.fetchone() staticmethod def create(employee_data): with Database.get_connection() as conn: cursor conn.cursor() cursor.execute( INSERT INTO employees (first_name, last_name, age, department, salary) VALUES (%s, %s, %s, %s, %s), employee_data ) conn.commit() return cursor.lastrowid7.2 使用配置文件管理数据库连接# config.py import os from dotenv import load_dotenv load_dotenv() class Config: DB_CONFIG { user: os.getenv(DB_USER), password: os.getenv(DB_PASSWORD), host: os.getenv(DB_HOST), database: os.getenv(DB_NAME), pool_name: mypool, pool_size: 5, charset: utf8mb4 } # app.py from config import Config from db import Database # 初始化数据库连接池 Database.initialize(**Config.DB_CONFIG) # 使用示例 from employee_dao import EmployeeDAO employees EmployeeDAO.get_all() for emp in employees: print(emp[first_name], emp[last_name])7.3 异步数据库访问对于高性能应用可以使用异步MySQL驱动如aiomysqlimport asyncio import aiomysql async def fetch_data(): pool await aiomysql.create_pool( hostlocalhost, userusername, passwordpassword, dbtestdb, minsize5, maxsize10 ) async with pool.acquire() as conn: async with conn.cursor() as cursor: await cursor.execute(SELECT * FROM employees) result await cursor.fetchall() print(result) pool.close() await pool.wait_closed() # 运行 loop asyncio.get_event_loop() loop.run_until_complete(fetch_data())8. 测试与调试技巧8.1 单元测试数据库操作import unittest from mysql.connector import Error from yourmodule import EmployeeDAO class TestEmployeeDAO(unittest.TestCase): classmethod def setUpClass(cls): # 初始化测试数据库连接 cls.setup_test_database() classmethod def tearDownClass(cls): # 清理测试数据 cls.cleanup_test_database() def test_create_employee(self): test_data (Test, User, 30, QA, 50000) emp_id EmployeeDAO.create(test_data) self.assertIsNotNone(emp_id) employee EmployeeDAO.get_by_id(emp_id) self.assertEqual(employee[first_name], Test) classmethod def setup_test_database(cls): # 创建测试数据库和表 pass classmethod def cleanup_test_database(cls): # 删除测试数据 pass if __name__ __main__: unittest.main()8.2 使用数据库迁移工具对于项目演进使用迁移工具管理数据库变更安装Alembicpip install alembic初始化迁移环境alembic init migrations配置alembic.ini[alembic] script_location migrations sqlalchemy.url mysqlmysqlconnector://username:passwordlocalhost/testdb创建迁移脚本alembic revision -m create employees table编辑迁移文件def upgrade(): op.create_table( employees, sa.Column(id, sa.Integer, primary_keyTrue), sa.Column(first_name, sa.String(50), nullableFalse), # 其他列定义... ) def downgrade(): op.drop_table(employees)执行迁移alembic upgrade head8.3 性能分析与优化使用Python内置的cProfile分析数据库操作性能import cProfile import pstats from io import StringIO from yourmodule import EmployeeDAO def profile_query(): pr cProfile.Profile() pr.enable() # 要分析的数据库操作 employees EmployeeDAO.get_all() pr.disable() s StringIO() ps pstats.Stats(pr, streams).sort_stats(cumulative) ps.print_stats() print(s.getvalue()) if __name__ __main__: profile_query()9. 扩展阅读与资源推荐9.1 官方文档MySQL Connector/Python官方文档PyMySQL文档SQLAlchemy文档Python DB-API 2.0规范9.2 推荐书籍《Python数据库编程实战》《MySQL必知必会》《SQL进阶教程》《高性能MySQL》9.3 在线课程Coursera: Python Database AccessUdemy: Complete Python Developer in 2024edX: Introduction to Databases10. 个人实践经验分享在实际项目中使用Python连接MySQL多年总结出以下几点经验连接管理总是使用上下文管理器或try-finally块确保连接被正确关闭避免连接泄漏。参数化查询永远使用参数化查询而非字符串拼接这是防止SQL注入的第一道防线。连接池对于Web应用连接池是必须的可以显著提高性能。但要注意合理设置pool_size。ORM选择对于简单项目直接使用驱动即可复杂项目可以考虑SQLAlchemy等ORM但要了解其性能特点。批量操作当需要处理大量数据时使用executemany()或LOAD DATA INFILE比循环单条插入快几个数量级。错误处理正确处理数据库错误特别是事务中的错误确保失败时能正确回滚。编码问题始终使用utf8mb4字符集以支持完整的Unicode字符如emoji。测试数据为测试准备专门的数据库或使用事务回滚避免测试污染生产数据。监控为关键数据库操作添加日志和监控便于性能分析和故障排查。备份定期备份数据库并测试恢复流程。代码可以重写数据丢失可能是灾难性的。