Playhouse，Peewee 的擴充功能

Peewee 帶有許多擴充模組，這些模組收集在 playhouse 命名空間下。儘管名稱有點傻，但有一些非常實用的擴充功能，特別是那些公開特定供應商資料庫功能的擴充功能，例如 SQLite 擴充功能 和 Postgresql 擴充功能。

以下您將找到構成 playhouse 的各種模組的鬆散組織列表。

資料庫驅動程式/特定供應商資料庫功能

高階功能

資料庫管理和框架整合

Sqlite 擴充功能

Sqlite 擴充功能已移至 它們自己的頁面。

SqliteQ

playhouse.sqliteq 模組提供 SqliteExtDatabase 的子類別，它將序列化對 SQLite 資料庫的並行寫入。SqliteQueueDatabase 可以作為常規 SqliteDatabase 的直接替代品使用，如果您想要從**多個執行緒**簡單地**讀寫**存取 SQLite 資料庫。

SQLite 一次只允許一個連線寫入資料庫。因此，如果您有一個多執行緒的應用程式（例如 Web 伺服器）需要寫入資料庫，您可能會看到偶爾的錯誤，當一個或多個嘗試寫入的執行緒無法取得鎖定時。

SqliteQueueDatabase 旨在透過單一的、長時間存在的連線傳送所有寫入查詢來簡化操作。好處是您可以在沒有衝突或逾時的情況下，讓多個執行緒寫入資料庫。然而，缺點是您無法發出包含多個查詢的寫入交易 – 本質上，所有寫入都在自動提交模式下執行。

注意

該模組的名稱來自於所有寫入查詢都被放入執行緒安全的佇列中。單一的工作執行緒會監聽佇列並執行所有傳送給它的查詢。

交易

由於所有查詢都由單一工作執行緒序列化和執行，因此來自不同執行緒的事務 SQL 可能會以亂序執行。在下面的範例中，執行緒「B」開始的事務被執行緒「A」回滾（會造成不良後果！）

  • 執行緒 A：UPDATE transplants SET organ='liver', ...;

  • 執行緒 B：BEGIN TRANSACTION;

  • 執行緒 B：UPDATE life_support_system SET timer += 60 …;

  • 執行緒 A：ROLLBACK; – 糟了…。

由於來自不同交易的查詢有可能被穿插，因此 transaction() 和 atomic() 方法在 SqliteQueueDatabase 上被禁用。

如果您希望臨時從不同的執行緒寫入資料庫，您可以使用 pause() 和 unpause() 方法。這些方法會阻塞呼叫者，直到寫入器執行緒完成目前的工作負載。然後，寫入器會斷開連線，呼叫者會接管，直到呼叫 unpause。

stop()、start() 和 is_stopped() 方法也可以用來控制寫入器執行緒。

注意

請查看 SQLite 的 隔離 文件，以獲取有關 SQLite 如何處理並行連線的更多資訊。

程式碼範例

建立資料庫實例不需要任何特殊處理。但是，SqliteQueueDatabase 接受一些您應該注意的特殊參數。如果您使用 gevent，您必須在實例化資料庫時指定 use_gevent=True – 這樣 Peewee 才知道使用適當的物件來處理佇列、執行緒建立和鎖定。

from playhouse.sqliteq import SqliteQueueDatabase

db = SqliteQueueDatabase(
    'my_app.db',
    use_gevent=False,  # Use the standard library "threading" module.
    autostart=False,  # The worker thread now must be started manually.
    queue_max_size=64,  # Max. # of pending writes that can accumulate.
    results_timeout=5.0)  # Max. time to wait for query to be executed.

如果 autostart=False，如上面的範例所示，您需要呼叫 start() 來啟動將執行實際寫入查詢的工作執行緒。

@app.before_first_request
def _start_worker_threads():
    db.start()

如果您計劃執行 SELECT 查詢或通常想要存取資料庫，您需要呼叫 connect() 和 close()，就像您使用任何其他資料庫實例一樣。

當您的應用程式準備終止時，請使用 stop() 方法來關閉工作執行緒。如果存在未完成的工作，則此方法將會阻塞，直到所有待處理的工作完成（儘管不允許新增任何新工作）。

import atexit

@atexit.register
def _stop_worker_threads():
    db.stop()

最後，is_stopped() 方法可用於確定資料庫寫入器是否已啟動並執行。

Sqlite 使用者定義函式

sqlite_udf playhouse 模組包含許多使用者定義的函式、聚合函式和表值函式，您可能會覺得有用。這些函式分組在集合中，您可以單獨、按集合或註冊所有這些使用者定義的擴充功能。

純量函式是接受多個參數並傳回單一值的函式。例如，將字串轉換為大寫，或計算 MD5 十六進位摘要。

聚合函式類似於對多行資料進行操作並產生單一結果的純量函式。例如，計算整數列表的總和，或找到特定欄中的最小值。

表值函式只是可以傳回多行資料的函式。例如，傳回給定字串中所有符合項目的正規表示式搜尋函式，或是接受兩個日期並產生所有中間日期的函式。

注意

若要使用表值函數，您需要建置 playhouse._sqlite_ext C 擴充功能。

註冊使用者定義的函數

db = SqliteDatabase('my_app.db')

# Register *all* functions.
register_all(db)

# Alternatively, you can register individual groups. This will just
# register the DATE and MATH groups of functions.
register_groups(db, 'DATE', 'MATH')

# If you only wish to register, say, the aggregate functions for a
# particular group or groups, you can:
register_aggregate_groups(db, 'DATE')

# If you only wish to register a single function, then you can:
from playhouse.sqlite_udf import gzip, gunzip
db.register_function(gzip, 'gzip')
db.register_function(gunzip, 'gunzip')

使用函式庫函數（“hostname”）

# Assume we have a model, Link, that contains lots of arbitrary URLs.
# We want to discover the most common hosts that have been linked.
query = (Link
         .select(fn.hostname(Link.url).alias('host'), fn.COUNT(Link.id))
         .group_by(fn.hostname(Link.url))
         .order_by(fn.COUNT(Link.id).desc())
         .tuples())

# Print the hostname along with number of links associated with it.
for host, count in query:
    print('%s: %s' % (host, count))

函數列表，依集合名稱

純量函數以 (f) 表示，聚合函數以 (a) 表示，表值函數以 (t) 表示。

CONTROL_FLOW

if_then_else(cond, truthy[, falsey=None])

簡單的三元運算子，根據 cond 參數的真值，將傳回 truthy 或 falsey 值。

DATE

strip_tz(date_str)
參數

date_str – 以字串編碼的日期時間。

傳回值

已移除任何時區資訊的日期時間。

時間不會以任何方式調整，只是移除時區。

humandelta(nseconds[, glue=', '])
參數

  • nseconds (int) – 時間差的總秒數。

  • glue (str) – 用於連接值的片段。

傳回值

易於閱讀的時間差描述。

範例，86471 -> “1 天，1 分鐘，11 秒”

mintdiff(datetime_value)
參數

datetime_value – 日期時間。

傳回值

列表中任何兩個值之間的最小差異。

計算任何兩個日期時間之間最小差異的聚合函數。

avgtdiff(datetime_value)
參數

datetime_value – 日期時間。

傳回值

列表中值之間的平均差異。

計算列表中連續值之間平均差異的聚合函數。

duration(datetime_value)
參數

datetime_value – 日期時間。

傳回值

列表中最小值到最大值的持續時間，以秒為單位。

計算列表中最小值到最大值的持續時間的聚合函數，以秒為單位傳回。

date_series(start, stop[, step_seconds=86400])
參數

  • start (datetime) – 開始日期時間

  • stop (datetime) – 結束日期時間

  • step_seconds (int) – 構成一個步驟的秒數。

表值函數，傳回從開始到結束迭代時所遇到的日期/時間值所組成的列，每次遞增 step_seconds。

此外，如果 start 沒有時間元件，且 step_seconds 大於或等於一天（86400 秒），則傳回的值將是日期。相反地，如果 start 沒有日期元件，則值將以時間形式傳回。否則，值將以日期時間形式傳回。

範例

SELECT * FROM date_series('2017-01-28', '2017-02-02');

value
-----
2017-01-28
2017-01-29
2017-01-30
2017-01-31
2017-02-01
2017-02-02

FILE

file_ext(filename)
參數

filename (str) – 要從中提取副檔名的檔案名稱。

傳回值

傳回檔案副檔名，包括開頭的 “.”。

file_read(filename)
參數

filename (str) – 要讀取的檔案名稱。

傳回值

檔案的內容。

HELPER

gzip(data[, compression=9])
參數

  • data (bytes) – 要壓縮的資料。

  • compression (int) – 壓縮等級（9 為最大）。

傳回值

壓縮的二進位資料。

gunzip(data)
參數

data (bytes) – 壓縮的資料。

傳回值

解壓縮的二進位資料。

hostname(url)
參數

url (str) – 要從中提取主機名稱的 URL。

傳回值

URL 的主機名稱部分

toggle(key)
參數

key – 要切換的鍵。

在 True/False 狀態之間切換鍵。範例

>>> toggle('my-key')
True
>>> toggle('my-key')
False
>>> toggle('my-key')
True
setting(key[, value=None])
參數

  • key – 要設定/擷取的鍵。

  • value – 要設定的值。

傳回值

與鍵關聯的值。

在記憶體中儲存/擷取設定，並在應用程式生命週期中保持。若要取得目前的值，只需指定鍵。若要設定新值，請使用鍵和新值呼叫。

clear_toggles()

清除與 toggle() 函數關聯的所有狀態。

clear_settings()

清除與 setting() 函數關聯的所有狀態。

MATH

randomrange(start[, stop=None[, step=None]])
參數

  • start (int) – 範圍的開始（包含）

  • end (int) – 範圍的結束（不包含）

  • step (int) – 傳回值的間隔。

傳回 [start, end) 之間的一個隨機整數。

gauss_distribution(mean, sigma)
參數

  • mean (float) – 平均值

  • sigma (float) – 標準差

sqrt(n)

計算 n 的平方根。

tonumber(s)
參數

s (str) – 要轉換為數字的字串。

傳回值

整數、浮點數或失敗時為 NULL。

mode(val)
參數

val – 列表中的數字。

傳回值

眾數，或最常觀察到的數字。

計算數值眾數的聚合函數。

minrange(val)
參數

val – 值

傳回值

兩個值之間的最小差異。

計算序列中兩個數字之間最小距離的聚合函數。

avgrange(val)
參數

val – 值

傳回值

值之間的平均差異。

計算序列中兩個連續數字之間平均距離的聚合函數。

range(val)
參數

val – 值

傳回值

序列中最小值到最大值的範圍。

傳回觀察到的值範圍的聚合函數。

median(val)
參數

val – 值

傳回值

序列中的中位數或中間值。

計算序列中中間值的聚合函數。

注意

只有在您編譯 _sqlite_udf 擴充功能時才可用。

STRING

substr_count(haystack, needle)

傳回 needle 在 haystack 中出現的次數。

strip_chars(haystack, chars)

從 haystack 的開頭和結尾移除 chars 中的任何字元。

damerau_levenshtein_dist(s1, s2)

使用 Levenshtein 演算法的 Damerau 變體計算從 s1 到 s2 的編輯距離。

注意

只有在您編譯 _sqlite_udf 擴充功能時才可用。

levenshtein_dist(s1, s2)

使用 Levenshtein 演算法計算從 s1 到 s2 的編輯距離。

注意

只有在您編譯 _sqlite_udf 擴充功能時才可用。

str_dist(s1, s2)

使用標準函式庫 SequenceMatcher 的演算法計算從 s1 到 s2 的編輯距離。

注意

只有在您編譯 _sqlite_udf 擴充功能時才可用。

參數

  • regex (str) – 正規表示式

  • search_string (str) – 要搜尋符合正規表示式字串的字串。

表值函式，在字串中搜尋符合提供的 regex 的子字串。為每個找到的符合項傳回列。

範例

SELECT * FROM regex_search('\w+', 'extract words, ignore! symbols');

value
-----
extract
words
ignore
symbols

apsw，一個進階的 sqlite 驅動程式

apsw_ext 模組包含一個資料庫類別，適用於 apsw sqlite 驅動程式。

APSW 專案頁面： https://github.com/rogerbinns/apsw

APSW 是一個非常棒的函式庫，它在 SQLite 的 C 介面上提供了一個輕薄的封裝，使得可以使用 SQLite 的所有進階功能。

以下是使用 APSW 的幾個原因，摘自文件

  • APSW 提供 SQLite 的所有功能，包括虛擬表格、虛擬檔案系統、blob i/o、備份和檔案控制。

  • 連線可以在執行緒之間共享，而無需任何額外的鎖定。

  • 交易由您的程式碼明確管理。

  • APSW 可以處理巢狀交易。

  • 正確處理 Unicode。

  • APSW 速度更快。

如需更多關於 apsw 和 pysqlite 之間差異的資訊，請查看 apsw 文件。

如何使用 APSWDatabase

from apsw_ext import *

db = APSWDatabase(':memory:')

class BaseModel(Model):
    class Meta:
        database = db

class SomeModel(BaseModel):
    col1 = CharField()
    col2 = DateTimeField()

apsw_ext API 註解

APSWDatabase 擴展了 SqliteExtDatabase 並繼承其進階功能。

class APSWDatabase(database, **connect_kwargs)
參數

  • database (string) – sqlite 資料庫的檔案名稱

  • connect_kwargs – 開啟連線時傳遞給 apsw 的關鍵字引數

register_module(mod_name, mod_inst)

提供一種全域註冊模組的方法。如需更多資訊，請參閱 關於虛擬表格的文件。

參數

  • mod_name (string) – 模組使用的名稱

  • mod_inst (object) – 一個實作 虛擬表格 介面的物件

