Mysql 8/5 Python 操作自动化脚本封装

2022-11-28 08:32:53

（脚本测试Mysql 版本 8.0.23）
（旧版测试Mysql 版本 5.？）

（本人小白学习开发的）
（求一键三连，点赞，收藏）

有问题或者其他东西可以在留言区留言

查阅资料链接在后面，包括Linux下的安装（博主超级推荐）
包括一系列的安装问题都可以解决

先安装依赖库

pip install mysql-connector

pip install MySQL-connector-python

然后接下来是模块代码（仅测试了Mysql 8）文件命名为_Mysql.py

import mysql.connector


class _MySql:
    def __init__(self, user="root", password="root", host="localhost", database="", auth_plugin="mysql_native_password"):
        self.user = user
        self.password = password
        self.host = host
        self.database = database
        self.auth_plugin = auth_plugin
        if database != "":
            self.mydb = mysql.connector.connect(
                host=host,       # 数据库主机地址
                user=user,    # 数据库用户名
                passwd=password,   # 数据库密码
                database=database,  # 链接的数据库
                auth_plugin=auth_plugin
            )
        else:
            self.mydb = mysql.connector.connect(
                host=host,       # 数据库主机地址
                user=user,    # 数据库用户名
                passwd=password,   # 数据库密码
                auth_plugin=auth_plugin
            )
    def GoROOT(self):
         self.mydb = mysql.connector.connect(
                host=self.host,       # 数据库主机地址
                user=self.user,    # 数据库用户名
                passwd=self.password,   # 数据库密码
                auth_plugin=self.auth_plugin
            )
    def usedb(self,name): #相当于重构连接器
        self.mydb = mysql.connector.connect(
                host=self.host,       # 数据库主机地址
                user=self.user,    # 数据库用户名
                passwd=self.password,   # 数据库密码
                database=name,  # 链接的数据库
                auth_plugin=self.auth_plugin
            )
    def DelConteer(self):
        del self.mydb
        print("如需要进行操作可以使用usedb或者GoROOT重新创建db对象")
    def CreateBase(self, name):  # 创建数据库
        mycursor = self.mydb.cursor()
        return mycursor.execute("CREATE DATABASE "+name)

    def ShowDataBaseList(self):  # 显示数据库名称
        mycursor = self.mydb.cursor()
        mycursor.execute("show databases")
        mapdata = []
        try:
            for data_name in mycursor:
                print(data_name[0])
                mapdata.append(data_name[0])
        except:
            print("[-]  Not Have Print!")
        return mapdata

    def ShowTableList(self):  # 显示数据表名称
        mycursor = self.mydb.cursor()
        mycursor.execute("show tables")
        mapdata = []
        try:
            for data_name in mycursor:
                print(data_name[0])
                mapdata.append(data_name[0])
        except:
            print("[-]  Not Have Print!")
        return mapdata

    # 创建数据表  viewCode -> id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255), url VARCHAR(255)
    def CreateTable(self, viewCodeOfDic, TableName):
        mycursor = self.mydb.cursor()
        View = ""
        for i in viewCodeOfDic:
            View = View + "`"+str(i)+"`" + " "+str(viewCodeOfDic[i])+","
        code = "CREATE TABLE `"+TableName+"` ("+View[:-1]+");"
        print(code)
        return mycursor.execute(code)

    def SqlConsole(self, sqlcode, selete=False):  # 如果没引用数据库可以在这里用组合命令 如果使用selete语句要标识
        mycursor = self.mydb.cursor()
        if selete == False:
            return mycursor.execute(sqlcode)
        else:
            mycursor.execute(sqlcode)
            result = mycursor.fetchall()
            return result

    # [('Google', 'https://www.google.com'),('Github', 'https://www.github.com'),('Taobao', 'https://www.taobao.com'),('*', 'https://www.*.com/')]
    # table_data_list ->["name","url"]
    def CIN_DATA(self, table_name, table_data_list, table_data):
        # table_data - >[('Google', 'https://www.google.com'),('Github', 'https://www.github.com'),('Taobao', 'https://www.taobao.com'),('*', 'https://www.*.com/')]
        if self.mydb == None:
            print("please connect!")
        else:
            data_code = ""
            data_code_2 = ""
            flage = False
            for data_name in table_data_list:
                if flage == False:
                    data_code = data_code+"`"+data_name+"`"
                    data_code_2 = data_code_2+"%s"
                    flage = True
                else:
                    data_code = data_code + ","+"`"+data_name+"`"
                    data_code_2 = data_code_2+", %s"
            sql_code = "INSERT INTO `"+table_name + \
                "` ("+data_code+") VALUES ("+data_code_2+")"
            # print(sql_code)
            mycursor = self.mydb.cursor()
            mycursor.executemany(sql_code, table_data)
            self.mydb.commit()

    def _DelTable(self, TableName):
        mycursor = self.mydb.cursor()
        sql_code = "DROP TABLE IF EXISTS "+TableName
        return mycursor.execute(sql_code)
    def _DelBase(self,BaseName):
        mycursor = self.mydb.cursor()
        sql_code = "DROP DATABASE "+BaseName+";"
        return mycursor.execute(sql_code)

使用测试代码

from Spider._Mysql import _MySql
SqlDraver = _MySql("root","***********")

