ORM 工具¶

这些模块在 Peewee 的核心 ORM 之上提供了更高级别的抽象，并可与任何数据库后端配合使用。

本页内容

快捷方式
Pydantic 集成
混合属性
键/值存储
信号
DataSet
额外的字段类型
Flask 工具

快捷方式 ¶

playhouse.shortcuts 提供了用于将模型实例序列化为字典或从字典反序列化的助手，解析复合查询以及线程安全的数据库切换。

模型序列化¶

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) – 关注外键并包含相关对象作为嵌套字典（默认值：True）。
backrefs (bool) – 关注反向引用，并将相关集合包含为字典的嵌套列表。
only – 一个字段实例的列表或集合，用于排他性包含。
exclude – 一个字段实例的列表或集合，用于排除。
extra_attrs – 一个属性或方法名称列表，将包含在输出字典中。
fields_from_query (Select) – 将序列化限制为仅生成查询中显式选择的字段。
max_depth (int) – 关注关系的最高深度。
manytomany (bool) – 包含多对多字段。

示例

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

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

t = Tweet.create(user=user, content='hello')
model_to_dict(t)
# {'id': 1, 'content': 'hello', 'user': {'id': 1, 'username': 'alice'}}

model_to_dict(t, recurse=False)
# {'id': 1, 'content': 'hello', 'user': 1}

model_to_dict(user, backrefs=True)
# {'id': 1, 'tweets': [{'id': 1, 'content': 'hello'}], 'username': 'alice'}

注意

如果您的用例不寻常，请编写一个小的自定义函数，而不是尝试用复杂的参数组合来强制使用 model_to_dict。

dict_to_model(model_class, data, ignore_unknown=False)¶

从字典构建模型实例。外键可以提供为嵌套字典；反向引用可以提供为字典列表。

参数

model_class (Model) – 要构建的模型类。
data (dict) – 数据字典。外键可以包含为嵌套字典，反向引用可以包含为字典列表。
ignore_unknown (bool) – 允许不对应模型上任何字段的键。

user = dict_to_model(User, {'id': 1, 'username': 'alice'})
user.username   # 'alice'

# Nested foreign key:
tweet = dict_to_model(Tweet, {
    'id': 1, 'content': 'hi',
    'user': {'id': 1, 'username': 'alice'}})
tweet.user.username   # 'alice'

update_model_from_dict(instance, data, ignore_unknown=False)¶

使用字典中的值更新现有模型实例。遵循与 dict_to_model() 相同的规则。

参数

instance (Model) – 要更新的模型实例。
data (dict) – 数据字典。外键可以包含为嵌套字典，反向引用可以包含为字典列表。
ignore_unknown (bool) – 允许不对应模型上任何字段的键。

复合查询解析¶

resolve_multimodel_query(query, key='_model_identifier')¶

将复合 UNION 或类似查询的行解析到正确的模型类。当两个表联合并且您需要每行作为适当模型的实例时非常有用。

参数

query – 复合 SelectQuery。
key (str) – 用于标识模型的列名。

返回

一个产生正确类型模型实例的可迭代对象。

线程安全的数据库切换¶

class ThreadSafeDatabaseMetadata¶

模型 Metadata 实现，它允许在多线程应用程序中安全地更改 database 属性。当您的应用程序可能在运行时跨线程切换活动数据库（例如主数据库/读副本）时，请使用此功能。

from playhouse.shortcuts import ThreadSafeDatabaseMetadata

primary  = PostgresqlDatabase('main')
replica  = PostgresqlDatabase('replica')

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

# Safe to do at runtime from any thread:
BaseModel._meta.database = replica

Pydantic 集成 ¶

playhouse.pydantic_utils 使用 to_pydantic() 函数从 Peewee Model 类生成 Pydantic v2 模型。

示例¶

import datetime
from peewee import *
from playhouse.pydantic_utils import to_pydantic

db = SqliteDatabase(':memory:')

class User(db.Model):
    name = CharField(verbose_name='Full Name', help_text='Display name')
    age = IntegerField()
    active = BooleanField(default=True)
    bio = TextField(null=True)
    status = CharField(
        verbose_name='Status',
        help_text='Record status',
        choices=[
            ('active', 'Active'),
            ('archived', 'Archived'),
            ('deleted', 'Deleted'),
        ])
    created = DateTimeField(default=datetime.datetime.now)