unregister_module(mod_name)

取消註冊模組。

參數

mod_name (string) – 模組使用的名稱

注意

請務必使用 apsw_ext 模組中定義的 Field 子類別，因為它們會正確處理資料類型以進行儲存。

例如，請務必匯入並使用 playhouse.apsw_ext.DateTimeField，而不是使用 peewee.DateTimeField。

Sqlcipher 後端

注意

雖然此擴充功能的程式碼很短，但尚未經過適當的同儕審查，可能存在漏洞。

另請注意，此程式碼依賴於 sqlcipher3（python 綁定）和 sqlcipher，而且那裡的程式碼也可能存在漏洞，但由於這些是廣泛使用的加密模組，我們可以預期那裡會有「短暫的零日漏洞」。

sqlcipher_ext API 註解

class SqlCipherDatabase(database, passphrase, **kwargs)

SqliteDatabase 的子類別，用於儲存加密的資料庫。它不是使用標準的 sqlite3 後端，而是使用 sqlcipher3：一個 sqlcipher 的 python 封裝，而 sqlcipher 本身是 sqlite3 的加密封裝，因此 API 與 SqliteDatabase 的 API *相同*，除了物件建構參數之外

參數

  • database – 要開啟 [或建立] 的加密資料庫檔案名稱路徑。

  • passphrase – 資料庫加密密碼：長度應至少為 8 個字元，但*強烈建議*在您的實作中加強更好的 密碼強度 準則。

  • 如果 database 檔案不存在，將會使用從 passhprase 衍生出的金鑰*建立*加密檔案。

  • 嘗試開啟現有資料庫時，passhprase 應該與建立時使用的密碼相同。如果密碼不正確，第一次嘗試存取資料庫時會引發錯誤。

rekey(passphrase)
參數

passphrase (str) – 資料庫的新密碼。

變更資料庫的密碼。

注意

可以使用許多擴充 PRAGMA 來設定 SQLCipher。PRAGMA 及其描述的清單可以在 SQLCipher 文件中找到。

例如，指定金鑰衍生 (在 SQLCipher 3.x 中為 64K，在 SQLCipher 4.x 中預設為 256K) 的 PBKDF2 迭代次數

# Use 1,000,000 iterations.
db = SqlCipherDatabase('my_app.db', pragmas={'kdf_iter': 1000000})

使用 16KB 的密碼頁面大小和 10,000 頁的快取大小

db = SqlCipherDatabase('my_app.db', passphrase='secret!!!', pragmas={
    'cipher_page_size': 1024 * 16,
    'cache_size': 10000})  # 10,000 16KB pages, or 160MB.

提示使用者輸入密碼的範例

db = SqlCipherDatabase(None)

class BaseModel(Model):
    """Parent for all app's models"""
    class Meta:
        # We won't have a valid db until user enters passhrase.
        database = db

# Derive our model subclasses
class Person(BaseModel):
    name = TextField(primary_key=True)

right_passphrase = False
while not right_passphrase:
    db.init(
        'testsqlcipher.db',
        passphrase=get_passphrase_from_user())

    try:  # Actually execute a query against the db to test passphrase.
        db.get_tables()
    except DatabaseError as exc:
        # This error indicates the password was wrong.
        if exc.args[0] == 'file is encrypted or is not a database':
            tell_user_the_passphrase_was_wrong()
            db.init(None)  # Reset the db.
        else:
            raise exc
    else:
        # The password was correct.
        right_passphrase = True

另請參閱：一個稍加詳細的 範例。

Postgresql 擴充功能

postgresql 擴充功能模組提供許多「僅限 postgres」的函式，目前

未來我希望加入對更多 postgresql 功能的支援。如果有一些您希望加入的特定功能，請開啟一個 Github 問題。

警告

為了開始使用以下描述的功能，您需要使用 PostgresqlExtDatabase 類別，而不是 PostgresqlDatabase 類別。

以下程式碼將假設您正在使用以下資料庫和基礎模型

from playhouse.postgres_ext import *

ext_db = PostgresqlExtDatabase('peewee_test', user='postgres')

class BaseExtModel(Model):
    class Meta:
        database = ext_db

JSON 支援

peewee 透過 JSONField 的形式，對 Postgres 的原生 JSON 資料類型提供基本支援。從 2.4.7 版本開始，peewee 也透過 BinaryJSONField 支援 Postgres 9.4 的二進位 json jsonb 類型。

警告

Postgres 從 9.2 版本開始原生支援 JSON 資料類型（9.3 版本提供完整支援）。為了使用此功能，您必須使用正確版本的 Postgres，並搭配 psycopg2 2.5 或更高版本。

要使用 BinaryJSONField（它具有許多效能和查詢優勢），您必須使用 Postgres 9.4 或更高版本。

注意

為了使用 JSONField，您必須確保您的資料庫是 PostgresqlExtDatabase 的實例。

以下範例說明如何宣告具有 JSON 欄位的模型

import json
import urllib2
from playhouse.postgres_ext import *

db = PostgresqlExtDatabase('my_database')

class APIResponse(Model):
    url = CharField()
    response = JSONField()

    class Meta:
        database = db

    @classmethod
    def request(cls, url):
        fh = urllib2.urlopen(url)
        return cls.create(url=url, response=json.loads(fh.read()))

APIResponse.create_table()

# Store a JSON response.
offense = APIResponse.request('http://crime-api.com/api/offense/')
booking = APIResponse.request('http://crime-api.com/api/booking/')

# Query a JSON data structure using a nested key lookup:
offense_responses = APIResponse.select().where(
    APIResponse.response['meta']['model'] == 'offense')

# Retrieve a sub-key for each APIResponse. By calling .as_json(), the
# data at the sub-key will be returned as Python objects (dicts, lists,
# etc) instead of serialized JSON.
q = (APIResponse
     .select(
       APIResponse.data['booking']['person'].as_json().alias('person'))
     .where(APIResponse.data['meta']['model'] == 'booking'))

for result in q:
    print(result.person['name'], result.person['dob'])

BinaryJSONField 的運作方式與常規的 JSONField 相同，並支援相同的操作，但它提供了一些額外的操作來測試 **包含性**。使用二進位 json 欄位，您可以測試您的 JSON 資料是否包含其他部分 JSON 結構（contains()、contains_any()、contains_all()），或者它是否為較大 JSON 文件的一個子集（contained_by()）。

如需更多範例，請參閱下方的 JSONField 和 BinaryJSONField API 文件。

hstore 支援

Postgresql hstore 是一個嵌入式的鍵/值儲存區。透過 hstore，您可以將任意的鍵/值對儲存在資料庫中，以及結構化的關聯資料。

要使用 hstore，您需要在實例化 PostgresqlExtDatabase 時指定一個額外的參數

# Specify "register_hstore=True":
db = PostgresqlExtDatabase('my_db', register_hstore=True)

目前，postgres_ext 模組支援以下操作

  • 儲存和檢索任意字典

  • 依鍵或部分字典進行篩選

  • 更新/新增一個或多個鍵至現有字典

  • 從現有字典中刪除一個或多個鍵

  • 選取鍵、值或將鍵和值壓縮在一起

  • 檢索鍵/值的一部分

  • 測試某個鍵是否存在

  • 測試某個鍵是否具有非 NULL 值

使用 hstore

首先，您需要從 playhouse.postgres_ext 導入自訂資料庫類別和 hstore 函數（請參閱上面的程式碼片段）。然後，只需在您的模型中新增一個 HStoreField 即可

class House(BaseExtModel):
    address = CharField()
    features = HStoreField()

您現在可以在 House 實例上儲存任意的鍵/值對

>>> h = House.create(
...     address='123 Main St',
...     features={'garage': '2 cars', 'bath': '2 bath'})
...
>>> h_from_db = House.get(House.id == h.id)
>>> h_from_db.features
{'bath': '2 bath', 'garage': '2 cars'}

您可以依個別鍵、多個鍵或部分字典進行篩選

>>> query = House.select()
>>> garage = query.where(House.features.contains('garage'))
>>> garage_and_bath = query.where(House.features.contains(['garage', 'bath']))
>>> twocar = query.where(House.features.contains({'garage': '2 cars'}))

假設您想要對房屋執行原子更新

>>> new_features = House.features.update({'bath': '2.5 bath', 'sqft': '1100'})
>>> query = House.update(features=new_features)
>>> query.where(House.id == h.id).execute()
1
>>> h = House.get(House.id == h.id)
>>> h.features
{'bath': '2.5 bath', 'garage': '2 cars', 'sqft': '1100'}

或者，也可以執行原子刪除

>>> query = House.update(features=House.features.delete('bath'))
>>> query.where(House.id == h.id).execute()
1
>>> h = House.get(House.id == h.id)
>>> h.features
{'garage': '2 cars', 'sqft': '1100'}

可以同時刪除多個鍵

>>> query = House.update(features=House.features.delete('garage', 'sqft'))

您可以僅選取鍵、僅選取值，或將兩者壓縮在一起

>>> for h in House.select(House.address, House.features.keys().alias('keys')):
...     print(h.address, h.keys)

123 Main St [u'bath', u'garage']

>>> for h in House.select(House.address, House.features.values().alias('vals')):
...     print(h.address, h.vals)

123 Main St [u'2 bath', u'2 cars']

>>> for h in House.select(House.address, House.features.items().alias('mtx')):
...     print(h.address, h.mtx)

123 Main St [[u'bath', u'2 bath'], [u'garage', u'2 cars']]

您可以檢索一部分資料，例如，所有車庫資料

>>> query = House.select(House.address, House.features.slice('garage').alias('garage_data'))
>>> for house in query:
...     print(house.address, house.garage_data)

123 Main St {'garage': '2 cars'}

您可以檢查某個鍵是否存在，並據此篩選列

>>> has_garage = House.features.exists('garage')
>>> for house in House.select(House.address, has_garage.alias('has_garage')):
...     print(house.address, house.has_garage)

123 Main St True

>>> for house in House.select().where(House.features.exists('garage')):
...     print(house.address, house.features['garage'])  # <-- just houses w/garage data

123 Main St 2 cars

間隔支援

Postgres 透過 INTERVAL 資料類型 (文件) 支援持續時間。

class IntervalField([null=False[, ...]])

能夠儲存 Python datetime.timedelta 實例的欄位類別。

範例

from datetime import timedelta

from playhouse.postgres_ext import *

db = PostgresqlExtDatabase('my_db')

class Event(Model):
    location = CharField()
    duration = IntervalField()
    start_time = DateTimeField()

    class Meta:
        database = db

    @classmethod
    def get_long_meetings(cls):
        return cls.select().where(cls.duration > timedelta(hours=1))

伺服器端游標

當 psycopg2 執行查詢時，通常後端會擷取所有結果並將其返回給用戶端。當進行大型查詢時，這可能會導致您的應用程式使用大量記憶體。使用伺服器端游標，結果會一次返回一點點（預設為 2000 筆記錄）。如需最終參考，請參閱 psycopg2 文件。

注意

若要使用伺服器端（或具名）游標，您必須使用 PostgresqlExtDatabase。

若要使用伺服器端游標執行查詢，只需使用 ServerSide() 協助程式封裝您的 select 查詢

large_query = PageView.select()  # Build query normally.

# Iterate over large query inside a transaction.
for page_view in ServerSide(large_query):
    # do some interesting analysis here.
    pass

# Server-side resources are released.

如果您希望所有 SELECT 查詢自動使用伺服器端游標，您可以在建立 PostgresqlExtDatabase 時指定此選項

from postgres_ext import PostgresqlExtDatabase

ss_db = PostgresqlExtDatabase('my_db', server_side_cursors=True)

注意

伺服器端游標的存活時間僅與交易相同，因此 peewee 不會在執行 SELECT 查詢後自動呼叫 commit()。如果您在迭代完成後不 commit，您將不會釋放伺服器端資源，直到連線關閉（或稍後提交交易）。此外，由於 peewee 預設會快取游標傳回的列，因此當迭代大型查詢時，您應始終呼叫 .iterator()。

如果您正在使用 ServerSide() 協助程式，交易和對 iterator() 的呼叫將會以透明方式處理。

全文檢索

Postgresql 使用特殊資料類型 (tsvector 和 tsquery) 提供複雜的全文檢索。文件應儲存或轉換為 tsvector 類型，而搜尋查詢應轉換為 tsquery。

對於簡單的情況，您可以直接使用 Match() 函數，它將自動執行適當的轉換，並且不需要任何結構描述變更

def blog_search(search_term):
    return Blog.select().where(
        (Blog.status == Blog.STATUS_PUBLISHED) &
        Match(Blog.content, search_term))

Match() 函數會自動將左側運算元轉換為 tsvector，並將右側運算元轉換為 tsquery。為了獲得更好的效能，建議您在您計劃搜尋的欄位上建立 GIN 索引

CREATE INDEX blog_full_text_search ON blog USING gin(to_tsvector(content));

或者，您可以使用 TSVectorField 來維護一個專用欄位，用於儲存 tsvector 資料

class Blog(Model):
    content = TextField()
    search_content = TSVectorField()

注意

TSVectorField 將會自動使用 GIN 索引建立。

當插入或更新 search_content 欄位時，您需要明確地將傳入的文字資料轉換為 tsvector

content = 'Excellent blog post about peewee ORM.'
blog_entry = Blog.create(
    content=content,
    search_content=fn.to_tsvector(content))

若要執行全文檢索，請使用 TSVectorField.match()

terms = 'python & (sqlite | postgres)'
results = Blog.select().where(Blog.search_content.match(terms))

如需更多資訊，請參閱Postgres 全文檢索文件。

postgres_ext API 說明

