测试中简单的数据库操作-pymysql入门-木盒主机

什么是 「PyMySQL」？PyMySQL 是在 Python3.x 版本中用于连接 MySQL 服务器的一个库，Python2 中则使用 mysqldb。

本文目录

1 使用的目的

测试前操作数据库准备测试数据。
测试后查询数据库数据入库情况。
测试结束以后清除测试数据，恢复之前的状态。

2 PyMySQL 安装

$ pip3 install PyMySQL

❝以下来自菜鸟教程

实例：

❞

<code style="margin-left:0">实例(Python 3.0+)
#!/usr/bin/python3
 
import pymysql
 
# 打开数据库连接
db = pymysql.connect(host='localhost',
                     user='testuser',
                     password='test123',
                     database='TESTDB')
 
# 使用 cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()
 
# 使用 execute()  方法执行 SQL 查询 
cursor.execute("SELECT VERSION()")
 
# 使用 fetchone() 方法获取单条数据.
data = cursor.fetchone()
 
print ("Database version : %s " % data)
 
# 关闭数据库连接
db.close()
</code>

可以看到查询需要用到的信息，主要是数据库ip，用户名，密码，数据库名，还有没有列出来的端口号
这些我们可以配置到yaml文件里面维护

3 使用yaml文件配置数据库连接信息

我们先建一个表，字段分别是id，busi_order订单，state状态，crate_time创建时间，describe描述。

创建一个工具类

<code style="margin-left:0">#-*- codeing = utf-8 -*-
#@Time: 2022/10/30 19:01
#@Author: 怪盗LLYL
#@File: db_unit.py
#@Software: PyCharm
import pymysql

from conf.config import FilePath
from unit.conf_yaml import YamlReader

class DateBase(object):
#初始化
    def __init__(self,env,yamlf_name=FilePath.DB_CONF_FILE):
        self._connect_conf = YamlReader(yamlf_name).data()[env]
    def basedb(slef,key:str):
        '''根据key选择数据库表
        :param key:订单号最后一位
        :return 数据库表连接:type:pymysql.connections.Connection
        '''
        key=int(key)
        database_count = len(slef._connect_conf['db_order']['database'])
        selected_database = (key)%database_count
        return pymysql.connect(host=slef._connect_conf['db_order']['host'],
                               port=slef._connect_conf['db_order']['port'],
                               user=slef._connect_conf['db_order']['user'],
                               password=slef._connect_conf['db_order']['password'],
                               database=slef._connect_conf['db_order']['database'][selected_database])
</code>

这里假设我们有两个库，一库和二库，根据尾数分别入了不同的库里，我们可以初选择不同数据库连接信息，具体情况请根据读者公司实际逻辑编写。
YamlReader类请看上一篇文章。

4 添加一个查询方法:

「注意datetime字段需要特殊处理下， cursor.description 是用来获取列名的。」

<code style="margin-left:0">    def select(slef,db,sql):
        # 打开数据库连接
        # 使用 cursor() 方法创建一个游标对象 cursor
        cursor = db.cursor()
        # 使用 execute()  方法执行 SQL 查询
        cursor.execute(sql)
        # 使用 fetchone() 方法获取单条数据.
        # 使用 fetchall() 方法获取所有数据.
        result = cursor.fetchall()
        data_list = []
        for z in result:
            i = 0
            data_dict = {}
            for field in cursor.description:
                if isinstance(z[i], datetime.datetime) or isinstance(z[i], datetime.date):
                    data_dict[field[0]] = str(z[i])
                else:
                    data_dict[field[0]] = z[i]
                i = i + 1
            data_list.append(data_dict)

        # 关闭数据库连接
        db.close()
        return data_list
    def get_date_by_busi_order(slef,busi_order:str,table:str):
        '''根据busi_order查询订单信息
        :param busi_order:订单号
        :param table:表名
        :return df List[dic]:返回订单信息,字典内key：字段 value：值
        '''
        key = int(busi_order[-1])
        db=slef.basedb(key)
        sql=f"select * from {table} where busi_order='{busi_order}'"
        df=slef.select(db,sql)
        return df
</code>

先手动插入数据

<code style="margin-left:0">if __name__ == '__main__':
    datebase=DateBase("dev")
    busi_order = "test123456789"
    table = "table_order"
    df = datebase.get_date_by_busi_order(busi_order=busi_order,table=table)
    print(df)
    busi_order = "test12345678"
    table = "table_order"
    df = datebase.get_date_by_busi_order(busi_order=busi_order,table=table)
    print(df)
</code>

可以看到两条不同库的数据都查出来了

5 插入操作:

创建插入方法,其中订单数据我临时使用到一个字典数据，读者可以按照实际情况修改。