# Generate a Pydantic model in one call:
UserSchema = to_pydantic(User)

UserSchema 是一个标准的 Pydantic BaseModel。您可以验证数据，序列化实例，或从用户数据填充实例。

# Validate a dict (e.g. from an HTTP request body).
data = UserSchema.model_validate({'name': 'Huey', 'age': 14, 'status': 'active'})
print(data.model_dump())
# {'name': 'Huey', 'age': 14, 'active': True, 'bio': None, 'score': None,
#  'status': 'active', 'created': datetime.datetime(...)}

# Populate an instance from the validated data.
user = User(**validated.dict())

# Validate directly from a Peewee model instance:
huey = User.create(name='Huey', age=14, status='active')
data = UserSchema.model_validate(huey)

字段元数据如何映射¶

to_pydantic() 读取您已在 Peewee 字段上设置的元数据，并将其转换为 Pydantic 的等效项。

Peewee 属性	Pydantic 效果
`choices`	生成的字段使用限制为选择值的 `Literal` 类型，并且可用选择值会附加到字段描述中。
`default` / `default=callable`	在 Pydantic 字段上设置 `default` 或 `default_factory`，以便在输入数据中它不是必需的。
`null=True`	将类型包装在 `Optional[...]` 中，并在没有其他默认值时默认为 `None`。
`verbose_name`	成为 JSON schema 中的 `title`。
`help_text`	成为 JSON schema 中的 `description`。

没有默认值且 null=False（默认）的字段在生成的 Pydantic 模型中是 **必需** 的。

字段类型映射¶

Peewee 字段类型被映射到 Pydantic 用于验证的 Python 类型。

Peewee 字段	Python 类型
`CharField`, `FixedCharField`, `TextField`	`str`
`IntegerField`, `SmallIntegerField`, `BigIntegerField`	`int`
`AutoField`, `BigAutoField`	`int`
`FloatField`, `DoubleField`	`float`
`DecimalField`	`Decimal`
`BooleanField`	`bool`
`DateTimeField`	`datetime.datetime`
`DateField`	`datetime.date`
`TimeField`	`datetime.time`
`BlobField`	`bytes`
`UUIDField`	`uuid.UUID`
`JSONField`, `BinaryJSONField` (SQLite 或 Postgres 扩展)	`dict`
`IntervalField` (Postgres)	`datetime.timedelta`
`ForeignKeyField`	related PK 的类型

AutoField 和 BigAutoField 默认从生成的 schema 中排除（exclude_autofield=True）- 可以通过传递 exclude_autofield=False 来包含它们。

ForeignKeyField 通过相关模型的primary-key字段解析，因此指向具有 AutoField PK 的模型的外键变为 int。当您通过 relationships 参数提供嵌套 schema 时，这将被覆盖。

在映射中未找到 field_type 的任何字段都将回退到 Any，这意味着 Pydantic 将接受任何值而无需验证。如果您使用自定义字段类型并希望进行严格验证，请确保它们设置了受认可的 field_type 或自行处理转换。

当字段具有定义的 choices 时，上述映射的 Python 类型将被限制为选择值的 Literal **替换**，而不管底层字段类型如何。

API 参考¶

to_pydantic(model_cls, exclude=None, include=None, exclude_autofield=True, model_name=None, relationships=None, base_model=None)¶

从 Peewee 模型生成 Pydantic BaseModel 类。

参数

model_cls (Model) – Peewee 模型类。
exclude (set 或 list) – 要从生成的 schema 中排除的字段名称。
include (set 或 list) – 如果提供，则 **仅** 这些字段名称会出现在生成的 schema 中。所有其他字段都将被排除。
exclude_autofield (bool) – 当为 True（默认值）时，自增主键字段将从 schema 中省略。当您需要在响应中包含 id 字段时，将其设置为 False。
model_name (str) – 为生成的 Pydantic 类指定的名称。默认为 <ModelName>Schema。
relationships (dict) – 一个映射，告诉 to_pydantic 如何将外键或反向引用字段处理为嵌套的 Pydantic 模型，而不是扁平的标量值。请参阅下面的嵌套关系。
base_model – 用户提供的 Pydantic BaseModel 子类，用作生成模型的基类。

