创建数据库

create database gx character set utf8mb4;

连接数据库

#!/usr/bin/python3import mysql as pymysql# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')# 使用 cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()# 使用 execute()  方法执行 SQL 查询
cursor.execute("SELECT VERSION()")# 使用 fetchone() 方法获取单条数据.
data = cursor.fetchone()print("Database version : %s " % data)# 关闭数据库连接
db.close()

创建数据库表

#!/usr/bin/python3import mysql as pymysql# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')# 使用 cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()# 使用 execute() 方法执行 SQL，如果表存在则删除
cursor.execute("drop table if exists user")# 使用预处理语句创建表
sql = "create table user(id int primary key auto_increment, name varchar(36), age int)"cursor.execute(sql)# 关闭数据库连接
db.close()

新增数据

基本用法

#!/usr/bin/python3import mysql as pymysql# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')# 使用 cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()# 使用 execute() 方法执行 SQL，如果表存在则删除
cursor.execute("drop table if exists user")# 使用预处理语句创建表
sql = "create table user(id int primary key auto_increment, name varchar(36), age int)"
try:# 执行sql语句cursor.execute(sql)# 提交到数据库执行db.commit()
except:# 如果发生错误则回滚db.rollback()# 使用预处理语句创建表
sql = "insert into user(name,age) values(%s,%s)"
args = ["张三", 23]
try:# 执行sql语句cursor.execute(sql, args)# 提交到数据库执行db.commit()
except:# 如果发生错误则回滚db.rollback()# 关闭数据库连接
db.close()

封装execute方法

import mysql as pymysql# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')def execute(db, sql, args=None):# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.execute(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()# 使用 execute() 方法执行 SQL，如果表存在则删除
sql = "drop table if exists user"
execute(db, sql)# 使用预处理语句创建表
sql = "create table user(id int primary key auto_increment, name varchar(36), age int)"
execute(db, sql)# 使用预处理语句创建表
sql = "insert into user(name,age) values(%s,%s)"
args = ["张三", 23]
execute(db, sql, args)# 关闭数据库连接
db.close()

批量新增

import mysql as pymysql# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')def execute(db, sql, args=None):# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.execute(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def executemany(db, sql, args=None):if type(args) is not list:return# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.executemany(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()# 使用 execute() 方法执行 SQL，如果表存在则删除
sql = "drop table if exists user"
execute(db, sql)# 使用预处理语句创建表
sql = "create table user(id int primary key auto_increment, name varchar(36), age int)"
execute(db, sql)# 使用预处理语句创建表
sql = "insert into user(name,age) values(%s,%s)"
args = [("张三1", 23),("张三2", 33),("张三3", 24),
]
executemany(db, sql, args)# 关闭数据库连接
db.close()

查询数据

查询单条数据

import mysql as pymysql# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')def execute(db, sql, args=None):# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.execute(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def fetchone(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchone()def executemany(db, sql, args=None):if type(args) is not list:return# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.executemany(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()# 使用 execute() 方法执行 SQL，如果表存在则删除
sql = "drop table if exists user"
execute(db, sql)# 使用预处理语句创建表
sql = "create table user(id int primary key auto_increment, name varchar(36), age int)"
execute(db, sql)# 使用预处理语句创建表
sql = "insert into user(name,age) values(%s,%s)"
args = [("张三1", 23),("张三2", 33),("张三3", 24),
]
executemany(db, sql, args)# 查询
sql = "select * from user where id = %s"
args = [1]
print(fetchone(db, sql, args))# 关闭数据库连接
db.close()

查询多条数据

import mysql as pymysql# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')def execute(db, sql, args=None):# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.execute(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def fetchone(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchone()def fetchall(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchall()def executemany(db, sql, args=None):if type(args) is not list:return# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.executemany(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()# 使用 execute() 方法执行 SQL，如果表存在则删除
sql = "drop table if exists user"
execute(db, sql)# 使用预处理语句创建表
sql = "create table user(id int primary key auto_increment, name varchar(36), age int)"
execute(db, sql)# 使用预处理语句创建表
sql = "insert into user(name,age) values(%s,%s)"
args = [("张三1", 23),("张三2", 33),("张三3", 24),
]
executemany(db, sql, args)# 查询
sql = "select * from user"
print(fetchall(db, sql))# 关闭数据库连接
db.close()

更新数据

更新单条数据

import mysql as pymysql# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')def execute(db, sql, args=None):# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.execute(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def fetchone(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchone()def fetchall(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchall()def executemany(db, sql, args=None):if type(args) is not list:return# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.executemany(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()# 使用 execute() 方法执行 SQL，如果表存在则删除
sql = "drop table if exists user"
execute(db, sql)# 使用预处理语句创建表
sql = "create table user(id int primary key auto_increment, name varchar(36), age int)"
execute(db, sql)# 批量新增
sql = "insert into user(name,age) values(%s,%s)"
args = [("张三1", 23),("张三2", 33),("张三3", 24),
]
executemany(db, sql, args)# 更新
sql = "update user set name=%s where id = %s"
args = ["张三333", 1]
execute(db, sql, args)# 查询
sql = "select * from user"
print(fetchall(db, sql))# 关闭数据库连接
db.close()

