Postgresql¶

playhouse.postgres_ext 模块公开了在标准的 PostgresqlDatabase 中不可用的 Postgresql 特定字段类型和功能。

本页内容

入门
PostgresqlExtDatabase
JSON 支持
HStore
数组
Interval
DateTimeTZ 字段
全文搜索
服务器端游标
CockroachDB

入门 ¶

要开始使用，请导入 playhouse.postgres_ext 模块并使用 PostgresqlExtDatabase 数据库类

from playhouse.postgres_ext import *

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

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

PostgresqlExtDatabase ¶

class PostgresqlExtDatabase(database, server_side_cursors=False, register_hstore=False, prefer_psycopg3=False, **kwargs)¶

扩展了 PostgresqlDatabase，并且是使用...

ArrayField
DateTimeTZField
JSONField / BinaryJSONField
HStoreField
TSVectorField
服务器端游标

参数

database (str) – 要连接的数据库名称。
server_side_cursors (bool) – SELECT 查询是否应使用服务器端游标。
register_hstore (bool) – 注册 hstore 扩展。
prefer_psycopg3 (bool) – 如果同时安装了 psycopg2 和 psycopg3，则指示 Peewee 优先使用 psycopg3。

使用 server_side_cursors 时，请务必用 ServerSide() 包装您的查询。

class PooledPostgresqlExtDatabase(database, **kwargs)¶: PostgresqlExtDatabase 的连接池变体。

class Psycopg3Database(database, **kwargs)¶: 与 PostgresqlExtDatabase 相同，但指定了 prefer_psycopg3=True。

class PooledPsycopg3Database(database, **kwargs)¶: Psycopg3Database 的连接池变体。

JSON 支持 ¶

Peewee 为 Postgresql 提供了两种 JSON 字段类型

BinaryJSONField - 以高效的二进制 jsonb 格式存储 JSON。支持键/项访问、包含操作。
JSONField - 将 JSON 存储为文本，支持键/项访问。

大多数应用程序会希望使用 BinaryJSONField (JSONB)

更快的查询：无需每次都解析整个 JSON 文档即可直接访问数据元素。
索引支持：支持通过 GiST 或 GIN 进行索引。
更新更快，无需重写整个文档。

JSONField 仅在您必须按原样存储确切的 JSON 数据（空格、对象键顺序）时才更可取。

from playhouse.postgres_ext import PostgresqlExtDatabase, BinaryJSONField

db = PostgresqlExtDatabase('my_app')

class Event(Model):
    data = BinaryJSONField()
    class Meta:
        database = db

# Store data:
Event.create(data={
    'type': 'login',
    'user_id': 42,
    'request': {'ip': '1.2.3.4'},
    'success': True})

# Filter using a nested key:
query = (Event
         .select()
         .where(Event.data['request']['ip'] == '1.2.3.4'))

# Select, group and order-by JSON values.
query = (Event
         .select(Event.data['user_id'],
                 fn.COUNT(Event.id))
         .group_by(Event.data['user_id'])
         .order_by(Event.data['user_id'])
         .tuples())

# Retrieve JSON objects.
query = (Event
         .select(Event.data['request'].as_json().alias('request'))
         .where(Event.data['user_id'] == 42))
for event in query:
    print(event.request['ip'])

提示

请参阅 Postgresql JSON 文档，了解有关使用 JSON 和 JSONB 的深入讨论和示例。

BinaryJSONField 和 JSONField¶

class BinaryJSONField(dumps=None, *args, **kwargs)¶

参数: dumps – json.dumps 的自定义实现

扩展了 JSONField 以用于 jsonb 类型。

默认情况下，BinaryJSONField 将使用 GiST 索引。要禁用此功能，请使用 index=False 初始化字段。

as_json()¶: 反序列化并返回给定路径的 JSON 值。

concat(data)¶

将字段值与 data 连接。请注意，这是一个浅层操作，不会深度合并嵌套对象。