返回

一个配置了 from_attributes=True 的 Pydantic BaseModel 子类。

为给定的 Peewee model_cls 生成 Pydantic Model。生成的模型将保留 Peewee 字段元数据

choices - 限制字段的可接受值。
default - 为字段提供默认值。
verbose_name - 为字段提供人类可读的标题。
help_text - 为字段提供人类可读的描述。
null - 控制字段是可选的还是必需的。

外键字段通过底层列名公开，并接受标量值，**除非**您通过 relationships 参数指定关系 schema。请参阅下方示例。

外键处理¶

默认情况下，外键字段通过其 **底层列名**（例如 user_id 而不是 user）公开，并接受纯标量值，通常是整数主键。这使得 schema 保持扁平，并且在接受输入数据时非常有用。

class Tweet(db.Model):
    user = ForeignKeyField(User, backref='tweets')
    content = TextField()
    timestamp = DateTimeField(default=datetime.datetime.now)
    is_published = BooleanField(default=True)

TweetSchema = to_pydantic(Tweet)

# The schema exposes the column name "user_id", not "user":
data = TweetSchema.model_validate({'user_id': 1, 'content': 'hello'})
print(data.model_dump())
# {'user_id': 1,
#  'content': 'hello',
#  'timestamp': datetime.datetime(...),
#  'is_published: True}

# Works when validating from a model instance too:
tweet = Tweet.create(user=huey, content='hello')
data = TweetSchema.model_validate(tweet)
print(data.model_dump())
# {'user_id': 1,
#  'content': 'hello',
#  'timestamp': datetime.datetime(...),
#  'is_published: True}

嵌套关系¶

当您希望嵌入相关对象而不是仅 ID 时，请传递一个 relationships 字典，该字典将 Peewee ForeignKeyField（或反向引用）映射到应为嵌套对象使用的 Pydantic schema。

嵌套外键

# Include the id field so it appears in the response.
UserSchema = to_pydantic(User, exclude_autofield=False)

TweetResponse = to_pydantic(
    Tweet,
    exclude_autofield=False,
    relationships={Tweet.user: UserSchema})

tweet = Tweet.create(user=huey, content='hello')

data = TweetResponse.model_validate(tweet)
print(data.model_dump())
# {'id': 1,
#  'content': 'hello',
#  'user': {'id': 1, 'name': 'Huey', 'age': 14, ...},
#  'timestamp': datetime.datetime(...),
#  'is_published': True}

注意

从模型实例进行验证将访问 tweet.user，如果关系尚未加载，这将触发 SELECT 查询。为了避免额外的查询，请使用 join。

tweet = (Tweet
         .select(Tweet, User)
         .join(User)
         .get())
data = TweetResponse.model_validate(tweet)  # No additional query.

嵌套反向引用

反向引用以相同的方式工作，但 schema 必须包装在 List[...] 中，因为反向引用可能包含 0..n 条记录。

from typing import List

# Exclude the "user" FK from the tweet schema to avoid circular nesting.
TweetResponse = to_pydantic(Tweet, exclude={'user'}, exclude_autofield=False)

UserDetail = to_pydantic(
    User,
    exclude_autofield=False,
    relationships={User.tweets: List[TweetResponse]})

user = User.create(name='Huey', age=14, status='active')
Tweet.create(user=user, content='tweet 0')
Tweet.create(user=user, content='tweet 1')

data = UserDetail.model_validate(user)
print(data.model_dump())
# {'id': 1, 'name': 'Huey', ...,
#  'tweets': [{'id': 1, 'content': 'tweet 0', ...},
#             {'id': 2, 'content': 'tweet 1', ...}]}

注意

与外键一样，访问反向引用会触发查询。使用 prefetch() 提前加载集合。

users = (User
         .select()
         .where(User.id == 123)
         .prefetch(Tweet))

data = UserDetail.model_validate(users[0])  # No additional query.

JSON schema 输出¶

由于生成的类是常规的 Pydantic 模型，因此您可以调用 model_json_schema() 来获取适合 OpenAPI 文档的 JSON-schema 字典。

import json
print(json.dumps(UserSchema.model_json_schema(), indent=2))

