Python连接MySQL执行SQL查询的完整代码示例

一、前置准备：驱动安装与测试数据表

连接 MySQL 需要在 Python 环境中安装兼容 PEP 249 标准的数据库驱动。最常用的驱动是 PyMySQL，其优势在于纯 Python 编写，无需依赖本地 C 语言编译环境或复杂的第三方客户端动态链接库，直接通过 pip 即可完成安装：

pip install pymysql

为了让后续的代码示例能够真实运行并验证输出，假定我们在 MySQL 中创建了一个名为 demo_db 的数据库，并有一张员工表 employees。其结构与测试数据可以通过以下 SQL 语句进行初始化：

-- 创建测试数据库并指定字符集为 utf8mb4
CREATE DATABASE IF NOT EXISTS demo_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE demo_db;

-- 创建员工信息表
CREATE TABLE IF NOT EXISTS employees (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(50) NOT NULL,
    department VARCHAR(50) NOT NULL,
    salary DECIMAL(10, 2) NOT NULL,
    hire_date DATE NOT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 插入基础测试数据
INSERT INTO employees (name, department, salary, hire_date) VALUES
('张伟', '技术部', 15000.00, '2022-03-15'),
('李娜', '运营部', 9500.00, '2021-07-01'),
('王强', '技术部', 18500.00, '2020-11-20'),
('赵敏', '产品部', 13000.00, '2023-02-10');

二、标准操作流程与生命周期

一次完整的数据库查询生命周期应严格保证“用后即关”，防止连接句柄挂起导致数据库连接池耗尽。标准步骤包括：定义配置字典、建立连接、获取游标、执行带参查询、读取数据、关闭游标与连接。

三、核心代码实现：基础查询与上下文管理器

在编写生产级代码时，推荐使用 Python 的 with 上下文管理器来管理游标和连接，它可以在代码执行完毕或中途发生异常时自动触发 close()，避免资源泄漏。下面是推荐的完整查询脚本：

# -*- coding: utf-8 -*-
import pymysql
from pymysql.cursors import DictCursor

# 1. 定义数据库连接参数字典
db_config = {
    "host": "127.0.0.1",        # 数据库主机IP或域名
    "port": 3306,               # MySQL 服务端口，默认 3306
    "user": "root",             # 数据库登录用户名
    "password": "YourPassword", # 数据库登录密码
    "database": "demo_db",      # 目标数据库名
    "charset": "utf8mb4",       # 务必指定 utf8mb4，兼容完整中文字符与Emoji
    "cursorclass": DictCursor,  # 使用字典游标，使查询结果以字典键值对返回
    "connect_timeout": 10       # 连接超时时间（秒）
}

def query_employees_by_department(dept_name, min_salary):
    """
    查询指定部门且薪资高于指定金额的员工记录
    """
    connection = None
    try:
        # 2. 建立数据库物理连接
        connection = pymysql.connect(**db_config)
        
        # 3. 使用 with 语句创建游标对象（执行完毕后自动关闭游标）
        with connection.cursor() as cursor:
            # 4. 编写带占位符的 SQL 模板，严禁使用 % 或 + 拼接字符串
            sql = """
                SELECT id, name, department, salary, hire_date
                FROM employees
                WHERE department = %s AND salary >= %s
                ORDER BY salary DESC
            """
            params = (dept_name, min_salary)

            # 5. 执行参数化查询
            rows_matched = cursor.execute(sql, params)
            print(f"[INFO] 查询命中记录数: {rows_matched}")

            # 6. 获取查询结果
            results = cursor.fetchall()
            return results

    except pymysql.MySQLError as err:
        print(f"[ERROR] 数据库操作异常: {err}")
        return []
    finally:
        # 7. 在 finally 块中确保连接无论如何都被安全关闭
        if connection and connection.open:
            connection.close()
            print("[INFO] 数据库连接已安全关闭。")

# 示例调用
if __name__ == "__main__":
    dept = "技术部"
    salary_threshold = 16000.00
    print(f"正在查询【{dept}】部门中薪资大于等于 {salary_threshold} 的员工...")
    
    emp_list = query_employees_by_department(dept, salary_threshold)
    for emp in emp_list:
        # 因为使用了 DictCursor，每一行都是可以直接按列名索引的字典
        print(f"ID: {emp['id']} | 姓名: {emp['name']} | 薪资: {emp['salary']} | 入职: {emp['hire_date']}")

运行输出（示例输出）：

正在查询【技术部】部门中薪资大于等于 16000.0 的员工...
[INFO] 查询命中记录数: 1
ID: 3 | 姓名: 王强 | 薪资: 18500.00 | 入职: 2020-11-20
[INFO] 数据库连接已安全关闭。

四、结果集读取模式：fetchone、fetchall 与 fetchmany

执行 cursor.execute() 成功后，数据已从 MySQL 数据库传输到客户端的通信缓存中。此时必须通过游标的获取方法读取行数据。不同方法适用于不同的场景与数据量级：

例如，对于批量数据处理，使用 fetchmany 分批循环迭代的代码模式如下：

with connection.cursor() as cursor:
    cursor.execute("SELECT id, name FROM employees")
    batch_size = 2
    while True:
        rows = cursor.fetchmany(batch_size)
        if not rows:
            break  # 所有批次读取完毕
        for row in rows:
            print(f"处理员工: {row['name']}")

五、安全红线：参数化查询与 SQL 注入防范

在任何需要代入外部变量（如表单输入、URL 参数）的 SQL 查询中，严禁使用 Python 的 % 操作符、format() 方法或 f-string 字符串插值拼接 SQL。

字符串拼接将用户输入直接融入语法解析，攻击者可输入 ' OR '1'='1 篡改逻辑。而参数化查询将 SQL 语句和参数作为独立单元传递给驱动，驱动层会自动对参数进行严格转义并作为纯字面量（Literal Value）处理，从根本上杜绝语法逻辑被篡改。

六、生产环境常见问题与排错指南

在实际部署与运行 Python 连接 MySQL 代码时，常见的报错原因和排查方式如下：

• 报错 OperationalError: (1045, "Access denied for user...")：
  账号或密码不正确，或者该账号未被允许从当前客户端 IP 登录。可登录 MySQL 服务器检查 mysql.user 表中的 host 字段配置。
• 报错 OperationalError: (2003, "Can't connect to MySQL server...")：
  通常是网络不通、MySQL 服务未启动、服务器防火墙阻挡了 3306 端口，或者 MySQL 配置中的 bind-address 被限制为 127.0.0.1 导致无法远程访问。
• 中文乱码或 Emoji 存储截断报错 Incorrect string value：
  连接参数中的 charset 务必明确设置为 utf8mb4。MySQL 历史上的 utf8 仅支持最多 3 字节字符，当遇到 4 字节的 Emoji 或生僻汉字时会报错。
• 关于 conn.commit() 的调用时机：
  执行 SELECT 查询操作时，属于只读操作，不需要调用 connection.commit()；只有执行 INSERT、UPDATE、DELETE 等涉及写事务的操作时，才必须显式调用 conn.commit() 提交事务，否则数据更改不会持久化到磁盘。

七、总结与生产速查

掌握 Python 与 MySQL 的查询交互，关键在于建立规范的编码习惯：严格采用参数化传递防止安全隐患、根据查询规模合理挑选获取模式、并在上下文管理器中完备处理连接与游标的生命周期。