PyMySQL

概述

在之前的文章中介绍过数据库及其操作，接下来这篇文章，我们来看看在python中怎么来操作数据库

Python 操作 Mysql 主要包含下面 3 种方式：

  • MySQLdb：对MySQL的C Client封装实现，仅支持 Python2，不推荐
  • PyMySQL：纯 Python 语言编写的 Mysql 操作客户端，安装方便，支持 Python3
  • SQLAlchemy：是一个非常强大的 ORM 框架，不提供底层的数据库操作，主要是通过定义模型对应数据表结构，在 Python Web 编程领域应用广泛
    因此，这里主要介绍PyMySQL和SQLAlchemy（其中SQLAlchemy的介绍请参看Python操作数据库之SQLAlchemy

PyMySQL

项目地址

PyMySQL安装

pip install pymysql

PyMysql操作及示例

  • 数据库安装及操作等基本知识可参考文章数据库及其操作
  • 操作数据库一般流程：
    • 建立连接
    • 获取游标
    • 执行SQL
    • 提交事务
    • 释放资源

数据库连接pymysql.connect

  • pymysql.connect()方法返回的是Connections模块下的Connection类实例（查看源码可知connect方法的传参就是给Connection类的 __init__提供参数）

游标cursor

  • 操作数据库，必须使用游标，需要先获取一个游标对象
  • Connection.cursor(cursor=None) 方法返回一个新的游标对象
  • 连接没有关闭前，游标对象可以反复使用

操作数据库

  • 数据库操作需要使用Cursor类的实例，提供 execute() 方法，执行SQL语句，成功返回影响的行数
  • 数据库操作示例
    如数据库school中的student表如下，我们想新增一条数据：
id name age
1 jerry 28
2 tom 30
3 tina 29
4 cherry 18
5 qiumy 10 
import pymysql

HOST = '10.211.55.3'
USER = 'root'
PASSWORD = 'root'
DATABASE = 'school'
PORT = 3306

conn = None
cursor = None

try:
    conn = pymysql.connect(host=HOST, user=USER, password=PASSWORD, database=DATABASE, port=PORT)
    cursor = conn.cursor()

    sql = "insert into student (name, age) value ('lucy', 20)"
    r = cursor.execute(sql) # 返回影响的行数
    conn.commit() # 注释掉提交的代码，发现数据并没有插入，为什么呢？

except Exception as e:
    print(e, '~~~~')
    conn.rollback()

finally:
    if cursor:
        cursor.close()
    if conn:
        conn.close()

如果没有提交事务conn.commit()，发现数据并没有插入，为什么呢？查看源码：在Connection类的 __init__方法的注释中有:param autocommit: Autocommit mode. None means use server default.(default: False)，也就是说在建立的conn中事务是默认不自动提交的
这里一般不建议设置autocommit参数为True，而是手动管理事务

事务管理

  • Connection类有三个方法：begin 开始事务、commit 将变更提交、rollback 回滚事务
  • 当游标建立之时，就自动开始了一个隐形的数据库事务，因此不需要调用begin方法来开始事务

数据库查询

我们知道cursor.execute()执行sql语句返回的是影响的行数，如果是执行查询，也就不能得到select后的结果。这里cursor提供如下方法来获取查询结果：

  • fetchone()：获取结果集的下一行
  • fetchmany(size=None)：size指定返回的行数的行，默认size为None表示取1行(cursor.arraysize=1)
  • fetchall()：返回剩余所有行，如果走到末尾，就返回空元组，否则返回一个元组，其元素是每一行的记录封装的一个元组

注：fetch操作的是结果集，结果集是保存在客户端的，也就是说fetch的时候，查询已经结束了

  • cursor.rownumber：返回当前行号。可以修改，支持负数
  • cursor.rowcount：返回的总行数
import pymysql

HOST = '10.211.55.3'
USER = 'root'
PASSWORD = 'root'
DATABASE = 'school'
PORT = 3306

conn = None
cursor = None

try:
    conn = pymysql.connect(host=HOST, user=USER, password=PASSWORD, database=DATABASE, port=PORT)
    cursor = conn.cursor()

    sql = 'select * from student'
    r = cursor.execute(sql)
    # conn.commit()  # 仅仅是查询，可以不用提交事务

    print(1, r)  # 影响的行数
    print(2, cursor.fetchone())
    print(3, cursor.fetchone())
    print(4, cursor.rowcount, cursor.rownumber)
    print(5, cursor.fetchmany(3))
    print(6, cursor.rowcount, cursor.rownumber)
    print(7, cursor.fetchall())
    print(8, cursor.rowcount, cursor.rownumber)
    print(9, cursor.fetchall())

    cursor.rownumber = 0
    print(10, cursor.fetchall())

except Exception as e:
    print(e)
    conn.rollback()
finally:
    if cursor:
        cursor.close()
    if conn:
        conn.close()
  • 带列名的查询
    • conn.cursor(cursor=DictCursor)，默认为None，使用的是Cursor
    • 返回一行，是一个字典
    • 返回多行，放在列表中，元素是字典，代表一行
import pymysql
from pymysql.cursors import DictCursor

HOST = '10.211.55.3'
USER = 'root'
PASSWORD = 'root'
DATABASE = 'school'
PORT = 3306

conn = None
cursor = None

try:
    conn = pymysql.connect(host=HOST, user=USER, password=PASSWORD, database=DATABASE, port=PORT)
    cursor = conn.cursor(DictCursor)
    conn.begin()

    sql = 'select * from student'
    r = cursor.execute(sql)

    print(cursor.fetchmany(2))

