Playhouse,擴展到Peewee?

Peewee帶有許多擴展模塊,這些擴展模塊收集在 playhouse 命名空間。盡管名字很傻,但還是有一些非常有用的擴展,特別是那些公開特定于供應商的數據庫功能的擴展,如 sqlite擴展PostgreSQL擴展 擴展。

下面您將看到一個松散組織的各種模塊列表,這些模塊構成了 playhouse .

數據庫驅動程序/特定于供應商的數據庫功能

高級功能

數據庫管理和框架集成

sqlite擴展?

sqlite擴展已移動到 their own page .

SqliteQ?

這個 playhouse.sqliteq 模塊提供的子類 SqliteExtDatabase ,它將序列化對sqlite數據庫的并發寫入。 SqliteQueueDatabase 可作為常規 SqliteDatabase 如果你想簡單點 read and write 從訪問sqlite數據庫 多線程.

SQLite在任何給定的時間只允許一個連接寫入數據庫。因此,如果您有一個多線程應用程序(例如Web服務器)需要寫入數據庫,當一個或多個嘗試寫入的線程無法獲取鎖時,您可能會偶爾看到錯誤。

SqliteQueueDatabase 旨在通過一個長期存在的連接發送所有寫查詢,從而簡化操作。好處是,您可以看到多個線程在向數據庫寫入時沒有沖突或超時。但是,缺點是,您不能發出包含多個查詢的寫事務——本質上,所有寫操作都在自動提交模式下運行。

備注

模塊的名稱來自這樣一個事實:所有寫查詢都被放入線程安全隊列中。單個工作線程監聽隊列并執行發送到隊列的所有查詢。

交易?

因為所有查詢都是由一個工作線程序列化和執行的,所以來自不同線程的事務性SQL可能會無序執行。在下面的示例中,由線程“B”啟動的事務由線程“A”回滾(結果很糟糕?。?/p>

  • 線程A:更新移植集器官='肝',…;

  • B線:開始交易;

  • 線程B:更新life_支持_系統設置計時器+=60…;

  • 線程A:回滾;--哦不…

由于有可能對來自不同事務的查詢進行交錯,因此 transaction()atomic() 方法在上被禁用 SqliteQueueDatabase .

對于希望從其他線程臨時寫入數據庫的情況,可以使用 pause()unpause() 方法。在編寫器線程完成其當前工作負載之前,這些方法會阻止調用方。然后作者斷開連接,調用者接管直到 unpause 被稱為。

這個 stop() , start()is_stopped() 方法還可以用于控制編寫器線程。

備注

看看sqlite的 isolation 有關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 參數,或者 truthyfalsey 將返回值。

DATE

strip_tz(date_str)?
參數

date_str -- 日期時間,編碼為字符串。

返回

除去所有時區信息的日期時間。

時間不會以任何方式調整,時區只會被刪除。

humandelta(nseconds[, glue=', '])?
參數
  • nseconds (int) -- 以時間增量表示的總秒數。

  • glue (str) -- 連接值的片段。

返回

易于閱讀的TimeDelta描述。

例如,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沒有日期組件,則值將作為times返回。否則,值將作為日期時間返回。

例子:

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 -- 切換鍵。

在真/假狀態之間切換鍵。例子:

>>> 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) -- 要轉換為數字的字符串。

返回

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

mode(val)?
參數

val -- 列表中的數字。

返回

觀察到的模式或最常見的數字。

計算的聚合函數 mode 價值觀。

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)?

刪除中的任何字符 chars 從開始和結束 haystack .

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實例的字符串。

表值函數,在字符串中搜索與提供的 regex . 返回找到的每個匹配項的行。

例子:

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

value
-----
extract
words
ignore
symbols

高級sqlite驅動程序apsw?

這個 apsw_ext 模塊包含一個適用于APSWSQLite驅動程序的數據庫類。

APSW項目頁面:https://github.com/rogerbins/apsw

APSW是一個非常整潔的庫,它在sqlite的C接口上提供了一個很薄的包裝器,使得使用sqlite的所有高級功能成為可能。

