本文使用案例是基于 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。

1. 簡(jiǎn)單、輕巧、富有表現(xiàn)力（原詞 expressive ）的ORM
2. 支持python版本 2.6+ 和 3.2+
3. 支持?jǐn)?shù)據(jù)庫(kù)包括：sqlite， mysql and postgresql
4. 包含一堆實(shí)用的擴(kuò)展在 playhouse 模塊中

總而言之，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形如：

1. mysql://user:passwd@ip:port/my_db 將創(chuàng)建一個(gè) 本地 MySQL 的 my_db 數(shù)據(jù)庫(kù)的實(shí)例（will create a MySQLDatabase instance）
2. mysql+pool://user:passwd@ip:port/my_db?charset=utf8&max_connections=20&stale_timeout=300 將創(chuàng)建一個(gè)本地 MySQL 的 my_db 的連接池，最大連接數(shù)為20（In a multi-threaded application, up to max_connections will be opened. Each thread (or, if using gevent, greenlet) will have it's own connection. ），超時(shí)時(shí)間為300秒（will create a PooledMySQLDatabase instance）

注意：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：

1. apsw: APSWDatabase
2. mysql: MySQLDatabase
3. mysql+pool: PooledMySQLDatabase
4. postgres: PostgresqlDatabase
5. postgres+pool: PooledPostgresqlDatabase
6. postgresext: PostgresqlExtDatabase
7. postgresext+pool: PooledPostgresqlExtDatabase
8. sqlite: SqliteDatabase
9. sqliteext: SqliteExtDatabase
10. sqlite+pool: PooledSqliteDatabase
11. sqliteext+pool: PooledSqliteExtDatabase

3.1.2 推薦姿勢(shì)

from playhouse.db_url import connect

from dock.common import config

＃ url: mysql+pool://root:root@127.0.0.1:3306/appmanage?max_connections=300&stale_timeout=300
mysql_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í)例，如果有的話(huà)就判斷是否超時(shí)；如果沒(méi)有的話(huà)就新建。

然而，使用中的數(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í)的話(huà)，關(guān)閉數(shù)據(jù)庫(kù)連接
    else:
      logger.debug('Returning %s to pool.', key)
      heapq.heappush(self._connections, (ts, conn)) # 沒(méi)有超時(shí)的話(huà)，放回到pool中

3.2.2 如果不顯式的關(guān)閉連接，會(huì)出現(xiàn)的問(wèn)題

如果不調(diào)用_close方法的話(huà)，使用后 的數(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)求的話(huà)，MySQL Server 就會(huì)關(guān)閉這個(gè)連接，但是，peewee的已建立（后面會(huì)解釋為什么特指已建立的）的連接實(shí)例，并不知道 MySQL Server 已經(jīng)關(guān)閉了，再去通過(guò)這個(gè)連接請(qǐng)求數(shù)據(jù)的話(huà)，就會(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_request
def _db_connect():
  database.connect()
#
# This hook ensures that the connection is closed when we've finished
# processing the request.
@app.teardown_request
def _db_close(exc):
  if not database.is_closed():
    database.close()
#
#
# 更優(yōu)雅的用法：
from playhouse.flask_utils import FlaskDB
from 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 處理查詢(xún)結(jié)果

這里沒(méi)有什么大坑，就是有兩點(diǎn)需要注意：

首先，查詢(xú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_dict
from model import HelloGitHub

def 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

四、常見(jiàn)錯(cuò)誤及解決辦法

4.1 'buffer' object has no attribute 'translate'

1. 錯(cuò)誤信息： "'buffer' object has no attribute 'translate'"
2. 場(chǎng)景：BlobField 字段存儲(chǔ)zlib compress壓縮的數(shù)據(jù)
3. 解決辦法：需要指定pymysql的版本小于0.6.7 否則會(huì)報(bào)錯(cuò)
4. 參考

4.2 Can't connect to MySQL server Lost connection to MySQL server during query

1. 錯(cuò)誤信息：Can't connect to MySQL server Lost connection to MySQL server during query
2. 場(chǎng)景：向 RDS 中插入數(shù)據(jù)
3. 解決辦法：因?yàn)檎?qǐng)求的連接數(shù)過(guò)多，達(dá)到了 RDS 設(shè)置的連接數(shù)，所以需要調(diào)高 RDS 連接數(shù)
4. 參考

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

最新評(píng)論