1、环境import pymysql # 链接MySQL
db = pymysql.connect(host='rm-bp1zq3879r28p726lco.mysql.rds.aliyuncs.com', port=3306, user='qwe8403000', password='Qwe8403000', db='mytest', charset='utf8') print(db)
配置
pip3 config set global.index-url https://repo.huaweicloud.com/repository/pypi/simple
pip3 config list
pip3 install --upgrade pip
pip3 install pymysql
2、环境引入与连接测试
数据库是我阿里的数据库，可以用到23年。做测试没问题。
输入顺序：host连接地址，port连接端口号，user连接用户名，password连接密码，db连接数据库，charset连接的编码格式。
3、创建数据库表
核心编码
conn = pymysql.connect() # 创建连接 cursor = conn.cursor() #获取游标对象 rows = cursor.execute(sql) #执行SQL语句 conn.commit() #提交数据库连接，如果是增、删、改操作，则必须提交 cursor.close() #关闭游标对象 conn.close() #关闭数据库连接
import pymysql #创建数据库连接 conn = pymysql.connect(host='rm-bp1zq3879r28p726lco.mysql.rds.aliyuncs.com', port=3306, user='qwe8403000', password='Qwe8403000', db='mytest', charset='utf8') cursor = conn.cursor() #获取游标对象 #定义创建user表的sql语句，使用三引号'''表示格式化字符串 sql = """CREATE TABLE mytestuser20220829 ( userid int(8) NOT NULL auto_increment, username varchar(50) NOT NULL, password varchar(50) default NULL, PRIMARY KEY (userid))""" rows = cursor.execute(sql) #执行SQL语句 conn.commit() #提交数据库连接，如果是增、删、改操作，则必须提交 print("mytestuser20220829表创建成功") cursor.close() #关闭游标对象 conn.close() #关闭数据库连接
4、增删改语句测试
游标可以多次执行sql语句，最终进行commit即可。
import pymysql #创建数据库连接 conn = pymysql.connect(host='rm-bp1zq3879r28p726lco.mysql.rds.aliyuncs.com', port=3306, user='qwe8403000', password='Qwe8403000', db='mytest', charset='utf8') cursor = conn.cursor() #获取游标对象 #定义创建user表的sql语句，使用三引号'''表示格式化字符串 sqlInsert1 = "insert into mytestuser20220829 values(0,'{0}',{1})".format("王语嫣", "123") sqlInsert2 = "insert into mytestuser20220829 values(0,'{0}',{1})".format("小龙女","123") sqlInsert3 = "insert into mytestuser20220829 values(0,'{0}',{1})".format("赵灵儿", "321") sqlInsert4 = "insert into mytestuser20220829 values(0,'{0}',{1})".format("删除测试", "321") cursor.execute(sqlInsert1) cursor.execute(sqlInsert2) cursor.execute(sqlInsert3) cursor.execute(sqlInsert4) sqlUpdate = "update mytestuser20220829 set password='{0}' where username='{1}'".format("潇洒的姑娘", "赵灵儿") cursor.execute(sqlUpdate) sqlDelete = "delete from mytestuser20220829 where username='{0}'".format("删除测试") cursor.execute(sqlDelete) # 提交 conn.commit() # 关闭游标 cursor.close() # 关闭数据库 conn.close()
5、查询语句测试
使用fetchall函数就可以获取游标中的数据了。
import pymysql #创建数据库连接 conn = pymysql.connect(host='rm-bp1zq3879r28p726lco.mysql.rds.aliyuncs.com', port=3306, user='qwe8403000', password='Qwe8403000', db='mytest', charset='utf8') cursor = conn.cursor() #获取游标对象 sql = "select * from mytestuser20220829" # 执行SQL cursor.execute(sql) # 回去返回集合 data = cursor.fetchall() for item in data: print(item[0]) print(item[1]) print(item[2]) # 关闭游标 cursor.close() # 关闭数据库 conn.close()
6、事务测试
ACID_MySQL事务的四大特性详解(MySQL高频面试题)_红目香薰的博客-CSDN博客
异常提交
异常回滚测试，我这里一个正确的insert一个错误的，正确的肯定是能添加成功的，但是有由于第二个语句错误，所以回滚的时候一同回滚。
import pymysql #创建数据库连接 conn = pymysql.connect(host='rm-bp1zq3879r28p726lco.mysql.rds.aliyuncs.com', port=3306, user='qwe8403000', password='Qwe8403000', db='mytest', charset='utf8') cursor = conn.cursor() #获取游标对象 try: sql1 = "insert into mytestuser20220829 values(0,'{0}',{1})".format("事务测试1", "666") cursor.execute(sql1) # 回去返回集合 sql2 = "insert into mytestuser20220829 values(0,'{0}',{1},{2})".format("事务测试2", "666",66) cursor.execute(sql2) # 回去返回集合 conn.commit() except: print("有异常，数据库回滚") conn.rollback() #回滚事务，之前的修改操作都不生效 # 关闭游标 cursor.close() # 关闭数据库 conn.close()
异常回滚，可以看到没有任何数据添加成功。
无异常提交
两个sql语句都是正确的sql，一同提交并无异常。
import pymysql #创建数据库连接 conn = pymysql.connect(host='rm-bp1zq3879r28p726lco.mysql.rds.aliyuncs.com', port=3306, user='qwe8403000', password='Qwe8403000', db='mytest', charset='utf8') cursor = conn.cursor() #获取游标对象 try: sql1 = "insert into mytestuser20220829 values(0,'{0}',{1})".format("事务测试1", "666") cursor.execute(sql1) # 回去返回集合 sql2 = "insert into mytestuser20220829 values(0,'{0}',{1})".format("事务测试2", "666") cursor.execute(sql2) # 回去返回集合 conn.commit() except: print("有异常，数据库回滚") conn.rollback() #回滚事务，之前的修改操作都不生效 # 关闭游标 cursor.close() # 关闭数据库 conn.close()
两条sql都添加成功。
7、封装DBHelper.py
主要进行增删改、查询两个函数的封装，其中init是初始化参数，getConnect是获取游标
import pymysql class DBHelper(): def __init__(self): # 数据库连接参数 self.host = "rm-bp1zq3879r28p726lco.mysql.rds.aliyuncs.com" self.user = "qwe8403000" self.pwd = "Qwe8403000" self.db = "mytest" self.charset = "utf-8" # 获取游标 def getConnect(self): if not self.db: raise(NameError, "没有设置数据库信息") self.conn = pymysql.connect(host=self.host, port=3306, user=self.user, passwd=self.pwd, db=self.db, charset="utf8mb4") # 按照字典的方式返回 cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor) if not cursor: raise(NameError, "连接数据库失败") else: return cursor # 查 def query(self, sql): cursor = self.getConnect() cursor.execute(sql) result = cursor.fetchall() # 获取查询的所有数据 # 查询完毕后必须关闭连接 self.conn.close() return result # 增删改 def excute(self, sql): cursor = self.getConnect() cursor.execute(sql) self.conn.commit() self.conn.close()
使用DBHelper测试
from DBHelper import DBHelper db=DBHelper() db.excute("insert into mytestuser20220829 values(0,'{0}',{1})".format("DBHelper测试数据1", "666")) db.excute("insert into mytestuser20220829 values(0,'{0}',{1})".format("DBHelper测试数据2", "666")) db.excute("insert into mytestuser20220829 values(0,'{0}',{1})".format("DBHelper测试数据3", "666")) db.excute("insert into mytestuser20220829 values(0,'{0}',{1})".format("DBHelper测试数据4", "666")) db.excute("insert into mytestuser20220829 values(0,'{0}',{1})".format("DBHelper测试数据5", "666")) result=db.query("select * from mytestuser20220829") print(result)
封装完毕。
