怪盗LYL

V1

2022/10/30阅读:15主题:草原绿

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

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

1 使用的目的

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

2 PyMySQL 安装

  • $ pip3 install PyMySQL

以下来自菜鸟教程

  • 实例:
实例(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()
  • 可以看到查询需要用到的信息,主要是数据库ip,用户名,密码,数据库名,还有没有列出来的端口号
  • 这些我们可以配置到yaml文件里面维护

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

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

  • 创建一个工具类
#-*- 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])
  • 这里假设我们有两个库,一库和二库,根据尾数分别入了不同的库里,我们可以初选择不同数据库连接信息,具体情况请根据读者公司实际逻辑编写。
  • YamlReader类请看上一篇文章。

4 添加一个查询方法:

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

    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
  • 先手动插入数据
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)
  • 可以看到两条不同库的数据都查出来了

5 插入操作:

  • 创建插入方法,其中订单数据我临时使用到一个字典数据,读者可以按照实际情况修改。
    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订单信息插入数据库
        :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
if __name__ == '__main__':
    datebase=DateBase("dev")
    table_order = {"busi_order":"test654321721","state":"200","create_time":"2022-10-30 20:06:25","describe":"测试插入数据库"}
    table = "table_order"
    code = datebase.insert_order(table_order=table_order,table=table)
    print(code)
  • 可以看到拼接了插入语句,插入成功返回200。
  • 查看数据库,插入成功了
  • 再次执行因为订单号重复的原因插入失败了,并且没有入数据库

6 更新和删除操作:

  • 更新,删除和插入没有什么不同,所以可以把insrt改成update其他方法都调用这个。
    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)
  • 更新成功,插入成功,删除成功

7 结语

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

分类:

后端

标签:

Python

作者介绍

怪盗LYL
V1