class PostgresqlExtDatabase(database[, server_side_cursors=False[, register_hstore=False[, ...]]])

與 PostgresqlDatabase 相同，但需要此類別以支援

參數

  • database (str) – 要連線的資料庫名稱。

  • server_side_cursors (bool) – SELECT 查詢是否應使用伺服器端游標。

  • register_hstore (bool) – 是否將 HStore 擴充功能註冊到連線。

如果您想要使用 HStore 擴充功能，您必須指定 register_hstore=True。

如果使用 server_side_cursors，也請務必使用 ServerSide() 包裝您的查詢。

ServerSide(select_query)
參數

select_query – SelectQuery 實例。

Rtype generator

將指定的 select 查詢包裝在交易中，並呼叫其 iterator() 方法以避免快取資料列實例。為了釋放伺服器端資源，請務必耗盡產生器 (迭代所有資料列)。

用法

large_query = PageView.select()
for page_view in ServerSide(large_query):
    # Do something interesting.
    pass

# At this point server side resources are released.
class ArrayField([field_class=IntegerField[, field_kwargs=None[, dimensions=1[, convert_values=False]]]])
參數

  • field_class – Field 的子類別，例如 IntegerField。

  • field_kwargs (dict) – 初始化 field_class 的引數。

  • dimensions (int) – 陣列的維度。

  • convert_values (bool) – 將 field_class 值轉換套用至陣列資料。

能夠儲存提供的 field_class 陣列的欄位。

注意

預設情況下，ArrayField 會使用 GIN 索引。若要停用此功能，請使用 index=False 初始化欄位。

您可以儲存和擷取清單 (或清單的清單)

class BlogPost(BaseModel):
    content = TextField()
    tags = ArrayField(CharField)


post = BlogPost(content='awesome', tags=['foo', 'bar', 'baz'])

此外，您可以使用 __getitem__ API 來查詢資料庫中的值或切片

# Get the first tag on a given blog post.
first_tag = (BlogPost
             .select(BlogPost.tags[0].alias('first_tag'))
             .where(BlogPost.id == 1)
             .dicts()
             .get())

# first_tag = {'first_tag': 'foo'}

取得值的切片

# Get the first two tags.
two_tags = (BlogPost
            .select(BlogPost.tags[:2].alias('two'))
            .dicts()
            .get())
# two_tags = {'two': ['foo', 'bar']}
contains(*items)
參數

items – 給定陣列欄位中必須存在的一或多個項目。

# Get all blog posts that are tagged with both "python" and "django".
Blog.select().where(Blog.tags.contains('python', 'django'))
contains_any(*items)
參數

items – 要在給定陣列欄位中搜尋的一或多個項目。

與 contains() 類似，但會比對陣列中包含*任何*指定項目的資料列。

# Get all blog posts that are tagged with "flask" and/or "django".
Blog.select().where(Blog.tags.contains_any('flask', 'django'))
class DateTimeTZField(*args, **kwargs)

DateTimeField 的時區感知子類別。

class HStoreField(*args, **kwargs)

用於儲存和擷取任意鍵/值對的欄位。如需詳細使用方式，請參閱 hstore 支援。

注意

若要使用 HStoreField，您需要確保 hstore 擴充功能已註冊到連線。若要完成此操作，請使用 register_hstore=True 實例化 PostgresqlExtDatabase。

注意

預設情況下，HStoreField 會使用 GiST 索引。若要停用此功能，請使用 index=False 初始化欄位。

keys()

傳回給定資料列的鍵。

>>> for h in House.select(House.address, House.features.keys().alias('keys')):
...     print(h.address, h.keys)

123 Main St [u'bath', u'garage']
values()

傳回給定資料列的值。

>>> for h in House.select(House.address, House.features.values().alias('vals')):
...     print(h.address, h.vals)

123 Main St [u'2 bath', u'2 cars']
items()

與 Python 的 dict 類似，以清單的清單形式傳回鍵和值

>>> for h in House.select(House.address, House.features.items().alias('mtx')):
...     print(h.address, h.mtx)

123 Main St [[u'bath', u'2 bath'], [u'garage', u'2 cars']]
slice(*args)

傳回給定鍵清單的資料切片。

>>> for h in House.select(House.address, House.features.slice('garage').alias('garage_data')):
...     print(h.address, h.garage_data)

123 Main St {'garage': '2 cars'}
exists(key)

查詢指定的鍵是否存在。

>>> for h in House.select(House.address, House.features.exists('garage').alias('has_garage')):
...     print(h.address, h.has_garage)

123 Main St True

>>> for h in House.select().where(House.features.exists('garage')):
...     print(h.address, h.features['garage']) # <-- just houses w/garage data

123 Main St 2 cars
defined(key)

查詢指定的鍵是否具有關聯的值。

update(**data)

對給定資料列或多個資料列的鍵/值執行原子更新。

>>> query = House.update(features=House.features.update(
...     sqft=2000,
...     year_built=2012))
>>> query.where(House.id == 1).execute()
delete(*keys)

刪除給定資料列或多個資料列的指定鍵。

注意

我們將使用 UPDATE 查詢。



>>> query = House.update(features=House.features.delete(
...     'sqft', 'year_built'))
>>> query.where(House.id == 1).execute()
contains(value)
參數

value – dict、鍵的 list 或單一鍵。

查詢具有下列其中一項的資料列

  • 部分字典。

  • 鍵的清單。

  • 單一鍵。

>>> query = House.select()
>>> has_garage = query.where(House.features.contains('garage'))
>>> garage_bath = query.where(House.features.contains(['garage', 'bath']))
>>> twocar = query.where(House.features.contains({'garage': '2 cars'}))
contains_any(*keys)
參數

keys – 要搜尋的一或多個鍵。

查詢是否存在*任何*鍵的資料列。

class JSONField(dumps=None, *args, **kwargs)
參數

dumps – 預設為呼叫 json.dumps() 或 dumps 函數。您可以覆寫此方法以建立自訂的 JSON 包裝函式。

適用於儲存和查詢任意 JSON 的欄位類別。當在模型上使用此欄位時，請將欄位的值設定為 Python 物件 (可以是 dict 或 list)。當您從資料庫擷取值時，它將會以 Python 資料結構的形式傳回。

注意

您必須使用 Postgres 9.2 / psycopg2 2.5 或更高版本。

注意

如果您使用的是 Postgres 9.4，強烈建議您改用 BinaryJSONField，因為它可提供更好的效能和更強大的查詢選項。

模型宣告範例

db = PostgresqlExtDatabase('my_db')

class APIResponse(Model):
    url = CharField()
    response = JSONField()

    class Meta:
        database = db

儲存 JSON 資料的範例

url = 'http://foo.com/api/resource/'
resp = json.loads(urllib2.urlopen(url).read())
APIResponse.create(url=url, response=resp)

APIResponse.create(url='http://foo.com/baz/', response={'key': 'value'})

若要查詢，請使用 Python 的 [] 運算子來指定巢狀鍵或陣列查閱

APIResponse.select().where(
    APIResponse.response['key1']['nested-key'] == 'some-value')

為了說明 [] 運算子的用法，假設我們在 APIResponse 中儲存了下列資料

{
  "foo": {
    "bar": ["i1", "i2", "i3"],
    "baz": {
      "huey": "mickey",
      "peewee": "nugget"
    }
  }
}

以下是一些查詢的結果

def get_data(expression):
    # Helper function to just retrieve the results of a
    # particular expression.
    query = (APIResponse
             .select(expression.alias('my_data'))
             .dicts()
             .get())
    return query['my_data']

# Accessing the foo -> bar subkey will return a JSON
# representation of the list.
get_data(APIResponse.data['foo']['bar'])
# '["i1", "i2", "i3"]'

# In order to retrieve this list as a Python list,
# we will call .as_json() on the expression.
get_data(APIResponse.data['foo']['bar'].as_json())
# ['i1', 'i2', 'i3']

# Similarly, accessing the foo -> baz subkey will
# return a JSON representation of the dictionary.
get_data(APIResponse.data['foo']['baz'])
# '{"huey": "mickey", "peewee": "nugget"}'

# Again, calling .as_json() will return an actual
# python dictionary.
get_data(APIResponse.data['foo']['baz'].as_json())
# {'huey': 'mickey', 'peewee': 'nugget'}

# When dealing with simple values, either way works as
# you expect.
get_data(APIResponse.data['foo']['bar'][0])
# 'i1'

# Calling .as_json() when the result is a simple value
# will return the same thing as the previous example.
get_data(APIResponse.data['foo']['bar'][0].as_json())
# 'i1'
class BinaryJSONField(dumps=None, *args, **kwargs)
參數

dumps – 預設為呼叫 json.dumps() 或 dumps 函數。您可以覆寫此方法以建立自訂的 JSON 包裝函式。

儲存和查詢任意 JSON 文件。資料應使用一般的 Python dict 和 list 物件儲存，而當從資料庫傳回資料時，也會使用 dict 和 list 傳回。

關於基本查詢操作的範例，請參閱上方 JSONField 的程式碼範例。以下範例查詢將使用上述的相同 APIResponse 模型。

注意

預設情況下，BinaryJSONField 將使用 GiST 索引。若要停用此功能，請使用 index=False 初始化欄位。

注意

您必須使用 Postgres 9.4 / psycopg2 2.5 或更新版本。如果您使用的是 Postgres 9.2 或 9.3，則可以使用一般的 JSONField 作為替代方案。

contains(other)

測試給定的 JSON 資料是否包含給定的 JSON 片段或鍵。

範例

search_fragment = {
    'foo': {'bar': ['i2']}
}
query = (APIResponse
         .select()
         .where(APIResponse.data.contains(search_fragment)))

# If we're searching for a list, the list items do not need to
# be ordered in a particular way:
query = (APIResponse
         .select()
         .where(APIResponse.data.contains({
             'foo': {'bar': ['i2', 'i1']}})))

我們也可以傳入簡單的鍵。要找到頂層包含鍵 foo 的 APIResponses

APIResponse.select().where(APIResponse.data.contains('foo'))

我們也可以使用方括號搜尋子鍵

APIResponse.select().where(
    APIResponse.data['foo']['bar'].contains(['i2', 'i1']))
contains_any(*items)

搜尋是否存在一個或多個給定的項目。

APIResponse.select().where(
    APIResponse.data.contains_any('foo', 'baz', 'nugget'))

和 contains() 一樣，我們也可以搜尋子鍵

APIResponse.select().where(
    APIResponse.data['foo']['bar'].contains_any('i2', 'ix'))
contains_all(*items)

搜尋是否存在所有給定的項目。

APIResponse.select().where(
    APIResponse.data.contains_all('foo'))

和 contains_any() 一樣，我們也可以搜尋子鍵

APIResponse.select().where(
    APIResponse.data['foo']['bar'].contains_all('i1', 'i2', 'i3'))
contained_by(other)

測試給定的 JSON 文件是否被給定的 JSON 文件所包含（是其子集）。此方法與 contains() 的作用相反。

big_doc = {
    'foo': {
        'bar': ['i1', 'i2', 'i3'],
        'baz': {
            'huey': 'mickey',
            'peewee': 'nugget',
        }
    },
    'other_key': ['nugget', 'bear', 'kitten'],
}
APIResponse.select().where(
    APIResponse.data.contained_by(big_doc))
concat(data)

串連兩個欄位資料和提供的資料。請注意，此操作不會合併或執行「深層串連」。

has_key(key)

測試鍵是否存在於 JSON 物件的頂層。

remove(*keys)

從 JSON 物件的頂層移除一個或多個鍵。

Match(field, query)

產生全文搜尋表達式，自動將左側運算元轉換為 tsvector，將右側運算元轉換為 tsquery。

範例

def blog_search(search_term):
    return Blog.select().where(
        (Blog.status == Blog.STATUS_PUBLISHED) &
        Match(Blog.content, search_term))
class TSVectorField

適合儲存 tsvector 資料的欄位類型。此欄位將自動使用 GIN 索引建立，以提高搜尋效能。

注意

儲存在此欄位中的資料仍然需要手動轉換為 tsvector 類型。

注意

預設情況下，TSVectorField 將使用 GIN 索引。若要停用此功能，請使用 index=False 初始化欄位。

範例用法

class Blog(Model):
    content = TextField()
    search_content = TSVectorField()

content = 'this is a sample blog entry.'
blog_entry = Blog.create(
    content=content,
    search_content=fn.to_tsvector(content))  # Note `to_tsvector()`.
match(query[, language=None[, plain=False]])
參數

  • query (str) – 全文搜尋查詢。

  • language (str) – 語言名稱（選填）。

  • plain (bool) – 使用 plain（簡單）剖析器剖析搜尋查詢。

傳回值

表示全文搜尋/比對的表達式。

範例

# Perform a search using the "match" method.
terms = 'python & (sqlite | postgres)'
results = Blog.select().where(Blog.search_content.match(terms))

Cockroach Database

peewee 良好支援 CockroachDB (CRDB)。

from playhouse.cockroachdb import CockroachDatabase

db = CockroachDatabase('my_app', user='root', host='10.1.0.8')

如果您使用的是 Cockroach Cloud，您可能會發現使用連線字串指定連線參數會更容易

db = CockroachDatabase('postgresql://root:secret@host:26257/defaultdb...')

注意

CockroachDB 需要 psycopg2 (postgres) Python 驅動程式。

注意

CockroachDB 安裝和入門指南可在此處找到：https://www.cockroachlabs.com/docs/stable/install-cockroachdb.html

SSL 設定

強烈建議在執行 Cockroach 叢集時使用 SSL 憑證。Psycopg2 支援開箱即用的 SSL，但在初始化資料庫時，您可能需要指定一些額外的選項

