整理了一下前面3期学的内容后，现在练习使用python去操作数据库
#!python3
# coding:utf-8
import pymysql
class mysql_option():
def __init__(self, host, port, username, pwd, **kwargs):
'''
如果直接连接到具体的数据库，只需要传kwargs，即db = dbname
:param host:
:param port:
:param username:
:param pwd:
:param kwargs:
:return:
'''
self.conn = self.mysql_connect(host, port, username, pwd, **kwargs)
self.cursor = self.conn.cursor() @staticmethod
def mysql_connect(host, port, username, pwd, **kwargs):
try:
connect = pymysql.connect(host = host, port = int(port), user = username, passwd = pwd, charset = 'utf8', **kwargs)
print("Mysql数据库连接成功")
return connect
except Exception as ex:
print("Mysql连接异常,具体报错如下：\n%s"%(ex))
return False def create_database(self, dbname):
'''
创建数据库
:param dbname: 数据库名称
:return:
'''
try:
create_db_sql = "create database if not exists %s default character set utf8 collate utf_8_general_ci;"%(dbname)
self.cursor.execute(create_db_sql)
result = self.cursor.fetcall()
print("创建%s数据库成功%s"%(dbname, result))
except Exception as ex:
print("创建%s数据库失败，具体原因如下：\n%s"%(dbname, ex))
self.cursor.close()
return False def select_db(self, db_name):
try:
self.conn.select_db(db_name)
print("连接数据库%s成功"%(db_name))
except Exception as ex:
print("连接数据库%s失败,具体原因如下：\n%s"%(db_name, ex))
self.cursor.close()
return False def create_table(self, table_name):
create_table_sql = "create table if not exists %s"%(table_name)
try:
self.cursor.execute(create_table_sql)
result = self.cursor.fetcall()
print("创建%s数据表成功%s"%(table_name, result))
return True
except Exception as ex:
print("创建%s数据表失败,具体原因如下：\n%s"%(table_name, ex))
self.cursor.close()
return False def sql_option(self, option):
try:
self.cursor.execute(option)
except Exception as ex:
print("操作语句执行失败,具体原因如下：\n%s"%(ex))
self.conn.rollback() #事务回滚
return False
else:
self.conn.commit()
print("SQL事物提交成功，提交结果：%s"%(self.cursor.rowcount))
return True
finally:
self.cursor.close() #关闭连接 def show_db(self):
self.cursor.execute("show databases;")
result = self.cursor.fetchall()
return result def show_table(self):
self.cursor.execute("show tables;")
result = self.cursor.fetchall()
return result def show_desc(self, table_name):
sql_op = "desc %s"%(table_name)
self.cursor.execute(sql_op)
result = self.cursor.fetchall()
return result def drop_db(self, db_name):
sql_op = "drop database %s"%(db_name)
print("警告：您将删除数据库%s，请再次确认删除！(Y)确认 (N)取消"%(db_name))
confirm = input("请选择：")
if confirm == "Y" or confirm == "y":
try:
print("开始删除……")
self.cursor.execute(sql_op)
print("删除数据库%s成功"%(db_name))
return True
except Exception as ex:
print("删除数据库%s失败，具体原因：%s"%(db_name, ex))
return False
finally:
self.cursor.close()
else:
print("本次操作已取消！")
return False def drop_table(self, table_name):
sql_op = "drop table %s"%(table_name)
print("警告：您将删除数据表%s，请再次确认删除！(Y)确认 (N)取消"%(table_name))
confirm = input("请选择：")
if confirm == "Y" or confirm == "y":
print("开始删除……")
try:
self.cursor.execute(sql_op)
print("删除数据表%s成功%s"%(table_name))
return True
except Exception as ex:
print("删除数据表%s失败,具体原因：%s"%(table_name, ex))
return False
finally:
self.cursor.close() #关闭连接 else:
print("本次操作已取消！") def close_mysql(self):
self.cursor.close() if __name__ == '__main__':
mysql_option = mysql_option("192.168.183.128", "3306", "root", "Abc123!", db = 'testdb')
# mysql_option = mysql_option("192.168.183.128", "3306", "root", "Abc123!")
# print(mysql_option.show_db()[4][0])
# db = mysql_option.show_db()[4][0]
# mysql_option.select_db(db)
# # mysql_option.show_table()
# # mysql_option.close_mysql()
# print(mysql_option.show_desc("test_tables"))
# mysql_option.drop_db("testdb02")
SQL = "select *from user where id=1"
mysql_option.sql_option(SQL)