菜鸟教程 – 学的不仅是技术，更是梦想！

技术网站，更是交流网站

Python 3 教程

【超详细教程】Python 连接 MySQL 全流程实战

likuolei

【超详细教程】Python 连接 MySQL 全流程实战（2026 最新版）

Python 连接 MySQL 是数据处理、Web 开发、自动化脚本中最常见的场景之一。本教程基于 Python 3.10+ 和 MySQL 8.0+，从零开始一步步带你实战。无论你是初学者还是有经验的开发者，这里都会覆盖从安装到高级操作的全流程。

我会优先使用 mysql-connector-python（MySQL 官方驱动），因为它纯 Python 实现、易用、安全。备选 pymysql（社区版，轻量级）。

注意：

  • 本教程假设你有基本的 Python 知识。
  • 操作数据库需谨慎，避免在生产环境测试。
  • 示例代码已测试通过（基于本地 MySQL 环境）。

一、准备工作（环境搭建）

1. 安装 MySQL 数据库

  • Windows/Mac/Linux：下载 MySQL Community Server（免费版）从官网 mysql.com。
  • Windows：用 MSI 安装器，一路下一步，记住 root 密码。
  • Mac：用 Homebrew brew install mysql。
  • Linux（Ubuntu）：sudo apt update && sudo apt install mysql-server。
  • 启动 MySQL 服务：
  • Windows：服务管理器启动 “MySQL80″。
  • Mac/Linux：mysql.server start 或 sudo systemctl start mysql。
  • 测试安装：命令行运行 mysql -u root -p，输入密码登录成功即 OK。

2. 创建测试数据库和用户（安全起见，不要用 root）

  • 登录 MySQL：mysql -u root -p
  • 执行 SQL：
  CREATE DATABASE test_db;  -- 创建数据库

  CREATE USER 'test_user'@'localhost' IDENTIFIED BY 'your_password';  -- 创建用户

  GRANT ALL PRIVILEGES ON test_db.* TO 'test_user'@'localhost';  -- 授予权限

  FLUSH PRIVILEGES;  -- 刷新权限
  • 退出：EXIT;

3. 安装 Python MySQL 驱动

  • 推荐：mysql-connector-python（官方，纯 Python，无需额外依赖）
  pip install mysql-connector-python
  • 备选：pymysql（轻量级，纯 Python）
  pip install pymysql

二、连接 MySQL 数据库（核心步骤）

1. 基本连接代码（mysql-connector-python 版）

import mysql.connector
from mysql.connector import Error

try:
    # 连接参数（替换为你的实际值）
    connection = mysql.connector.connect(
        host='localhost',       # MySQL 主机，通常 localhost
        database='test_db',     # 数据库名
        user='test_user',       # 用户名
        password='your_password'  # 密码
    )

    if connection.is_connected():
        db_info = connection.get_server_info()
        print(f"成功连接到 MySQL 服务器版本：{db_info}")

        # 获取游标（用于执行 SQL）
        cursor = connection.cursor()
        cursor.execute("SELECT DATABASE();")
        record = cursor.fetchone()
        print(f"当前数据库：{record[0]}")

except Error as e:
    print(f"连接失败：{e}")

finally:
    # 关闭连接（必须！）
    if connection.is_connected():
        cursor.close()
        connection.close()
        print("连接已关闭")
  • 解释：
  • connect()：建立连接，参数包括 host、port（默认3306）、user、password、database。
  • is_connected()：检查连接是否成功。
  • cursor()：创建游标对象，用于执行 SQL 和获取结果。
  • fetchone() / fetchall()：获取单条/所有结果。
  • finally 块：确保连接关闭，避免资源泄漏。
  • 运行输出示例：
  成功连接到 MySQL 服务器版本：8.0.35
  当前数据库：test_db
  连接已关闭

2. 备选：pymysql 版连接（更轻量，类似）

import pymysql
from pymysql import Error

try:
    connection = pymysql.connect(
        host='localhost',
        database='test_db',
        user='test_user',
        password='your_password'
    )

    print(f"成功连接到 MySQL 服务器版本：{connection.server_version}")

    with connection.cursor() as cursor:
        cursor.execute("SELECT DATABASE();")
        record = cursor.fetchone()
        print(f"当前数据库：{record[0]}")

except Error as e:
    print(f"连接失败：{e}")

finally:
    if connection.open:
        connection.close()
        print("连接已关闭")
  • 差异：pymysql 支持上下文管理器（with），更 Pythonic。

3. 常见连接问题及解决

  • 错误 1045：用户名/密码错 → 检查 GRANT 权限。
  • 错误 2003：无法连接主机 → 检查 MySQL 服务是否启动，防火墙是否允许 3306 端口。
  • 错误 1049：数据库不存在 → 先创建数据库。
  • 远程连接：host=’远程 IP’，MySQL 需配置远程访问（修改 bind-address，GRANT TO ‘user’@’%’）。

三、数据库操作实战（CRUD 全流程）

假设我们创建一个“员工表” employees（id, name, age, salary）。

