1.2 SQL操作

1.2.1 【必须】使用参数化查询

• 使用参数化SQL语句，强制区分数据和命令，避免产生SQL注入漏洞。

# 错误示例
import mysql.connector
mydb = mysql.connector.connect(
... ...
)
cur = mydb.cursor()
userid = get_id_from_user()
# 使用%直接格式化字符串拼接SQL语句
cur.execute("SELECT `id`, `password` FROM `auth_user` WHERE `id`=%s " % (userid,)) 
myresult = cur.fetchall()

# 安全示例
import mysql.connector
mydb = mysql.connector.connect(
... ...
)
cur = mydb.cursor()
userid = get_id_from_user()
# 将元组以参数的形式传入
cur.execute("SELECT `id`, `password` FROM `auth_user` WHERE `id`=%s " , (userid,))
myresult = cur.fetchall()

• 推荐使用ORM框架来操作数据库，如：使用SQLAlchemy。

# 安装sqlalchemy并初始化数据库连接
# pip install sqlalchemy
from sqlalchemy import create_engine
# 初始化数据库连接，修改为你的数据库用户名和密码
engine = create_engine('mysql+mysqlconnector://user:password@host:port/DATABASE')

# 引用数据类型
from sqlalchemy import Column, String, Integer, Float
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
# 定义 Player 对象:
class Player(Base):
    # 表的名字:
    __tablename__ = 'player'
    # 表的结构:
    player_id = Column(Integer, primary_key=True, autoincrement=True)
    team_id = Column(Integer)
    player_name = Column(String(255))
    height = Column(Float(3, 2))

# 增删改查
from sqlalchemy.orm import sessionmaker
# 创建 DBSession 类型:
DBSession = sessionmaker(bind=engine)
# 创建 session 对象:
session = DBSession()
# 增:
new_player = Player(team_id=101, player_name="Tom", height=1.98)
session.add(new_player)
# 删:
row = session.query(Player).filter(Player.player_name=="Tom").first()
session.delete(row)
# 改:
row = session.query(Player).filter(Player.player_name=="Tom").first()
row.height = 1.99
# 查:
rows = session.query(Player).filter(Player.height >= 1.88).all()
# 提交即保存到数据库:
session.commit()
# 关闭 session:
session.close()

1.2.2 【必须】对参数进行过滤

• 将接受到的外部参数动态拼接到SQL语句时，必须对参数进行安全过滤。

def sql_filter(sql, max_length=20):
    dirty_stuff = ["\"", "\\", "/", "*", "'", "=", "-", "#", ";", "<", ">", "+", 
                   "&", "$", "(", ")", "%", "@", ","]
    for stuff in dirty_stuff:
        sql = sql.replace(stuff, "x")
    return sql[:max_length]