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

数据查询类封装

1. 功能分析

• 可以毗连不同sql数据库
  
• 查一条数据，多条数据
  
• 可以获取不同格式的数据
  

2. 封装成数据库查询类

封装思路:       

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

• 废话不多说，直接上代码！！！
   
  2. # -*- coding: utf-8 -*-
  3. # @Time : 2019/11/13 14:51
  4. # @Author : kira
  5. # @Email : 262667641@qq.com
  6. # @File : do_mysql.py
  7. # @Project : risk_project
  9. import pymysql
  10. from pymysql.cursors import DictCursor
  13. class DoMysql:
  14.     """
  15.     sql数据库查询类
  16.     """
  18.     def __init__(self, db_config: dict):
  19.         # 创建连接
  20.         engine = db_config.pop('engine', 'mysql')
  21.         if engine.lower() == 'mysql':
  22.             self.conn = pymysql.connect(**db_config)
  23.         elif engine.lower() == 'oracle':
  24.             pass
  26.     def __execute(self, sql, action, res_type='t', *args):
  27.         """
  28.         :param sql:
  29.         :param action: 字符串，指定执行cursor对应的方法
  30.         :param res_type: 返回数据类型
  31.         :param args:
  32.         :return:
  33.         """
  34.         if res_type == 't':
  35.             cursor = self.conn.cursor()
  36.         else:
  37.             cursor = self.conn.cursor(DictCursor)
  38.         try:
  39.             cursor.execute(sql)
  40.             return getattr(cursor, action)(*args)
  41.         except Exception as e:
  42.             raise e
  43.         finally:
  44.             cursor.close()
  46.     def get_one(self, sql, res_type='t'):
  47.         """
  48.         获取一条数据
  49.         :param sql:
  50.         :param res_type: 返回数据的类型，默认为t表示以元组返回，'d'表示以字典的形式返回
  51.         :return: 元组/字典
  52.         """
  53.         # 数据库若断开即重连
  54.         self.reConnect()
  55.         return self.__execute(sql, 'fetchone', res_type)
  57.     def get_many(self, sql, size, res_type='t'):
  58.         # 数据库若断开即重连
  59.         self.reConnect()
  60.         return self.__execute(sql, 'fetchmany', res_type, size)
  62.     def get_all(self, sql, res_type='t'):
  63.         # 数据库若断开即重连
  64.         self.reConnect()
  65.         return self.__execute(sql, 'fetchall', res_type)
  67.     def exist(self, sql):
  68.         if self.get_one(sql):
  69.             return True
  70.         else:
  71.             return False
  73.     def reConnect(self):
  74.         """
  75.         重连机制
  76.         :@return
  77.         """
  78.         try:
  79.             self.conn.ping()
  80.         except:
  81.             self.conn()
  83.     def __del__(self):
  84.         """
  85.         对象销毁的时候自动会被调用
  86.         :return:
  87.         """
  88.         self.conn.close()
  91. if __name__ == '__main__':
  92.     db = {
  93.         'engine': 'mysql',  # 指定mysql数据
  94.         'host': '127.0.0.1',
  95.         'user': 'admin',
  96.         'password': '12345',
  97.         'port': 3306,
  98.         'db': 'mysql',
  99.         'charset': 'utf8'
  100.     }
  101.     db = DoMysql(db)
  102.     sql = 'select id, reg_name, mobile_phone from member limit 10'
  103.     # res = db.get_one(sql)
  104.     res = db.get_many(sql, size=5)
  105.     print(res)

  复制代码

同时，在这我为各人预备了一份软件测试视频教程（含口试、接口、主动化、性能测试等），就在下方，需要的可以直接去观看。
     【2025最新版】字节大牛讲的最全最细的主动化测试全套教程！永久白嫖，拿走不谢，全程干货无废话！逼自己15天内学完，从软件测试基础到项目实战一套全通关！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。