示例

# Add object - if "result" key existed before it is overwritten.
(Event
 .update(data=Event.data.concat({'result': {'success': True}}))
 .execute())

嵌套数据也可以使用 concat()

# Select the result subkey and merge with additional data:
# {'ip': '1.2.3.4'} --> {'ip': '1.2.3.4', 'status': 'ok'}

Event.select(Event.data['result'].concat({'status': 'ok'}))

contains(other)¶

测试此字段的值是否包含 other（作为子集）。other 可以是部分 dict、list 或标量值。对于匹配部分 JSON 对象或检查项是否在数组中很有用。

Event.create(data={
    'type': 'rename',
    'name': 'new name',
    'metadata': {'old_name': 'the old name'},
    'tags': ['t1', 't2', 't3']})

# These queries match the above row:

# Search by partial object:
Event.select().where(Event.data.contains({'type': 'rename'}))

# Partial object and partial array:
Event.select().where(Event.data.contains({
    'type': 'rename',
    'tags': ['t2', 't1'],
}))

# Partial array, irrespective of ordering:
Event.select().where(Event.data['tags'].contains(['t2', 't1']))

# Search array by individual item:
Event.select().where(Event.data['tags'].contains('t1'))

要测试一个**键**是否简单存在，请使用 has_key()

Event.select().where(Event.data.has_key('name'))

# Or search a sub-key.
Event.select().where(Event.data['metadata'].has_key('old_name'))

contains_any(*keys)¶

测试 JSON 值中是否存在 keys 中的任何一个。

Event.create(data={
    'type': 'rename',
    'name': 'new name',
    'metadata': {'old_name': 'the old name'},
    'tags': ['t1', 't2', 't3']})

# These queries match the above row:

Event.select().where(Event.data.contains_any('name', 'other'))

# Search a nested object:
Event.select().where(
    Event.data['metadata'].contains_any('old_name', 'old_status'))

# Search nested object for items in an array:
Event.select().where(Event.data['tags'].contains_any('t3', 'tx'))

contains_all(*keys)¶

测试 JSON 值中是否存在所有 keys。

Event.create(data={
    'type': 'rename',
    'name': 'new name',
    'metadata': {'old_name': 'the old name'},
    'tags': ['t1', 't2', 't3']})

# These queries match the above row:

Event.select().where(Event.data.contains_all('name', 'tags'))

# Search nested object for items in an array:
Event.select().where(Event.data['tags'].contains_all('t3', 't2'))

contained_by(other)¶

测试此字段的值是否是 other 的子集。

Event.create(data={
    'type': 'login',
    'result': {'success': True}})

Event.create(data={
    'type': 'rename',
    'name': 'new name',
    'metadata': {'old_name': 'the old name'},
    'tags': ['t1', 't2', 't3']})

# Matches the login row.
(Event
 .select()
 .where(Event.data.contained_by({
     'type': 'login',
     'result': {'success': True, 'message': 'OK'}})))