db = CockroachDatabase(
    'my_app',
    user='root',
    host='10.1.0.8',
    sslmode='verify-full',  # Verify the cert common-name.
    sslrootcert='/path/to/root.crt')


# Or, alternatively, specified as part of a connection-string:
db = CockroachDatabase('postgresql://root:secret@host:26257/dbname'
                       '?sslmode=verify-full&sslrootcert=/path/to/root.crt'
                       '&options=--cluster=my-cluster-xyz')

有關用戶端驗證的更多詳細資訊，請參閱 libpq 文件。

Cockroach 擴充 API

playhouse.cockroachdb 擴充模組提供以下類別和輔助程式

使用 CRDB 時可能很有用的特殊欄位類型

  • UUIDKeyField - 使用 CRDB 的 UUID 類型及預設隨機產生的 UUID 的主鍵欄位實作。

  • RowIDField - 使用 CRDB 的 INT 類型及預設 unique_rowid() 的主鍵欄位實作。

  • JSONField - 與 Postgres BinaryJSONField 相同，因為 CRDB 將 JSON 視為 JSONB。

  • ArrayField - 與 Postgres 擴充功能相同（但不支援多維陣列）。

CRDB 與 Postgres 的連線協定相容，並公開非常相似的 SQL 介面，因此可以使用 PostgresqlDatabase 與 CRDB 搭配使用（雖然不建議）。

  1. CRDB 不支援巢狀交易（儲存點），因此 atomic() 方法已實作，以便在使用 CockroachDatabase 時強制執行此行為。如需更多資訊，請參閱 CRDB 交易。

  2. CRDB 在欄位類型、日期函式和 Postgres 的內省方面可能存在細微差異。

  3. CRDB 特定的功能由 CockroachDatabase 公開，例如指定交易優先順序或 AS OF SYSTEM TIME 子句。

CRDB 交易

CRDB 不支援巢狀交易（儲存點），因此 atomic() 方法在 CockroachDatabase 上已修改為，如果遇到無效的巢狀結構，則會引發例外狀況。如果您想要能夠巢狀化交易程式碼，您可以使用 transaction() 方法，這將確保最外層區塊會管理交易（例如，退出巢狀區塊將不會導致提早提交）。

範例

@db.transaction()
def create_user(username):
    return User.create(username=username)

def some_other_function():
    with db.transaction() as txn:
        # do some stuff...

        # This function is wrapped in a transaction, but the nested
        # transaction will be ignored and folded into the outer
        # transaction, as we are already in a wrapped-block (via the
        # context manager).
        create_user('some_user@example.com')

        # do other stuff.

    # At this point we have exited the outer-most block and the transaction
    # will be committed.
    return

CRDB 提供用戶端交易重試，可使用特殊的 run_transaction() 輔助程式。此輔助方法接受可呼叫物件，該物件負責執行任何可能需要重試的交易陳述式。

run_transaction() 的最簡單範例

def create_user(email):
    # Callable that accepts a single argument (the database instance) and
    # which is responsible for executing the transactional SQL.
    def callback(db_ref):
        return User.create(email=email)

    return db.run_transaction(callback, max_attempts=10)

huey = create_user('huey@example.com')

注意

如果在給定次數的嘗試後無法提交交易，則會引發 cockroachdb.ExceededMaxAttempts 例外狀況。如果 SQL 格式錯誤、違反約束等等，則函式會將例外狀況引發給呼叫者。

使用 run_transaction() 實作用戶端重試的範例，用於將金額從一個帳戶轉移到另一個帳戶的交易

from playhouse.cockroachdb import CockroachDatabase

db = CockroachDatabase('my_app')


def transfer_funds(from_id, to_id, amt):
    """
    Returns a 3-tuple of (success?, from balance, to balance). If there are
    not sufficient funds, then the original balances are returned.
    """
    def thunk(db_ref):
        src, dest = (Account
                     .select()
                     .where(Account.id.in_([from_id, to_id])))
        if src.id != from_id:
            src, dest = dest, src  # Swap order.

        # Cannot perform transfer, insufficient funds!
        if src.balance < amt:
            return False, src.balance, dest.balance

        # Update each account, returning the new balance.
        src, = (Account
                .update(balance=Account.balance - amt)
                .where(Account.id == from_id)
                .returning(Account.balance)
                .execute())
        dest, = (Account
                 .update(balance=Account.balance + amt)
                 .where(Account.id == to_id)
                 .returning(Account.balance)
                 .execute())
        return True, src.balance, dest.balance

    # Perform the queries that comprise a logical transaction. In the
    # event the transaction fails due to contention, it will be auto-
    # matically retried (up to 10 times).
    return db.run_transaction(thunk, max_attempts=10)

CRDB API

class CockroachDatabase(database[, **kwargs])

CockroachDB 的實作，基於 PostgresqlDatabase 並使用 psycopg2 驅動程式。

額外的關鍵字參數會傳遞給 psycopg2 連線建構子，可用來指定資料庫的 user、port 等等。

或者，連線詳細資訊也可以用 URL 形式指定。

run_transaction(callback[, max_attempts=None[, system_time=None[, priority=None]]])
參數

  • callback – 可呼叫的物件，接受單一 db 參數（此參數將會是呼叫此方法的資料庫實例）。

  • max_attempts (int) – 放棄前嘗試的最大次數。

  • system_time (datetime) – 執行交易時，根據給定的值使用 AS OF SYSTEM TIME。

  • priority (str) – "low"、"normal" 或 "high" 其中一個。

傳回值

返回 callback 所返回的值。

引發

如果超過 max_attempts，則引發 ExceededMaxAttempts。

在交易中執行 SQL，並自動進行客戶端重試。

使用者提供的 callback

  • 必須接受一個參數，即代表交易正在執行的連線的 db 實例。

  • 不得嘗試 commit、rollback 或以其他方式管理交易。

  • 可能會被呼叫多次。

  • 應該理想情況下僅包含 SQL 操作。

此外，資料庫在呼叫此函數時不得有任何開啟的交易，因為 CRDB 不支援巢狀交易。嘗試這樣做會引發 NotImplementedError。

最簡單的範例

def create_user(email):
    def callback(db_ref):
        return User.create(email=email)

    return db.run_transaction(callback, max_attempts=10)

user = create_user('huey@example.com')
class PooledCockroachDatabase(database[, **kwargs])

CockroachDB 連線池實作，基於 PooledPostgresqlDatabase。實作與 CockroachDatabase 相同的 API，但會執行客戶端連線池。

run_transaction(db, callback[, max_attempts=None[, system_time=None[, priority=None]]])

在交易中執行 SQL，並自動進行客戶端重試。有關詳細資訊，請參閱 CockroachDatabase.run_transaction()。

參數

  • db (CockroachDatabase) – 資料庫實例。

  • callback – 可呼叫的物件，接受單一 db 參數（與上面傳遞的值相同）。

注意

此函數等效於 CockroachDatabase 類別上名稱相同的方法。

class UUIDKeyField

UUID 主鍵欄位，使用 CRDB 的 gen_random_uuid() 函數自動填入初始值。

class RowIDField

自動遞增整數主鍵欄位，使用 CRDB 的 unique_rowid() 函數自動填入初始值。

另請參閱

  • Postgresql 擴充功能中的 BinaryJSONField（在 cockroachdb 擴充模組中提供，並別名為 JSONField）。

  • Postgresql 擴充功能中的 ArrayField。

MySQL 擴充功能

Peewee 提供了一個替代的資料庫實作，用於使用 mysql-connector 驅動程式或 mariadb-connector。這些實作可以在 playhouse.mysql_ext 中找到。

class MySQLConnectorDatabase(database, **kwargs)

使用 mysql-connector 的資料庫實作。完整支援的連線參數列表。

mysql-connector 的使用範例

from playhouse.mysql_ext import MySQLConnectorDatabase

# MySQL database implementation that utilizes mysql-connector driver.
db = MySQLConnectorDatabase('my_database', host='1.2.3.4', user='mysql')
class MariaDBConnectorDatabase(database, **kwargs)

使用 mariadb-connector 的資料庫實作。完整支援的連線參數列表。

mariadb-connector 的使用範例

from playhouse.mysql_ext import MariaDBConnectorDatabase

# MySQL database implementation that utilizes mysql-connector driver.
db = MariaDBConnectorDatabase('my_database', host='1.2.3.4', user='mysql')

注意

MariaDBConnectorDatabase 不接受以下參數

  • charset （它始終是 utf8mb4）

  • sql_mode

  • use_unicode

其他特定於 MySQL 的輔助程式

class JSONField

擴充 TextField 並實作 Python 中透明的 JSON 編碼和解碼。

extract(path)
參數

path (str) – JSON 路徑，例如 $.key1

從給定路徑的 JSON 文件中提取值。

Match(columns, expr[, modifier=None])
參數

  • columns – 單一 Field 或多個欄位的元組。

  • expr (str) – 全文搜尋表示式。

  • modifier (str) – 搜尋的可選修飾詞，例如 *'in boolean mode'*。

用於建構以下形式的 MySQL 全文搜尋查詢的輔助類別

MATCH (columns, ...) AGAINST (expr[ modifier])

DataSet

dataset 模組包含一個高階 API，用於處理仿照流行的同名專案建模的資料庫。 dataset 模組的目的是提供

  • 一個簡化的 API，用於處理關聯式資料，類似於處理 JSON。

  • 一種將關聯式資料匯出為 JSON 或 CSV 的簡單方法。

  • 一種將 JSON 或 CSV 資料匯入關聯式資料庫的簡單方法。

一個最小化的資料載入指令碼可能如下所示

from playhouse.dataset import DataSet

db = DataSet('sqlite:///:memory:')

table = db['sometable']
table.insert(name='Huey', age=3)
table.insert(name='Mickey', age=5, gender='male')

huey = table.find_one(name='Huey')
print(huey)
# {'age': 3, 'gender': None, 'id': 1, 'name': 'Huey'}

for obj in table:
    print(obj)
# {'age': 3, 'gender': None, 'id': 1, 'name': 'Huey'}
# {'age': 5, 'gender': 'male', 'id': 2, 'name': 'Mickey'}

您也可以使用字典 API 插入、更新或刪除

huey = table.find_one(name='Huey')
# {'age': 3, 'gender': None, 'id': 1, 'name': 'Huey'}

# Perform an update by supplying a partial record of changes.
table[1] = {'gender': 'male', 'age': 4}
print(table[1])
# {'age': 4, 'gender': 'male', 'id': 1, 'name': 'Huey'}

# Or insert a new record:
table[3] = {'name': 'Zaizee', 'age': 2}
print(table[3])
# {'age': 2, 'gender': None, 'id': 3, 'name': 'Zaizee'}

# Or delete a record:
del table[3]  # Remove the row we just added.

您可以使用 freeze() 和 thaw() 匯出或匯入資料

# Export table content to the `users.json` file.
db.freeze(table.all(), format='json', filename='users.json')

# Import data from a CSV file into a new table. Columns will be automatically
# created for each field in the CSV file.
new_table = db['stats']
new_table.thaw(format='csv', filename='monthly_stats.csv')

開始使用

DataSet 物件透過傳入格式為 dialect://user:password@host/dbname 的資料庫 URL 來初始化。請參閱 資料庫 URL 章節，以取得連接到各種資料庫的範例。

# Create an in-memory SQLite database.
db = DataSet('sqlite:///:memory:')

儲存資料

要儲存資料，我們必須先取得對表格的參考。如果表格不存在，它將會自動建立。

# Get a table reference, creating the table if it does not exist.
table = db['users']

我們現在可以將新的列 insert() 到表格中。如果欄位不存在，它們將會自動建立。

table.insert(name='Huey', age=3, color='white')
table.insert(name='Mickey', age=5, gender='male')

要更新表格中現有的條目，請傳入一個包含新值和篩選條件的字典。要用作篩選器的欄位列表在 columns 參數中指定。如果未指定篩選欄位，則所有列都將被更新。

# Update the gender for "Huey".
table.update(name='Huey', gender='male', columns=['name'])

# Update all records. If the column does not exist, it will be created.
table.update(favorite_orm='peewee')

匯入資料

要從外部來源（例如 JSON 或 CSV 檔案）匯入資料，您可以使用 thaw() 方法。預設情況下，將為遇到的任何屬性建立新的欄位。如果您只想填入表格上已定義的欄位，您可以傳入 strict=True。

# Load data from a JSON file containing a list of objects.
table = dataset['stock_prices']
table.thaw(filename='stocks.json', format='json')
table.all()[:3]

# Might print...
[{'id': 1, 'ticker': 'GOOG', 'price': 703},
 {'id': 2, 'ticker': 'AAPL', 'price': 109},
 {'id': 3, 'ticker': 'AMZN', 'price': 300}]

使用交易

DataSet 支援使用簡單的上下文管理器來巢狀交易。

table = db['users']
with db.transaction() as txn:
    table.insert(name='Charlie')

    with db.transaction() as nested_txn:
        # Set Charlie's favorite ORM to Django.
        table.update(name='Charlie', favorite_orm='django', columns=['name'])

        # jk/lol
        nested_txn.rollback()

檢視資料庫

您可以使用 tables() 方法來列出目前資料庫中的表格

>>> print(db.tables)
['sometable', 'user']

對於給定的表格，您可以列印欄位

>>> table = db['user']
>>> print(table.columns)
['id', 'age', 'name', 'gender', 'favorite_orm']

我們也可以找出表格中有多少列

>>> print(len(db['user']))
3

讀取資料

要檢索所有列，您可以使用 all() 方法

# Retrieve all the users.
users = db['user'].all()

# We can iterate over all rows without calling `.all()`
for user in db['user']:
    print(user['name'])

可以使用 find() 和 find_one() 來檢索特定的物件。

