本文使用案例是基于 python2.7 實(shí)現(xiàn)
以下內(nèi)容均為個(gè)人使用 peewee 的經(jīng)驗(yàn)和遇到的坑，不會(huì)涉及過(guò)多的基本操作。所以，沒(méi)有使用過(guò) peewee，可以先閱讀文檔
正確性和覆蓋面有待提高，如果遇到新的問(wèn)題歡迎討論。
一、介紹
Peewee 是一個(gè)簡(jiǎn)單、輕巧的 Python ORM。
總而言之，peewee 可以完全可以應(yīng)付個(gè)人或企業(yè)的中小型項(xiàng)目的 Model 層，上手容易，功能很強(qiáng)大。
二、基本使用方法
from peewee import *db = SqliteDatabase('people.db')class BaseModel(Model): class Meta: database = db # This model uses the "people.db" database.class Person(BaseModel): name = CharField() birthday = DateField() is_relative = BooleanField() 基本的使用方法，推薦閱讀文檔－－quickstart
三、推薦使用姿勢(shì)
下面介紹一些我在使用過(guò)程的經(jīng)驗(yàn)和遇到的坑，希望可以幫助大家更好的使用 peewee。
3.1 連接數(shù)據(jù)庫(kù)
連接數(shù)據(jù)庫(kù)時(shí)，推薦使用 playhouse 中的 db_url 模塊。db_url 的 connect 方法可以通過(guò)傳入的 URL 字符串，生成數(shù)據(jù)庫(kù)連接。
3.1.1 connect(url, **connect_params)
通過(guò)傳入的 url 字符串，創(chuàng)建一個(gè)數(shù)據(jù)庫(kù)實(shí)例
url形如：
注意：charset 默認(rèn)為utf8。如需要支持 emoji ，charset 設(shè)置為utf8mb4，同時(shí)保證創(chuàng)建數(shù)據(jù)庫(kù)時(shí)的字符集設(shè)置正確CREATE DATABASE mydatabase CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;。
支持的 schemes：
3.1.2 推薦姿勢(shì)
from playhouse.db_url import connectfrom dock.common import config＃ url: mysql+pool://root:root@127.0.0.1:3306/appmanage?max_connections=300&stale_timeout=300mysql_config_url = config_dict.get('config').get('mysql').get('url')db = connect(url=mysql_config_url)查看更多詳情請(qǐng)移步官方文檔：db-url
3.2 連接池的使用
peewee 的連接池，使用時(shí)需要顯式的關(guān)閉連接。下面先說(shuō)下為什么，最后會(huì)給出推薦的使用方法，避免進(jìn)坑。
3.2.1 為什么要顯式的關(guān)閉連接
Connections will not be closed exactly when they exceed their stale_timeout. Instead, stale connections are only closed when a new connection is requested.
這里引用官方文檔的提示。大致說(shuō)：“超時(shí)連接不會(huì)自動(dòng)關(guān)閉，只會(huì)在有新的請(qǐng)求時(shí)是才會(huì)關(guān)閉”。這里的request是指‘web 框架處理的請(qǐng)求'，peewee 源碼片段：
def _connect(self, *args, **kwargs): while True: try: # Remove the oldest connection from the heap. ts, conn = heapq.heappop(self._connections) # _connections是連接實(shí)例的list(pool) key = self.conn_key(conn) except IndexError: ts = conn = None logger.debug('No connection available in pool.') break else: if self._is_closed(key, conn): # This connecton was closed, but since it was not stale # it got added back to the queue of available conns. We # then closed it and marked it as explicitly closed, so # it's safe to throw it away now. # (Because Database.close() calls Database._close()). logger.debug('Connection %s was closed.', key) ts = conn = None self._closed.discard(key) elif self.stale_timeout and self._is_stale(ts): # If we are attempting to check out a stale connection, # then close it. We don't need to mark it in the "closed" # set, because it is not in the list of available conns # anymore. logger.debug('Connection %s was stale, closing.', key) self._close(conn, True) self._closed.discard(key) ts = conn = None else: break if conn is None: if self.max_connections and ( len(self._in_use) >= self.max_connections): raise ValueError('Exceeded maximum connections.') conn = super(PooledDatabase, self)._connect(*args, **kwargs) ts = time.time() key = self.conn_key(conn) logger.debug('Created new connection %s.', key) self._in_use[key] = ts # 使用中的數(shù)據(jù)庫(kù)連接實(shí)例dict return conn 根據(jù) pool 庫(kù)中的 _connect 方法的代碼可知：每次在建立數(shù)據(jù)庫(kù)連接時(shí)，會(huì)檢查連接實(shí)例是否超時(shí)。但是需要注意一點(diǎn)：使用中的數(shù)據(jù)庫(kù)連接實(shí)例（_in_use dict中的數(shù)據(jù)庫(kù)連接實(shí)例)，是不會(huì)在創(chuàng)建數(shù)據(jù)庫(kù)連接時(shí)，檢查是否超時(shí)的。
因?yàn)檫@段代碼中，每次創(chuàng)建連接實(shí)例，都是在 _connections(pool) 取實(shí)例，如果有的話就判斷是否超時(shí)；如果沒(méi)有的話就新建。
然而，使用中的數(shù)據(jù)庫(kù)連接并不在 _connections 中，所以每次創(chuàng)建數(shù)據(jù)庫(kù)連接實(shí)例時(shí)，并沒(méi)有檢測(cè)使用中的數(shù)據(jù)庫(kù)連接實(shí)例是否超時(shí)。
只有調(diào)用連接池實(shí)例的 _close 方法。執(zhí)行這個(gè)方法后，才會(huì)把使用后的連接實(shí)例放回到 _connections (pool)。
def _close(self, conn, close_conn=False): key = self.conn_key(conn) if close_conn: self._closed.add(key) super(PooledDatabase, self)._close(conn) # 關(guān)閉數(shù)據(jù)庫(kù)連接的方法 elif key in self._in_use: ts = self._in_use[key] del self._in_use[key] if self.stale_timeout and self._is_stale(ts): # 到這里才會(huì)判斷_in_use中的連接實(shí)例是否超時(shí) logger.debug('Closing stale connection %s.', key) super(PooledDatabase, self)._close(conn) # 超時(shí)的話，關(guān)閉數(shù)據(jù)庫(kù)連接 else: logger.debug('Returning %s to pool.', key) heapq.heappush(self._connections, (ts, conn)) # 沒(méi)有超時(shí)的話，放回到pool中3.2.2 如果不顯式的關(guān)閉連接，會(huì)出現(xiàn)的問(wèn)題
如果不調(diào)用_close方法的話，使用后 的數(shù)據(jù)庫(kù)連接就一直不會(huì)關(guān)閉(兩個(gè)含義：回到pool中和關(guān)閉數(shù)據(jù)庫(kù)連接)，這樣會(huì)造成兩個(gè)問(wèn)題：
1.每次都是新建數(shù)據(jù)庫(kù)連接，因?yàn)?pool 中沒(méi)有數(shù)據(jù)庫(kù)連接實(shí)例。會(huì)導(dǎo)致稍微有一點(diǎn)并發(fā)量就會(huì)返回Exceeded maximum connections.錯(cuò)誤
2.MySQL也是有 timeout 的，如果一個(gè)連接長(zhǎng)時(shí)間沒(méi)有請(qǐng)求的話，MySQL Server 就會(huì)關(guān)閉這個(gè)連接，但是，peewee的已建立（后面會(huì)解釋為什么特指已建立的）的連接實(shí)例，并不知道 MySQL Server 已經(jīng)關(guān)閉了，再去通過(guò)這個(gè)連接請(qǐng)求數(shù)據(jù)的話，就會(huì)返回 Error 2006: “MySQL server has gone away”錯(cuò)誤，根據(jù)官方文檔
3.2.3 推薦姿勢(shì)
所以，每次操作完數(shù)據(jù)庫(kù)就關(guān)閉連接實(shí)例。
用法1：使用with
def send_rule(): with db.execution_context(): # A new connection will be opened or, if using a connection pool, # pulled from the pool of available connections. Additionally, a # transaction will be started. for user in get_all_user(): user_id = user['id'] rule = Rule(user_id) rule_dict = rule.slack_rule(index) .....do something.....
用法2：使用Flask hook
@app.before_requestdef _db_connect(): database.connect()## This hook ensures that the connection is closed when we've finished# processing the request.@app.teardown_requestdef _db_close(exc): if not database.is_closed(): database.close()### 更優(yōu)雅的用法：from playhouse.flask_utils import FlaskDBfrom dock_fastgear.model.base import db#app = Flask(__name__)FlaskDB(app, db) # 這樣就自動(dòng)做了上面的事情（具體實(shí)現(xiàn)可查看http://docs.peewee-orm.com/en/latest/peewee/playhouse.html?highlight=Flask%20DB#flask-utils）
查看更多詳情請(qǐng)移步官方文檔：pool-apis
3.3 處理查詢結(jié)果
這里沒(méi)有什么大坑，就是有兩點(diǎn)需要注意：
首先，查詢的結(jié)果都是該 Model 的 object，注意不是 dict。如果想讓結(jié)果為 dict，需要 playhouse 模塊的工具方法進(jìn)行轉(zhuǎn)化：from playhouse.shortcuts import model_to_dict
其次，get方法只會(huì)返回一條記錄
3.3.1 推薦姿勢(shì)
from playhouse.shortcuts import model_to_dictfrom model import HelloGitHubdef read_from_db(input_vol): content_list = [] category_object_list = HelloGitHub.select(HelloGitHub.category).where(HelloGitHub.vol == input_vol)/ .group_by(HelloGitHub.category).order_by(HelloGitHub.category) for fi_category_object in category_object_list: hellogithub = HelloGitHub.select()/ .where((HelloGitHub.vol == input_vol) & (HelloGitHub.category == fi_category_object.category))/ .order_by(HelloGitHub.create_time) for fi_hellogithub in hellogithub: content_list.append(model_to_dict(fi_hellogithub)) return content_list
四、常見錯(cuò)誤及解決辦法
4.1 'buffer' object has no attribute 'translate'
4.2 Can't connect to MySQL server Lost connection to MySQL server during query
以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持VEVB武林網(wǎng)。
新聞熱點(diǎn)
疑難解答
圖片精選