# Match events that have a result w/success=True and/or
# error=False:
Event.select().where(Event.data['result'].contained_by({
    'success': True,
    'error': False})

# Check that tags are subset of the popular tags (matches rename row).
popular_tags = ['t3', 't2', 't1', 'tx', 'ty']
Event.select().where(Event.data['tags'].contained_by(popular_tags))

has_key(key)¶

测试 key 是否存在。

Event.select().where(Event.data.has_key('result'))

Event.select().where(Event.data['result'].has_key('success'))

remove(*keys)¶

从 JSON 对象中删除一个或多个键。

# Atomically remove key:
Event.update(data=Event.data.remove('result')).execute()

# Equivalent to above:
Event.update(data=Event.data['result'].remove()).execute()

# Remove deeply-nested item:
Event.update(data=Event.data['metadata']['prior'].remove())

length()¶

返回给定路径的 JSON 数组的长度。

Event.select().where(Event.data['tags'].length() > 3)

extract(*path)¶

提取给定路径的 JSON 数据。

Event.select().where(Event.data.extract('tags', 0) == 'first_tag')

Event.select().where(Event.data.extract('result', 'success') == True)

# Equivalent to above.
Event.select().where(Event.data['result'].extract('success') == True)

class JSONField(dumps=None, *args, **kwargs)¶

参数: dumps – json.dumps 的自定义实现

存储和检索 JSON 数据的字段。支持 __getitem__ 键访问，用于过滤和子对象检索。

请考虑改用 BinaryJSONField，因为它提供了更好的性能和更强大的查询选项。

as_json()¶: 反序列化并返回给定路径的 JSON 值。

concat(data)¶

将字段值与 data 连接。请注意，这是一个浅层操作，不会深度合并嵌套对象。

请参阅 BinaryJSONField.concat() 的用法示例。

length()¶

返回给定路径的 JSON 数组的长度。

请参阅 BinaryJSONField.length() 的用法示例。

extract(*path)¶

提取给定路径的 JSON 数据。

请参阅 BinaryJSONField.extract() 的用法示例。

HStore ¶

Postgresql 的 hstore 扩展将任意键值对存储在单个列中。通过在初始化数据库时传递 register_hstore=True 来启用它

db = PostgresqlExtDatabase('my_app', register_hstore=True)

class Event(Model):
    data = HStoreField()
    class Meta:
        database = db

HStoreField 支持以下操作

存储和检索任意字典
按键或部分字典过滤
更新/添加一个或多个键到现有字典
从现有字典中删除一个或多个键
选择键、值，或打包键和值
检索键/值的切片
测试键是否存在
测试键是否具有非 NULL 值

示例

# Create a record with arbitrary attributes:
Event.create(data={
    'type': 'register',
    'ip': '1.2.3.4',
    'email': 'charles@example.com',
    'result': 'success',
    'referrer': 'google.com'})

Event.create(data={
    'type': 'login',
    'ip': '1.2.3.4',
    'email': 'charles@example.com',
    'result': 'success'})

# Lookup nested values in the data:
Event.select().where(Event.data['type'] == 'login')

# Filter by a key/value pair:
Event.select().where(Event.data.contains({'result': 'success'})

# Filter by key existence:
Event.select().where(Event.data.exists('referrer'))

# Atomic update - adds new keys, updates existing ones:
new_data = Event.data.update({
    'result': 'ok',
    'status': 'success'})
(Event
 .update(data=new_data)
 .where(Event.data['result'] == 'success')
 .execute())

# Atomic key deletion:
(Event
 .update(data=Event.data.delete('referrer'))
 .where(Event.data['referrer'] == 'google.com')
 .execute())

# Retrieve keys or values as a list:
for event in Event.select(Event.id, Event.data.keys().alias('k')):
    print(event.id, event.k)

# Prints:
# 1 ['ip', 'type', 'email', 'result', 'status']

# Retrieve a subset of data:
query = (Event
         .select(Event.id,
                 Event.data.slice('ip', 'email').alias('source'))
         .order_by(Event.data['ip']))
for event in query:
    print(event.id, event.source)

# Prints:
# 1 {'ip': '1.2.3.4', 'email': 'charles@example.com'}

HStoreField API¶

class HStoreField¶

默认情况下，HStoreField 将使用 *GiST* 索引。要禁用此功能，请使用 index=False 初始化字段。

__getitem__(key)¶

参数: key (str) – 获取给定键的值。

示例

Event.select().where(Event.data['type'] == 'login')

contains(value)¶

参数: value (dict, list, tuple or string key.) – 要搜索的值。

测试 HStore 数据是否包含给定的 dict（匹配键和值）、list/tuple（匹配键）或 str 键。

示例

# Contains key/value pairs:
Event.select().where(Event.data.contains({'result': 'success'}))

# Contains a list of keys:
Event.select().where(Event.data.contains(['result', 'status']))

# Contains a single key:
Event.select().where(Event.data.contains('result'))

contains_any(*keys)¶: 测试 HStore 是否包含给定的任何键。

exists(key)¶: 测试数据中是否存在该键。

defined(key)¶: 测试数据中该键是否为非 NULL。

update(__data=None, **data)¶

参数

__data (dict) – 指定更新为 dict。
data – 指定更新为关键字参数。

执行就地、原子更新。

# Atomic update - adds new keys, updates existing ones:
new_data = Event.data.update({
    'result': 'ok',
    'status': 'success'})
(Event
 .update(data=new_data)
 .where(Event.data['result'] == 'success')
 .execute())

delete(*keys)¶

参数: keys – 要从数据中删除的一个或多个键。

# Atomic key deletion:
(Event
 .update(data=Event.data.delete('referrer'))
 .where(Event.data['referrer'] == 'google.com')
 .execute())

slice(*keys)¶

参数: keys (str) – 要检索的键。

仅检索提供的键/值对

query = (Event
         .select(Event.id,
                 Event.data.slice('ip', 'email').alias('source'))
         .order_by(Event.data['ip']))
for event in query:
    print(event.id, event.source)

# 1 {'ip': '1.2.3.4', 'email': 'charles@example.com'}

keys()¶

将键作为列表返回。

query = Event.select(Event.data.keys().alias('keys'))
for event in query:
    print(event.keys)

# ['ip', 'type', 'email', 'result', 'status']

values()¶: 将值作为列表返回。

items()¶

将键值对作为二维列表返回。

query = Event.select(Event.data.items().alias('items'))
for event in query:
    print(event.items)

# [['ip', '1.2.3.4'],
#  ['type', 'register'],
#  ['email', 'charles@example.com'],
#  ['result', 'ok'],
#  ['status', 'success']]

数组 ¶

class ArrayField(field_class=IntegerField, field_kwargs=None, dimensions=1, convert_values=False)¶

存储给定字段类型的 Postgresql 数组。

参数

field_class – Field 的子类，例如 IntegerField。
field_kwargs (dict) – 初始化 field_class 的参数。
dimensions (int) – 数组的维度数。
convert_values (bool) – 将 field_class 值转换应用于检索到的数据。

默认情况下，ArrayField 将使用 GIN 索引。要禁用此功能，请使用 index=False 初始化字段。

示例

class Post(Model):
    tags = ArrayField(CharField)

Post.create(tags=['python', 'peewee', 'postgresql'])
Post.create(tags=['python', 'sqlite'])

# Get an item by index.
Post.select(Post.tags[0].alias('first_tag'))

# Get a slice:
Post.select(Post.tags[:2].alias('first_two'))

多维数组示例

class Outline(Model):
    points = ArrayField(IntegerField, dimensions=2)

Outline.create(points=[[1, 1], [1, 5], [5, 5], [5, 1]])

contains(*items)¶

过滤包含所有给定值的数组的行。

参数: items – 必须存在于给定数组字段中的一个或多个项。

Post.select().where(Post.tags.contains('postgresql', 'python'))

contains_any(*items)¶

过滤数组中包含给定值的任何一项的行。

参数: items – 要在给定数组字段中搜索的一个或多个项。

Post.select().where(Post.tags.contains('postgresql', 'python'))

Interval ¶

class IntervalField(**kwargs)¶

使用 Postgresql 的本机 INTERVAL 类型存储 Python datetime.timedelta 实例。

from datetime import timedelta

class Subscription(Model):
    duration = IntervalField()

Subscription.create(duration=timedelta(days=30))

(Subscription
 .select()
 .where(Subscription.duration > timedelta(days=10)))

DateTimeTZ 字段 ¶

class DateTimeTZField(**kwargs)¶

使用 Postgresql 的 TIMESTAMP WITH TIME ZONE 类型进行时区感知的 datetime 字段。

class Event(Model):
    timestamp = DateTimeTZField()

now = datetime.datetime.now().astimezone(datetime.timezone.utc)

Event.create(timestamp=now)

event = Event.get()
print(event.timestamp)
# 2026-01-02 03:04:05.012345+00:00

全文搜索 ¶

Postgresql 全文搜索使用 tsvector 和 tsquery 类型。Peewee 提供两种方法：简单的 Match() 函数（无需架构更改）和用于专用搜索列的 TSVectorField（性能更好）。

简单方法 - 无需架构更改

from playhouse.postgres_ext import Match

def search_posts(term):
    return Post.select().where(
        (Post.status == 'published') &
        Match(Post.body, term))

Match() 函数会自动将左侧操作数转换为 tsvector，将右侧操作数转换为 tsquery。为了获得更好的性能，请创建一个 GIN 索引

CREATE INDEX posts_fts ON post USING gin(to_tsvector('english', body));

专用列 - 性能更好

class Post(Model):
    body = TextField()
    search_content = TSVectorField()  # Automatically gets a GIN index.

# Store a post and populate the search vector:
Post.create(
    body=body_text,
    search_content=fn.to_tsvector(body_text))

# Search:
Post.select().where(Post.search_content.match('python postgresql'))

# Search using expressions:
terms = 'python & (sqlite | postgres)'
Post.select().where(Post.search_content.match(terms))

有关更多信息，请参阅 Postgres 全文搜索文档。

Match(field, query)¶: 生成一个全文搜索表达式，该表达式会自动将 field 转换为 tsvector，将 query 转换为 tsquery。

class TSVectorField¶

用于存储预计算的 tsvector 数据的字段类型。自动创建 GIN 索引（使用 index=False 禁用）。

写入时必须使用 fn.to_tsvector() 显式将数据转换为 tsvector。

示例

class Post(Model):
    body = TextField()
    search_content = TSVectorField()

Post.create(
    body=body_text,
    search_content=fn.to_tsvector(body_text))

(Post
 .select()
 .where(Post.search_content.match('python & (sqlite | postgres)')))

match(query, language=None, plain=False)¶

参数

query (str) – 全文搜索查询。
language (str) – 可选的语言名称。
plain (bool) – 使用 plain（简单）查询解析器，而不是默认解析器，后者支持 &、| 和 ! 运算符。

服务器端游标 ¶

对于大型结果集，服务器端（命名）游标会从服务器流式传输行，而不是将整个结果加载到内存中。行会在您迭代时从服务器透明地获取。

有关详细信息，请参阅您的驱动程序文档

要使用服务器端（或命名）游标，您必须使用 PostgresqlExtDatabase。

用 ServerSide() 包装任何 SELECT 查询

from playhouse.postgres_ext import ServerSide

# Must be in a transaction to use server-side cursors.
with db.atomic():

    # Create a normal SELECT query.
    large_query = PageView.select()

    # Then wrap in `ServerSide` and iterate.
    for page_view in ServerSide(large_query):
        # Do something interesting.
        pass

    # At this point server side resources are released.

有关更精细的控制或显式关闭游标

with db.atomic():
    large_query = PageView.select().order_by(PageView.id.desc())

    # Rows will be fetched 1000 at-a-time, but iteration is transparent.
    query = ServerSideQuery(query, array_size=1000)

    # Read 9500 rows then close server-side cursor.
    accum = []
    for i, obj in enumerate(query.iterator()):
        if i == 9500:
            break
        accum.append(obj)

    # Release server-side resource.
    query.close()

警告

服务器端游标仅在事务内有效。如果您使用的是 psycopg2（而不是 psycopg3），则游标声明为 WITH HOLD，并且必须完全耗尽或显式关闭才能释放服务器资源。

ServerSide(select_query)¶

参数: select_query – 一个 SelectQuery 实例。
返回类型：生成器

在事务中包装 select_query，并使用 iterator() 进行迭代（禁用行缓存）。

CockroachDB ¶

CockroachDB (CRDB) 与 Postgresql 的线协议兼容，并且得到了 Peewee 的良好支持。使用专用的 CockroachDatabase 类而不是 PostgresqlDatabase 类，以获得 CRDB 特定的处理。

from playhouse.cockroachdb import CockroachDatabase

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

如果您正在使用 Cockroach Cloud，您可能会发现使用连接字符串指定连接参数更方便

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

SSL 配置

db = CockroachDatabase(
    'my_app',
    user='root',
    host='10.8.0.1',
    sslmode='verify-full',
    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')

与 Postgresql 的主要区别¶

不支持嵌套事务。 CRDB 不支持保存点，因此在另一个 atomic() 块内调用 atomic() 会引发异常。改用 transaction()，它会忽略嵌套调用，仅在外层块退出时提交。
客户端重试。 CRDB 可能会因争用而中止事务。使用 run_transaction() 进行自动重试。

使用 CRDB 时可能有用的特殊字段类型

UUIDKeyField - 一个主键字段实现，使用 CRDB 的 UUID 类型，并带有默认随机生成的 UUID。
RowIDField - 一个主键字段实现，使用 CRDB 的 INT 类型，并带有默认 unique_rowid()。
JSONField - 与 Postgres 的 BinaryJSONField 相同，因为 CRDB 将所有 JSON 视为 JSONB。
ArrayField - 与 Postgres 扩展相同（但不支持多维数组）。

事务

# transaction() is safe to nest; the outer block manages the commit.
@db.transaction()
def create_user(username):
    return User.create(username=username)

with db.transaction():
    create_user('alice')   # Nested call is folded into outer transaction.
    create_user('bob')
# Transaction is committed here.

客户端重试

from playhouse.cockroachdb import CockroachDatabase

db = CockroachDatabase('my_app')

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

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

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

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

CRDB API¶

class CockroachDatabase(database, **kwargs)¶

用于 CockroachDB 的 PostgresqlDatabase 子类。

run_transaction(callback, max_attempts=None, system_time=None, priority=None)¶

参数

callback – 接受单个 db 参数的可调用对象。不得自行管理事务。可能会被多次调用。
max_attempts (int) – 重试限制。
system_time (datetime) – AS OF SYSTEM TIME 语句，相对于给定值执行。
priority (str) – 'low'、'normal' 或 'high'。

引发

ExceededMaxAttempts – 当超过 max_attempts 时。

在具有自动客户端重试的事务中执行 SQL。

用户提供的 callback

必须接受一个参数，即表示事务正在运行的连接的 db 实例。
不得尝试提交、回滚或以其他方式管理事务。
可能被多次调用。
理想情况下应只包含 SQL 操作。

此外，在调用此函数时，数据库不得有任何未完成的事务，因为 CRDB 不支持嵌套事务。尝试这样做将引发 NotImplementedError。

class PooledCockroachDatabase(database, **kwargs)¶: CockroachDatabase 的连接池变体。

run_transaction(db, callback, max_attempts=None, system_time=None, priority=None)¶

在具有自动客户端重试的事务中运行 SQL。有关详细信息，请参阅 CockroachDatabase.run_transaction()。

此函数等同于 CockroachDatabase 类中同名的方法。

CRDB 特定的字段类型

class UUIDKeyField: 使用 CRDB 的 gen_random_uuid() 自动填充的 UUID 主键。

class RowIDField: 使用 CRDB 的 unique_rowid() 自动填充的整数主键。

Postgresql¶

入门 ¶

PostgresqlExtDatabase ¶

JSON 支持 ¶

BinaryJSONField 和 JSONField¶

HStore ¶

HStoreField API¶

数组 ¶

Interval ¶

DateTimeTZ 字段 ¶

全文搜索 ¶

服务器端游标 ¶

CockroachDB ¶

与 Postgresql 的主要区别¶

CRDB API¶

peewee

导航

相关主题