import mysql.connector
class database:
def __init__(self,ip,port,user,psw,dbname):
try:
con=mysql.connector.connect(
host=ip,
user=user,
password=psw,
port=port,
database=dbname,
charset='utf8',
buffered=True
)
print('数据库连接成功')
self.con=con #con在其他类方法中还要多次调用，所以定义为成员变量
#cursor=con.cursor()
except mysql.connector.Error as e:
print('连接失败',str(e))
def create_tb(self,sql):
try:
cursor=self.con.cursor()#获取游标
cursor.execute(sql)#执行sql
print('创建成功')
except mysql.connector.Error as e:
print('创建失败',str(e))
finally:
cursor.close()#关闭游标
def insert_tb(self,sql,data):
try:
cursor=self.con.cursor()
cursor.executemany(sql,data)
self.con.commit()
print('数据插入成功')
except mysql.connector.Error as e:
self.con.rollback()
print('插入失败',str(e))
cursor.close()
def select_tb(self,sql):
try:
cursor=self.con.cursor(dictionary=True)
cursor.execute(sql)
result1=cursor.fetchall()
print('查询全部结果：',result1)
except mysql.connector.Error as e:
print('查询失败',str(e))
finally:
cursor.close()
def select_tb_one(self,sql):
try:
cursor=self.con.cursor(dictionary=True)
cursor.execute(sql)
result2=cursor.fetchone()
print('查询一条结果：',result2)
except mysql.connector.Error as e:
print('查询失败',str(e))
finally:
cursor.close()
def select_tb_many(self,sql,count):
try:
cursor=self.con.cursor(dictionary=True)
cursor.execute(sql)
result3=cursor.fetchmany(count)
print('查询结果：',result3)
except mysql.connector.Error as e:
print('查询失败',str(e))
finally:
cursor.close()
#连接数据库
db=database('127.0.0.1','3306','root','vertrigo','mysql')
#创建表
sql_create='create table student_5(id int(10) not null auto_increment, name varchar(10) default null, age int(3) default null, primary key (id))engine=myisam default charset=utf8;'
db.create_tb(sql_create)
# #插入数据
sql_insert='insert into student_5(id,name,age) values(%s,%s,%s)'
data_insert=[(1,'guozhen',18),(2,'ss',19),(3,'alen',30)]
db.insert_tb(sql_insert,data_insert)
#查询全部数据
sql_select='select * from student_5'
db.select_tb(sql_select)
#查询多条数据
db.select_tb_many(sql_select,2)
#查询一条数据
db.select_tb_one(sql_select)
#关闭数据库连接
db.con.close()