以下是從文檔中獲取的使用APSW的幾個原因:

  • APSW提供了SQLite的所有功能,包括虛擬表、虛擬文件系統、BLOB I/O、備份和文件控制。

  • 連接可以跨線程共享,而無需任何附加鎖定。

  • 事務由代碼顯式管理。

  • APSW可以處理嵌套事務。

  • Unicode處理正確。

  • APSW速度更快。

有關apsw和pysqlite之間差異的詳細信息,請檢查 the apsw docs .

如何使用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)?

提供全局注冊模塊的方法。有關詳細信息,請參閱 documentation on virtual tables .

參數
  • mod_name (string) -- 用于模塊的名稱

  • mod_inst (object) -- 實現 Virtual Table 界面

unregister_module(mod_name)?

注銷模塊。

參數

mod_name (string) -- 用于模塊的名稱

備注

一定要使用 Field 在中定義的子類 apsw_ext 模塊,因為它們將正確處理適應存儲的數據類型。

例如,而不是使用 peewee.DateTimeField ,確保正在導入和使用 playhouse.apsw_ext.DateTimeField .

sqlcipher后端?

備注

盡管這個擴展的代碼很短,但它還沒有經過適當的同行評審,可能引入了漏洞。

另請注意,此代碼依賴于 sqlcipher3 (Python綁定)和 sqlcipher, 那里的代碼也可能有漏洞,但由于這些都是廣泛使用的密碼模塊,我們可以預期那里會有“短暫的零日”。

sqlcipher_ext api說明?

class SqlCipherDatabase(database, passphrase, **kwargs)?

的子類 SqliteDatabase 它存儲加密的數據庫。不是標準 sqlite3 后端,它使用 sqlcipher3: 一個python包裝器,用于 sqlcipher, 這反過來又是一個加密的包裝器 sqlite3 ,所以API是 identicalSqliteDatabase 的,但對象構造參數除外:

參數
  • database -- 要打開[或創建]的加密數據庫文件名的路徑。

  • passphrase -- 數據庫加密密碼:長度至少為8個字符,但 strongly advised 更好地執行 passphrase strength 實現中的標準。

  • 如果 database 文件不存在,將 created 通過從中派生的密鑰進行加密 passhprase .

  • 嘗試打開現有數據庫時, passhprase 應該與創建時使用的相同。如果密碼短語不正確,則在首次嘗試訪問數據庫時將引發錯誤。

rekey(passphrase)?
參數

passphrase (str) -- 數據庫的新密碼。

更改數據庫的密碼。

備注

可以使用許多擴展pragma配置sqlcipher。pragma及其描述的列表可以在 SQLCipher documentation .

例如,要指定密鑰派生的pbkdf2迭代次數(默認情況下,sqlcipher 3.x中為64K,sqlcipher 4.x中為256K):

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

要使用16KB的密碼頁大小和10000頁的緩存大小,請執行以下操作:

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

另見:稍微詳細一點 example .

PostgreSQL擴展?

PostgreSQL擴展模塊提供了許多“僅Postgres”功能,目前:

以后我想增加對PostgreSQL更多功能的支持。如果您希望看到添加的特定功能,請 open a Github issue .

警告

為了開始使用下面描述的功能,您需要使用擴展 PostgresqlExtDatabase 類而不是 PostgresqlDatabase .

下面的代碼將假定您正在使用以下數據庫和基本模型:

from playhouse.postgres_ext import *

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

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

JSON支持?

Peewee基本支持Postgres的原生JSON數據類型,其形式為 JSONField . 從2.4.7版開始,Peewee還支持Postgres9.4二進制JSON jsonb 通過類型 BinaryJSONField .

警告

Postgres從9.2開始支持本地JSON數據類型(在9.3中完全支持)。要使用此功能,必須使用正確版本的Postgres psycopg2 2.5或更高版本。

使用 BinaryJSONField 具有許多性能和查詢優勢,您必須擁有Postgres9.4或更高版本。

