昨天 17:23 发布于广东

关注

Python 链接数据库与基础增删改查（CRUD）操作详解

在 Python 开发中，数据库交互是后端开发、数据处理的核心能力 —— 无论是 Web 应用的用户数据存储、数据分析中的数据读写，还是自动化脚本的数据持久化，都需要通过 Python 与数据库建立连接并执行操作。本文以 MySQL 数据库 为例，从环境准备、数据库连接、基础 CRUD（创建 / 读取 / 更新 / 删除）操作三个维度，手把手教你实现 Python 与数据库的交互，适合 Python 初学者快速入门。

一、环境准备：搭建基础开发环境

在编写代码前，需确保本地已配置好以下工具和依赖，避免后续出现兼容性或依赖缺失问题。

1.1 核心工具清单

Python 解释器	执行 Python 代码	3.8 及以上（LTS）
MySQL 数据库	提供数据存储服务的关系型数据库	5.7 或 8.0
代码编辑器	编写 Python 代码（如 VS Code、PyCharm）	任意支持 Python 的编辑器
MySQL 驱动包	Python 与 MySQL 通信的桥梁（ `pymysql` ）	最新稳定版

1.2 依赖安装（核心步骤）

Python 连接 MySQL 需安装第三方驱动包 pymysql（轻量、易用，支持 Python 3.x），通过 pip 快速安装：

bash

运行

# 打开命令行，执行以下命令
pip install pymysql

验证安装：安装完成后，在命令行输入 python -c "import pymysql"，若无报错则说明安装成功。

1.3 提前创建数据库和表

为测试 CRUD 操作，先在 MySQL 中创建示例数据库和表（以「用户表 user」为例）：

打开 MySQL 管理工具（如 Navicat、phpMyAdmin，或 MySQL 命令行）。
执行以下 SQL 语句创建数据库和表：

sql

-- 创建数据库（若不存在），指定编码为 utf8mb4（支持中文和表情）
CREATE DATABASE IF NOT EXISTS python_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE python_db; -- 切换到该数据库

-- 创建用户表（user）
CREATE TABLE IF NOT EXISTS user (
    id INT PRIMARY KEY AUTO_INCREMENT, -- 自增主键（唯一标识用户）
    username VARCHAR(50) NOT NULL,     -- 用户名（非空）
    age INT COMMENT '年龄（可选）',     -- 年龄（允许为 NULL）
    email VARCHAR(100) UNIQUE NOT NULL,-- 邮箱（唯一且非空，避免重复注册）
    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP -- 创建时间（自动填充）
);

二、Python 链接数据库：核心流程与工具类封装

Python 通过 pymysql 库实现与 MySQL 的连接，核心流程为：建立连接 → 创建游标 → 执行 SQL → 处理结果 → 关闭资源。为避免重复编写连接逻辑，我们封装一个 DBUtil 工具类，统一管理连接配置和资源释放。

2.1 封装数据库连接工具类

创建 db_util.py 文件，代码如下：

python

运行

import pymysql
from pymysql.err import OperationalError, ProgrammingError

class DBUtil:
    # 数据库配置（根据你的本地环境修改）
    CONFIG = {
        'host': 'localhost',    # 数据库主机（默认 localhost）
        'port': 3306,           # MySQL 端口（默认 3306）
        'user': 'root',         # MySQL 用户名（默认 root）
        'password': '123456',   # MySQL 密码（根据你的实际密码修改，XAMPP 默认为空）
        'database': 'python_db',# 数据库名（已创建的 python_db）
        'charset': 'utf8mb4'    # 编码（与数据库一致）
    }

    @staticmethod
    def get_connection():
        """获取数据库连接对象"""
        connection = None
        try:
            # 建立数据库连接
            connection = pymysql.connect(**DBUtil.CONFIG)
            print("数据库连接成功！")
        except OperationalError as e:
            print(f"数据库连接失败：{e}")
        return connection

    @staticmethod
    def close_resource(connection, cursor):
        """关闭资源（游标 → 连接，后创建的先关闭）"""
        try:
            if cursor:
                cursor.close()
            if connection:
                connection.close()
                print("数据库连接已关闭！")
        except Exception as e:
            print(f"关闭资源失败：{e}")

