Recipes

收集了常见实际问题的模式。每个 recipes 都假定您熟悉 Querying、Writing Data 和 Relationships and Joins。

所有示例都使用以下模型

import datetime
from peewee import *

db = SqliteDatabase(':memory:')

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

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

class Tweet(BaseModel):
    user = ForeignKeyField(User, backref='tweets')
    content = TextField()
    timestamp = DateTimeField(default=datetime.datetime.now)
    created_date = DateTimeField(default=datetime.datetime.now)

Optimistic Locking

Optimistic locking 通过在每行中记录一个版本号来避免在读-改-写周期内持有数据库锁。在写操作时，更新会以版本号未改变为条件。如果在此期间另一进程修改了该行，更新将不匹配任何行，并且冲突会在应用程序代码中检测到。

这是 SELECT FOR UPDATE (Postgresql) 或 BEGIN IMMEDIATE (SQLite) 的一种更轻量级的替代方案，适用于预期锁竞争较低的情况。

一个可重用的基类

class ConflictDetectedException(Exception):
    pass

class BaseVersionedModel(BaseModel):
    version = IntegerField(default=1, index=True)

    def save_optimistic(self):
        if not self.id:
            # This is a new record, so the default logic is to perform an
            # INSERT. Ideally your model would also have a unique
            # constraint that made it impossible for two INSERTs to happen
            # at the same time.
            return self.save()

        # Update any data that has changed and bump the version counter.
        field_data = dict(self.__data__)
        current_version = field_data.pop('version', 1)
        self._populate_unsaved_relations(field_data)
        field_data = self._prune_fields(field_data, self.dirty_fields)
        if not field_data:
            raise ValueError('No changes have been made.')

        ModelClass = type(self)
        field_data['version'] = ModelClass.version + 1  # Atomic increment.

        updated = (ModelClass
                   .update(**field_data)
                   .where(
                       (ModelClass.version == current_version) &
                       (ModelClass.id == self.id))
                   .execute())
        if updated == 0:
            # No rows were updated, indicating another process has saved
            # a new version.
            raise ConflictDetectedException()
        else:
            # Increment local version to match what is now in the db.
            self.version += 1
            return True

用法

class UserProfile(BaseVersionedModel):
    username = TextField(unique=True)
    bio = TextField(default='')

>>> u = UserProfile(username='charlie')
>>> u.save_optimistic()
True

>>> u.bio = 'Python developer'
>>> u.save_optimistic()
True
>>> u.version
2

# Simulate a concurrent modification:
>>> u2 = UserProfile.get(UserProfile.username == 'charlie')
>>> u2.bio = 'Changed by another process'
>>> u2.save_optimistic()
True

# The original instance's version is now stale:
>>> u.bio = 'My update'
>>> u.save_optimistic()
ConflictDetectedException

Get-or-Create Safely

get_or_create() 很方便，但在行尚不存在时，SELECT 和 INSERT 之间存在一个小的竞争窗口。两个并发进程都可能在 SELECT 中失败，并都尝试 INSERT，导致一个因 IntegrityError 而失败。

安全的模式是先尝试 INSERT，然后在遇到 IntegrityError 时回退到 GET。

def get_or_create_user(username):
    try:
        with db.atomic():
            return User.create(username=username), True
    except IntegrityError:
        return User.get(User.username == username), False

user, created = get_or_create_user('charlie')

db.atomic() 包装器很重要：它确保在 IntegrityError 发生时的回滚仅影响此操作，而不影响任何周围的事务。

Top Item Per Group

这些示例查找每个用户最常出现的单条推文。有关更通用的 N 条记录问题，请参阅下面的 Top N Per Group。

最可移植的方法是在非相关子查询中使用 MAX() 聚合，然后通过用户和时间戳连接回推文表。

# When referencing a table multiple times, we'll call Model.alias() to create
# a secondary reference to the table.
TweetAlias = Tweet.alias()

# Create a subquery that will calculate the maximum Tweet created_date for
# each user.
subquery = (TweetAlias
            .select(
                TweetAlias.user,
                fn.MAX(TweetAlias.created_date).alias('max_ts'))
            .group_by(TweetAlias.user)
            .alias('tweet_max'))

# Query for tweets and join using the subquery to match the tweet's user
# and created_date.
query = (Tweet
         .select(Tweet, User)
         .join(User)
         .switch(Tweet)
         .join(subquery, on=(
             (Tweet.created_date == subquery.c.max_ts) &
             (Tweet.user == subquery.c.user_id))))