備注

必須確保數據庫是 PostgresqlExtDatabase 為了使用 JSONField.

下面是一個使用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()

有關更多示例,請參見 JSONFieldBinaryJSONField 以下為API文件。

HStand支持?

Postgresql hstore 是嵌入的密鑰/值存儲。使用hstore,您可以在數據庫中存儲任意鍵/值對以及結構化關系數據。

使用 hstore ,在實例化 PostgresqlExtDatabase

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

目前 postgres_ext 模塊支持以下操作:

  • 存儲和檢索任意字典

  • 按關鍵字或部分字典篩選

  • 更新/向現有詞典添加一個或多個鍵

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

  • 選擇鍵、值或zip鍵和值

  • 檢索鍵/值的切片

  • 測試是否存在密鑰

  • 測試鍵是否具有非空值

使用HSt鋪?

首先,您需要從導入自定義數據庫類和hstore函數 playhouse.postgres_ext (請參見上面的代碼段)。然后,只需添加一個 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 數據類型( docs

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 documentation .

備注

要使用服務器端(或命名的)光標,必須使用 PostgresqlExtDatabase .

要使用服務器端光標執行查詢,只需使用 ServerSide() 幫手:

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不會自動調用 commit() 在執行 SELECT 查詢。如果你不 commit 迭代完成后,在連接關閉(或稍后提交事務)之前,不會釋放服務器端資源。此外,由于peewee將默認緩存光標返回的行,因此應始終調用 .iterator() 迭代大型查詢時。

如果您正在使用 ServerSide() 幫助程序、事務和調用 iterator() 將透明處理。

Postgres擴展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 實例。

R型發生器

將給定的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() ,但將與數組包含的行匹配 any 給定項目的。

# 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)?

用于存儲和檢索任意鍵/值對的字段。有關用法的詳細信息,請參閱 HStand支持 .

注意

使用 HStoreField 你需要確定 hstore 擴展已注冊到連接。要完成此操作,請實例化 PostgresqlExtDatabase 具有 register_hstore=True .

備注

默認情況下 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 ,A 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 -- 要搜索的一個或多個鍵。

查詢行是否存在 any 關鍵。

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

dumps -- 默認值是調用json.dumps()或dumps函數。您可以重寫這個方法來創建一個定制的JSON包裝器。

適用于存儲和查詢任意JSON的字段類。在模型上使用這個時,將字段的值設置為python對象(或者 dict 或A list )從數據庫中檢索值時,它將作為python數據結構返回。

備注

您必須使用Postgres9.2/psycopg2 2.5或更高版本。

備注

如果您使用的是Postgres9.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存儲數據 dictlist 對象,當從數據庫返回數據時,將使用 dictlist 也。

有關基本查詢操作的示例,請參見上面的代碼示例 JSONField . 下面的示例查詢將使用相同的 APIResponse 上述型號。

備注

默認情況下,binaryjsonfield將使用gist索引。要禁用此功能,請使用初始化字段 index=False .

備注

您必須使用Postgres9.4/psycopg2 2.5或更高版本。如果您使用的是Postgres9.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']}})))

我們也可以傳遞簡單的鍵。查找包含鍵的APIResponses foo 在頂層:

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)?

總結兩個現場數據和提供的數據。請注意,此操作不會合并或執行“Deep Concat”。

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) -- 使用普通(簡單)分析器分析搜索查詢。

返回

表示全文搜索/匹配的表達式。

例子:

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

Cockroach數據庫?

CockroachDB (CRDB)得到了peewee的大力支持。

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

安全套接字層配置?

在運行Cockroach群集時,強烈建議使用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 docs 。

蟑螂擴展接口?

這個 playhouse.cockroachdb 擴展模塊提供以下類和助手:

使用CRDB時可能有用的特殊字段類型:

  • UUIDKeyField -使用CRDB的主鍵字段實現 UUID 使用默認隨機生成的UUID鍵入。

  • RowIDField -使用CRDB的主鍵字段實現 INT 使用默認值鍵入 unique_rowid() .

  • JSONField -和博士后一樣 BinaryJSONField ,因為CRDB將JSON視為JSONB。

  • ArrayField -與Postgres擴展相同(但不支持多維數組)。