# 测试连接（可选，运行该文件可验证连接是否正常）
if __name__ == "__main__":
    conn = DBUtil.get_connection()
    DBUtil.close_resource(conn, None)

2.2 核心说明

连接配置：CONFIG 字典中的 password 需根据你的本地 MySQL 密码修改（如 XAMPP 默认为空，可改为 ''）。
异常处理：捕获 OperationalError（连接相关错误，如密码错误、端口错误），便于排查问题。
资源关闭：游标（cursor）和连接（connection）是稀缺资源，必须关闭，避免内存泄漏。

三、基础 CRUD 操作实现

以下所有操作均基于 DBUtil 工具类，通过「游标执行 SQL」实现 CRUD，同时使用 参数化查询 避免 SQL 注入风险。

3.1 新增操作（Create）：插入用户数据

需求：向 user 表插入一条用户数据（username=「张三」，age=25，email=「zhangsan@xxx.com」）。

创建 create_user.py 文件，代码如下：

python

运行

from db_util import DBUtil

def add_user(username, age, email):
    """
    新增用户
    :param username: 用户名
    :param age: 年龄
    :param email: 邮箱
    :return: 成功返回新增用户的 id，失败返回 None
    """
    connection = DBUtil.get_connection()
    cursor = None
    user_id = None
    try:
        if not connection:
            return None
        
        # 创建游标（默认返回元组类型，可指定 cursorclass=pymysql.cursors.DictCursor 返回字典）
        cursor = connection.cursor()
        
        # SQL 插入语句（%s 是参数占位符，并非字符串格式化，避免 SQL 注入）
        sql = """INSERT INTO user (username, age, email) 
                 VALUES (%s, %s, %s)"""
        
        # 执行 SQL（参数需传入元组格式）
        affected_rows = cursor.execute(sql, (username, age, email))
        
        # 提交事务（插入/更新/删除操作需手动提交，查询无需）
        connection.commit()
        
        if affected_rows > 0:
            # 获取新增记录的自增 id
            user_id = cursor.lastrowid
            print(f"新增用户成功！用户 ID：{user_id}")
        else:
            print("新增用户失败！")
    except ProgrammingError as e:
        # 捕获 SQL 语法错误、表不存在等问题
        print(f"新增用户失败（SQL 错误）：{e}")
        connection.rollback()  # 出错时回滚事务
    finally:
        # 关闭资源
        DBUtil.close_resource(connection, cursor)
    return user_id

# 测试新增操作
if __name__ == "__main__":
    add_user("张三", 25, "zhangsan@xxx.com")

验证结果：运行脚本后，若输出「新增用户成功！用户 ID：1」，可在 MySQL 中查询 user 表，确认数据已插入。

3.2 查询操作（Read）：查询用户数据

需求：两种常见查询场景：

单条查询：根据 id 查询指定用户。
批量查询：查询所有用户或符合条件的用户。

3.2.1 单条查询（按 ID 查询）

创建 query_user_by_id.py 文件，代码如下：

python

运行

from db_util import DBUtil

def get_user_by_id(user_id):
    """
    根据 ID 查询用户
    :param user_id: 用户 ID
    :return: 成功返回用户字典，失败返回 None
    """
    connection = DBUtil.get_connection()
    cursor = None
    user = None
    try:
        if not connection:
            return None
        
        # 创建游标，指定返回字典类型（便于通过字段名获取数据）
        cursor = connection.cursor(cursor=pymysql.cursors.DictCursor)
        
        # SQL 查询语句
        sql = """SELECT id, username, age, email, create_time 
                 FROM user WHERE id = %s"""
        
        # 执行 SQL
        cursor.execute(sql, (user_id,))  # 参数元组只有一个元素时，末尾需加逗号
        
        # 获取单条结果（fetchone() 返回一条记录，fetchall() 返回所有记录）
        user = cursor.fetchone()
        
        if user:
            print(f"查询到用户：{user}")
        else:
            print(f"未查询到 ID 为 {user_id} 的用户！")
    except Exception as e:
        print(f"查询用户失败：{e}")
    finally:
        DBUtil.close_resource(connection, cursor)
    return user