# Find all the users who like peewee.
peewee_users = db['user'].find(favorite_orm='peewee')

# Find Huey.
huey = db['user'].find_one(name='Huey')

匯出資料

要匯出資料，請使用 freeze() 方法，並傳入您要匯出的查詢

peewee_users = db['user'].find(favorite_orm='peewee')
db.freeze(peewee_users, format='json', filename='peewee_users.json')

API

class DataSet(url, **kwargs)
參數

DataSet 類別提供用於處理關聯式資料庫的高階 API。

tables

傳回儲存在資料庫中的表格列表。此列表會在每次存取時動態計算。

__getitem__(table_name)

提供指定表格的 Table 參考。如果表格不存在，將會建立它。

query(sql[, params=None[, commit=True]])
參數

  • sql (str) – SQL 查詢。

  • params (list) – 查詢的可選參數。

  • commit (bool) – 查詢是否應在執行時提交。

傳回值

資料庫游標。

針對資料庫執行提供的查詢。

transaction()

建立一個代表新交易（或儲存點）的上下文管理器。

freeze(query[, format='csv'[, filename=None[, file_obj=None[, encoding='utf8'[, **kwargs]]]]])
參數

  • query – 使用 all() 或 ~Table.find 產生的 SelectQuery。

  • format – 輸出格式。預設情況下，支援 csv 和 json。

  • filename – 要將輸出寫入的檔案名稱。

  • file_obj – 要將輸出寫入的類似檔案的物件。

  • encoding (str) – 檔案編碼。

  • kwargs – 匯出特定功能的任意參數。

thaw(table[, format='csv'[, filename=None[, file_obj=None[, strict=False[, encoding='utf8'[, **kwargs]]]]]])
參數

  • table (str) – 要將資料載入的表格名稱。

  • format – 輸入格式。預設情況下，支援 csv 和 json。

  • filename – 要從中讀取資料的檔案名稱。

  • file_obj – 要從中讀取資料的類似檔案的物件。

  • strict (bool) – 是否要為表格上尚未存在的欄位儲存值。

  • encoding (str) – 檔案編碼。

  • kwargs – 匯入特定功能的任意參數。

connect()

開啟與基礎資料庫的連線。如果沒有明確開啟連線，則會在第一次執行查詢時開啟連線。

close()

關閉與基礎資料庫的連線。

class Table(dataset, name, model_class)
Noindex

提供用於處理給定表格中的列的高階 API。

columns

傳回給定表格中欄位的列表。

model_class

動態建立的 Model 類別。

create_index(columns[, unique=False])

在給定的欄位上建立索引

# Create a unique index on the `username` column.
db['users'].create_index(['username'], unique=True)
insert(**data)

將給定的資料字典插入表格中，並在需要時建立新的欄位。

update(columns=None, conjunction=None, **data)

使用提供的資料更新表格。如果 columns 參數中指定了一或多個欄位，則 data 字典中這些欄位的值將被用來決定要更新哪些列。

# Update all rows.
db['users'].update(favorite_orm='peewee')

# Only update Huey's record, setting his age to 3.
db['users'].update(name='Huey', age=3, columns=['name'])
find(**query)

查詢表格中符合指定等式條件的列。如果沒有指定查詢條件，則會傳回所有列。

peewee_users = db['users'].find(favorite_orm='peewee')
find_one(**query)

傳回符合指定等式條件的單一列。如果沒有找到符合的列，則會傳回 None。

huey = db['users'].find_one(name='Huey')
all()

傳回指定表格中的所有列。

delete(**query)

刪除所有符合指定等式條件的列。如果沒有提供查詢條件，則會刪除所有列。

# Adios, Django!
db['users'].delete(favorite_orm='Django')

# Delete all the secret messages.
db['secret_messages'].delete()
freeze([format='csv'[, filename=None[, file_obj=None[, **kwargs]]]])
參數

  • format – 輸出格式。預設情況下，支援 csv 和 json。

  • filename – 要將輸出寫入的檔案名稱。

  • file_obj – 要將輸出寫入的類似檔案的物件。

  • kwargs – 匯出特定功能的任意參數。

thaw([format='csv'[, filename=None[, file_obj=None[, strict=False[, **kwargs]]]]])
參數

  • format – 輸入格式。預設情況下，支援 csv 和 json。

  • filename – 要從中讀取資料的檔案名稱。

  • file_obj – 要從中讀取資料的類似檔案的物件。

  • strict (bool) – 是否要為表格上尚未存在的欄位儲存值。

  • kwargs – 匯入特定功能的任意參數。

欄位

這些欄位可以在 playhouse.fields 模組中找到。

class CompressedField([compression_level=6[, algorithm='zlib'[, **kwargs]]])
參數

  • compression_level (int) – 介於 0 到 9 的值。

  • algorithm (str) – 'zlib' 或 'bz2' 其中一個。

使用指定的演算法儲存壓縮資料。這個欄位擴展了 BlobField，在資料庫中透明地儲存資料的壓縮表示。

class PickleField

透過透明地封裝 (pickling) 和解封裝 (un-pickling) 儲存在欄位中的資料來儲存任意 Python 資料。這個欄位擴展了 BlobField。如果 cPickle 模組可用，將會使用它。

混合屬性

混合屬性封裝了在 Python *和* SQL 層級運作的功能。混合屬性的想法來自於 SQLAlchemy 中同名的功能。考慮以下範例

class Interval(Model):
    start = IntegerField()
    end = IntegerField()

    @hybrid_property
    def length(self):
        return self.end - self.start

    @hybrid_method
    def contains(self, point):
        return (self.start <= point) & (point < self.end)

混合屬性的名稱來自於以下事實：length 屬性會根據它是透過 Interval 類別或 Interval 實例存取而有不同的行為。

如果透過實例存取，它的行為就會如您預期。

然而，如果透過 Interval.length 類別屬性存取，長度計算將會以 SQL 表達式表示。例如

query = Interval.select().where(Interval.length > 5)

這個查詢將會等同於以下的 SQL

SELECT "t1"."id", "t1"."start", "t1"."end"
FROM "interval" AS t1
WHERE (("t1"."end" - "t1"."start") > 5)

playhouse.hybrid 模組也包含一個裝飾器，用於實作可以接受參數的混合方法。與混合屬性一樣，當透過模型實例存取時，該函數會像寫入的那樣正常執行。然而，當在類別上呼叫混合方法時，它會產生一個 SQL 表達式。

範例

query = Interval.select().where(Interval.contains(2))

這個查詢等同於以下的 SQL

SELECT "t1"."id", "t1"."start", "t1"."end"
FROM "interval" AS t1
WHERE (("t1"."start" <= 2) AND (2 < "t1"."end"))

對於 Python 實作與 SQL 實作略有不同的情況，還有一個額外的 API。讓我們在 Interval 模型中新增一個 radius 方法。由於這個方法會計算絕對值，我們將會對實例部分使用 Python 的 abs() 函數，而對類別部分使用 fn.ABS() SQL 函數。

class Interval(Model):
    start = IntegerField()
    end = IntegerField()

    @hybrid_property
    def length(self):
        return self.end - self.start

    @hybrid_property
    def radius(self):
        return abs(self.length) / 2

    @radius.expression
    def radius(cls):
        return fn.ABS(cls.length) / 2

有趣的是，radius 的兩個實作都參考了 length 混合屬性！當透過 Interval 實例存取時，半徑計算會在 Python 中執行。當透過 Interval 類別呼叫時，我們會得到適當的 SQL。

範例

query = Interval.select().where(Interval.radius < 3)

這個查詢等同於以下的 SQL

SELECT "t1"."id", "t1"."start", "t1"."end"
FROM "interval" AS t1
WHERE ((abs("t1"."end" - "t1"."start") / 2) < 3)

很棒吧？感謝 SQLAlchemy 的酷炫想法！

混合 API

class hybrid_method(func[, expr=None])

方法裝飾器，允許定義具有實例層級和類別層級行為的 Python 物件方法。

範例

class Interval(Model):
    start = IntegerField()
    end = IntegerField()

    @hybrid_method
    def contains(self, point):
        return (self.start <= point) & (point < self.end)

當使用 Interval 實例呼叫時，contains 方法會如您預期般運作。然而，當作為類別方法呼叫時，將會產生一個 SQL 表達式。

query = Interval.select().where(Interval.contains(2))

會產生以下 SQL

SELECT "t1"."id", "t1"."start", "t1"."end"
FROM "interval" AS t1
WHERE (("t1"."start" <= 2) AND (2 < "t1"."end"))
expression(expr)

方法裝飾器，用於指定產生 SQL 表達式的方法。

class hybrid_property(fget[, fset=None[, fdel=None[, expr=None]]])

方法裝飾器，允許定義具有實例層級和類別層級行為的 Python 物件屬性。

範例

class Interval(Model):
    start = IntegerField()
    end = IntegerField()

    @hybrid_property
    def length(self):
        return self.end - self.start

    @hybrid_property
    def radius(self):
        return abs(self.length) / 2

    @radius.expression
    def radius(cls):
        return fn.ABS(cls.length) / 2

當在 Interval 實例上存取時，length 和 radius 屬性會如您預期般運作。然而，當作為類別屬性存取時，則會產生一個 SQL 表達式。

query = (Interval
         .select()
         .where(
             (Interval.length > 6) &
             (Interval.radius >= 3)))

會產生以下 SQL

SELECT "t1"."id", "t1"."start", "t1"."end"
FROM "interval" AS t1
WHERE (
    (("t1"."end" - "t1"."start") > 6) AND
    ((abs("t1"."end" - "t1"."start") / 2) >= 3)
)

鍵/值儲存

playhouse.kv 模組包含持久字典的實作。

class KeyValue([key_field=None[, value_field=None[, ordered=False[, database=None[, table_name='keyvalue']]]]])
參數

  • key_field (Field) – 用於鍵的欄位。預設為 CharField。必須具有 primary_key=True。

  • value_field (Field) – 用於值的欄位。預設為 PickleField。

  • ordered (bool) – 資料應以鍵排序的順序傳回。

  • database ( Database) – 儲存鍵/值資料的資料庫。如果未指定，則會使用記憶體中的 SQLite 資料庫。

  • table_name (str) – 用於資料儲存的表格名稱。

類似字典的 API，用於儲存鍵/值資料。如同字典，支援預期的 API，但還增加了接受用於獲取、設定和刪除項目的表達式的功能。

當 KeyValue 實例化時，表格會自動建立 (如果不存在)。

使用高效的 upsert 實作來設定和更新/覆寫鍵/值對。

基本範例

# Create a key/value store, which uses an in-memory SQLite database
# for data storage.
KV = KeyValue()

# Set (or overwrite) the value for "k1".
KV['k1'] = 'v1'

# Set (or update) multiple keys at once (uses an efficient upsert).
KV.update(k2='v2', k3='v3')

# Getting values works as you'd expect.
assert KV['k2'] == 'v2'

# We can also do this:
for value in KV[KV.key > 'k1']:
    print(value)

# 'v2'
# 'v3'

# Update multiple values at once using expression:
KV[KV.key > 'k1'] = 'vx'

# What's stored in the KV?
print(dict(KV))

# {'k1': 'v1', 'k2': 'vx', 'k3': 'vx'}

# Delete a single item.
del KV['k2']

# How many items are stored in the KV?
print(len(KV))
# 2

# Delete items that match the given condition.
del KV[KV.key > 'k1']
__contains__(expr)
參數

expr – 單一鍵或表達式

傳回值

布林值，表示鍵/表達式是否存在。

範例

>>> kv = KeyValue()
>>> kv.update(k1='v1', k2='v2')

>>> 'k1' in kv
True
>>> 'kx' in kv
False

>>> (KV.key < 'k2') in KV
True
>>> (KV.key > 'k2') in KV
False
__len__()
傳回值

儲存的項目計數。

__getitem__(expr)
參數

expr – 單一鍵或表達式。

傳回值

對應於鍵/表達式的值。

引發

如果給定單一鍵但未找到，則引發 KeyError。

範例

>>> KV = KeyValue()
>>> KV.update(k1='v1', k2='v2', k3='v3')

>>> KV['k1']
'v1'
>>> KV['kx']
KeyError: "kx" not found

>>> KV[KV.key > 'k1']
['v2', 'v3']
>>> KV[KV.key < 'k1']
[]
__setitem__(expr, value)
參數

  • expr – 單一鍵或表達式。

  • value – 要為鍵設定的值

為給定鍵設定值。如果 expr 是一個表達式，則任何符合該表達式的鍵的值都會被更新。

範例

>>> KV = KeyValue()
>>> KV.update(k1='v1', k2='v2', k3='v3')

>>> KV['k1'] = 'v1-x'
>>> print(KV['k1'])
'v1-x'

>>> KV[KV.key >= 'k2'] = 'v99'
>>> dict(KV)
{'k1': 'v1-x', 'k2': 'v99', 'k3': 'v99'}
__delitem__(expr)
參數

expr – 單一鍵或表達式。

刪除給定的鍵。如果給定一個表達式，則刪除所有符合該表達式的鍵。

範例

>>> KV = KeyValue()
>>> KV.update(k1=1, k2=2, k3=3)

>>> del KV['k1']  # Deletes "k1".
>>> del KV['k1']
KeyError: "k1" does not exist

>>> del KV[KV.key > 'k2']  # Deletes "k3".
>>> del KV[KV.key > 'k99']  # Nothing deleted, no keys match.
keys()
傳回值

表格中所有鍵的可迭代物件。

values()
傳回值

表格中所有值的可迭代物件。

items()
傳回值

表格中所有鍵/值對的可迭代物件。

update([__data=None[, **mapping]])

高效地批量插入或取代給定的鍵/值對。

範例

