?

SanicDB 是為 Python的異步 Web 框架 Sanic 方便操作MySQL而開發的工具，是對 aiomysql.Pool 的輕量級封裝。Sanic 是異步IO的Web框架，同時用異步IO讀寫MySQL才能更大發揮它的效率。雖然這個模塊叫做 SanicDB，但是它可以用在任何異步IO操作MySQL的地方。

SanicDB的靈感來源于tornado里面的對MySQLdb（MySQL的C語言接口的Python封裝）的操作，后來tornado把它去除了，就有人把這部分代碼寫成一個單獨的模塊叫做 torndb，torndb是對python-mysql的封裝，不能進行異步操作。但是它很簡潔，用起來已經非常習慣了。

在用Sanic的時候，發現有異步IO的aiomysql可以用，但是用起來還有點麻煩，于是就著手對aiomysql進行封裝，既然aiomysql支持連接池，那就直接封裝aiomysql.Pool吧。

代碼倉庫：?https://github.com/veelion/sanicdb

首先，看看它的初始化

class SanicDB: """A lightweight wrapper around aiomysql.Pool for easy to use """ def __init__(self, host, database, user, password,loop=None, sanic=None,minsize=3, maxsize=5,return_dict=True,pool_recycle=7*3600,autocommit=True,charset = "utf8mb4", **kwargs): '''

?

前面4個參數是對應數據庫的；

loop 是你的應用程序中的事件循環（event_loop）；

sanic 是你的Sanic 應用中的sanic 對象；

loop 和 sanic 兩者只需提供一個即可，當同時存在時，連接數據庫其實用的是sanic里面的loop，這在__init__()代碼中可以看到。

minisize 和 maxsize 是連接池的個數限制；

return_dict 是返回的數據一條記錄為一個dict，key是MySQL表的字段名，value是字段的值；

pool_recycle 是連接池重連的時間間隔，MySQL默認的連接閑置時間是8小時；

對MySQL的讀寫操作方法：

async def query(self, query, *parameters, **kwparameters):?
async def execute(self, query, *parameters, **kwparameters):
async def get(self, query, *parameters, **kwparameters):

從代碼中可以看到，以上三個函數都是調用的cursor的execute() 執行 SQL 命令，但是query(), get() 用于讀操作返回數據，query返回全部數據（list），get只返回一條數據。

execute() 方法用來執行寫操作，返回受影響的行ID。

上面三個函數都是一樣的：

query 是要執行的SQL語句， 后面兩個參數是用來參數化執行sql語句的。參數化執行sql是為了防止SQL注入的，那么參數化執行SQL是什么意思呢，舉個例子就目標了。

比如，進行用戶登錄時，我們要執行一條SQL，Python拼接SQL如下：

sql = "select * from user where name='%s' and password='%s'" % (input_name, input_password)

?

input_name 和 input_password 都是用戶輸入的，如果用戶想搗鬼，他把input_name 輸入成 “ ‘ or 1 = 1 –”，拼接出來的sql就成了：

select * from user where name='' or 1 = 1 --' and password='anything'

?

or 后面的 1=1 就是True，而 — 后面都成了注釋，這條語句不管input_name, input_password 是什么都可以執行，通過用戶驗證，達到了注入目的。

參數化執行SQL就是把拼接SQL的任務交給SanicDB（實際上是它后的aiomysql后面的pymysql），我們只把要拼接的參數當做*parameters傳給讀寫方法即可：

mydb.query("select * from user where user=%s and password=%s", input_name, input_password)

?

更高級的封裝：

query(), get(), execute() 就是基本的操作，在此基礎上，做了進一步封裝，讓寫應用更方便：

# 檢查一個table中是否含有某字段為某值的記錄 async def table_has(self, table_name, field, value):# 把一個item插入一個表，item為dict，key是字段名，# ignore_duplicated為True時，遇到唯一索引重復時不報錯 async def table_insert(self, table_name, item, ignore_duplicated=True):# 更新一個table的記錄，updates是dict，key為字段名 async def table_update(self, table_name, updates, field_where, value_where):

?

?

示例

代碼倉庫中有兩個示例：

test.py?是在普通asyncio程序中使用異步MySQL的例子

example.py?是在Sanic Web中使用的例子

特別說明一下，Sanic 對象 app 中還沒有生成 event_loop 對象（它是在app.run() 之后生成的），所以在Sanic中初始化SanicDB時（app.run()）之前，運行sql語句會報錯。如果需要創建連接后初始化運行一些SQL操作，可以利用Sanic的Listner：

@app.listener('before_server_start') async def setup_db(app, loop):app.db = await db_setup()

?

?

文章首發于我的技術博客猿人學Python基礎教程

轉載于:https://www.cnblogs.com/amiza/p/10313153.html

總結

以上是生活随笔為你收集整理的python自学笔记之开源小工具：SanicDB介绍的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。