<code style="margin-left:0">    def insert(self,db,sql):
        '''根据db和sql执行插入操作
        :param db:数据库连接
        :param sql:sql语句 插入语句
        :return code int:操作结果 200：成功 500：失败
        '''
        # 使用 cursor() 方法创建一个游标对象 cursor
        cursor = db.cursor()
        # 使用 execute()  方法执行 SQL 查询
        code = 0
        try:
            # 执行sql语句
            cursor.execute(sql)
            # 提交到数据库执行
            db.commit()
            code = 200
        except:
            # 如果发生错误则回滚
            db.rollback()
            code = 500
        # 关闭数据库连接
        db.close()
        return code
    def insert_order(slef,table_order:dict,table:str):
        '''根据table_order订单信息插入<a href="https://www.mhzhuji.com/kwck/57"  class="lar_link lar_link_outgoing" data-linkid="219" data-postid="1485"  title="数据库MySQL"  rel="nofollow" target="_blank" >数据库</a>
        :param table_order dir：订单信息
        :param table:表名
        :return code int:操作结果 200：成功 500：失败
        '''
        print(table_order)
        key = int(table_order["busi_order"][-1])
        db=slef.basedb(key)

        sql=f"INSERT INTO {table} "+str(tuple(table_order.keys())).replace("'","`")+" VALUES "+str(tuple(table_order.values())) #注意mysql语句中字段不是单引号而是反引号
        code=slef.insert(db,sql)
        return code
</code>

<code style="margin-left:0">if __name__ == '__main__':
    datebase=DateBase("dev")
    table_order = {"busi_order":"test654321721","state":"200","create_time":"2022-10-30 20:06:25","describe":"测试插入<a href="https://www.mhzhuji.com/kwck/57"  class="lar_link lar_link_outgoing" data-linkid="219" data-postid="1485"  title="数据库MySQL"  rel="nofollow" target="_blank" >数据库</a>"}
    table = "table_order"
    code = datebase.insert_order(table_order=table_order,table=table)
    print(code)
</code>

可以看到拼接了插入语句，插入成功返回200。

查看数据库，插入成功了

再次执行因为订单号重复的原因插入失败了，并且没有入数据库

6 更新和删除操作:

更新，删除和插入没有什么不同，所以可以把insrt改成update其他方法都调用这个。

<code style="margin-left:0">    def update (self,db,sql):
        '''根据db和sql执行插入，更新，删除操作
        :param db:数据库连接
        :param sql:sql语句 执行语句
        :return code int:操作结果 200：成功 500：失败
        '''
        # 使用 cursor() 方法创建一个游标对象 cursor
        cursor = db.cursor()
        # 使用 execute()  方法执行 SQL 查询
        code = 0
        try:
            # 执行sql语句
            cursor.execute(sql)
            # 提交到数据库执行
            db.commit()
            code = 200
        except Exception as e:
            # 如果发生错误则回滚
            print("sql出错:",sql)
            print(e.__str__())
            db.rollback()
            code = 500
        # 关闭数据库连接
        db.close()
        return code
    def insert_order(slef,table_order:dict,table:str):
        '''根据table_order订单信息插入数据库
        :param table_order dir：订单信息
        :param table:表名
        :return code int:操作结果 200：成功 500：失败
        '''
        key = int(table_order["busi_order"][-1])
        db=slef.basedb(key)

        sql=f"INSERT INTO {table} "+str(tuple(table_order.keys())).replace("'","`")+" VALUES "+str(tuple(table_order.values())) #注意mysql语句中字段不是单引号而是反引号
        code=slef.update(db,sql)
        return code
    def update_order(slef,table_order:dict,table:str):
        '''根据table_order订单信息修改记录
        :param table_order dir：订单信息
        :param table:表名
        :return code int:操作结果 200：成功 500：失败
        '''
        key = int(table_order["busi_order"][-1])
        db=slef.basedb(key)
        sql = f"UPDATE {table} SET "
        for i in table_order:
            sql = sql + f"`{i}`= '{str(table_order[i])}',"#注意mysql语句中字段不是单引号而是反引号
        sql = sql[:-1]
        sql = sql + f" WHERE busi_order = '{table_order['busi_order']}';"
        code=slef.update(db,sql)
        return code
    def delete_order(slef,table_order:dict,table:str):
        '''根据table_order订单信息删除记录
        :param table_order dir：订单信息
        :param table:表名
        :return code int:操作结果 200：成功 500：失败
        '''
        key = int(table_order["busi_order"][-1])
        db=slef.basedb(key)
        sql = f"DELETE FROM {table} WHERE  busi_order = '{table_order['busi_order']}';"
        code=slef.update(db,sql)
        return code
if __name__ == '__main__':
    datebase=DateBase("dev")
    table_order = {"busi_order":"test654321721","state":"200","create_time":"2022-10-30 22:13:25","describe":"更新操作"}
    table = "table_order"
    code = datebase.update_order(table_order=table_order,table=table)
    print(code)
</code>

更新成功，插入成功，删除成功

7 结语

这样pymysql入门操作就介绍完了。大家可以根据自己实际情况在自己的项目里使用，或者使用orm。

未经允许不得转载：木盒主机 » 测试中简单的数据库操作-pymysql入门

测试中简单的数据库操作-pymysql入门

1 使用的目的

2 PyMySQL 安装

3 使用yaml文件配置数据库连接信息

4 添加一个查询方法:

5 插入操作:

6 更新和删除操作:

7 结语

相关推荐

热门推荐

DMIT 美国/香港/日本 CN2 GIA

搬瓦工限量版CN2 GIA整理

随便看看

热门标签

分类