CRDB與Postgres的wire協議兼容,并且公開了一個非常相似的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() 幫手。這個helper方法接受callable,它負責執行任何可能需要重試的事務語句。

最簡單的例子 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應用程序接口?

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) -- “低”、“正?!被颉案摺?。

返回

返回回調返回的值。

加薪

ExceededMaxAttempts 如果 max_attempts 超過了。

在事務中運行SQL,并在客戶端自動重試。

用戶提供 callback

  • Must 接受一個參數 db 實例,該實例表示正在運行事務的連接。

  • Must 不嘗試提交、回滾或以其他方式管理事務。

  • May 不止一次地被召喚。

  • 應該 理想情況下只包含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 . 實現與相同的API CockroachDatabase ,但將執行客戶端連接池。

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

在事務中運行SQL,并在客戶端自動重試??吹搅藛?CockroachDatabase.run_transaction() 有關詳細信息。

參數
  • db (CockroachDatabase) -- 數據庫實例。

  • callback -- 可贖回的,接受單一 db 參數(與上面傳遞的值相同)。

備注

此函數等效于 CockroachDatabase 班級。

class UUIDKeyField?

使用CRDB的UUID主鍵字段 gen_random_uuid() 函數自動填充初始值。

class RowIDField?

使用CRDB的自動遞增整數主鍵字段 unique_rowid() 函數自動填充初始值。

參見:

  • BinaryJSONField 從Postgresql擴展(在 cockroachdb 擴展模塊,并別名為 JSONField

  • ArrayField 從Postgresql擴展。

MySQL擴展?

Peewee為使用 mysql-connector 司機或 mariadb-connector 。這些實現可以在 playhouse.mysql_ext 。

class MySQLConnectorDatabase(database, **kwargs)?

使用以下工具實現數據庫 mysql-connector 。支持的完整列表 connection parameters 。

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 。支持的完整列表 connection parameters 。

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 會嗎? not 接受以下參數:

  • charset (始終為utf8mb4)

  • sql_mode

  • use_unicode

其他特定于MySQL的幫助程序:

class JSONField?

延伸 TextField 并在python中實現透明的JSON編碼和解碼。

Match(columns, expr[, modifier=None])?
參數
  • columns -- 單人間 Field 或多個字段的元組。

  • expr (str) -- 全文搜索表達式。

  • modifier (str) -- 搜索的可選修飾符,例如 '在布爾模式' .

用于構造表單的MySQL全文搜索查詢的幫助程序類:

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

DataSet?

這個 dataset 模塊包含一個高級API,用于處理根據流行的 project of the same name . 目的 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 通過傳入格式為的數據庫URL初始化對象。 dialect://user:password@host/dbname . 見 數據庫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}]

使用事務?

數據集支持使用簡單的上下文管理器嵌套事務。

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 -- A SelectQuery ,使用生成 all() 或`~表。查找`。

  • format -- 輸出格式。默認情況下, csvjson 支持。

  • filename -- 要寫入輸出的文件名。

  • file_obj -- 類似文件的對象,用于寫入輸出。

  • encoding (str) -- 文件編碼。

  • kwargs -- 用于導出特定功能的任意參數。

thaw(table[, format='csv'[, filename=None[, file_obj=None[, strict=False[, encoding='utf8'[, **kwargs]]]]]])?
參數
  • table (str) -- 要將數據加載到的表的名稱。

  • format -- 輸入格式。默認情況下, csvjson 支持。

  • filename -- 要從中讀取數據的文件名。

  • file_obj -- 從中讀取數據的類似文件的對象。

  • strict (bool) -- 是否存儲表中不存在的列的值。

  • encoding (str) -- 文件編碼。

  • kwargs -- 導入特定功能的任意參數。

connect()?

打開與基礎數據庫的連接。如果未顯式打開連接,則第一次執行查詢時將打開一個連接。

close()?

關閉與基礎數據庫的連接。

class Table(dataset, name, model_class)?
諾因德克斯

提供用于處理給定表中的行的高級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 -- 輸出格式。默認情況下, csvjson 支持。

  • filename -- 要寫入輸出的文件名。

  • file_obj -- 類似文件的對象,用于寫入輸出。

  • kwargs -- 用于導出特定功能的任意參數。

thaw([format='csv'[, filename=None[, file_obj=None[, strict=False[, **kwargs]]]]])?
參數
  • format -- 輸入格式。默認情況下, csvjson 支持。

  • 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?

通過透明的酸洗和非酸洗存儲在字段中的數據來存儲任意的python數據。此字段擴展 BlobField . 如果 cPickle 模塊可用,將使用。

混合屬性?

混合屬性封裝了在兩個Python上運行的功能 and SQL級別?;旌蠈傩缘南敕▉碜杂? same name in 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)