1. 创建表（Create）

import mysql.connector
from mysql.connector import Error

def create_table():
    try:
        connection = mysql.connector.connect(
            host='localhost', database='test_db', user='test_user', password='your_password'
        )

        cursor = connection.cursor()

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

        cursor.execute(create_table_query)
        print("员工表创建成功！")

    except Error as e:
        print(f"创建表失败：{e}")

    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()

create_table()

2. 插入数据（Insert）

  • 单条插入：
  insert_query = "INSERT INTO employees (name, age, salary) VALUES (%s, %s, %s)"
  employee = ("张三", 30, 5000.00)
  cursor.execute(insert_query, employee)
  connection.commit()  # 提交事务（必须！否则不保存）
  print(f"插入记录 ID：{cursor.lastrowid}")
  • 批量插入：
  employees = [
      ("李四", 25, 4500.50),
      ("王五", 28, 6000.00)
  ]
  cursor.executemany(insert_query, employees)
  connection.commit()
  print(f"批量插入 {cursor.rowcount} 条记录")

3. 查询数据（Read）

select_query = "SELECT * FROM employees WHERE age > %s"
cursor.execute(select_query, (25,))
results = cursor.fetchall()  # 或 fetchone() / fetchmany(size=5)

for row in results:
    print(row)  # 输出如：(1, '张三', 30, Decimal('5000.00'))
  • 带列名查询（字典形式）：
  cursor = connection.cursor(dictionary=True)
  cursor.execute("SELECT * FROM employees")
  for row in cursor.fetchall():
      print(row['name'])  # 直接用键访问

4. 更新数据（Update）

update_query = "UPDATE employees SET salary = salary + %s WHERE name = %s"
cursor.execute(update_query, (500.00, "张三"))
connection.commit()
print(f"更新 {cursor.rowcount} 条记录")

5. 删除数据（Delete）

delete_query = "DELETE FROM employees WHERE age < %s"
cursor.execute(delete_query, (26,))
connection.commit()
print(f"删除 {cursor.rowcount} 条记录")

6. 完整 CRUD 示例函数

def full_crud_demo():
    connection = mysql.connector.connect(host='localhost', database='test_db', user='test_user', password='your_password')
    cursor = connection.cursor()

    # Insert
    cursor.execute("INSERT INTO employees (name, age, salary) VALUES ('赵六', 35, 7000.00)")
    connection.commit()

    # Read
    cursor.execute("SELECT * FROM employees")
    print("所有员工：")
    for row in cursor.fetchall():
        print(row)

    # Update
    cursor.execute("UPDATE employees SET age = 36 WHERE name = '赵六'")
    connection.commit()

    # Delete
    cursor.execute("DELETE FROM employees WHERE name = '赵六'")
    connection.commit()

    cursor.close()
    connection.close()

full_crud_demo()

四、错误处理与最佳实践

1. 错误处理

  • 用 try-except Error 捕获。
  • 常见：ProgrammingError（SQL 语法错）、OperationalError（连接问题）。
  • 回滚：connection.rollback()（事务失败时）。

2. 最佳实践

  • 参数化查询：用 %s 占位符防 SQL 注入。
  • 上下文管理器：用 with connection.cursor() as cursor: 自动关闭游标。
  • 连接池：大数据场景用 mysql.connector.pooling：
  from mysql.connector.pooling import MySQLConnectionPool
  pool = MySQLConnectionPool(pool_name="mypool", pool_size=5, host='localhost', ...)
  conn = pool.get_connection()
  • 性能优化：批量操作用 executemany；大结果集用 buffered=True。
  • ORM 替代：复杂项目用 SQLAlchemy 或 Django ORM。
  • 安全：生产环境用环境变量存密码（os.getenv(‘DB_PASSWORD’)）。

五、常见问题 FAQ

  • Unicode 问题：加 charset='utf8mb4' 到 connect()。
  • 时区问题：MySQL 设置 SET time_zone = '+08:00';。
  • 大文件导入：用 pandas + to_sql() 或 LOAD DATA INFILE。
  • 异步连接：用 aiomysql（asyncio 版 pymysql）。

六、扩展：结合 pandas 操作 MySQL

import pandas as pd
import mysql.connector

connection = mysql.connector.connect(...)  # 如上

# 查询到 DataFrame
df = pd.read_sql("SELECT * FROM employees", connection)
print(df)

# DataFrame 写回 MySQL（需 sqlalchemy）
from sqlalchemy import create_engine
engine = create_engine('mysql+mysqlconnector://test_user:your_password@localhost/test_db')
df.to_sql('new_table', engine, if_exists='replace', index=False)

掌握这些，你就能轻松用 Python 操作 MySQL 了！如果有具体场景（如 Web 应用集成、数据迁移），或代码报错，欢迎提供细节，我帮你调试。

网站 http://yingjuxia.com
文章已创建 4424

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

相关文章

开始在上面输入您的搜索词，然后按回车进行搜索。按ESC取消。

返回顶部