详细介绍一下Python连接MySQL数据库的完整步骤

以下是 Python 连接 MySQL 数据库的完整步骤，包含环境准备、连接建立、数据操作、错误处理和性能优化等内容：
一、环境准备

1. 安装 MySQL 服务器
   Windows/macOS：下载安装包 MySQL Installer
   Linux：
   bash

Ubuntu/Debian

sudo apt-get install mysql-server

CentOS/RHEL

sudo yum install mysql-server

2. 启动 MySQL 服务
   bash

Linux/macOS

sudo service mysql start

Windows

net start mysql

3. 创建测试数据库和用户
   sql
   – 登录MySQL
   mysql -u root -p

– 创建数据库
CREATE DATABASE testdb;

– 创建用户并授权
CREATE USER ‘testuser’@‘localhost’ IDENTIFIED BY ‘testpassword’;
GRANT ALL PRIVILEGES ON testdb.* TO ‘testuser’@‘localhost’;
FLUSH PRIVILEGES;
4. 安装 Python 驱动
bash
pip install mysql-connector-python
二、建立数据库连接

1. 基本连接示例
   python
   import mysql.connector
   from mysql.connector import Error

try:
connection = mysql.connector.connect(
host=‘localhost’,
user=‘testuser’,
password=‘testpassword’,
database=‘testdb’,
port=3306 # 默认端口
)

if connection.is_connected():db_info = connection.get_server_info()print(f"连接成功，MySQL服务器版本: {db_info}")cursor = connection.cursor()cursor.execute("SELECT DATABASE();")database = cursor.fetchone()print(f"当前数据库: {database}")

except Error as e:
print(f"连接错误: {e}")

finally:
if connection.is_connected():
cursor.close()
connection.close()
print(“数据库连接已关闭”)
2. 连接参数详解
python
connection = mysql.connector.connect(
host=‘localhost’, # 主机地址
user=‘testuser’, # 用户名
password=‘testpassword’, # 密码
database=‘testdb’, # 数据库名
port=3306, # 端口号
charset=‘utf8mb4’, # 字符集
autocommit=True, # 自动提交事务
connection_timeout=10 # 连接超时时间（秒）
)
三、执行 SQL 操作

1. 创建表
   python
   try:
   connection = mysql.connector.connect(…)
   cursor = connection.cursor()

   create_table_query = “”"
   CREATE TABLE IF NOT EXISTS employees (
   id INT AUTO_INCREMENT PRIMARY KEY,
   name VARCHAR(255) NOT NULL,
   department VARCHAR(255) NOT NULL,
   salary DECIMAL(10, 2) NOT NULL,
   hire_date DATE NOT NULL
   )
   “”"

   cursor.execute(create_table_query)
   print(“表创建成功”)

except Error as e:
print(f"表创建失败: {e}")
2. 插入数据
python
try:
connection = mysql.connector.connect(…)
cursor = connection.cursor()

# 单条插入
insert_query = """
INSERT INTO employees (name, department, salary, hire_date)
VALUES (%s, %s, %s, %s)
"""
employee_data = ("John Doe", "IT", 5000.00, "2023-01-15")
cursor.execute(insert_query, employee_data)# 批量插入
employees_data = [("Jane Smith", "HR", 6000.00, "2023-02-20"),("Robert Johnson", "Finance", 5500.00, "2023-03-10")
]
cursor.executemany(insert_query, employees_data)connection.commit()  # 提交事务
print(f"插入成功，影响行数: {cursor.rowcount}")

except Error as e:
print(f"插入失败: {e}")
connection.rollback() # 回滚事务
3. 查询数据
python
try:
connection = mysql.connector.connect(…)
cursor = connection.cursor()

# 查询所有记录
select_query = "SELECT * FROM employees"
cursor.execute(select_query)
records = cursor.fetchall()
print(f"共查询到 {cursor.rowcount} 条记录")for row in records:print(f"ID: {row[0]}, 姓名: {row[1]}, 部门: {row[2]}, 薪水: {row[3]}")# 参数化查询
select_salary_query = "SELECT * FROM employees WHERE salary > %s"
cursor.execute(select_salary_query, (5000,))
high_salary_employees = cursor.fetchall()
print(f"高薪员工: {len(high_salary_employees)} 人")