>>> KV = KeyValue()
>>> KV.update(k1=1, k2=2)  # Sets 'k1'=1, 'k2'=2.

>>> dict(KV)
{'k1': 1, 'k2': 2}

>>> KV.update(k2=22, k3=3)  # Updates 'k2'->22, sets 'k3'=3.

>>> dict(KV)
{'k1': 1, 'k2': 22, 'k3': 3}

>>> KV.update({'k2': -2, 'k4': 4})  # Also can pass a dictionary.

>>> dict(KV)
{'k1': 1, 'k2': -2, 'k3': 3, 'k4': 4}
get(expr[, default=None])
參數

  • expr – 單一鍵或表達式。

  • default – 如果找不到鍵，則使用預設值。

傳回值

給定鍵/表達式的值，如果找不到單一鍵則使用預設值。

取得給定鍵的值。如果鍵不存在，則傳回預設值，除非鍵是一個表達式，在這種情況下將傳回一個空列表。

pop(expr[, default=Sentinel])
參數

  • expr – 單一鍵或表達式。

  • default – 如果鍵不存在，則使用預設值。

傳回值

給定鍵/表達式的值，如果找不到單一鍵則使用預設值。

取得值並刪除給定的鍵。如果鍵不存在，則傳回預設值，除非鍵是一個表達式，在這種情況下將傳回一個空列表。

clear()

從鍵值表格中移除所有項目。

快捷方式

此模組包含輔助函數，用於表達使用 peewee 的 API 在其他情況下會有些冗長或繁瑣的事情。還有用於將模型序列化為字典以及反向操作的輔助函數。

model_to_dict(model[, recurse=True[, backrefs=False[, only=None[, exclude=None[, extra_attrs=None[, fields_from_query=None[, max_depth=None[, manytomany=False]]]]]]]])
參數

  • recurse (bool) – 是否應遞迴處理外來鍵。

  • backrefs (bool) – 是否應遞迴處理相關物件的列表。

  • only – 應包含在結果字典中的欄位實例的列表 (或集合)。

  • exclude – 應從結果字典中排除的欄位實例的列表 (或集合)。

  • extra_attrs – 應包含在字典中的實例上的屬性或方法名稱的列表。

  • fields_from_query ( Select) – 建立此模型實例的 SelectQuery 。只會序列化查詢明確選取的欄位和值。

  • max_depth (int) – 遞迴時的最大深度。

  • manytomany (bool) – 處理多對多欄位。

將模型實例 (以及選擇性的任何相關實例) 轉換為字典。

範例

>>> user = User.create(username='charlie')
>>> model_to_dict(user)
{'id': 1, 'username': 'charlie'}

>>> model_to_dict(user, backrefs=True)
{'id': 1, 'tweets': [], 'username': 'charlie'}

>>> t1 = Tweet.create(user=user, message='tweet-1')
>>> t2 = Tweet.create(user=user, message='tweet-2')
>>> model_to_dict(user, backrefs=True)
{
  'id': 1,
  'tweets': [
    {'id': 1, 'message': 'tweet-1'},
    {'id': 2, 'message': 'tweet-2'},
  ],
  'username': 'charlie'
}

>>> model_to_dict(t1)
{
  'id': 1,
  'message': 'tweet-1',
  'user': {
    'id': 1,
    'username': 'charlie'
  }
}

>>> model_to_dict(t2, recurse=False)
{'id': 1, 'message': 'tweet-2', 'user': 1}

model_to_dict 的實作相當複雜，因為它試圖支援各種用法。如果您有特殊的用法，我強烈建議您不要嘗試將一些瘋狂的參數組合塞入此函數。只要寫一個簡單的函數來完成您試圖做的事情即可。

dict_to_model(model_class, data[, ignore_unknown=False])
參數

  • model_class ( Model) – 要建構的模型類別。

  • data (dict) – 資料字典。外來鍵可以作為巢狀字典包含，回溯參考可以作為字典列表包含。

  • ignore_unknown (bool) – 是否允許無法辨識 (非欄位) 的屬性。

將資料字典轉換為模型實例，並在適當的情況下建立相關實例。

範例

>>> user_data = {'id': 1, 'username': 'charlie'}
>>> user = dict_to_model(User, user_data)
>>> user
<__main__.User at 0x7fea8fa4d490>

>>> user.username
'charlie'

>>> note_data = {'id': 2, 'text': 'note text', 'user': user_data}
>>> note = dict_to_model(Note, note_data)
>>> note.text
'note text'
>>> note.user.username
'charlie'

>>> user_with_notes = {
...     'id': 1,
...     'username': 'charlie',
...     'notes': [{'id': 1, 'text': 'note-1'}, {'id': 2, 'text': 'note-2'}]}
>>> user = dict_to_model(User, user_with_notes)
>>> user.notes[0].text
'note-1'
>>> user.notes[0].user.username
'charlie'
update_model_from_dict(instance, data[, ignore_unknown=False])
參數

  • instance ( Model) – 要更新的模型實例。

  • data (dict) – 資料字典。外來鍵可以作為巢狀字典包含，回溯參考可以作為字典列表包含。

  • ignore_unknown (bool) – 是否允許無法辨識 (非欄位) 的屬性。

使用給定的資料字典更新模型實例。

resolve_multimodel_query(query[, key='_model_identifier'])
參數

  • query – 複合式 select 查詢。

  • key (str) – 用於儲存模型識別碼的鍵

傳回值

一個可迭代的游標，為複合式 select 查詢中選取的每一列產生正確的模型實例。

用於將複合式 select 查詢中傳回的列解析為正確模型實例類型的輔助函數。例如，如果您有兩個不同表格的聯集，則此輔助函數會在迭代查詢結果時將每一列解析為正確的模型。

class ThreadSafeDatabaseMetadata

Model Metadata 實作，可提供對 database 屬性的執行緒安全存取，允許應用程式在多執行緒應用程式中安全地在執行期間交換資料庫。

用法

from playhouse.shortcuts import ThreadSafeDatabaseMetadata

# Our multi-threaded application will sometimes swap out the primary
# for the read-replica at run-time.
primary = PostgresqlDatabase(...)
read_replica = PostgresqlDatabase(...)

class BaseModel(Model):
    class Meta:
        database = primary
        model_metadata_class = ThreadSafeDatabaseMetadata

訊號支援

在 playhouse.signals 中提供了具有訊號掛鉤 (類似於 django) 的模型。若要使用訊號，您需要將專案的所有模型設為 playhouse.signals.Model 的子類別，這會覆寫必要的函數，以提供對各種訊號的支援。

from playhouse.signals import Model, post_save


class MyModel(Model):
    data = IntegerField()

@post_save(sender=MyModel)
def on_save_handler(model_class, instance, created):
    put_data_in_cache(instance.data)

警告

由於我希望是顯而易見的原因，當您使用 Model.insert()、 Model.update() 或 Model.delete() 函數時，Peewee 訊號將無法運作。這些函數會產生在 ORM 範圍之外執行的查詢，而且 ORM 不知道查詢執行時可能會或可能不會影響哪些模型實例。

訊號透過掛鉤到更高等級的 peewee API (如 Model.save() 和 Model.delete_instance() ) 來運作，其中受影響的模型實例是預先知道的。

提供了以下訊號

pre_save

在物件儲存至資料庫之前立即呼叫。提供一個額外的關鍵字引數 created，指示模型是第一次儲存還是更新。

post_save

在物件儲存至資料庫之後立即呼叫。提供一個額外的關鍵字引數 created，指示模型是第一次儲存還是更新。

pre_delete

當使用 Model.delete_instance() 時，在物件從資料庫刪除之前立即呼叫。

post_delete

當使用 Model.delete_instance() 時，在物件從資料庫刪除之後立即呼叫。

pre_init

當模型類別首次實例化時呼叫

連接處理器

每當派送訊號時，它將呼叫已註冊的任何處理器。這允許完全獨立的程式碼回應模型儲存和刪除等事件。

Signal 類別提供了一個 connect() 方法，該方法接受一個回呼函數和兩個可選參數「sender」和「name」。如果指定，「sender」參數應為單個模型類別，並允許您的回呼僅接收來自該模型的訊號。「name」參數用作方便的別名，如果您希望取消註冊訊號處理器。

範例用法

from playhouse.signals import *

def post_save_handler(sender, instance, created):
    print('%s was just saved' % instance)

# our handler will only be called when we save instances of SomeModel
post_save.connect(post_save_handler, sender=SomeModel)

所有訊號處理器都接受其前兩個引數 sender 和 instance，其中 sender 是模型類別，而 instance 是實際被操作的模型。

如果您願意，也可以使用裝飾器來連接訊號處理器。這在功能上與上面的範例相同

@post_save(sender=SomeModel)
def post_save_handler(sender, instance, created):
    print('%s was just saved' % instance)

訊號 API

class Signal

儲存接收器（回呼）的列表，並在呼叫「send」方法時呼叫它們。

connect(receiver[, name=None[, sender=None]])
參數

  • receiver (可呼叫的) – 一個可呼叫的物件，它至少接受兩個參數，一個「sender」，它是觸發訊號的模型子類別，和一個「instance」，它是實際的模型實例。

  • name (字串) – 一個簡短的別名

  • sender (模型) – 如果指定，則只有此模型類別的實例才會觸發接收器回呼。

將接收器新增至接收器的內部列表，該列表將在每次發送訊號時呼叫。

from playhouse.signals import post_save
from project.handlers import cache_buster

post_save.connect(cache_buster, name='project.cache_buster')
disconnect([receiver=None[, name=None[, sender=None]]])
參數

  • receiver (可呼叫的) – 要斷開連接的回呼

  • name (字串) – 一個簡短的別名

  • sender (模型) – 斷開特定模型的處理器。

斷開指定的接收器（或具有指定名稱別名的接收器），以便不再呼叫它。必須提供接收器或名稱。

post_save.disconnect(name='project.cache_buster')
send(instance, *args, **kwargs)
參數

instance – 模型實例

迭代接收器，並按照它們連接的順序呼叫它們。如果接收器指定了發送者，則只有在實例是發送者的實例時才會呼叫它。

pwiz，一個模型產生器

pwiz 是一個隨 peewee 一起提供的小腳本，能夠內省現有資料庫並產生適合與底層資料互動的模型程式碼。如果您已經有一個資料庫，pwiz 可以透過產生具有正確欄親和性和外鍵的骨架程式碼來給您帶來很好的提升。

如果您使用 setup.py install 安裝 peewee，pwiz 將作為「腳本」安裝，您可以直接執行

python -m pwiz -e postgresql -u postgres my_postgres_db

這將把一堆模型列印到標準輸出。因此，您可以執行此操作

python -m pwiz -e postgresql my_postgres_db > mymodels.py
python # <-- fire up an interactive shell

>>> from mymodels import Blog, Entry, Tag, Whatever
>>> print([blog.name for blog in Blog.select()])

命令列選項

pwiz 接受以下命令列選項

選項

意義

範例

-h

顯示說明

-e

資料庫後端

-e mysql

-H

要連線的主機

-H remote.db.server

-p

要連線的埠

-p 9001

-u

資料庫使用者

-u postgres

-P

資料庫密碼

-P (將提示輸入密碼)

-s

結構描述

-s public

-t

要產生的表格

-t tweet,users,relationships

-v

產生 VIEW 的模型

(無引數)

-i

將資訊中繼資料新增到產生的檔案

(無引數)

-o

保留表格欄順序

(無引數)

以下是 engine (-e) 的有效參數

  • sqlite

  • mysql

  • postgresql

警告

如果需要密碼才能存取您的資料庫，系統將提示您使用安全提示輸入密碼。

密碼將包含在輸出中。具體來說，在檔案的頂部，將定義一個 Database 以及任何需要的參數，包括密碼。

pwiz 範例

內省各種資料庫的範例

# Introspect a Sqlite database.
python -m pwiz -e sqlite path/to/sqlite_database.db

# Introspect a MySQL database, logging in as root. You will be prompted
# for a password ("-P").
python -m pwiz -e mysql -u root -P mysql_db_name

# Introspect a Postgresql database on a remote server.
python -m pwiz -e postgres -u postgres -H 10.1.0.3 pg_db_name

完整範例

$ sqlite3 example.db << EOM
CREATE TABLE "user" ("id" INTEGER NOT NULL PRIMARY KEY, "username" TEXT NOT NULL);
CREATE TABLE "tweet" (
    "id" INTEGER NOT NULL PRIMARY KEY,
    "content" TEXT NOT NULL,
    "timestamp" DATETIME NOT NULL,
    "user_id" INTEGER NOT NULL,
    FOREIGN KEY ("user_id") REFERENCES "user" ("id"));
CREATE UNIQUE INDEX "user_username" ON "user" ("username");
EOM

$ python -m pwiz -e sqlite example.db

產生以下輸出

from peewee import *

database = SqliteDatabase('example.db', **{})

class UnknownField(object):
    def __init__(self, *_, **__): pass

class BaseModel(Model):
    class Meta:
        database = database

class User(BaseModel):
    username = TextField(unique=True)

    class Meta:
        table_name = 'user'

class Tweet(BaseModel):
    content = TextField()
    timestamp = DateTimeField()
    user = ForeignKeyField(column_name='user_id', field='id', model=User)

    class Meta:
        table_name = 'tweet'

觀察

  • 外鍵 Tweet.user_id 被偵測並正確對應。

  • 偵測到 User.username UNIQUE 約束。

  • 每個模型都明確宣告其表格名稱，即使在沒有必要的情況下也是如此（因為 Peewee 會自動將類別名稱轉換為適當的表格名稱）。

  • ForeignKeyField 的所有參數都會明確宣告，即使它們遵循 Peewee 預設使用的慣例。

注意

