pymysql的使用

1 驱动

MySQL基于TCP协议之上开发，但是网络连接后，传输的数据必须遵循MySQL的协议。
封装好MySQL协议的包，就是驱动程序。

MySQL的驱动：

• MySQLdb: 最有名的库。对MySQL的C Client封装实现，支持Python2，不更新了，不支持Python3

• MySQL官方Connector

• pymysql: 语法兼容MySQLdb，使用Python写的库，支持Python3

2 pymysql的使用

连接数据库：pymysql.connect方法，返回Connections模块下的Connection类的实例。

• Connection类的事务管理：Connection.begin，开始事务， Connection.commit，提交， Connection.rollback，回滚。

游标cursor：操作数据库必须使用游标。需先获取一个游标对象：

• Connection.cursor(cursor=None)方法返回一个新的游标对象： cursor参数，可以指定一个Cursor类，比如DictCursor类,默认为Cursor类。
• Cursor类：Cursor.fetchone，获取结果集的下一行；Cursor.fetchmang，获取结果集的指定行数；Cursor.fetchall，返回结果集的所有行。
• DictCursor类：会将查询数据库结果集的每一行转换为字典，key为列名，value为值。

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

数据库交互的一般流程：
建立连接 – 获取游标 – 执行sql – 提交事务 – 释放资源

import pymysql
from pymysql.cursors import DictCursordef update_database():global connconn = Nonecursor = Nonetry:# 建立连接, 支持上下文with pymysql.connect(host="127.0.0.1", user="root", password="cli*963.", database="test") as conn:print(conn.ping(False))  # 测试数据库连接是否活着；参数reconnect，表示如果是断开时，是否重连# 获取一个cursorcursor = conn.cursor()for i in range(10):  # 批量提交，一般commit一般放到最后统一commit，效率更高。sql = "insert into tee values (12, 'zhonghua_{}', 21)".format(i)res = cursor.execute(sql)print(res)conn.commit()  # connection默认不commit：autocommit=Falseexcept Exception as err:print(err)conn.rollback()  # 异常回滚finally:if cursor:cursor.close()def get_database():sql = "select * from tee"try:with pymysql.connect(host="127.0.0.1", user="root", password="cli*963.", database="test") as _conn:with _conn.cursor() as cursor:line = cursor.execute(sql)print(line)print(cursor.fetchone())print(cursor.fetchone())print(cursor.fetchmany(2))print(cursor.fetchmany(2))print(cursor.fetchall())print(cursor.rownumber)cursor.rownumber = 0  # 改变游标位置，指向初始位置cursor.rownumber = -2  # 支持负向索引print(cursor.fetchone())print(cursor.rowcount)  # 返回总行数# fetch操作的是结果集，结果集是保存在客户端的，也就是说fetch的时候，查询结果已经结束了。print("*---------------------------*")with _conn.cursor(cursor=DictCursor) as cursor:  # Cursor类有一个Mixin的子类DictCursor，将返回结果保证为dictcursor.execute(sql)print(cursor.fetchone())print(cursor.fetchone())print(cursor.fetchmany(2))print(cursor.fetchmany(2))print(cursor.fetchall())except Exception as err:print("error:", err)

3 sql注入攻击

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

例如：使用字符串拼接sql语句select * from tee where id={}".format("10 or 1=1")进行查询。结果为select * from tee where id=10 or 1=1，where子句永远为真，导致整个表格被查出来。

永远不要相信客户端传来的数据是规范的及安全的！！！

如何解决注入攻击？
参数化查询，可以有效防止注入攻击，并提高查询的效率。： Cursor.execute(query, args=None)

• args，必须是元组、列表或字典。如果查询字符串使用%(name)s，就必须使用字典。

import pymysql
from pymysql.cursors import DictCursordef sql_injection_test():sql = "select * from tee where id={}".format("10 or 1=1")  # 字符串拼接，使where子句永远为真，导致整个表格被查出来。sql1 = "select * from tee where id=%s"# 解决注入攻击：参数化查询，可以有效防止注入攻击，并提高查询的效率try:with pymysql.connect(host="127.0.0.1", user="root", password="cli*963.", database="test") as _conn:with _conn.cursor(DictCursor) as cursor:cursor.execute(sql)  # 导致注入攻击print("aaaaaaaaaaaaaaa")print(cursor.fetchall())args = ("10 or 1=1",)cursor.execute(sql1, args=args)  # 参数化查询防止sql注入。 args可以是元组、列表、字典；为字典时，sql1 查询字符串必须是%(name)s格式print("bbbbbbbbbbbbbbb")print(cursor.fetchall())except Exception as err:print(err)