這個 hybrid attribute 它的名字來源于 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。讓我們添加一個 radius 方法到 Interval 模型。因為這個方法計算絕對值,所以我們將使用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 實例 lengthradius 屬性的行為將如您所期望的那樣。但是,當作為類屬性訪問時,將生成一個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 . Must have 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()?
返回

表中所有值的I表。

items()?
返回

表中所有鍵/值對的ITable。

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 -- 未找到鍵時的默認值。

返回

給定鍵/expr的值,如果找不到單個鍵,則為默認值。

獲取給定鍵的值。如果鍵不存在,則返回默認值,除非鍵是表達式,在這種情況下,將返回空列表。

pop(expr[, default=Sentinel])?
參數
  • expr -- 單個鍵或表達式。

  • default -- 鍵不存在時的默認值。

返回

給定鍵/expr的值,如果找不到單個鍵,則為默認值。

獲取值并刪除給定的鍵。如果鍵不存在,則返回默認值,除非鍵是表達式,在這種情況下返回空列表。

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 是相當復雜的,因為它試圖支持的各種用法。如果你有特殊用法,我強烈建議你這樣做 not 嘗試在這個函數中加入一些瘋狂的參數組合。只需編寫一個簡單的函數,它就可以精確地完成您想要做的事情。

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 -- 復合選擇查詢。

  • key (str) -- 用于存儲模型標識符的鍵

返回

一個可迭代的游標,為在復合選擇查詢中選擇的每一行生成正確的模型實例。

幫助程序,用于將復合選擇查詢中返回的行解析為正確的模型實例類型。例如,如果您有兩個不同表的聯合,則當迭代查詢結果時,此助手將把每一行解析為適當的模型。

class ThreadSafeDatabaseMetadata?

型號 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

信號支持?

帶有信號鉤(A-LA Django)的型號見 playhouse.signals . 要使用這些信號,您需要將項目的所有模型都作為 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() 方法。這些方法生成在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”參數應該是單個模型類,并且允許回調只接收來自該模型類的信號?!皀ame”參數在您想要注銷信號處理程序的事件中用作方便的別名。

示例用法:

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)

所有信號處理程序都接受作為前兩個參數 senderinstance 在哪里 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[, sender=None[, name=None]])?
參數
  • receiver (callable) -- 一個至少接受兩個參數的可調用文件,“sender”是觸發信號的模型子類,“instance”是實際的模型實例。

  • sender (Model) -- 如果指定,則只有此模型類的實例才會觸發接收器回調。

  • name (string) -- 短別名

將接收器添加到接收器的內部列表中,在發送信號時將調用該列表。

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]])?
參數
  • receiver (callable) -- 要斷開連接的回調

  • name (string) -- 短別名

斷開給定接收器(或具有給定名稱別名的接收器)的連接,使其不再被調用。必須提供接收器或名稱。

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

instance -- 模型實例

循環訪問接收器,并按連接順序調用它們。如果接收者指定了發送者,則只有當實例是發送者的實例時才會調用它。

Pwiz,模型發生器?