{
  "properties": {
    "name": {
      "description": "Display name",
      "title": "Full Name",
      "type": "string"
    },
    "age": {
      "title": "Age",
      "type": "integer"
    },
    "active": {
      "default": true,
      "title": "Active",
      "type": "boolean"
    },
    "bio": {
      "anyOf": [{"type": "string"}, {"type": "null"}],
      "default": null,
      "title": "Bio"
    },
    "status": {
      "description": "Record status | Choices: 'active' = Active, 'archived' = Archived, 'deleted' = Deleted",
      "enum": ["active", "archived", "deleted"],
      "title": "Status",
      "type": "string"
    },
    "created": {
      "format": "date-time",
      "title": "Created",
      "type": "string"
    }
  },
  "required": ["name", "age", "status"],
  "title": "UserSchema",
  "type": "object"
}

请注意，name、age 和 status 是唯一的必填字段。所有其他字段都有默认值（active 默认为 True，bio 默认为 None，而 created 使用 default_factory）。

混合属性 ¶

一个 *混合属性* 在访问模型 **实例**（执行 Python 逻辑）或模型 **类**（生成 SQL 表达式）时表现不同。这允许您编写既作为 Python 计算又作为可组合 SQL 子句的 Python 方法。

该概念借用于 SQLAlchemy 的 hybrid 扩展。

from playhouse.hybrid import hybrid_property, hybrid_method

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)

在实例上，Python 算术运行

i = Interval(start=1, end=5)
i.length  # 4 (Python arithmetic)
i.contains(3)  # True (Python comparison)

在类上，SQL 被生成

Interval.select().where(Interval.length > 5)
# WHERE ("end" - "start") > 5

Interval.select().where(Interval.contains(2))
# WHERE ("start" <= 2) AND (2 < "end")

当 Python 和 SQL 实现不同时，提供单独的 expression 覆盖

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

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

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

示例

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)

class hybrid_property(fget, fset=None, fdel=None, expr=None)¶

装饰器，用于定义具有独立实例和类行为的属性。使用 @prop.expression 指定 SQL 形式（当它与 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)
)

class hybrid_method(func, expr=None)¶

装饰器，用于定义具有独立实例和类行为的方法。使用 @method.expression 指定 SQL 形式。

示例

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

键/值存储 ¶

playhouse.kv.KeyValue 提供了一个由 Peewee 数据库实例支持的持久化字典。

from playhouse.kv import KeyValue

KV = KeyValue()   # Defaults to an in-memory SQLite database.

KV['k1'] = 'v1'
KV.update(k2='v2', k3='v3')

assert KV['k2'] == 'v2'
print(dict(KV))   # {'k1': 'v1', 'k2': 'v2', 'k3': 'v3'}

# Expression-based access:
for value in KV[KV.key > 'k1']:
    print(value)    # 'v2', 'v3'

# Expression-based bulk update:
KV[KV.key > 'k1'] = 'updated'

# Expression-based deletion:
del KV[KV.key > 'k1']

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) – 底层表的名称。

表在构造时会自动创建（如果尚不存在）。支持标准的字典接口以及基于表达式的访问。