except Exception as e:
    print(e)
    conn.rollback()
finally:
    if cursor:
        cursor.close()
    if conn:
        conn.close()
----------输出结果--------------
[{'id': 1, 'name': 'jerry', 'age': 28}, {'id': 2, 'name': 'tom', 'age': 30}]

SQL注入及参数化查询

SQL注入攻击

猜测后台数据库的查询语句使用拼接字符串等方式，从而经过设计为服务端传参，令其拼接出特殊字符串的SQL语句，返回攻击者想要的结果

  • SQL注入示例
import pymysql
from pymysql.cursors import DictCursor

HOST = '10.211.55.3'
USER = 'root'
PASSWORD = 'root'
DATABASE = 'school'
PORT = 3306

conn = None
cursor = None

try:
    conn = pymysql.connect(host=HOST, user=USER, password=PASSWORD, database=DATABASE, port=PORT)
    cursor = conn.cursor(DictCursor)

    name = "abc' or '1=1"
    # name = "'abc' or 1=1"
    # sql = 'select * from student where name={}'.format('abc') # 并没有name为abc的数据
    sql = "select * from student where name='{}'".format(name)   # 通过拼接字符串可查询到所有数据，尽管没有name为abc的数据
    print(sql)
    r = cursor.execute(sql)
    print(cursor.fetchall())

except Exception as e:
    print(e)
    conn.rollback()
finally:
    if cursor:
        cursor.close()
    if conn:
        conn.close()
------------输出结果-----------------
select * from student where name='abc' or '1=1'
[{'id': 1, 'name': 'jerry', 'age': 28}, {'id': 2, 'name': 'tom', 'age': 30}, {'id': 3, 'name': 'tina', 'age': 29}, {'id': 4, 'name': 'cherry', 'age': 18}, {'id': 5, 'name': 'qiumy', 'age': 10}, {'id': 6, 'name': 'lucy', 'age': 20}]

参数化查询

参数化查询，可以有效防止注入攻击，并提高查询的效率

  • execute(query, args=None)
  • args，必须是元组、列表或字典。如果查询字符串使用%(name)s，就必须使用字典
def execute(self, query, args=None):
    """Execute a query
    :param str query: Query to execute.
    :param args: parameters used with query. (optional)
    :type args: tuple, list or dict
    :return: Number of affected rows
    :rtype: int

    If args is a list or tuple, %s can be used as a placeholder in the query.
    If args is a dict, %(name)s can be used as a placeholder in the query.
    """
......
  • SQL注入解决示例（参数查询）
import pymysql
from pymysql.cursors import DictCursor

HOST = '10.211.55.3'
USER = 'root'
PASSWORD = 'root'
DATABASE = 'school'
PORT = 3306

conn = None
cursor = None

try:
    conn = pymysql.connect(host=HOST, user=USER, password=PASSWORD, database=DATABASE, port=PORT)
    cursor = conn.cursor(DictCursor)

    name = "'abc' or '1=1"
    sql = "select * from student where name=%s"
    print(sql)
    cursor.execute(sql, (name,))
    print(cursor.fetchall())

    sql = "select * from student where name like %(name)s and age>%(age)s"
    cursor.execute(sql, {'name': 't%', 'age': 20})
    print(cursor.fetchall())

except Exception as e:
    print(e)
    conn.rollback()
finally:
    if cursor:
        cursor.close()
    if conn:
        conn.close()

参数化查询为什么提高效率？SQL语句缓存
数据库服务器一般会对SQL语句编译和缓存，编译只对SQL语句部分，所以参数中就算有SQL指令也不会被当做指令执行，这就防止了注入风险
服务端会先查找是否对同一条查询语句进行了缓存，如果缓存未失效，则不需要再次编译，从而降低了编译的成本，降低了内存消耗

批量执行

  • cursor.executemany()，这里不做介绍，请参考官方文档
  • pymysql不支持分号分割的多条sql大字符串execute执行如"sql1;sql2;sql3"
  • 也可使用for循环来execute多条

connection和cursor的上下文管理

  • 查看源码
# Connection类和Cursor类的__enter__和__exit__
def __enter__(self):
    return self

def __exit__(self, *exc_info):
    del exc_info
    self.close()

进入时都返回类示例对象，退出时，都会关闭类示例对象

  • 演示示例（使用上下文管理简化如上代码）
import pymysql
from pymysql.cursors import DictCursor

HOST = '10.211.55.3'
USER = 'root'
PASSWORD = 'root'
DATABASE = 'school'
PORT = 3306

try:
    conn = pymysql.connect(host=HOST, user=USER, password=PASSWORD, database=DATABASE, port=PORT, cursorclass=DictCursor)   # 根据需要设置cursorclass参数为DictCursor，则结果为带列名的查询
    with conn:
        try:
            # with conn.cursor(DictCursor) as cursor:  # 如果不在conn中设置cursorclass参数，也可以在conn.cusror中设置带列名的查询
            with conn.cursor() as cursor:
                for i in range(5):
                    sql = f"insert into student (name, age) value ('lily{i}', 20+{i})"
                    # print(sql)
                    cursor.execute(sql)
            conn.commit()
            with conn.cursor() as cursor:
                name = 'jerry'
                sql = "select * from student where name=%s"
                cursor.execute(sql, (name,))
                print(cursor.fetchall())

        except Exception as e:
            print(e, '~~~~')
            conn.rollback()
except Exception as e:
    print(e)

参考

  • magedu