from Spider import SqlDraver
Draver = SqlDraver
baseData = Draver.ShowDataBaseList() #显示数据库列表 List返回类型
Draver.CreateBase("novel_caption") #创建数据库
Draver.usedb("novel_caption") #进入数据库 可以对此usedb到别的数据库
argsmap = {
            "test":"VARCHAR(255)",
            "t2":"VARCHAR(255)",
            "t3":"VARCHAR(255)",
            "t4":"VARCHAR(255)",
            } #数据表类型
tablist = Draver.ShowTableList() #显示数据表列表 List返回累心
Draver.CreateTable(argsmap,"testta") #创建数据表
li = ["test","t2","t3","t4"] #表参数类型
Data = [("ok","ok","ok","ok"),("ok1","ok1","ok1","ok1")] #写入的数据
Draver.CIN_DATA("test",li,data)#将数据写入
Draver.SqlConsole("此处可以执行自己的SQL命令")
res = Draver.SqlConsole("selete .....",selete=True) #筛选命令可以返回结果
Draver._DelTable("testta") #删除数据表
Draver._DelBase("test") #删除数据库

Mysql 其他旧版本（我以前弄的，不知道能不能使用了）

import mysql.connector


class _MySql:
    def __init__(self, user="root", password="root", host="localhost", database="", auth_plugin="mysql_native_password"):
        if database != "":
            self.mydb = mysql.connector.connect(
                host=host,       # 数据库主机地址
                user=user,    # 数据库用户名
                passwd=password,   # 数据库密码
                database=database,  # 链接的数据库
                auth_plugin=auth_plugin
            )
        else:
            self.mydb = mysql.connector.connect(
                host=host,       # 数据库主机地址
                user=user,    # 数据库用户名
                passwd=password,   # 数据库密码
                auth_plugin=auth_plugin
            )

    def CreateBase(self, name):  # 创建数据库
        mycursor = self.mydb.cursor()
        return mycursor.execute("CREATE DATABASE "+name)

    def ShowDataBaseList(self):  # 显示数据库名称
        mycursor = self.mydb.cursor()
        mycursor.execute("show databases")
        mapdata = []
        try:
            for data_name in mycursor:
                print(data_name[0])
                mapdata.append(data_name[0])
        except:
            print("[-]  Not Have Print!")
        return mapdata

    def ShowTableList(self):  # 显示数据表名称
        mycursor = self.mydb.cursor()
        mycursor.execute("show tables")
        mapdata = []
        try:
            for data_name in mycursor:
                print(data_name[0])
                mapdata.append(data_name[0])
        except:
            print("[-]  Not Have Print!")
        return mapdata

    # 创建数据表  viewCode -> id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255), url VARCHAR(255)
    def CreateTable(self, viewCodeOfDic, TableName):
        mycursor = self.mydb.cursor()
        View = ""
        for i in viewCodeOfDic:
            View = View + str(i) + " "+str(viewCodeOfDic[i])+","
        print("CREATE TABLE "+TableName+"("+View[:-1]+")")
        return mycursor.execute("CREATE TABLE "+TableName+"("+View[:-1]+")")

    def SqlConsole(self, sqlcode, selete=False):  # 如果没引用数据库可以在这里用组合命令 如果使用selete语句要标识
        mycursor = self.mydb.cursor()
        if selete == False:
            return mycursor.execute(sqlcode)
        else:
            mycursor.execute(sqlcode)
            result = mycursor.fetchall()
            return result

    # [('Google', 'https://www.google.com'),('Github', 'https://www.github.com'),('Taobao', 'https://www.taobao.com'),('*', 'https://www.*.com/')]
    # table_data_list ->["name","url"]
    def CIN_DATA(self, table_name, table_data_list, table_data):
        # table_data - >[('Google', 'https://www.google.com'),('Github', 'https://www.github.com'),('Taobao', 'https://www.taobao.com'),('*', 'https://www.*.com/')]
        if self.mydb == None:
            print("please connect!")
        else:
            data_code = ""
            data_code_2 = ""
            flage = False
            for data_name in table_data_list:
                if flage == False:
                    data_code = data_code+data_name
                    data_code_2 = data_code_2+"%s"
                    flage = True
                else:
                    data_code = data_code + ","+data_name
                    data_code_2 = data_code_2+", %s"
            sql_code = "INSERT INTO "+table_name + \
                " ("+data_code+") VALUES ("+data_code_2+")"
            # print(sql_code)
            mycursor = self.mydb.cursor()
            mycursor.executemany(sql_code, table_data)
            self.mydb.commit()

    def _DelTable(self, TableName):
        mycursor = self.mydb.cursor()
        sql_code = "DROP TABLE IF EXISTS "+TableName
        return mycursor.execute(sql_code)

参考文章：
MySQL CONNECTOR基本用法

---->>>>点我转跳<<<<----

Linux 下安装MySql超全教程（写的非常好）

---->>>>点我转跳<<<<----

Linux 下安装记得给权限

chown -R mysql:mysql /usr/local/mysql/data/
chmod -R 777 /usr/local/

不方便启动可以用这个脚本来快捷点

import os
while True:
    com = input("Command-> ")
    if com =="console":
        os.system("sudo /usr/local/mysql/bin/mysql -u root -p")
    elif com == "start":
        os.system("sudo service mysql start")
    elif com =="restart":
        os.system("sudo service mysql restart")
    elif com == "stop":
        os.system("sudo service mysql stop")
    elif com == "exit":
        break

码农公寓

相关文章