__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'
print(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.

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

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

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

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

print(dict(KV))  # {'k1': 1, 'k2': -2, 'k3': 3, 'k4': 4}

get(expr, default=None)¶

参数

expr – 单个键或表达式。
default – 如果找不到键，则为默认值。

返回

给定键/表达式的值，或在找不到单个键时为默认值。

获取给定键处的值。如果键不存在，则返回默认值，除非键是表达式，在这种情况下将返回一个空列表。

pop(expr, default=Sentinel)¶

参数

expr – 单个键或表达式。
default – 如果键不存在，则为默认值。

返回

给定键/表达式的值，或在找不到单个键时为默认值。

获取值并删除给定的键。如果键不存在，则返回默认值，除非键是表达式，在这种情况下将返回一个空列表。

clear()¶: 从键值表中删除所有项。

信号 ¶

playhouse.signals 添加了 Django 风格的模型生命周期信号。模型必须继承 playhouse.signals.Model（而不是 peewee.Model）才能触发钩子。

from playhouse.signals import Model, post_save

class MyModel(Model):
    data = IntegerField()
    class Meta:
        database = db

@post_save(sender=MyModel)
def on_save(model_class, instance, created):
    if created:
        notify_new(instance)

提供的信号如下

pre_save: 在对象保存到数据库之前立即调用。提供一个额外的关键字参数 created，指示模型是第一次保存还是更新。
post_save: 在对象保存到数据库之后立即调用。提供一个额外的关键字参数 created，指示模型是第一次保存还是更新。
pre_delete: 在使用 Model.delete_instance() 从数据库删除对象之前立即调用。
post_delete: 在使用 Model.delete_instance() 从数据库删除对象之后立即调用。
pre_init: 在模型类首次实例化时调用

警告

信号仅通过高级实例方法（save(), delete_instance()）触发。通过 insert(), update() 和 delete() 进行批量操作不会触发信号，因为没有模型实例参与。

连接处理程序¶

每当分派信号时，它都会调用已注册的任何处理程序。这允许完全独立的代码响应诸如模型保存和删除之类的事件。

Signal 类提供了一个 connect() 方法，该方法接受一个回调函数和两个可选参数“sender”和“name”。如果指定，则“sender”参数应为单个模型类，并允许您的回调仅接收来自该模型类的信号。“name”参数用作一个方便的别名，以防您需要取消注册信号处理程序。

示例

@post_save(sender=MyModel, name='project.cache_buster')
def cache_bust(sender, instance, created):
    cache.delete(make_cache_key(instance))

或手动连接

def on_delete(sender, instance):
    audit_log(instance)

pre_delete.connect(on_delete, sender=MyModel)

按名称或引用断开连接

post_save.disconnect(name='project.cache_buster')
pre_delete.disconnect(on_delete)

信号回调签名

pre_init(sender, instance)
pre_save(sender, instance, created)
post_save(sender, instance, created)
pre_delete(sender, instance)
post_delete(sender, instance)

class Signal¶

存储接收者（回调）列表，并在调用“send”方法时调用它们。

connect(receiver, name=None, sender=None)¶

参数

receiver (callable) – 一个至少接受两个参数的可调用对象，“sender”，它是触发信号的模型子类，以及“instance”，它是实际的模型实例。
name (string) – 一个简短的别名
sender (Model) – 如果指定，只有该模型类的实例才会触发接收者回调。

将接收者添加到内部接收者列表中，该列表将在信号发送时被调用。

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

param callable receiver

要断开的回调

param string name

一个简短的别名

param Model sender

断开模型特定的处理程序。

断开给定接收者（或具有给定别名的接收者）的连接，使其不再被调用。必须提供接收者或名称。
post_save.disconnect(name='project.cache_buster')

send(instance, *args, **kwargs)¶

参数: instance – 模型实例

迭代接收者，并按它们连接的顺序调用它们。如果接收者指定了发送者，则仅当实例是发送者实例时才调用它。

数据集 ¶

playhouse.dataset 提供了一个面向字典的 API 来处理关系数据，其模型基于 dataset library。它适用于快速脚本、数据加载以及 CSV/JSON 导入导出。

基本操作

from playhouse.dataset import DataSet

db = DataSet('sqlite:///data.db')

# Access a table (created automatically if it doesn't exist):
users = db['user']

# Insert rows with any columns:
users.insert(name='Alice', age=30)
users.insert(name='Bob', age=25, active=True)  # New column added automatically.

# Retrieve rows:
alice = users.find_one(name='Alice')
print(alice)   # {'id': 1, 'name': 'Alice', 'age': 30, 'active': None}

for user in users:
    print(user['name'])

for admin in users.find(active=True):
    print(admin['name'])  # Bob.

# Update:
users.update(name='Alice', age=31, columns=['name'])  # 'name' is the lookup.

# Update all records:
users.update(admin=False)

# Delete:
users.delete(name='Bob')

导出和导入数据

# Export to JSON:
db.freeze(users.all(), format='json', filename='users.json')

# Export CSV to stdout:
db.freeze(users.all(), format='csv', file_obj=sys.stdout)

# Import from CSV:
db.thaw('user', format='csv', filename='import.csv')

# Import a JSON file to a new table.
db.thaw('new_table', format='json', filename='json-data.json')

事务

# Transactions.
with db.transaction() as txn:
    users.insert(name='Charlie')

    with db.transaction() as nested_txn:
        table.update(name='Charlie', favorite_orm='sqlalchemy', columns=['name'])

        nested_txn.rollback()  # JK.

内省

print(db.tables)
# ['new_table', 'user']

print(db['user'].columns)
# ['id', 'age', 'name', 'active', 'admin', 'favorite_orm']

print(len(db['user']))
# 2

class DataSet(url, **kwargs)¶

参数

url – 数据库 URL 或 Database 实例。
kwargs – 在内省数据库时传递给 Introspector.generate_models() 的其他关键字参数。

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', iso8601_datetimes=False, base64_bytes=False, **kwargs)¶

参数

query – 一个 SelectQuery，使用 all() 或 ~Table.find 生成。
format – 输出格式。默认支持 *csv* 和 *json*。
filename – 要写入输出的文件名。
file_obj – 要写入输出的文件对象。
encoding (str) – 文件编码。
iso8601_datetimes (bool) – 以 ISO 8601 格式编码日期时间。
base64_bytes (bool) – 以 base64 编码二进制数据。默认使用十六进制。
kwargs – 特定于导出的任意参数。

将数据导出到文件。

thaw(table, format='csv', filename=None, file_obj=None, strict=False, encoding='utf8', iso8601_datetimes=False, base64_bytes=False, **kwargs)¶

参数

table (str) – 要将数据加载到的表的名称。
format – 输入格式。默认支持 *csv* 和 *json*。
filename – 要从中读取数据的文件名。
file_obj – 要从中读取数据的类文件对象。
strict (bool) – 是否存储不存在于表上的列的值。
encoding (str) – 文件编码。
iso8601_datetimes (bool) – 从 ISO 8601 格式解码日期和时间。
base64_bytes (bool) – 从 base64 解码 BLOB 字段数据。默认假定为十六进制。
kwargs – 特定于导入的任意参数。

将数据从文件导入到 table。如果 strict=False（默认值），则会自动添加新列。

connect()¶
close()¶: 打开或关闭底层数据库连接。

class Table(dataset, name, model_class)¶

提供处理表中行的 API。

columns¶: 列名列表。

model_class¶: 一个动态创建的 Model 类。

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)¶: 返回匹配相等条件的所有行（如果没有条件，则返回所有行）。