# 测试查询操作（查询 ID=1 的用户）
if __name__ == "__main__":
    get_user_by_id(1)

3.2.2 批量查询（查询所有用户）

创建 query_all_users.py 文件，代码如下：

python

运行

from db_util import DBUtil

def get_all_users():
    """
    查询所有用户
    :return: 成功返回用户列表（字典组成的列表），失败返回空列表
    """
    connection = DBUtil.get_connection()
    cursor = None
    user_list = []
    try:
        if not connection:
            return user_list
        
        cursor = connection.cursor(cursor=pymysql.cursors.DictCursor)
        
        # SQL 查询所有用户（按 id 降序排列）
        sql = """SELECT id, username, age, email, create_time 
                 FROM user ORDER BY id DESC"""
        
        cursor.execute(sql)  # 无参数时直接执行
        
        # 获取所有结果（fetchall() 返回所有匹配的记录）
        user_list = cursor.fetchall()
        
        if user_list:
            print(f"查询到 {len(user_list)} 个用户：")
            for user in user_list:
                print(user)
        else:
            print("暂无用户数据！")
    except Exception as e:
        print(f"查询所有用户失败：{e}")
    finally:
        DBUtil.close_resource(connection, cursor)
    return user_list

# 测试批量查询
if __name__ == "__main__":
    get_all_users()

3.3 更新操作（Update）：修改用户数据

需求：将 id=1 的用户年龄修改为 26，邮箱修改为「zhangsan_update@xxx.com」。

创建 update_user.py 文件，代码如下：

python

运行

from db_util import DBUtil

def update_user(user_id, update_data):
    """
    更新用户信息
    :param user_id: 用户 ID
    :param update_data: 要更新的字段（字典格式，如 {'age':26, 'email':'xxx'}）
    :return: 成功返回 True，失败返回 False
    """
    connection = DBUtil.get_connection()
    cursor = None
    is_success = False
    try:
        if not connection or not update_data:
            return is_success
        
        cursor = connection.cursor()
        
        # 动态拼接更新字段（避免字段固定，提高复用性）
        update_fields = []
        params = []
        for key, value in update_data.items():
            update_fields.append(f"{key} = %s")
            params.append(value)
        
        # 拼接 SQL 语句（最后加上 WHERE 条件，避免全表更新！）
        sql = f"UPDATE user SET {', '.join(update_fields)} WHERE id = %s"
        params.append(user_id)  # 将 user_id 加入参数列表
        
        # 执行 SQL
        affected_rows = cursor.execute(sql, tuple(params))
        connection.commit()  # 更新操作需提交事务
        
        if affected_rows > 0:
            print(f"更新用户成功！受影响行数：{affected_rows}")
            is_success = True
        else:
            print(f"更新失败（可能 ID 不存在或数据未变更）！")
    except Exception as e:
        print(f"更新用户失败：{e}")
        connection.rollback()  # 出错回滚事务
    finally:
        DBUtil.close_resource(connection, cursor)
    return is_success

# 测试更新操作（修改 ID=1 的用户年龄和邮箱）
if __name__ == "__main__":
    update_data = {
        'age': 26,
        'email': **********'
    }
    update_user(1, update_data)

3.4 删除操作（Delete）：删除用户数据

需求：删除 id=1 的用户数据。

创建 delete_user.py 文件，代码如下：

python

运行

from db_util import DBUtil

