有勇气的牛排博客
☰

PyMySQL 是什么？如何用 Python 操作 MySQL 数据库？（附详细案例）

有勇气的牛排 523 Python 2021-08-26 21:57:34
进群口令：博客

1 前言

PyMySQL 是一个纯 Python 编写的库，用于连接 MySQL 数据库并执行 SQL 操作。它完全遵循 DB-API 2.0 标准，适合用于各种 Python 应用程序中与 MySQL 数据库进行交互。由于它不依赖于底层 C 扩展，因此在环境配置和移植性方面非常方便。

安装

pip install pymysql==1.1.1

2 创建表

通用类封装，请见文章末尾。

# -*- coding: utf-8 -*-
"""
@file: test3.py
@author: 有勇气的牛排
@url: www.couragesteak.com
@description:
"""

from mysql.pymysql_test.MySQLHandler import MySQLHandler

if __name__ == "__main__":
    # 数据库配置
    db = MySQLHandler(
        host='127.0.0.1',
        port=3306,
        user='root',
        password='root',
        database='cs_test'
    )

    # 创建表
    create_table_sql = """
        CREATE TABLE IF NOT EXISTS article (
            id INT AUTO_INCREMENT PRIMARY KEY,
            ac_id INT,
            title VARCHAR(255),
            content VARCHAR(255),
            a_info VARCHAR(255)
        )
        """
    db.execute(create_table_sql)

    # 关闭数据库
    db.close()

PyMSQL创建表

3 增删改查

3.1 增加数据

3.1.1 增加多条数据

# 插入数据
insert_sql = "INSERT INTO article (ac_id, title, content, a_info) VALUES (%s, %s, %s, %s)"
users_data = [
    (1, "Python教程", '这是文章正文部分', '文章描述'),
    (1, "Python教程", '这是文章正文5%部分', '文章{}描述'),
]
db.executemany(insert_sql, users_data)

关于单条查询，自己可以探索。

3.2 删除数据

a_id = 1
delete_sql = f"DELETE FROM article WHERE id = {a_id}"
db.execute(delete_sql)

3.3 修改数据

a_id = 1
content = "这是文章正文部分, 测试5%"
update_sql = f"UPDATE article SET content='{content}' WHERE id = {a_id}"
db.execute(update_sql)

3.4 查询数据

a_id = 3
select_sql = f"SELECT * FROM article WHERE id = {a_id}"
results = db.execute(select_sql)
print("查询结果：", results)

PyMSQ查询数据

4 通用类封装

import pymysql

class MySQLHandler:
    """
    MySQL 数据库操作类
    """

    def __init__(self, host, port, user, password, database, charset='utf8mb4'):
        """
        初始化数据库连接
        """
        self.db_config = {
            'host': host,
            'port': port,
            'user': user,
            'password': password,
            'database': database,
            'charset': charset
        }
        self.connection = None
        self.connect()

    def connect(self):
        """
        建立数据库连接
        """
        try:
            self.connection = pymysql.connect(**self.db_config)
            print("数据库连接成功！")
        except Exception as e:
            print(f"数据库连接失败：{e}")
            self.connection = None

    def execute(self, sql, params=None):
        """
        执行单条 SQL 语句
        :param sql: SQL 语句
        :param params: 可选的参数列表
        :return: 受影响的行数或查询结果
        """
        if not self.connection:
            print("数据库未连接！")
            return None
        try:
            with self.connection.cursor() as cursor:
                cursor.execute(sql, params)
                if sql.strip().lower().startswith("select"):
                    return cursor.fetchall()
                else:
                    self.connection.commit()
                    return cursor.rowcount
        except Exception as e:
            print(f"执行 SQL 失败：{e}")
            self.connection.rollback()
            return None

    def executemany(self, sql, param_list):
        """
        批量执行 SQL 语句
        :param sql: SQL 语句
        :param param_list: 参数列表
        :return: 受影响的行数
        """
        if not self.connection:
            print("数据库未连接！")
            return None
        try:
            with self.connection.cursor() as cursor:
                cursor.executemany(sql, param_list)
                self.connection.commit()
                return cursor.rowcount
        except Exception as e:
            print(f"批量执行 SQL 失败：{e}")
            self.connection.rollback()
            return None

    def close(self):
        """
        关闭数据库连接
        """
        if self.connection:
            self.connection.close()
            print("数据库连接已关闭！")

5 总结

在MySQLHandler通用类上层，仍然可以进行进一步封装，比如插入数据、更新数据前传入字典数据，然后转化为sql语句，这样上层操作也会更加方便。

评论区

用户头像
有勇气的牛排 2025-08-08 10:46:41
1
×
×