except Error as e:
print(f"查询失败: {e}")
4. 更新数据
python
try:
connection = mysql.connector.connect(…)
cursor = connection.cursor()

update_query = "UPDATE employees SET salary = salary * 1.1 WHERE department = %s"
cursor.execute(update_query, ("IT",))
connection.commit()
print(f"更新成功，影响行数: {cursor.rowcount}")

except Error as e:
print(f"更新失败: {e}")
connection.rollback()
5. 删除数据
python
try:
connection = mysql.connector.connect(…)
cursor = connection.cursor()

delete_query = "DELETE FROM employees WHERE hire_date < %s"
cursor.execute(delete_query, ("2023-02-01",))
connection.commit()
print(f"删除成功，影响行数: {cursor.rowcount}")

except Error as e:
print(f"删除失败: {e}")
connection.rollback()
四、高级操作

1. 使用字典游标
   python
   cursor = connection.cursor(dictionary=True)
   cursor.execute(“SELECT * FROM employees”)
   records = cursor.fetchall()

for row in records:
print(f"姓名: {row[‘name’]}, 部门: {row[‘department’]}")
2. 事务处理
python
try:
connection = mysql.connector.connect(…)
connection.autocommit = False # 关闭自动提交

cursor = connection.cursor()# 执行多个操作
cursor.execute("INSERT INTO employees (...) VALUES (...)")
cursor.execute("UPDATE departments SET budget = budget - 10000")connection.commit()  # 提交事务
print("事务执行成功")

except Error as e:
print(f"事务失败: {e}")
connection.rollback() # 回滚事务
3. 连接池
python
from mysql.connector import pooling

创建连接池

connection_pool = pooling.MySQLConnectionPool(
pool_name=“mypool”,
pool_size=5,
host=‘localhost’,
user=‘testuser’,
password=‘testpassword’,
database=‘testdb’
)

从连接池获取连接

connection = connection_pool.get_connection()
cursor = connection.cursor()
cursor.execute(“SELECT * FROM employees”)
五、错误处理与最佳实践

1. 常见错误类型
   python
   try:
   connection = mysql.connector.connect(…)

   数据库操作

except mysql.connector.Error as e:
if e.errno == 1045: # 访问拒绝
print(“用户名或密码错误”)
elif e.errno == 1049: # 数据库不存在
print(“数据库不存在”)
elif e.errno == 2003: # 无法连接
print(“无法连接到MySQL服务器”)
else:
print(f"未知错误: {e}")
2. 资源管理
python

使用上下文管理器自动关闭连接和游标

with mysql.connector.connect(…) as connection:
with connection.cursor() as cursor:
cursor.execute(“SELECT * FROM employees”)
records = cursor.fetchall()
3. 性能优化
python

批量插入优化

data = [(f"User{i}", “Dept”, 5000) for i in range(1000)]
cursor.executemany(“INSERT INTO users VALUES (%s, %s, %s)”, data)

使用预编译语句

stmt = connection.prepare(“INSERT INTO users VALUES (?, ?, ?)”)
cursor.execute(stmt, (“User1”, “Dept”, 5000))
六、安全注意事项
避免 SQL 注入：
python

错误做法（不安全）

query = f"SELECT * FROM users WHERE name = ‘{name}’"

正确做法（使用参数化查询）

query = “SELECT * FROM users WHERE name = %s”
cursor.execute(query, (name,))

密码管理：
python

不要硬编码密码

推荐使用环境变量或配置文件

import os
password = os.environ.get(‘MYSQL_PASSWORD’)

权限最小化：
sql
– 为应用创建具有最小权限的用户
GRANT SELECT, INSERT, UPDATE ON testdb.* TO ‘appuser’@‘localhost’;