find_one(**query)¶: 返回第一个匹配的行，或者 None。

all()¶: 返回所有行。

delete(**query)¶: 删除匹配的行（如果没有条件，则删除所有行）。

create_index(columns, unique=False)¶

在给定列上创建索引

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

freeze(format='csv', filename=None, file_obj=None, encoding='utf8', iso8601_datetimes=False, base64_bytes=False, **kwargs)¶

参数

format – 输出格式。默认支持 *csv* 和 *json*。
filename – 要写入输出的文件名。
file_obj – 要写入输出的文件对象。
encoding (str) – 文件编码。
iso8601_datetimes (bool) – 以 ISO 8601 格式编码日期时间。
base64_bytes (bool) – 以 base64 编码二进制数据。默认使用十六进制。
kwargs – 特定于导出的任意参数。

thaw(format='csv', filename=None, file_obj=None, strict=False, encoding='utf8', iso8601_datetimes=False, base64_bytes=False, **kwargs)¶

参数

format – 输入格式。默认支持 *csv* 和 *json*。
filename – 要从中读取数据的文件名。
file_obj – 要从中读取数据的类文件对象。
strict (bool) – 是否存储不存在于表上的列的值。
encoding (str) – 文件编码。
iso8601_datetimes (bool) – 从 ISO 8601 格式解码日期和时间。
base64_bytes (bool) – 从 base64 解码 BLOB 字段数据。默认假定为十六进制。
kwargs – 特定于导入的任意参数。

附加字段类型 ¶

playhouse.fields 提供了两种通用字段类型。

class CompressedField(compression_level=6, algorithm='zlib', **kwargs)¶

使用 zlib 或 bz2 存储压缩的二进制数据。继承 BlobField；压缩和解压缩是透明的。

from playhouse.fields import CompressedField

class LogEntry(Model):
    payload = CompressedField(algorithm='zlib', compression_level=9)

参数

compression_level (int) – 0-9（9 为最大压缩）。
algorithm (str) – 'zlib' 或 'bz2'。

class PickleField¶

