python操作mysql进行封装，封装的好处我就不提了，以下是我做项目时的一个封装，大家可以根据实际需要进行自己的一个封装
我封装的内容：
1.数据库的配置文件
2.获取数据配置文件的地址
3.连接数据库的操作
4.操作mysql的语句
5.调用mysql的语句--执行
封装1：数据库配置文件
config中db_conf.ini
里面主要是一些mysql的配置文件
[mysqlconf]
host=127.0.0.1
port=3306
user=root
password=123456
db_name=数据库名
封装2：读取配置文件地址public_data.py
import os
# 整个项目的根目录绝对路劲
baseDir = os.path.dirname(os.path.dirname(__file__))
# 数据库配置你文件绝对路径
config_path = baseDir + "/config/db_config.ini"
print(config_path)
打印目录地址：
封装3：数据库的连接 config_handler.py
# -*- coding:utf-8 -*-
#@Time : 2020/6/14 11:01
#@Author: 张君
#@File : config_handler.py
from configparser import ConfigParser
from config.public_data import config_path
class ConfigParse(object):
def __init__(self):
pass
@classmethod
def get_db_config(cls):
#cls使用的类方法,cls就是指定本身
cls.cfp = ConfigParser()
cls.cfp.read(config_path)
host = cls.cfp.get("mysqlconf", "host")
port = cls.cfp.get("mysqlconf", "port")
user = cls.cfp.get("mysqlconf", "user")
password = cls.cfp.get("mysqlconf", "password")
db = cls.cfp.get("mysqlconf", "db_name")
return {"host":host, "port":port, "user":user, "password":password,"db":db}
if __name__ == '__main__':
cp=ConfigParse()
result=cp.get_db_config()
print(result)
封装4：sql语句的封装 db_handle.py
import datetime
import pymysql
from utils.config_handler import ConfigParse
class DB(object):
def __init__(self):
#获取mysql连接信息
self.db_conf = ConfigParse.get_db_config()
#获取连接对象
self.conn = pymysql.connect(
host = self.db_conf["host"],
port = int(self.db_conf["port"]),
user = self.db_conf["user"],
password = self.db_conf["password"],
database = self.db_conf["db"],
charset = "utf8"
)
#获取数据的游标
self.cur = self.conn.cursor()
def close_connect(self):
# 关闭数据连接
#提交，物理存储
self.conn.commit()
#游标关闭
self.cur.close()
#连接对象关闭
self.conn.close()
def get_api_list(self):
"""获取所有的对象数据"""
sqlStr = "select * from interface_api where status=1"
self.cur.execute(sqlStr)
data = self.cur.fetchall()
#转成一个list
apiList = list(data)
return apiList
def get_api_case(self, api_id):
"""获取某一条数据"""
sqlStr = "select * from interface_test_case where api_id=%s and status=1" %api_id
self.cur.execute(sqlStr)
api_case_list = list(self.cur.fetchall())
return api_case_list
def get_rely_data(self, api_id, case_id):
"""获取所有数据库中的某一条"""
sqlStr = "select data_store from interface_data_store where api_id=%s and case_id=%s" %(api_id, case_id)
self.cur.execute(sqlStr)
rely_data = eval(self.cur.fetchall()[0][0])
return rely_data
def write_check_result(self, case_id, errorInfo, res_data):
"""更新数据库表"""
sqlStr = "update interface_test_case set error_info=\"%s\", res_data=\"%s\" where id=%s" %(errorInfo, res_data, case_id)
self.cur.execute(sqlStr)
self.conn.commit()
def insert_dab(self):
sqlStr="INSERT INTO `interface_api` VALUES (4, '修改博文', 'http://39.106.41.11:8080/getBlogContent/', 'get', 'url','0', '2018-07-27 22:13:30')"
self.cur.execute(sqlStr)
def get_api_case(self, api_id):
"""获取表中id"""
sqlStr = "select * from interface_test_case where api_id=%s and status=1" % api_id
self.cur.execute(sqlStr)
api_case_list = list(self.cur.fetchall())
return api_case_list
def get_api_id(self, api_name):
"""获取表中id"""
sqlStr = "select api_id from interface_api where api_name='%s'" % api_name
self.cur.execute(sqlStr)
api_id = self.cur.fetchall()[0][0]
return api_id
def update_store_data(self, api_id, case_id, store_data):
sqlStr = "select data_store from interface_data_store where api_id=%s and case_id=%s" % (api_id, case_id)
self.cur.execute(sqlStr)
if self.cur.fetchall():
sqlStr = "update interface_data_store set data_store=\"%s\" where api_id=%s and case_id=%s" % (
store_data, api_id, case_id)
print(sqlStr)
self.cur.execute(sqlStr)
self.conn.commit()
else:
sqlStr = "insert into interface_data_store values(%s, %s, \"%s\", '%s')" % (
api_id, case_id, store_data, datetime.now())
self.cur.execute(sqlStr)
self.conn.commit()
if __name__ == '__main__':
db=DB()
print(db.get_api_list())
print(db.get_api_case(1))
# print(db.get_rely_data(1,1))
#print(db.insert_dab())
使用ing，我新建了一个test.py文件
from utils.db_handler import DB
from action.get_rely import GetRely
from utils.HttpClient import HttpClient
def main():
#连接数据库，获取连接实例对象
db=DB()
#从数据库中获取需要执行的api执行集合
api_list=db.get_api_list()
print(api_list)
main()
至此一整套封装，与使用就完成了，为什么要分为几个文件夹，我遵循的一个原则， 使用、配置、公共方法是用不的文件夹整理，其实最终也就是你使用怎么方便，维护比较好，就怎么用