更新多条数据

import mysql as pymysql# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')def execute(db, sql, args=None):# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.execute(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def fetchone(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchone()def fetchall(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchall()def executemany(db, sql, args=None):if type(args) is not list:return# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.executemany(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()# 使用 execute() 方法执行 SQL，如果表存在则删除
sql = "drop table if exists user"
execute(db, sql)# 使用预处理语句创建表
sql = "create table user(id int primary key auto_increment, name varchar(36), age int)"
execute(db, sql)# 批量新增
sql = "insert into user(name,age) values(%s,%s)"
args = [("张三1", 23),("张三2", 33),("张三3", 24),
]
executemany(db, sql, args)# 更新
sql = "update user set age=%s where id > %s"
args = [34, 1]
execute(db, sql, args)# 查询
sql = "select * from user"
print(fetchall(db, sql))# 关闭数据库连接
db.close()

删除数据

删除单条数据

import mysql as pymysql# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')def execute(db, sql, args=None):# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.execute(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def fetchone(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchone()def fetchall(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchall()def executemany(db, sql, args=None):if type(args) is not list:return# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.executemany(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()# 使用 execute() 方法执行 SQL，如果表存在则删除
sql = "drop table if exists user"
execute(db, sql)# 使用预处理语句创建表
sql = "create table user(id int primary key auto_increment, name varchar(36), age int)"
execute(db, sql)# 批量新增
sql = "insert into user(name,age) values(%s,%s)"
args = [("张三1", 23),("张三2", 33),("张三3", 24),
]
executemany(db, sql, args)# 删除
sql = "delete from user where id = %s"
args = [1]
execute(db, sql, args)# 查询
sql = "select * from user"
print(fetchall(db, sql))# 关闭数据库连接
db.close()

删除多条数据

import mysql as pymysql# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')def execute(db, sql, args=None):# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.execute(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def fetchone(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchone()def fetchall(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchall()def executemany(db, sql, args=None):if type(args) is not list:return# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.executemany(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()# 使用 execute() 方法执行 SQL，如果表存在则删除
sql = "drop table if exists user"
execute(db, sql)# 使用预处理语句创建表
sql = "create table user(id int primary key auto_increment, name varchar(36), age int)"
execute(db, sql)# 批量新增
sql = "insert into user(name,age) values(%s,%s)"
args = [("张三1", 23),("张三2", 33),("张三3", 24),
]
executemany(db, sql, args)# 删除
sql = "delete from user where id > %s"
args = [1]
execute(db, sql, args)# 查询
sql = "select * from user"
print(fetchall(db, sql))# 关闭数据库连接
db.close()

in查询

初步实现

这种方案，有比较严重的注入的风险。

import mysql as pymysql# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')def execute(db, sql, args=None):# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.execute(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def fetchone(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchone()def fetchall(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchall()def executemany(db, sql, args=None):if type(args) is not list:return# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.executemany(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def fetchmany(db, sql, args=None):if type(args) is not list:return# select * from user where id in (1,2,3)in_arg = ",".join([str(i) for i in args])print("in查询SQL：", sql)print("in查询参数：", in_arg)sql = sql % (in_arg,)print("in查询最终SQL：", sql)cursor = db.cursor()cursor.execute(sql)return cursor.fetchall()# 使用 execute() 方法执行 SQL，如果表存在则删除
sql = "drop table if exists user"
execute(db, sql)# 使用预处理语句创建表
sql = "create table user(id int primary key auto_increment, name varchar(36), age int)"
execute(db, sql)# 批量新增
sql = "insert into user(name,age) values(%s,%s)"
args = [("张三1", 23),("张三2", 33),("张三3", 24),
]
executemany(db, sql, args)# 根据id列表查询
sql = "select * from user where id in (%s)"
args = [1, 2, 4]
print(fetchmany(db, sql, args))# 查询
sql = "select * from user"
print(fetchall(db, sql))# 关闭数据库连接
db.close()

安全性优化

核心代码：

def is_safe_id(id):"""校验是否为安全的ID"""return re.match(r"^\w+$", str(id)) is not Nonedef fetchin(db, sql, args=None):if type(args) is not list:return# select * from user where id in (1,2,3)in_arg = ",".join([str(i) for i in args if is_safe_id(i)])print("in查询SQL：", sql)print("in查询参数：", in_arg)sql = sql % (in_arg,)print("in查询最终SQL：", sql)cursor = db.cursor()cursor.execute(sql)return cursor.fetchall()

完整代码：