def delete_user(user_id):
    """
    根据 ID 删除用户
    :param user_id: 用户 ID
    :return: 成功返回 True，失败返回 False
    """
    connection = DBUtil.get_connection()
    cursor = None
    is_success = False
    try:
        if not connection:
            return is_success
        
        cursor = connection.cursor()
        
        # SQL 删除语句（务必加 WHERE 条件，避免全表删除！）
        sql = "DELETE FROM user WHERE id = %s"
        
        # 执行 SQL
        affected_rows = cursor.execute(sql, (user_id,))
        connection.commit()  # 删除操作需提交事务
        
        if affected_rows > 0:
            print(f"删除用户成功！受影响行数：{affected_rows}")
            is_success = True
        else:
            print(f"删除失败（可能 ID 不存在）！")
    except Exception as e:
        print(f"删除用户失败：{e}")
        connection.rollback()  # 出错回滚事务
    finally:
        DBUtil.close_resource(connection, cursor)
    return is_success

# 测试删除操作（删除 ID=1 的用户）
if __name__ == "__main__":
    delete_user(1)

四、常见问题与注意事项

连接失败排查：确认 MySQL 已启动（Windows 用 net start mysql，Linux 用 systemctl status mysql，XAMPP 直接启动 MySQL 服务）。检查 db_util.py 中的 CONFIG 配置：user（默认 root）、password（你的 MySQL 密码）、database（python_db）是否正确。若报错「Unknown database 'python_db'」，需先执行创建数据库的 SQL。
SQL 注入防护：严禁直接拼接用户输入到 SQL 中（如 sql = f"SELECT * FROM user WHERE username = '{username}'"），恶意用户可输入 ' OR 1=1 # 绕过验证。必须使用参数化查询（execute(sql, params)），%s 是占位符，pymysql 会自动对参数进行转义，彻底避免 SQL 注入。
事务处理：插入、更新、删除操作需手动执行 connection.commit() 提交事务，否则数据不会写入数据库。若操作过程中报错，需执行 connection.rollback() 回滚事务，避免数据不一致。
编码问题（中文乱码）：数据库创建时指定 utf8mb4 编码（已配置）。db_util.py 中 charset 设为 utf8mb4（与数据库一致）。若查询结果中文乱码，可在 SQL 语句前添加 SET NAMES utf8mb4;。
游标使用技巧：cursor=pymysql.cursors.DictCursor：返回字典类型结果，可通过字段名（如 user['username']）获取数据，比元组更直观。fetchone()：获取一条记录，适用于单条查询。fetchall()：获取所有记录，适用于批量查询（数据量大时建议分页查询）。

五、进阶方向

本文讲解的是 Python 原生 pymysql 操作，实际开发中可通过框架简化代码，提升效率：

SQLAlchemy：Python 主流 ORM（对象关系映射）框架，将数据库表映射为 Python 类，无需编写 SQL 即可实现 CRUD（如 User(name='张三').save()）。
Django ORM：Django 框架内置的 ORM 工具，简化数据库操作，适合 Web 开发（如 User.objects.create(username='张三')、User.objects.get(id=1)）。
PyMySQL 进阶：学习连接池（如 DBUtils.PooledDB），避免频繁创建 / 关闭连接，提升高并发场景下的性能。

若你已掌握本文的基础操作，可进一步学习 ORM 框架和数据库进阶知识（如事务、索引、分页查询），应对复杂业务场景。

至此，Python 链接数据库与基础 CRUD 操作已全部讲解完毕。建议动手实践每一步，遇到报错时对照「常见问题」排查，逐步理解 Python 与数据库交互的核心逻辑。

www.khp6v.ipluskids.com

www.9nudn.ipluskids.com

look.gdpingjia.com

ocr.gdpingjia.com

ptd.gdpingjia.com

bn.gdpingjia.com

vl1z0.aijiu520.com

wcrdkx.aijiu520.com

5940s.aijiu520.com

e6tb.aijiu520.com