通过将任意 Python 对象 pickle 到 BlobField 来存储它们。

from playhouse.fields import PickleField

class CachedResult(Model):
    data = PickleField()

CachedResult.create(data={'nested': [1, 2, 3]})

Flask 工具 ¶

playhouse.flask_utils 简化了 Peewee 与 Flask 的集成。

FlaskDB 包装器¶

FlaskDB 处理三个样板任务：

从 Flask 的 app.config 创建 Peewee 数据库实例。
提供一个 Model 基类，其 Meta.database 已连接到 Peewee 实例。
注册 before_request / teardown_request 钩子，为每次请求打开和关闭连接。

基本设置

from flask import Flask
from playhouse.flask_utils import FlaskDB

app = Flask(__name__)
app.config['DATABASE'] = 'postgresql://postgres:pw@localhost/my_app'

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

访问底层 Peewee 数据库

peewee_db = db_wrapper.database

@app.route('/transfer', methods=['POST'])
def transfer():
    with peewee_db.atomic():
        # ... transactional logic ...
    return jsonify({'ok': True})

应用程序工厂模式

db_wrapper = FlaskDB()

class User(db_wrapper.Model):
    username = CharField(unique=True)

def create_app():
    app = Flask(__name__)
    app.config['DATABASE'] = 'sqlite:///my_app.db'
    db_wrapper.init_app(app)
    return app

通过字典或直接通过 Database 实例进行配置

# Dictionary-based (uses playhouse.db_url under the hood):
app.config['DATABASE'] = {
    'name': 'my_app',
    'engine': 'playhouse.pool.PooledPostgresqlDatabase',
    'user': 'postgres',
    'max_connections': 32,
}

# Pass a database object:
peewee_db = PostgresqlExtDatabase('my_app')
db_wrapper = FlaskDB(app, peewee_db)

排除路由以进行连接管理

app.config['FLASKDB_EXCLUDED_ROUTES'] = ('health_check', 'static')

class FlaskDB(app=None, database=None)¶

参数

app – Flask 应用程序实例（可选；对于工厂模式，请使用 init_app()）。
database – 数据库 URL 字符串、配置字典或 Database 实例。

database¶: 底层 Database 实例。

Model¶: 已绑定到此数据库实例的基类 Model。

init_app(app)¶: 绑定到 Flask 应用程序（工厂模式）。

查询助手¶

get_object_or_404(query_or_model, *query)¶

参数

query_or_model – 要么是 Model 类，要么是预先过滤的 SelectQuery。
query – Peewee 过滤表达式。

检索匹配给定查询的单个对象，如果没有匹配项则中止并返回 HTTP 404。

@app.route('/post/<slug>/')
def post_detail(slug):
    post = get_object_or_404(
        Post.select().where(Post.published == True),
        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 (str) – 要渲染的模板。
query – 要分页的 SelectQuery。
context_variable (str) – 对象页面的模板变量名（默认值：'object_list'）。
paginate_by (int) – 每页项数。
page_var (str) – 页码的 GET 参数名。
check_bounds (bool) – 是否为无效页码返回 404。
kwargs – 额外的模板上下文变量。

模板接收

object_list（或 context_variable）- 对象页面。
page - 当前页码。
pagination - 一个 PaginatedQuery 实例。

@app.route('/posts/')
def post_list():
    return object_list(
        'post_list.html',
        query=Post.select().where(Post.published == True),
        paginate_by=10)

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()¶: 返回当前页码（1-based；默认为 1）。

get_page_count()¶: 返回总页数。

get_object_list()¶: 返回请求页面的 SelectQuery，并应用适当的 LIMIT 和 OFFSET。如果 check_bounds=True 且页面为空，则返回 404。

ORM 工具¶

快捷方式 ¶

模型序列化¶

复合查询解析¶

线程安全的数据库切换¶

Pydantic 集成 ¶

示例¶

字段元数据如何映射¶

字段类型映射¶

API 参考¶

外键处理¶

嵌套关系¶

JSON schema 输出¶

混合属性 ¶

键/值存储 ¶

信号 ¶

连接处理程序¶

数据集 ¶

附加字段类型 ¶

Flask 工具 ¶

FlaskDB 包装器¶

查询助手¶

peewee

导航

相关主题