河曲智叟 发表于 2025-3-9 13:16:17

接口主动化如何封装mysql操作

数据查询类封装

1. 功能分析


[*]可以毗连不同sql数据库
[*]查一条数据,多条数据
[*]可以获取不同格式的数据
2. 封装成数据库查询类

封装思路:       

[*]数据库查询模块有多个功能,且需要复用,所以封装成类
[*]创建对象方法实现各种查询
[*]在构造方法中创建毗连 


[*] 废话不多说,直接上代码!!!
 
# -*- coding: utf-8 -*-
# @Time : 2019/11/13 14:51
# @Author : kira
# @Email : 262667641@qq.com
# @File : do_mysql.py
# @Project : risk_project

import pymysql
from pymysql.cursors import DictCursor


class DoMysql:
    """
    sql数据库查询类
    """

    def __init__(self, db_config: dict):
      # 创建连接
      engine = db_config.pop('engine', 'mysql')
      if engine.lower() == 'mysql':
            self.conn = pymysql.connect(**db_config)
      elif engine.lower() == 'oracle':
            pass

    def __execute(self, sql, action, res_type='t', *args):
      """
      :param sql:
      :param action: 字符串,指定执行cursor对应的方法
      :param res_type: 返回数据类型
      :param args:
      :return:
      """
      if res_type == 't':
            cursor = self.conn.cursor()
      else:
            cursor = self.conn.cursor(DictCursor)
      try:
            cursor.execute(sql)
            return getattr(cursor, action)(*args)
      except Exception as e:
            raise e
      finally:
            cursor.close()

    def get_one(self, sql, res_type='t'):
      """
      获取一条数据
      :param sql:
      :param res_type: 返回数据的类型,默认为t表示以元组返回,'d'表示以字典的形式返回
      :return: 元组/字典
      """
      # 数据库若断开即重连
      self.reConnect()
      return self.__execute(sql, 'fetchone', res_type)

    def get_many(self, sql, size, res_type='t'):
      # 数据库若断开即重连
      self.reConnect()
      return self.__execute(sql, 'fetchmany', res_type, size)

    def get_all(self, sql, res_type='t'):
      # 数据库若断开即重连
      self.reConnect()
      return self.__execute(sql, 'fetchall', res_type)

    def exist(self, sql):
      if self.get_one(sql):
            return True
      else:
            return False

    def reConnect(self):
      """
      重连机制
      :@return
      """
      try:
            self.conn.ping()
      except:
            self.conn()

    def __del__(self):
      """
      对象销毁的时候自动会被调用
      :return:
      """
      self.conn.close()


if __name__ == '__main__':
    db = {
      'engine': 'mysql',# 指定mysql数据
      'host': '127.0.0.1',
      'user': 'admin',
      'password': '12345',
      'port': 3306,
      'db': 'mysql',
      'charset': 'utf8'
    }
    db = DoMysql(db)
    sql = 'select id, reg_name, mobile_phone from member limit 10'
    # res = db.get_one(sql)
    res = db.get_many(sql, size=5)
    print(res)
同时,在这我为各人预备了一份软件测试视频教程(含口试、接口、主动化、性能测试等),就在下方,需要的可以直接去观看。
   【2025最新版】字节大牛讲的最全最细的主动化测试全套教程!永久白嫖,拿走不谢,全程干货无废话!逼自己15天内学完,从软件测试基础到项目实战一套全通关!

免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。
页: [1]
查看完整版本: 接口主动化如何封装mysql操作