pwiz 是一個與Peewee一起提供的小腳本,能夠內省現有數據庫并生成適合與基礎數據交互的模型代碼。如果您已經有了一個數據庫,那么pwiz可以通過使用正確的列關聯性和外鍵生成框架代碼來給您帶來很大的幫助。

如果安裝Peewee,請使用 setup.py install ,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遠程數據庫服務器

-P

要連接的端口

-P 9001

-U

數據庫用戶

-后郵局

-P

數據庫密碼

-P(將提示輸入密碼)

-S

圖式

公眾

-T

要生成的表

-推特、用戶、關系

-V

為視圖生成模型

(沒有參數)

-我

向生成的文件添加信息元數據

(沒有參數)

-O

保留表列順序

(沒有參數)

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

  • SQLite

  • MySQL

  • 波斯特雷斯爾

警告

如果訪問數據庫需要密碼,系統將提示您使用安全提示輸入密碼。

密碼將包含在輸出中. 具體來說,在文件的頂部 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 檢測到唯一約束。

  • 每個模型都顯式地聲明其表名,即使在不需要的情況下也是如此(因為peewee會自動將類名轉換為適當的表名)。

  • 的所有參數 ForeignKeyField 是顯式聲明的,即使它們在默認情況下遵循peewee使用的約定。

備注

這個 UnknownField 是在架構包含Peewee不知道如何映射到字段類的列聲明時使用的占位符。

架構遷移?

Peewee現在支持模式遷移,對PostgreSQL、SQLite和MySQL的支持測試良好。與其他模式遷移工具不同,Peewee的遷移不處理自省和數據庫“版本控制”。相反,peewee為生成和運行模式更改語句提供了許多幫助函數。這個引擎提供了一個更復雜的工具將來可以建立的基礎。

遷移可以寫成簡單的python腳本,并從命令行執行。因為遷移只依賴于您的應用程序 Database 對象,在不引入依賴項的情況下,應該可以輕松地管理更改模型定義和維護一組遷移腳本。

示例用法?

首先從 migrate 模塊:

from playhouse.migrate import *

實例化A 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'),
)

警告

遷移不在事務內部運行。如果希望在事務中運行遷移,則需要將調用包裝到 migrate 在一個 atomic() 上下文管理器,例如

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),
)

重命名字段:

# 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'))

備注

使用非標準模式時,Postgres用戶可能需要設置搜索路徑??砂慈缦路绞竭M行:

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)?

執行一個或多個架構更改操作。

用途:

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

database -- 一 Database 實例。

這個 SchemaMigrator 負責生成模式更改語句。

add_column(table, column_name, field)?
參數
  • table (str) -- 要添加列的表的名稱。

  • column_name (str) -- 新列的名稱。

  • field (Field) -- A 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) -- 要使其可以為空的列的名稱。

alter_column_type(table, column, field[, cast=None])?
參數
  • table (str) -- 表的名稱。

  • column_name (str) -- 要修改的列的名稱。

  • field (Field) -- Field 表示新數據類型的實例。

  • cast -- (僅限postgres)如果數據類型不兼容,請指定強制轉換表達式,例如。 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) -- 唯一約束的一列或多列。

class PostgresqlMigrator(database)?

為PostgreSQL數據庫生成遷移。

set_search_path(schema_name)?
參數

schema_name (str) -- 要使用的模式。

為后續操作設置搜索路徑(架構)。

class SqliteMigrator(database)?

為sqlite數據庫生成遷移。

sqlite對 ALTER TABLE 查詢,因此當前不支持對sqlite執行以下操作:

  • add_constraint

  • drop_constraint

  • add_unique

class MySQLMigrator(database)?

為MySQL數據庫生成遷移。

反射?

反射模塊包含用于自省現有數據庫的幫助程序。該模塊由Playhouse內的其他幾個模塊內部使用,包括 DataSetPwiz,模型發生器 .

generate_models(database[, schema=None[, **options]])?
參數
  • database (Database) -- 要自省的數據庫實例。

  • schema (str) -- 自省的可選模式。

  • 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) -- 可選架構(某些數據庫支持)。