import mysql as pymysql
import re# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')def execute(db, sql, args=None):# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.execute(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def fetchone(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchone()def fetchall(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchall()def executemany(db, sql, args=None):if type(args) is not list:return# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.executemany(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def is_safe_id(id):"""校验是否为安全的ID"""return re.match(r"^\w+$", str(id)) is not Nonedef fetchin(db, sql, args=None):if type(args) is not list:return# select * from user where id in (1,2,3)in_arg = ",".join([str(i) for i in args if is_safe_id(i)])print("in查询SQL：", sql)print("in查询参数：", in_arg)sql = sql % (in_arg,)print("in查询最终SQL：", sql)cursor = db.cursor()cursor.execute(sql)return cursor.fetchall()# 使用 execute() 方法执行 SQL，如果表存在则删除
sql = "drop table if exists user"
execute(db, sql)# 使用预处理语句创建表
sql = "create table user(id int primary key auto_increment, name varchar(36), age int)"
execute(db, sql)# 批量新增
sql = "insert into user(name,age) values(%s,%s)"
args = [("张三1", 23),("张三2", 33),("张三3", 24),
]
executemany(db, sql, args)# 根据id列表查询
sql = "select * from user where id in (%s)"
args = [1, 2, 4]
print(fetchin(db, sql, args))# 查询
sql = "select * from user"
print(fetchall(db, sql))# 关闭数据库连接
db.close()

根据ID列表删除

import mysql as pymysql
import re# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')def execute(db, sql, args=None):# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.execute(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def fetchone(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchone()def fetchall(db, sql, args=None):cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchall()def executemany(db, sql, args=None):if type(args) is not list:return# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()try:# 执行sql语句cursor.executemany(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def is_safe_id(id):"""校验是否为安全的ID"""return re.match(r"^\w+$", str(id)) is not Nonedef fetchin(db, sql, args=None):if type(args) is not list:return# select * from user where id in (1,2,3)in_arg = ",".join([str(i) for i in args if is_safe_id(i)])print("in查询SQL：", sql)print("in查询参数：", in_arg)sql = sql % (in_arg,)print("in查询最终SQL：", sql)cursor = db.cursor()cursor.execute(sql)return cursor.fetchall()def deletein(db, sql, args=None):if type(args) is not list:returnin_arg = ",".join([str(i) for i in args if is_safe_id(i)])sql = sql % (in_arg,)cursor = db.cursor()cursor.execute(sql)# 使用 execute() 方法执行 SQL，如果表存在则删除
sql = "drop table if exists user"
execute(db, sql)# 使用预处理语句创建表
sql = "create table user(id int primary key auto_increment, name varchar(36), age int)"
execute(db, sql)# 批量新增
sql = "insert into user(name,age) values(%s,%s)"
args = [("张三1", 23),("张三2", 33),("张三3", 24),
]
executemany(db, sql, args)# 根据id列表删除
sql = "delete from user where id in  (%s)"
args = [1, 2, 4]
deletein(db, sql, args)# 查询
sql = "select * from user"
print(fetchall(db, sql))# 关闭数据库连接
db.close()

使用with上下文执行SQL

import mysql as pymysql
import re# 打开数据库连接
db = pymysql.connect(host='localhost',port=3306,user='root',password='zhangdapeng520',database='gx')def execute(db, sql, args=None):# 使用 cursor() 方法创建一个游标对象 cursorwith db.cursor() as cursor:try:# 执行sql语句cursor.execute(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def fetchone(db, sql, args=None):with db.cursor() as cursor:cursor.execute(sql, args)return cursor.fetchone()def fetchall(db, sql, args=None):with db.cursor() as cursor:cursor = db.cursor()cursor.execute(sql, args)return cursor.fetchall()def executemany(db, sql, args=None):if type(args) is not list:returnwith db.cursor() as cursor:try:# 执行sql语句cursor.executemany(sql, args)# 提交到数据库执行db.commit()except:# 如果发生错误则回滚db.rollback()def is_safe_id(id):"""校验是否为安全的ID"""return re.match(r"^\w+$", str(id)) is not Nonedef fetchin(db, sql, args=None):if type(args) is not list:return# select * from user where id in (1,2,3)in_arg = ",".join([str(i) for i in args if is_safe_id(i)])print("in查询SQL：", sql)print("in查询参数：", in_arg)sql = sql % (in_arg,)print("in查询最终SQL：", sql)cursor = db.cursor()cursor.execute(sql)return cursor.fetchall()def deletein(db, sql, args=None):if type(args) is not list:returnin_arg = ",".join([str(i) for i in args if is_safe_id(i)])sql = sql % (in_arg,)cursor = db.cursor()cursor.execute(sql)# 使用 execute() 方法执行 SQL，如果表存在则删除
sql = "drop table if exists user"
execute(db, sql)# 使用预处理语句创建表
sql = "create table user(id int primary key auto_increment, name varchar(36), age int)"
execute(db, sql)# 批量新增
sql = "insert into user(name,age) values(%s,%s)"
args = [("张三1", 23),("张三2", 33),("张三3", 24),
]
executemany(db, sql, args)# 根据id列表删除
sql = "delete from user where id in  (%s)"
args = [1, 2, 4]
deletein(db, sql, args)# 查询
sql = "select * from user"
print(fetchall(db, sql))# 关闭数据库连接
db.close()