Mysql 8/5 Python 操作自动化脚本封装
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了Mysql 8/5 Python 操作自动化脚本封装，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含8250字，纯文字阅读大概需要12分钟。
内容图文
（脚本测试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'),('stackoverflow', 'https://www.stackoverflow.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'),('stackoverflow', 'https://www.stackoverflow.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'),('stackoverflow', 'https://www.stackoverflow.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'),('stackoverflow', 'https://www.stackoverflow.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
内容总结
以上是互联网集市为您收集整理的Mysql 8/5 Python 操作自动化脚本封装全部内容，希望文章能够帮你解决Mysql 8/5 Python 操作自动化脚本封装所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。
来源：【匿名】