参数化查询为什么提高效率？
原因就是–SQL语句缓存。
数据库服务器一般会对SQL语句编译和缓存，编译只对SQL语句部分，所以参数中就算有SQL指令也不会被执行。

编译过程，需要词法分析、语法分析、生成AST、优化、生成执行计划等过程，比较耗费资源。服务端会先查找是否对同一条查询语句进行了缓存，如果缓存未失效，则不需要再次编译，从而降低了编译的成本，降低了内存消耗。

可以认为SQL语句字符串就是一个key，如果使用拼接方案，每次发过去的SQL语句都不一样，都需要编译并缓存。

大量查询的时候，首选使用参数化查询，以节省资源。

开发时，应该使用参数化查询。

注意：这里说的是查询字符串的缓存，不是查询结果的缓存。

4 pysql连接池实现

设计一个连接池：可以设置池大小的容器，连接池存放着数据库的连接。使用时，从池中获取一个连接，用完归还。从而减少频繁的创建、销毁数据库连接的过程，提高性能。

设计：

• 设计一个池对象ConnPool。构建时，传入连接数据库的相关参数（用户名、密码、主机、端口、数据库名）。
• 考虑多线程使用
• 使用get从池中拿走一个连接，用完归还。

import queue
import threading
import time
import loggingimport pymysql
from pymysql.cursors import DictCursorlogging.basicConfig(level=logging.INFO)class ConnPool:def __init__(self, size: int = 10, *, host: str, user: str, password: str, database: str, **kwargs):if not isinstance(size, int) or size < 1:size = 10self.size = sizeself._pool = queue.Queue()  # 使用队列作为连接池的容器。get时有连接，则获取，否则阻塞。for _ in range(size):self._pool.put(pymysql.connect(host=host, user=user, password=password, database=database, **kwargs))self.local = threading.local()  # 使用threading.local，记录每一个线程获取的连接实例，用完归还def get_conn(self):# 一个线程多次拿连接场景，单个线程未归还只能拿同一个，保证threading.local.conn记录的是同一个连接if getattr(self.local, "conn", None) is None:_conn = self._pool.get()self.local.conn = _connreturn self.local.conndef return_conn(self, _conn: pymysql.connect):if isinstance(_conn, pymysql.connect):self._pool.put(_conn)self.local.conn = None  # 归还连接后，threading.local应置为None# threading.local只能解决不同线程使用conn的问题，线程内必须同步方式使用，自动拿连接并规划，自动提交或回滚，避免线程内多次拿连接，或update多次后没有commit# 通过上下文，实现自动连接、自动提交或回滚，并归还连接。def __enter__(self):return self.get_conn()def __exit__(self, exc_type, exc_val, exc_tb):# __exit__前，不知道归还的连接是哪一个，正好使用threading.local，记录当前线程获取的是哪一个连接if exc_type:  # 存在异常，回滚self.local.conn.rollback()else:  # 没有异常，提交self.local.conn.commit()self.return_conn(self.local.conn)def foo(_pool: ConnPool):# conn = _pool.get_conn()time.sleep(3)with _pool as cur_conn:  # 使用连接池的上下文with cur_conn.cursor(DictCursor) as cursor:cursor.execute("select * from tee where id=%s", args=(10,))logging.info("{}:{}".format(threading.current_thread().name, cursor.rowcount))for column in cursor:  # cursor是一个可迭代对象，迭代的是查询的每一条记录logging.info("{}:{}".format(threading.current_thread().name, column))if __name__ == '__main__':# update_database()# get_database()# sql_injection_test()pool = ConnPool(host="127.0.0.1", user="root", password="cli*963.", database="test")# 连接池连接获取的游标cursor不能跨线程使用，因为线程A用完cursor可能关闭，而线程B还没使用，这时候会抛异常。# 使用队列实现线程池，也可以改用信号量来实现。for i in range(8):threading.Thread(target=foo, args=(pool,), name="foo_{}".format(i)).start()

该连接池存在的问题：

• 连接池连接获取的游标cursor不能跨线程使用，因为线程A用完cursor可能关闭，而线程B还没使用，这时候会抛异常。
• 使用队列实现线程池，也可以改用信号量来实现。