UnknownField 是一個佔位符，如果您的結構描述包含 Peewee 不知道如何對應到欄位類別的欄宣告，則會使用它。

結構描述移轉

Peewee 現在支援結構描述移轉，並對 Postgresql、SQLite 和 MySQL 提供經過良好測試的支援。與其他結構描述移轉工具不同，peewee 的移轉不處理內省和資料庫「版本控制」。相反，peewee 提供了許多用於產生和執行結構描述變更陳述式的輔助函數。此引擎為未來建構更複雜的工具奠定了基礎。

移轉可以編寫為簡單的 Python 腳本，並從命令列執行。由於移轉僅取決於您的應用程式 Database 物件，因此應該很容易管理變更您的模型定義並維護一組移轉腳本，而不會引入依賴關係。

範例用法

首先從 migrate 模組匯入輔助程式

from playhouse.migrate import *

實例化一個 migrator。SchemaMigrator 類別負責產生結構描述變更操作，然後可以由 migrate() 輔助程式依序執行。

# Postgres example:
my_db = PostgresqlDatabase(...)
migrator = PostgresqlMigrator(my_db)

# SQLite example:
my_db = SqliteDatabase('my_database.db')
migrator = SqliteMigrator(my_db)

使用 migrate() 執行一個或多個操作

title_field = CharField(default='')
status_field = IntegerField(null=True)

migrate(
    migrator.add_column('some_table', 'title', title_field),
    migrator.add_column('some_table', 'status', status_field),
    migrator.drop_column('some_table', 'old_column'),
)

警告

移轉不在交易中執行。如果您希望移轉在交易中執行，您需要在 atomic() 內容管理員中包裝對 migrate 的呼叫，例如

with my_db.atomic():
    migrate(...)

支援的操作

在現有模型中新增欄位

# Create your field instances. For non-null fields you must specify a
# default value.
pubdate_field = DateTimeField(null=True)
comment_field = TextField(default='')

# Run the migration, specifying the database table, field name and field.
migrate(
    migrator.add_column('comment_tbl', 'pub_date', pubdate_field),
    migrator.add_column('comment_tbl', 'comment', comment_field),
)

注意

Peewee 遵循 Django 的慣例，預設會在給定的 ForeignKeyField 的欄位名稱後面附加 _id。當新增外鍵時，您會希望確保給予它正確的欄位名稱。例如，如果我想在 Tweet 模型中新增一個 user 外鍵

# Our desired model will look like this:
class Tweet(BaseModel):
    user = ForeignKeyField(User)  # I want to add this field.
    # ... other fields ...

# Migration code:
user = ForeignKeyField(User, field=User.id, null=True)
migrate(
    # Note that the column name given is "user_id".
    migrator.add_column(Tweet._meta.table_name, 'user_id', user),
)

重新命名欄位

# Specify the table, original name of the column, and its new name.
migrate(
    migrator.rename_column('story', 'pub_date', 'publish_date'),
    migrator.rename_column('story', 'mod_date', 'modified_date'),
)

刪除欄位

migrate(
    migrator.drop_column('story', 'some_old_field'),
)

將欄位設為可為空值或不可為空值

# Note that when making a field not null that field must not have any
# NULL values present.
migrate(
    # Make `pub_date` allow NULL values.
    migrator.drop_not_null('story', 'pub_date'),

    # Prevent `modified_date` from containing NULL values.
    migrator.add_not_null('story', 'modified_date'),
)

變更欄位的資料類型

# Change a VARCHAR(50) field to a TEXT field.
migrate(
    migrator.alter_column_type('person', 'email', TextField())
)

重新命名表格

migrate(
    migrator.rename_table('story', 'stories_tbl'),
)

新增索引

# Specify the table, column names, and whether the index should be
# UNIQUE or not.
migrate(
    # Create an index on the `pub_date` column.
    migrator.add_index('story', ('pub_date',), False),

    # Create a multi-column index on the `pub_date` and `status` fields.
    migrator.add_index('story', ('pub_date', 'status'), False),

    # Create a unique index on the category and title fields.
    migrator.add_index('story', ('category_id', 'title'), True),
)

刪除索引

# Specify the index name.
migrate(migrator.drop_index('story', 'story_pub_date_status'))

新增或刪除表格約束

# Add a CHECK() constraint to enforce the price cannot be negative.
migrate(migrator.add_constraint(
    'products',
    'price_check',
    Check('price >= 0')))

# Remove the price check constraint.
migrate(migrator.drop_constraint('products', 'price_check'))

# Add a UNIQUE constraint on the first and last names.
migrate(migrator.add_unique('person', 'first_name', 'last_name'))

為欄位新增或刪除資料庫層級的預設值

# Add a default value for a status column.
migrate(migrator.add_column_default(
    'entries',
    'status',
    'draft'))

# Remove the default.
migrate(migrator.drop_column_default('entries', 'status'))

# Use a function for the default value (does not work with Sqlite):
migrate(migrator.add_column_default(
    'entries',
    'timestamp',
    fn.now()))

# Or alternatively (works with Sqlite):
migrate(migrator.add_column_default(
    'entries',
    'timestamp',
    'now()'))

注意

Postgres 使用者在使用非標準的 schema 時可能需要設定 search-path。這可以透過以下方式完成

new_field = TextField(default='', null=False)
migrator = PostgresqlMigrator(db)
migrate(migrator.set_search_path('my_schema_name'),
        migrator.add_column('table', 'field_name', new_field))

遷移 API

migrate(*operations)

執行一個或多個 schema 變更操作。

用法

migrate(
    migrator.add_column('some_table', 'new_column', CharField(default='')),
    migrator.create_index('some_table', ('new_column',)),
)
class SchemaMigrator(database)
參數

database – 一個 Database 實例。

SchemaMigrator 負責產生變更 schema 的語句。

add_column(table, column_name, field)
參數

  • table (str) – 要新增欄位的表格名稱。

  • column_name (str) – 新欄位的名稱。

  • field (Field) – 一個 Field 實例。

將新欄位新增至提供的表格。提供的 field 將被用來產生適當的欄位定義。

注意

如果欄位不可為空值，則必須指定預設值。

注意

對於不可為空的欄位，欄位最初將新增為可為空的欄位，然後將執行 UPDATE 語句來使用預設值填充該欄位。最後，該欄位將被標記為不可為空。

drop_column(table, column_name[, cascade=True])
參數

  • table (str) – 要刪除欄位的表格名稱。

  • column_name (str) – 要刪除的欄位名稱。

  • cascade (bool) – 是否應使用 CASCADE 刪除該欄位。

rename_column(table, old_name, new_name)
參數

  • table (str) – 包含要重新命名欄位的表格名稱。

  • old_name (str) – 欄位的目前名稱。

  • new_name (str) – 欄位的新名稱。

add_not_null(table, column)
參數

  • table (str) – 包含欄位的表格名稱。

  • column (str) – 要設為不可為空值的欄位名稱。

drop_not_null(table, column)
參數

  • table (str) – 包含欄位的表格名稱。

  • column (str) – 要設為可為空值的欄位名稱。

add_column_default(table, column, default)
參數

  • table (str) – 包含欄位的表格名稱。

  • column (str) – 要新增預設值的欄位名稱。

  • default – 欄位的新預設值。請參閱以下注意事項。

如果預設值看起來像是字串文字，Peewee 會嘗試正確地加上引號。否則，預設值將被視為字面值。Postgres 和 MySQL 支援將預設值指定為 peewee 表達式，例如 fn.NOW()，但 Sqlite 使用者需要改為使用 default='now()'。

drop_column_default(table, column)
參數

  • table (str) – 包含欄位的表格名稱。

  • column (str) – 要移除預設值的欄位名稱。

alter_column_type(table, column, field[, cast=None])
參數

  • table (str) – 表格名稱。

  • column_name (str) – 要修改的欄位名稱。

  • field (Field) – 代表新資料類型的 Field 實例。

  • cast – （僅限 postgres）如果資料類型不相容，請指定 cast 表達式，例如 column_name::int。可以作為字串或 Cast 實例提供。

變更欄位的資料類型。此方法應謹慎使用，因為使用不相容的類型可能不受您的資料庫良好支援。

rename_table(old_name, new_name)
參數

  • old_name (str) – 表格的目前名稱。

  • new_name (str) – 表格的新名稱。

add_index(table, columns[, unique=False[, using=None]])
參數

  • table (str) – 要在其上建立索引的表格名稱。

  • columns (list) – 應建立索引的欄位列表。

  • unique (bool) – 新索引是否應指定唯一約束。

  • using (str) – 索引類型（在支援的情況下），例如 GiST 或 GIN。

drop_index(table, index_name)
參數

  • table (str) – 包含要刪除索引的表格名稱。

  • index_name (str) – 要刪除的索引名稱。

add_constraint(table, name, constraint)
參數

  • table (str) – 要新增約束的表格。

  • name (str) – 用於識別約束的名稱。

  • constraint – Check() 約束，或使用 SQL 來新增任意約束。

drop_constraint(table, name)
參數

  • table (str) – 要刪除約束的表格。

  • name (str) – 要刪除的約束名稱。

add_unique(table, *column_names)
參數

  • table (str) – 要新增約束的表格。

  • column_names (str) – 用於 UNIQUE 約束的一個或多個欄位。

class PostgresqlMigrator(database)

為 Postgresql 資料庫產生遷移。

set_search_path(schema_name)
參數

schema_name (str) – 要使用的 Schema。

設定後續操作的搜尋路徑（schema）。

class SqliteMigrator(database)

為 SQLite 資料庫產生遷移。

SQLite 對於 ALTER TABLE 查詢的支援有限，因此目前不支援 SQLite 的以下操作

  • add_constraint

  • drop_constraint

  • add_unique

class MySQLMigrator(database)

為 MySQL 資料庫產生遷移。

反射

反射模組包含用於內省現有資料庫的輔助工具。此模組在 playhouse 中的其他幾個模組內部使用，包括 DataSet 和 pwiz，一個模型產生器。

generate_models(database[, schema=None[, **options]])
參數

  • database (Database) – 要內省的資料庫實例。

  • schema (str) – 可選的要內省的 schema。

  • options – 任意選項，請參閱 Introspector.generate_models() 以了解詳細資訊。

傳回值

一個將表格名稱映射到模型類別的 dict。

為給定資料庫中的表格產生模型。有關如何使用此函數的範例，請參閱 互動式使用 Peewee 區段。

範例

>>> from peewee import *
>>> from playhouse.reflection import generate_models
>>> db = PostgresqlDatabase('my_app')
>>> models = generate_models(db)
>>> list(models.keys())
['account', 'customer', 'order', 'orderitem', 'product']

>>> globals().update(models)  # Inject models into namespace.
>>> for cust in customer.select():  # Query using generated model.
...     print(cust.name)
...

Huey Kitty
Mickey Dog
參數

model (Model) – 要列印的模型類別

傳回值

無回傳值

列印模型類別的易於理解的描述，適用於偵錯或互動式使用。目前，這會列印表格名稱，以及所有欄位及其資料類型。 互動式使用 Peewee 區段包含一個範例。

範例輸出

>>> from playhouse.reflection import print_model
>>> print_model(User)
user
  id AUTO PK
  email TEXT
  name TEXT
  dob DATE

index(es)
  email UNIQUE

>>> print_model(Tweet)
tweet
  id AUTO PK
  user INT FK: User.id
  title TEXT
  content TEXT
  timestamp DATETIME
  is_published BOOL

index(es)
  user_id
  is_published, timestamp
參數

model (Model) – 要列印的模型

傳回值

無回傳值

列印給定模型類別的 SQL CREATE TABLE，這對於偵錯或互動式使用可能很有用。有關範例用法，請參閱 互動式使用 Peewee 區段。請注意，索引和約束不包含在此函數的輸出中。

範例輸出

>>> from playhouse.reflection import print_table_sql
>>> print_table_sql(User)
CREATE TABLE IF NOT EXISTS "user" (
  "id" INTEGER NOT NULL PRIMARY KEY,
  "email" TEXT NOT NULL,
  "name" TEXT NOT NULL,
  "dob" DATE NOT NULL
)

>>> print_table_sql(Tweet)
CREATE TABLE IF NOT EXISTS "tweet" (
  "id" INTEGER NOT NULL PRIMARY KEY,
  "user_id" INTEGER NOT NULL,
  "title" TEXT NOT NULL,
  "content" TEXT NOT NULL,
  "timestamp" DATETIME NOT NULL,
  "is_published" INTEGER NOT NULL,
  FOREIGN KEY ("user_id") REFERENCES "user" ("id")
)
class Introspector(metadata[, schema=None])

可以通過實例化一個 Introspector 從資料庫中提取元數據。建議使用工廠方法 from_database()，而不是直接實例化此類別。

classmethod from_database(database[, schema=None])
參數

  • database – 一個 Database 實例。

  • schema (str) – 可選的 schema（某些資料庫支援）。

建立一個 Introspector 實例，適用於給定的資料庫。

用法

db = SqliteDatabase('my_app.db')
introspector = Introspector.from_database(db)
models = introspector.generate_models()

# User and Tweet (assumed to exist in the database) are
# peewee Model classes generated from the database schema.
User = models['user']
Tweet = models['tweet']
generate_models([skip_invalid=False[, table_names=None[, literal_column_names=False[, bare_fields=False[, include_views=False]]]]])
參數

  • skip_invalid (bool) – 跳過名稱為無效 Python 識別符號的表格。

  • table_names (list) – 要產生的表格名稱清單。如果未指定，則會為所有表格產生模型。

  • literal_column_names (bool) – 按原樣使用欄位名稱。預設情況下，欄位名稱會「python 化」，也就是說，混合大小寫會變成小寫。

  • bare_fields – 僅限 SQLite。不要為內省欄位指定資料類型。

  • include_views – 也為 VIEW 產生模型。

