簡單介紹我的開源小工具 SanicDB

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吧。

**倉庫：

首先，看看它的初始化

class sanicdb:
"""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中使用的例子

簡單介紹我的開源小工具 SanicDB

InSAR開源小工具

簡單小工具的使用

python小工具 tqdm介紹

簡單介紹我的開源小工具 SanicDB

InSAR開源小工具

簡單小工具的使用

python小工具 tqdm介紹

相關推薦