import pymysql from pymysql.cursors import DictCursor # 导入字典类型的游标对象 # 连接数据库 db = pymysql.connect(host='192.168.3.109', # 数据库IP地址 port=3306, # 数据库端口号 user='root', # 数据库用户名 password='123456', # 数据库用户密码 db='test') # 连接的数据库名称 # cursor() 方法可以创建一个游标对象 # 如果不指定游标返回的数据，则返回的数据默认为元组类型 # cursor = db.cursor() # 指定为dict类型游标，则返回的数据为字典类型 cursor = db.cursor(DictCursor) # execute() 方法可以执行所有的原生SQL,成功执行则结果返回数字 # 比如：创建表格，则返回0 ret = cursor.execute("CREATE TABLE testsheet (name VARCHAR(20), gender CHAR(1),age int(2))") # 创建表格 print(F"创建表格：{ret}") # 比如：插入数据，则返回插入记录数 ret = cursor.execute("INSERT INTO testsheet (name,gender,age) VALUES('张三', '男', 22),('三二一', '女', 22)") print(F"插入数据：{ret}") # 比如：查询数据，则返回记录数 ret = cursor.execute("select * from testsheet") print(F"记录数：{ret}") # 比如：删除表格，则返回0 ret = cursor.execute("DROP TABLE testsheet") print(F"删除表格：{ret}") # 使用fetchall()和fetchone()可以提取查询的数据,提取后数据删除 # fetchall() 提取所有数据，如果查询的数据为空，则返空，比如(),[]；一次查询，多次提取，则返空 cursor.execute("select * from info") ret = cursor.fetchall() print(F"fetchall提取数据：{ret}") ret = cursor.fetchall() print(F"fetchall再次提取数据：{ret}") # fetchone() 提取第一条数据，如果查询的数据为空，则返回None；一次查询，多次提取，则返回提取后的第一条数据，直到提取完，则返None。 cursor.execute("select * from info") ret = cursor.fetchone() print(ret) ret = cursor.fetchone() print(ret) # 强调：使用execute()执行有参数的sql语句时，如果sql语句直接使用字符串初始化的方法存在SQL注入风险；需要使用execute(sql, 参数)的方式执行,防止SQL注入 # 比如查询ID为1的数据 id = 1 sql = "select * from info where id='%s'" % (id,) cursor.execute(sql) ret = cursor.fetchall() print(F"查询ID为1的数据结果：{ret}") # 如果传入的值被恶意修改成下面的值，那么数据就存在脱库的安全问题 id = "1' or '1=1" sql = "select * from info where id='%s'" % (id,) cursor.execute(sql) ret = cursor.fetchall() print(F"SQL注入的查询结果：{ret}") # 防SQL注入的执行方式，如果使用上面的id参数，则会报SQL语句错误 sql = "select * from info where id='%s'" try: cursor.execute(sql, (id,)) ret = cursor.fetchall() except: ret = "执行报错" print(F"防SQL注入的查询结果：{ret}") # rollback() 回滚数据。只要没有commit,可以使用 rollback 回滚所有对数据库进行了修改的操作（增、删、改） # db.rollback() # 提交游标的操作到数据库。 db.commit() # 关闭游标 cursor.close() # 关闭数据库连接 db.close()
简单封装
import pymysql from pymysql.cursors import DictCursor # 导入游标类型对象 class Mysql(object): def __init__(self, host, user, password, port, db): # 连接数据库 self.database = pymysql.connect(host=host, user=user, password=password, port=port, db=db) print("连接数据库") # 获取游标对象 self.cursor = self.database.cursor() # 使用该游标返回的数据为元组类型 print(self.cursor) self.cursor = self.database.cursor(DictCursor) # 使用该游标返回的数据为字典类型 print(self.cursor) def __del__(self): """对象销毁进行资源回收""" # 关闭游标 self.cursor.close() # 关闭数据库连接 self.database.close() print("__del__被执行") def execute(self, query, args=None, force=None): """ 执行SQL :param query: sql语句 :param args: sql语句中的参数 :param force: 是否跳过对sql语句"where"的检查 :return:返回游标对象 """ # 对query传入的sql语句校验是否包含"where"，若包含，args又没传参数则抛异常。校验的目的是防SQL注入 if force is None: query = query.lower() if "where" in query and args is None: raise("检查到SQL中包含条件语句，但args参数为空，存在SQL注入的风险，若仍要执行，请将参数force的值修改为True") self.cursor.execute(query, args) # 返回游标对象就可以采用 execute().fetchone()的方式获取值 return self.cursor def commit(self): """提交数据，提交失败则回滚""" try: self.database.commit() return 0 except Exception as e: self.database.rollback() return -1 if __name__ == '__main__': db = Mysql(host="192.168.0.58", user='root', password='123456', port=3306, db="test") _id = "1 or 1=1" code = "000036" # SQL注入执行方式 sql = "select * from info WHERE id=%s or code=%s" % (_id, code) # ret = db.execute(sql).fetchall() ret = db.execute(sql, force=True).fetchall() print(ret) print(F"SQL注入获取的数据数：{len(ret)}") # 防SQL注入的执行方式 sql = "select * from info where id=%s or code=%s" ret = db.execute(sql, (_id, code)).fetchall() print(ret) print(F"防SQL注入获取的数据数：{len(ret)}") print("="*50) # 更新数据 _id = 1 code = "000011" sql = "UPDATE `d`.`info` SET `code` = %s WHERE `id` = %s" db.execute(sql, (code, _id)) ret = db.commit() if ret == 0: print("数据更新成功") else: print("数据更新失败") # 获取指定字段值 _id = 1 sql = "select * from info where id=%s" ret = db.execute(sql, (_id,)).fetchone() print(f"获取id为{_id}的code字段值：{ret.get('code')}")
执行结果