傳回值

一個將表格名稱映射到模型類別的字典。

內省資料庫，讀取表格、欄位和外鍵約束，然後產生一個字典，將每個資料庫表格映射到一個動態產生的 Model 類別。

資料庫 URL

此模組包含一個輔助函數，可從 URL 連接字串產生資料庫連接。

connect(url, **connect_params)

從給定的連接 URL 建立一個 Database 實例。

範例

  • sqlite:///my_database.db 將在目前目錄中為檔案 my_database.db 建立一個 SqliteDatabase 實例。

  • sqlite:///:memory: 將建立一個記憶體中的 SqliteDatabase 實例。

  • postgresql://postgres:my_password@localhost:5432/my_database 將建立一個 PostgresqlDatabase 實例。提供使用者名稱和密碼，以及要連接的主機和埠。

  • mysql://user:passwd@ip:port/my_db 將為本機 MySQL 資料庫 my_db 建立一個 MySQLDatabase 實例。

  • mysql+pool://user:passwd@ip:port/my_db?max_connections=20&stale_timeout=300 將為本機 MySQL 資料庫 my_db 建立一個 PooledMySQLDatabase 實例，其中 max_connections 設定為 20，stale_timeout 設定為 300 秒。

支援的配置

用法

import os
from playhouse.db_url import connect

# Connect to the database URL defined in the environment, falling
# back to a local Sqlite database if no database URL is specified.
db = connect(os.environ.get('DATABASE') or 'sqlite:///default.db')
parse(url)

將給定的 URL 中的資訊解析為包含 database、host、port、user 和/或 password 的字典。其他連線參數可以透過 URL 查詢字串傳遞。

如果您正在使用自訂的資料庫類別，您可以使用 parse() 函數從 URL 中提取資訊，然後將其傳遞到您的資料庫物件中。

register_database(db_class, *names)
參數

  • db_class – Database 的子類別。

  • names – 要在 URL 中用作方案的名稱列表，例如 'sqlite' 或 'firebird'

在指定的名稱下註冊其他資料庫類別。此函數可用於擴展 connect() 函數以支援其他方案。假設您有一個名為 FirebirdDatabase 的 Firebird 自訂資料庫類別。

from playhouse.db_url import connect, register_database

register_database(FirebirdDatabase, 'firebird')
db = connect('firebird://my-firebird-db')

連線池

pool 模組包含許多 Database 類別，這些類別為 PostgreSQL、MySQL 和 SQLite 資料庫提供連線池。連線池的工作方式是覆寫 Database 類別中開啟和關閉後端連線的方法。連線池可以指定逾時時間，之後連線將被回收，以及開啟連線數量的上限。

在多執行緒應用程式中，最多會開啟 max_connections 個連線。每個執行緒（或，如果使用 gevent，則為 greenlet）都會有自己的連線。

在單執行緒應用程式中，只會建立一個連線。它會持續回收，直到超出過時逾時時間或被明確關閉（使用 .manual_close()）。

預設情況下，您的應用程式只需要確保在使用完連線後將其關閉，它們就會返回到連線池。對於 Web 應用程式，這通常意味著在請求開始時，您會開啟一個連線，並且在您返回回應時，您將關閉連線。

簡單的 Postgres 連線池範例程式碼

# Use the special postgresql extensions.
from playhouse.pool import PooledPostgresqlExtDatabase

db = PooledPostgresqlExtDatabase(
    'my_app',
    max_connections=32,
    stale_timeout=300,  # 5 minutes.
    user='postgres')

class BaseModel(Model):
    class Meta:
        database = db

就是這樣！如果您想對連線池進行更精細的控制，請查看 連線管理 章節。

連線池 API

class PooledDatabase(database[, max_connections=20[, stale_timeout=None[, timeout=None[, **kwargs]]]])
參數

  • database (str) – 資料庫或資料庫檔案的名稱。

  • max_connections (int) – 最大連線數。提供 None 表示無限制。

  • stale_timeout (int) – 允許連線使用的秒數。

  • timeout (int) – 當連線池已滿時，阻塞的秒數。預設情況下，peewee 在連線池已滿時不會阻塞，而只是拋出異常。要無限期阻塞，請將此值設定為 0。

  • kwargs – 傳遞給資料庫類別的任意關鍵字引數。

旨在與 Database 的子類別一起使用的 Mixin 類別。

注意

連線不會在超出其 stale_timeout 時立即關閉。相反，只有在請求新的連線時才會關閉過時的連線。

注意

如果開啟的連線數超過 max_connections，則會引發 ValueError。

manual_close()

關閉目前開啟的連線，而不會將其返回到連線池。

close_idle()

關閉所有閒置連線。這不包括任何目前正在使用的連線，僅包括先前建立但已返回到連線池的連線。

close_stale([age=600])
參數

age (int) – 連線被視為過時的年齡。

傳回值

關閉的連線數。

關閉正在使用但超出給定年齡的連線。呼叫此方法時請小心！

close_all()

關閉所有連線。這包括當時可能正在使用的任何連線。呼叫此方法時請小心！

class PooledPostgresqlDatabase

PostgresqlDatabase 的子類別，它混合了 PooledDatabase 輔助類別。

class PooledPostgresqlExtDatabase

PostgresqlExtDatabase 的子類別，它混合了 PooledDatabase 輔助類別。PostgresqlExtDatabase 是 Postgresql 擴充功能 模組的一部分，並提供對許多 Postgres 特定功能的支持。

class PooledMySQLDatabase

MySQLDatabase 的子類別，它混合了 PooledDatabase 輔助類別。

class PooledSqliteDatabase

適用於 SQLite 應用程式的持續連線。

class PooledSqliteExtDatabase

適用於 SQLite 應用程式的持續連線，使用 SQLite 擴充功能 進階資料庫驅動程式 SqliteExtDatabase。

測試工具

包含測試 peewee 專案時有用的工具。

class count_queries([only_select=False])

將計算在內容中執行的查詢數量的上下文管理器。

參數

only_select (bool) – 僅計算 SELECT 查詢。

with count_queries() as counter:
    huey = User.get(User.username == 'huey')
    huey_tweets = [tweet.message for tweet in huey.tweets]

assert counter.count == 2
count

執行的查詢數量。

get_queries()

返回一個包含 SQL 查詢和參數列表的 2 元組列表。

assert_query_count(expected[, only_select=False])

函式或方法裝飾器，若被裝飾的函式中執行的查詢數量不等於預期的數量，則會引發 AssertionError。

class TestMyApp(unittest.TestCase):
    @assert_query_count(1)
    def test_get_popular_blogs(self):
        popular_blogs = Blog.get_popular()
        self.assertEqual(
            [blog.title for blog in popular_blogs],
            ["Peewee's Playhouse!", "All About Huey", "Mickey's Adventures"])

此函式也可以作為上下文管理器使用。

class TestMyApp(unittest.TestCase):
    def test_expensive_operation(self):
        with assert_query_count(1):
            perform_expensive_operation()

Flask 工具

playhouse.flask_utils 模組包含多個輔助程式，用於將 peewee 與 Flask 網頁框架整合。

資料庫包裝器

FlaskDB 類別是一個包裝器，用於從 Flask 應用程式內設定和引用 Peewee 資料庫。請別被它的名稱誤導：它**與 peewee 資料庫不同**。FlaskDB 的設計目的是要從您的 Flask 應用程式中移除以下樣板程式碼：

  • 根據應用程式設定資料動態建立 Peewee 資料庫實例。

  • 建立一個基底類別，您應用程式的所有模型都將繼承自該類別。

  • 在請求開始和結束時註冊鉤子，以處理開啟和關閉資料庫連線。

基本用法

import datetime
from flask import Flask
from peewee import *
from playhouse.flask_utils import FlaskDB

DATABASE = 'postgresql://postgres:password@localhost:5432/my_database'

# If we want to exclude particular views from the automatic connection
# management, we list them this way:
FLASKDB_EXCLUDED_ROUTES = ('logout',)

app = Flask(__name__)
app.config.from_object(__name__)

db_wrapper = FlaskDB(app)

class User(db_wrapper.Model):
    username = CharField(unique=True)

class Tweet(db_wrapper.Model):
    user = ForeignKeyField(User, backref='tweets')
    content = TextField()
    timestamp = DateTimeField(default=datetime.datetime.now)

上面的程式碼範例會建立並實例化一個由給定的資料庫 URL 指定的 peewee PostgresqlDatabase。請求鉤子將被設定為在收到請求時建立連線，並在傳送回應時自動關閉連線。最後，FlaskDB 類別會公開一個 FlaskDB.Model 屬性，可用作您應用程式模型的基本類別。

以下是如何存取由 FlaskDB 包裝器為您設定的已包裝 Peewee 資料庫實例：

# Obtain a reference to the Peewee database instance.
peewee_db = db_wrapper.database

@app.route('/transfer-funds/', methods=['POST'])
def transfer_funds():
    with peewee_db.atomic():
        # ...

    return jsonify({'transfer-id': xid})

注意

實際的 peewee 資料庫可以使用 FlaskDB.database 屬性存取。

這是使用 FlaskDB 設定 Peewee 資料庫的另一種方式：

app = Flask(__name__)
db_wrapper = FlaskDB(app, 'sqlite:///my_app.db')

雖然上面的範例顯示使用了資料庫 URL，但為了更進階的使用方式，您可以指定設定選項的字典，或者直接傳入 peewee Database 實例。

DATABASE = {
    'name': 'my_app_db',
    'engine': 'playhouse.pool.PooledPostgresqlDatabase',
    'user': 'postgres',
    'max_connections': 32,
    'stale_timeout': 600,
}

app = Flask(__name__)
app.config.from_object(__name__)

wrapper = FlaskDB(app)
pooled_postgres_db = wrapper.database

使用 peewee Database 物件：

peewee_db = PostgresqlExtDatabase('my_app')
app = Flask(__name__)
db_wrapper = FlaskDB(app, peewee_db)

與應用程式工廠搭配的資料庫

如果您偏好使用應用程式工廠模式，FlaskDB 類別會實作一個 init_app() 方法。

作為工廠使用：

db_wrapper = FlaskDB()

# Even though the database is not yet initialized, you can still use the
# `Model` property to create model classes.
class User(db_wrapper.Model):
    username = CharField(unique=True)


def create_app():
    app = Flask(__name__)
    app.config['DATABASE'] = 'sqlite:////home/code/apps/my-database.db'
    db_wrapper.init_app(app)
    return app

查詢工具

flask_utils 模組提供了多個輔助程式，用於管理您的網頁應用程式中的查詢。一些常見的模式包括：

get_object_or_404(query_or_model, *query)
參數

  • query_or_model – Model 類別或預先篩選的 SelectQuery。

  • query – 任意複雜的 peewee 表達式。

擷取符合給定查詢的物件，或傳回 404 未找到的回應。一個常見的使用案例可能是網誌的詳細資訊頁面。您想要擷取符合給定 URL 的文章，或傳回 404。

範例

@app.route('/blog/<slug>/')
def post_detail(slug):
    public_posts = Post.select().where(Post.published == True)
    post = get_object_or_404(public_posts, (Post.slug == slug))
    return render_template('post_detail.html', post=post)
object_list(template_name, query[, context_variable='object_list'[, paginate_by=20[, page_var='page'[, check_bounds=True[, **kwargs]]]]])
參數

  • template_name – 要呈現的範本名稱。

  • query – 要分頁的 SelectQuery 實例。

  • context_variable – 用於分頁物件清單的上下文變數名稱。

  • paginate_by – 每頁的物件數。

  • page_var – 包含頁面的 GET 引數的名稱。

  • check_bounds – 是否檢查給定的頁面是否為有效頁面。如果 check_bounds 為 True 且指定了無效的頁面，則會傳回 404。

  • kwargs – 要傳遞到範本上下文的任意鍵/值對。

擷取由給定查詢指定的分頁物件清單。分頁物件清單將使用給定的 context_variable 放入上下文，以及關於目前頁面和總頁數的中繼資料，最後還有作為關鍵字引數傳遞的任何任意上下文資料。

頁面是使用 page GET 引數指定的，例如 /my-object-list/?page=3 會傳回第三頁的物件。

範例

@app.route('/blog/')
def post_index():
    public_posts = (Post
                    .select()
                    .where(Post.published == True)
                    .order_by(Post.timestamp.desc()))

    return object_list(
        'post_index.html',
        query=public_posts,
        context_variable='post_list',
        paginate_by=10)

範本將具有以下上下文：

  • post_list，其中包含最多 10 篇文章的清單。

  • page，其中包含基於 page GET 參數值的目前頁面。

  • pagination，PaginatedQuery 實例。

class PaginatedQuery(query_or_model, paginate_by[, page_var='page'[, check_bounds=False]])
參數

  • query_or_model – Model 或包含您要分頁的記錄集合的 SelectQuery 實例。

  • paginate_by – 每頁的物件數。

  • page_var – 包含頁面的 GET 引數的名稱。

  • check_bounds – 是否檢查給定的頁面是否為有效頁面。如果 check_bounds 為 True 且指定了無效的頁面，則會傳回 404。

輔助類別，用於根據 GET 引數執行分頁。

get_page()

傳回目前選取的頁面，如 page_var GET 參數的值所示。如果沒有明確選取頁面，則此方法將傳回 1，表示第一頁。

get_page_count()

傳回可能頁面的總數。

get_object_list()

使用 get_page() 的值，傳回使用者請求的物件頁面。傳回值是帶有適當 LIMIT 和 OFFSET 子句的 SelectQuery。

如果 check_bounds 設定為 True 且請求的頁面不包含任何物件，則會引發 404。