pymysql

除了使用图形化工具以外，我们也可以使用编程语言来执行SQL从而操作数据库。

在Python中，使用第三方库：pymysql 来完成对MySQL数据库的操作。

安装： pip install pymysql

 或在pycharm中搜索pymysql插件安装

创建到MySQL的数据库链接

from pymysql import Connection
# 构建到MySQL数据库的链接
conn = Connection(
    host="localhost",   # 主机名IP
    port=3306,          # 端口
    user="root",        # 账户
    password="123123"  # 账号密码

)

# 打印MySQL数据库软件信息
print(conn.get_server_info())

# 关闭到数据库的链接
conn.close()

执行SQL语句

from pymysql import Connection
# 构建到MySQL数据库的链接
conn = Connection(
    host="localhost",   # 主机名IP
    port=3306,          # 端口
    user="root",        # 账户
    password="123123"  # 账号密码

)
# 打印MySQL数据库软件信息
# print(conn.get_server_info())
# 执行非查询性质SQL
cursor = conn.cursor()      # 获取到游标对象
# 选择数据库
conn.select_db("test")
# 执行SQL
cursor.execute("insert into student  values( id,'周杰伦', 31) ")    # 执行添加数据的方法
# 通过commit 确认
# conn.commit()       # 提交

# result = cursor.fetchall()      # 获取表中的数据
# print(result)
# for a in result:
#     print(a)

# 关闭链接
conn.close()

commit提交

经过执行是无法将数据插入到数据表student中的。

这是为什么呢？

pymysql在执行数据插入或其它产生数据更改的SQL语句时，默认是需要提交更改的，即，需要通过代码“确认”这种更改行为。

通过链接对象.commit() 即可确认此行为。

自动commit

如果不想手动commit确认，可以在构建链接对象的时候，设置自动commit的属性。

from pymysql import Connection
# 构建到MySQL数据库的链接
conn = Connection(
    host="localhost",   # 主机名IP
    port=3306,          # 端口
    user="root",        # 账户
    password="123123",  # 账号密码
    autocommit=True     # 设置自动提交
)

案例：

要求：

我们使用《面向对象》章节案例中的数据集，完成使用Python语言，读取数据，并将数据写入MySQL的功能

 

 

 

 

文件读取及数据处理

file_define.py

"""
和文件相关的类定义
"""
import json

from data_define import Record


# 先定义一个抽象类用来做顶层设计，确定有哪些功能要实现
class FileReader:
    def read_data(self) -> list[Record]:
        """
        读取文件的数据，读到的每一条数据都转为Record对象，
        将它们都封装到list内返回即可
        :return:
        """
        pass


class TextFileReader(FileReader):
    def __init__(self, path):
        self.path = path  # 定义成员变量记录文件的路径

    # 复写（实现抽象方法）父类的方法
    def read_data(self) -> list[Record]:
        f = open(self.path, "r", encoding="UTF-8")
        read_list: list[Record] = []
        for line in f.readlines():
            line = line.strip()  # 消除读取到的每一行数据中的\n
            data_list = line.split(",")
            record = Record(data_list[0], data_list[1], int(data_list[2]), data_list[3])
            read_list.append(record)
        f.close()
        return read_list


class JsonFileReader(FileReader):
    def __init__(self, path):
        self.path = path  # 定义成员变量记录文件的路径

    # 复写（实现抽象方法）父类的方法
    def read_data(self) -> list[Record]:
        f = open(self.path, "r", encoding="UTF-8")
        read_list: list[Record] = []
        for line in f.readlines():
            data_dict = json.loads(line)
            record = Record(data_dict["date"], data_dict["order_id"], int(data_dict["money"]), data_dict["province"])
            read_list.append(record)
        f.close()
        return read_list


if __name__ == '__main__':
    text_file_reader = TextFileReader("D:/2011年1月销售数据.txt")
    json_file_reader = JsonFileReader("D:/2011年2月销售数据JSON.txt")
    list1 = text_file_reader.read_data()
    list2 = json_file_reader.read_data()

    for l in list1:
        print(l)

    for l in list2:
        print(l)

数据定义

data_define.py

"""
数据定义的类
"""


class Record:
    def __init__(self, date, order, money, province):
        self.date = date            # 订单日期
        self.order = order          # 订单ID
        self.money = money          # 订单金额
        self.province = province    # 订单省份

    def __str__(self):  # 将地址数据转换为字符串
        return f"{self.date}, {self.order}, {self.money}, {self.province}"

将数据存入数据库

main.py

from file_define import FileReader, JsonFileReader, TextFileReader
from data_define import Record
from pymysql import Connection
import json
text_file_reader = TextFileReader("D:/2011年1月销售数据.txt")
json_file_reader = JsonFileReader("d:/2011年2月销售数据JSON.txt")

jan_data: list[Record] = text_file_reader.read_data()
feb_data: list[Record] = json_file_reader.read_data()

# 将俩个月份的数据合并为一个list来存储
all_data: list[Record] = jan_data + feb_data

# 构建MySQL链接对象
conn = Connection(
    host="localhost",   # 主机名IP
    port=3306,          # 端口
    user="root",        # 账户
    password="123123",  # 账号密码
    autocommit=True     # 提交事务（自动提交）
)
# 执行非查询性质SQL
cursor = conn.cursor()      # 获取到游标对象
# 选择数据库
conn.select_db("py_sql")
# 组织SQL语句
# 插入数据
for record in all_data:
    sql = f"insert into orders(order_date, order_id, money, province)"\
          f"values('{record.date}', '{record.order}', '{record.money}', '{record.province}')"
    # 执行SQL语句
    cursor.execute(sql)

# 关闭MySQL链接对象
conn.close()

从数据库读取数据并转为json格式

main.py

from file_define import FileReader, JsonFileReader, TextFileReader
from data_define import Record
from pymysql import Connection
import json
text_file_reader = TextFileReader("D:/2011年1月销售数据.txt")
json_file_reader = JsonFileReader("d:/2011年2月销售数据JSON.txt")

jan_data: list[Record] = text_file_reader.read_data()
feb_data: list[Record] = json_file_reader.read_data()

# 将俩个月份的数据合并为一个list来存储
all_data: list[Record] = jan_data + feb_data

# 构建MySQL链接对象
conn = Connection(
    host="localhost",   # 主机名IP
    port=3306,          # 端口
    user="root",        # 账户
    password="123123",  # 账号密码
    autocommit=True     # 提交事务（自动提交）
)
# 执行非查询性质SQL
cursor = conn.cursor()      # 获取到游标对象
# 选择数据库
conn.select_db("py_sql")
# 组织SQL语句
# 插入数据
# for record in all_data:
#     sql = f"insert into orders(order_date, order_id, money, province)"\
#           f"values('{record.date}', '{record.order}', '{record.money}', '{record.province}')"
#     # 执行SQL语句
#     cursor.execute(sql)

# 查询数据
# 组织SQL语句
cursor.execute("select * from orders")    # 执行查询方法
result = cursor.fetchall()      # 获取表中的数据
my_list = []
my_dict = {}
for a in result:
    time_faml = a[0].strftime("%Y-%m-%d")
    my_dict = {"order_date": time_faml, "order_id": a[1], "money": a[2], "province": a[3]}
    my_list.append(my_dict)

json_str = json.dumps(my_list, ensure_ascii=False)
print(json_str)
# 关闭MySQL链接对象
conn.close()