創建一個 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-ized”,即混合大小寫變為小寫。

  • bare_fields -- 僅SQLite. 不要為自省的列指定數據類型。

  • include_views -- 也為視圖生成模型。

返回

將表名映射到模型類的字典。

內省數據庫,讀取表、列和外鍵約束,然后生成一個字典,將每個數據庫表映射到動態生成的 Model 班級。

數據庫URL?

此模塊包含一個助手函數,用于從URL連接字符串生成數據庫連接。

connect(url, **connect_params)?

創建一個 Database 來自給定連接URL的實例。

實例:

  • sqlite:///my_database.db 將創建一個 SqliteDatabase 文件的實例 my_database.db 在當前目錄中。

  • *sqlite:///:內存:*將在內存中創建 SqliteDatabase 實例。

  • postgresql://postgres:my_password@localhost:5432/my_database 將創建一個 PostgresqlDatabase 實例。提供用戶名和密碼,以及要連接的主機和端口。

  • mysql://user:passwd@ip:port/my_db 將創建一個 MySQLDatabase 本地mysql數據庫實例 my_db.

  • mysql+pool://user:passwd@ip:port/my_db?max_connections=20&stale_timeout=300 將創建一個 PooledMySQLDatabase 本地mysql數據庫實例 my_db 最大連接數設置為20,過時超時設置為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() 支持其他方案的功能。假設您有一個自定義數據庫類用于 Firebird 已命名 FirebirdDatabase .

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 -- 傳遞給數據庫類的任意關鍵字參數。

Mixin類用于子類 Database .

備注

當連接超過其 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 幫手。這個 PostgresqlExtDatabasePostgreSQL擴展 并為許多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查詢和參數列表組成的兩個元組的列表。

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 Web框架。

數據庫包裝?

這個 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)

上面的代碼示例將創建并實例化一個peewee PostgresqlDatabase 由給定的數據庫URL指定。請求掛鉤將配置為在收到請求時建立連接,并在發送響應時自動關閉連接。最后, FlaskDB 類公開了 FlaskDB.Model 可以用作應用程序模型基礎的屬性。

下面是如何訪問由 FlaskDB 包裝:

# 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})

備注

可以使用 FlaskDB.database 屬性。

下面是另一種配置peewee數據庫的方法 FlaskDB

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)

帶應用程序工廠的數據庫?

如果您喜歡使用 application factory pattern , the 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 模塊提供了幾個幫助程序來管理Web應用程序中的查詢。一些常見的模式包括:

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 -- A SelectQuery 分頁實例。

  • context_variable -- 用于分頁對象列表的上下文變量名。

  • paginate_by -- 每頁的對象數。

  • page_var -- 的名字 GET 包含該頁的參數。

  • check_bounds -- 是否檢查給定頁是否為有效頁。如果 check_boundsTrue 并且指定了一個無效的頁面,那么將返回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 ,A PaginatedQuery 實例。

class PaginatedQuery(query_or_model, paginate_by[, page_var='page'[, check_bounds=False]])?
參數
  • query_or_model -- 要么是 Model 或A SelectQuery 包含要分頁的記錄集合的實例。

  • paginate_by -- 每頁的對象數。

  • page_var -- 的名字 GET 包含該頁的參數。

  • check_bounds -- 是否檢查給定頁是否為有效頁。如果 check_boundsTrue 并且指定了一個無效的頁面,那么將返回404。

根據幫助器類執行分頁 GET 參數。

get_page()?

返回當前選定的頁面,如 page_var GET 參數。如果沒有顯式選擇頁面,則此方法將返回1,指示第一頁。

get_page_count()?

返回可能的總頁數。

get_object_list()?

使用的值 get_page() ,返回用戶請求的對象頁。返回值為 SelectQuery 與適當的 LIMITOFFSET 條款。

如果 check_bounds 被設定為 True 并且請求的頁面不包含任何對象,那么將引發404。