SQLite 和 MySQL 允许使用一种更简洁的形式，按选定列的子集进行分组。

query = (Tweet
         .select(Tweet, User)
         .join(User)
         .group_by(Tweet.user)
         .having(Tweet.created_date == fn.MAX(Tweet.created_date)))

Postgresql 需要上述标准的子查询形式。

Top N Per Group

这些示例描述了几种相对高效地查询每个组的前 N 条记录的方法。有关各种技术的详细讨论，请参阅我的博客文章 Querying the top N objects per group with Peewee ORM。

Window functions

RANK() 窗口函数是最简洁的解决方案。按时间戳（最新的在前）为每个用户的推文排名，然后将外部查询过滤到前 N 名。

TweetAlias = Tweet.alias()

ranked = (TweetAlias
          .select(
              TweetAlias.content,
              User.username,
              fn.RANK().over(
                  partition_by=[TweetAlias.user],
                  order_by=[TweetAlias.created_date.desc()]
              ).alias('rnk'))
          .join(User, on=(TweetAlias.user == User.id))
          .alias('subq'))

query = (Tweet
         .select(ranked.c.content, ranked.c.username)
         .from_(ranked)
         .where(ranked.c.rnk <= 3))

Postgresql - lateral joins

LATERAL 连接会对驱动表的每一行执行一个相关的子查询。对于每个用户，它会选择最近的三条推文。

所需的 SQL 是

SELECT * FROM
  (SELECT id, username FROM user) AS uq
   LEFT JOIN LATERAL
  (SELECT message, created_date
   FROM tweet
   WHERE (user_id = uq.id)
   ORDER BY created_date DESC LIMIT 3)
  AS pq ON true

使用 peewee 实现这一点非常直接

subq = (Tweet
        .select(Tweet.message, Tweet.created_date)
        .where(Tweet.user == User.id)
        .order_by(Tweet.created_date.desc())
        .limit(3))

query = (User
         .select(User, subq.c.content, subq.c.created_date)
         .join(subq, JOIN.LEFT_LATERAL)
         .order_by(User.username, subq.c.created_date.desc()))

# We queried from the "perspective" of user, so the rows are User instances
# with the addition of a "content" and "created_date" attribute for each of
# the (up-to) 3 most-recent tweets for each user.
for row in query:
    print(row.username, row.content, row.created_date)

为了从 Tweet 模型“视角”实现一个等效的查询，我们可以改写为

# subq is the same as the above example.
subq = (Tweet
        .select(Tweet.message, Tweet.created_date)
        .where(Tweet.user == User.id)
        .order_by(Tweet.created_date.desc())
        .limit(3))

query = (Tweet
         .select(User.username, subq.c.content, subq.c.created_date)
         .from_(User)
         .join(subq, JOIN.LEFT_LATERAL)
         .order_by(User.username, subq.c.created_date.desc()))

# Each row is a "tweet" instance with an additional "username" attribute.
# This will print the (up-to) 3 most-recent tweets from each user.
for tweet in query:
    print(tweet.username, tweet.content, tweet.created_date)

Correlated subquery count

也可以使用一个相关的子查询来计算比当前行更新的推文数量。少于 N 条新推文的行属于前 N 条。

TweetAlias = Tweet.alias()

# Create a correlated subquery that calculates the number of
# tweets with a higher (newer) timestamp than the tweet we're
# looking at in the outer query.
subquery = (TweetAlias
            .select(fn.COUNT(TweetAlias.id))
            .where(
                (TweetAlias.created_date >= Tweet.created_date) &
                (TweetAlias.user == Tweet.user)))

# Wrap the subquery and filter on the count.
query = (Tweet
         .select(Tweet, User)
         .join(User)
         .where(subquery <= 3))

SQLite and MySQL - self-join

另一种方法：在 HAVING 子句中进行自连接并计数新推文。

TweetAlias = Tweet.alias()

query = (Tweet
         .select(Tweet.id, Tweet.content, Tweet.user, User.username)
         .join_from(Tweet, User)
         .join_from(Tweet, TweetAlias, on=(
             (TweetAlias.user == Tweet.user) &
             (TweetAlias.created_date >= Tweet.created_date)))
         .group_by(Tweet.id, Tweet.content, Tweet.user, User.username)
         .having(fn.COUNT(Tweet.id) <= 3))

