python封裝mysq操作，進行數據庫的增刪改

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()

　　

 

 至此一整套封裝，與使用就完成了，為什么要分為幾個文件夾，我遵循的一個原則， 使用、配置、公共方法是用不的文件夾整理，其實最終也就是你使用怎么方便，維護比較好，就怎么用