最后一个示例在相关的子查询中使用 LIMIT 子句。

TweetAlias = Tweet.alias()

# The subquery here will calculate, for the user who created the
# tweet in the outer loop, the three newest tweets. The expression
# will evaluate to `True` if the outer-loop tweet is in the set of
# tweets represented by the inner query.
query = (Tweet
         .select(Tweet, User)
         .join(User)
         .where(Tweet.id << (
             TweetAlias
             .select(TweetAlias.id)
             .where(TweetAlias.user == Tweet.user)
             .order_by(TweetAlias.created_date.desc())
             .limit(3))))

有关这些方法的详细基准比较，请参阅博客文章 Querying the top N objects per group with Peewee ORM。

Bulk-Loading with Explicit Primary Keys

当从已分配主键的外部源加载关系数据时，请使用包含 id 字段的 insert_many()。这避免了早期 peewee 版本中常见的 auto_increment 解决方法。

data = [(1, 'alice'), (2, 'bob'), (3, 'carol')]
fields = [User.id, User.username]

with db.atomic():
    User.insert_many(data, fields=fields).execute()

因为 insert_many 从不读取行，所以 INSERT 和 UPDATE 路径之间没有混淆。

Custom SQLite Functions

SQLite 可以用 Python 函数进行扩展，这些函数可以从 SQL 调用。这对于 SQLite 不原生支持的操作很有用。

使用 @db.func() 装饰器注册函数后，连接打开后即可立即使用该函数。

from hashlib import sha256
import os

db = SqliteDatabase('my_app.db')

def _hash_password(salt, password):
    return sha256((salt + password).encode()).hexdigest()

@db.func()
def make_password(raw_password):
    salt = os.urandom(8).hex()
    return salt + '$' + _hash_password(salt, raw_password)

@db.func()
def check_password(raw_password, stored):
    salt, hsh = stored.split('$', 1)
    return hsh == _hash_password(salt, raw_password)

存储哈希密码

User.insert(username='charlie',
            password=fn.make_password('s3cr3t')).execute()

登录时验证密码

def login(username, raw_password):
    try:
        return (User
                .select()
                .where(
                    (User.username == username) &
                    (fn.check_password(raw_password, User.password) == True))
                .get())
    except User.DoesNotExist:
        return None

另请参阅

SqliteDatabase.func()、SqliteDatabase.aggregate()、SqliteDatabase.window_function()。

Date Arithmetic Across Databases

每个数据库实现日期算术的方式都不同。本节展示了如何为每个后端表示“计划任务的下一次发生”——定义为 last_run + interval_seconds。

Schema

class Schedule(BaseModel):
    interval = IntegerField()   # Repeat every N seconds.

class Task(BaseModel):
    schedule = ForeignKeyField(Schedule, backref='tasks')
    command  = TextField()
    last_run = DateTimeField()

我们想要：任务满足 now >= last_run + interval。

我们期望的代码如下

next_occurrence = something  # ??? how do we define this ???

# We can express the current time as a Python datetime value, or we could
# alternatively use the appropriate SQL function/name.
now = Value(datetime.datetime.now())  # Or SQL('current_timestamp'), e.g.

query = (Task
         .select(Task, Schedule)
         .join(Schedule)
         .where(now >= next_occurrence))

Postgresql - 乘以类型化的间隔

one_second = SQL("INTERVAL '1 second'")
next_run = Task.last_run + (Schedule.interval * one_second)

now = Value(datetime.datetime.now())
tasks_due = (Task
             .select(Task, Schedule)
             .join(Schedule)
             .where(now >= next_run))

MySQL - 使用 DATE_ADD 和动态 INTERVAL 表达式

from peewee import NodeList

interval = NodeList((SQL('INTERVAL'), Schedule.interval, SQL('SECOND')))
next_run = fn.DATE_ADD(Task.last_run, interval)

now = Value(datetime.datetime.now())
tasks_due = (Task
             .select(Task, Schedule)
             .join(Schedule)
             .where(now >= next_run))

SQLite - 转换为 Unix 时间戳，添加秒数，再转换回来

next_ts = fn.strftime('%s', Task.last_run) + Schedule.interval
next_run = fn.datetime(next_ts, 'unixepoch')

now = Value(datetime.datetime.now())
tasks_due = (Task
             .select(Task, Schedule)
             .join(Schedule)
             .where(now >= next_run))