写入数据¶

本文档涵盖 INSERT、UPDATE 和 DELETE 查询。读取数据请参见查询。

所有示例均使用来自模型和字段的规范化模式

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)
    is_published = BooleanField(default=True)

将要讨论的方法

查询	方法
INSERT	`Model.create()` `Model.insert()` `Model.insert_many()` `Model.insert_from()` `Model.replace()` (Postgres 不支持) `Model.replace_many()` (Postgres 不支持)
UPDATE	`Model.save()` `Model.update()`
DELETE	`Model.delete_instance()` `Model.delete()`

另请参阅

丰富的 SQL / Peewee 示例库

插入记录¶

创建单行¶

create() 会插入一行并返回已保存的模型实例

>>> charlie = User.create(username='charlie')
>>> charlie.id
1

这将向数据库 INSERT 一条新记录。主键将自动检索并存储在模型实例中。

或者，实例化模型并调用 save()。对新实例的第一次 save() 调用执行 INSERT

>>> user = User(username='huey')
>>> user.save()
1  # Returns number of rows modified.
>>> user.id
2

第一次保存后，模型实例会持有其主键。任何后续对 save() 的调用将执行 UPDATE

>>> user.username = 'Huey'
>>> user.save()
1  # Returns number of rows updated.

对于外键字段，请传递关联的模型实例或其原始主键值

Tweet.create(user=huey, content='Hello!')
Tweet.create(user=2, content='Also valid.')

要插入而不构造模型实例，请使用 insert()。它将返回新记录的主键

>>> User.insert(username='mickey').execute()
3

批量插入¶

应避免在循环中调用 Model.create() 或 Model.save()

data = [
    {'username': 'alice'},
    {'username': 'bob'},
    {'username': 'carol'},
]

for data_dict in data:
    User.create(**data_dict)

上面的方法速度很慢

未将循环包装在事务中。 结果是每次 create() 都在自己的事务中进行。
Python 解释器 在起作用，并且每次 Insert 都必须生成并解析为 SQL。
大量数据 (以原始 SQL 字节数计) 可能被发送到数据库进行解析。
检索最后插入的 ID，这可能不是必需的。

通过使用 atomic() 将其包装在事务中，可以显著提高速度

with db.atomic():
    for data_dict in data:
        User.create(**data_dict)

插入多条记录最快的方法是 insert_many()。它接受字典或元组列表，并发出单个多行 INSERT

data = [
    {'username': 'alice'},
    {'username': 'bob'},
    {'username': 'carol'},
]
User.insert_many(data).execute()

# Tuples require an explicit field list:
data = [('alice',), ('bob',), ('carol',)]
User.insert_many(data, fields=[User.username]).execute()

可以选择将批量插入包装在事务中

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

INSERT 查询支持与 Postgresql 和 SQLite 一起使用 returning() 来获取已插入的记录

query = (User
         .insert_many([{'username': 'alice'}, {'username': 'bob'}])
         .returning(User))
for user in query:
    print(f'Added {user.username} with id = {user.id}')

批量处理大数据集¶

根据数据源中的行数，您可能需要将其分解成块。特别是 SQLite 可能有每查询变量限制为 32766 (因此批处理大小为 32766 // 行长度)。

您可以编写一个循环将数据分块。强烈建议您使用事务

from peewee import chunked

with db.atomic():
    for batch in chunked(data, 100):
        User.insert_many(batch).execute()

chunked() 可用于任何可迭代对象，包括生成器。

批量创建模型实例¶

bulk_create() 接受一个未保存的模型实例列表并高效地插入它们。传递 batch_size 以避免命中数据库限制

users = [User(username=f'user_{i}') for i in range(1000)]
with db.atomic():
    User.bulk_create(users, batch_size=100)

如果您使用的是 Postgresql (它支持 RETURNING 子句)，那么之前未保存的模型实例将自动填充其新的主键值。其他后端则不会。

从另一个表加载¶

insert_from() 生成一个 INSERT INTO ... SELECT 查询，将行从一个表复制到另一个表，而无需通过 Python 进行往返

res = (TweetArchive
       .insert_from(
           Tweet.select(Tweet.user, Tweet.message),
           fields=[TweetArchive.user, TweetArchive.message])
       .execute())

上述查询等同于以下 SQL

INSERT INTO "tweet_archive" ("user_id", "message")
SELECT "user_id", "message" FROM "tweet";

更新记录¶

更新模型实例¶

修改已获取实例的属性并调用 save() 来持久化更改

charlie = User.get(User.username == 'charlie')
charlie.username = 'charlie_admin'
charlie.save()   # Issues UPDATE WHERE id = charlie.id

默认情况下，save() 会重新保存所有字段。要仅发出已更改的字段，请在模型的 Meta 中设置 only_save_dirty = True，或者仅传递您要更新的字段

charlie.username = 'charlie_v2'
charlie.save(only=[User.username])

如果模型实例没有主键，第一次调用 save() 将执行 INSERT 查询。

一旦模型实例有了主键，后续调用 save() 将导致 *UPDATE*。

更新多行¶

update() 发出单个 UPDATE，影响 WHERE 子句匹配的所有行

# Publish all unpublished tweets older than one week.
one_week_ago = datetime.datetime.now() - datetime.timedelta(days=7)

nrows = (Tweet
         .update(is_published=True)
         .where(
             (Tweet.is_published == False) &
             (Tweet.timestamp < one_week_ago))
         .execute())

返回值是受影响的行数。

UPDATE 查询支持与 Postgresql 和 SQLite 一起使用 returning() 来获取已更新的记录

query = (User
         .update(spam=True)
         .where(User.username.contains('billing'))
         .returning(User))
for user in query:
    print(f'Marked {user.username} as spam')

因为 UPDATE 查询不支持 JOIN，所以我们可以使用子查询来根据相关表中的值更新行。例如，取消发布用户名中包含 'billing' 的用户的推文

spammers = User.select().where(User.username.contains('billing'))

(Tweet
 .update(is_published=False)
 .where(Tweet.user.in_(spammers))
 .execute())

原子更新¶

在 update() 中使用列表达式来修改值，而无需读-改-写周期。以原子方式执行更新可防止竞态条件

# WRONG: reads each row into Python, increments, then saves.
# Vulnerable to race conditions; slow on many rows.
for stat in Stat.select().where(Stat.url == url):
    stat.counter += 1
    stat.save()

# CORRECT: single UPDATE statement, atomic at the database level.
Stat.update(counter=Stat.counter + 1).where(Stat.url == url).execute()

任何 SQL 表达式在右侧都是有效的

# Give every employee a 10% salary bonus added to their existing bonus.
Employee.update(bonus=Employee.bonus + (Employee.salary * 0.10)).execute()

# Denormalize a count column from a subquery.
tweet_count = (Tweet
               .select(fn.COUNT(Tweet.id))
               .where(Tweet.user == User.id))
User.update(num_tweets=tweet_count).execute()

批量更新模型实例¶

当您有一系列已修改的模型实例并希望一次查询更新它们的所有特定字段时，请使用 bulk_update()

u1, u2, u3 = User.select().limit(3)
u1.username = 'u1-new'
u2.username = 'u2-new'
u3.username = 'u3-new'

User.bulk_update([u1, u2, u3], fields=[User.username])

这会发出一个使用 SQL CASE 表达式的 UPDATE。对于大型列表，请指定 batch_size 并将其包装在事务中

with db.atomic():
    User.bulk_update(users, fields=[User.username], batch_size=50)

bulk_update 可能比直接 UPDATE 查询慢，因为生成的 CASE 表达式会成比例增长。对于可以表示为单个 WHERE 子句的更新，直接的 update() 方法更快。

Upsert¶

Upsert (INSERT 或 UPDATE) 会插入一条新记录，如果违反唯一约束，则会更新现有记录。

Peewee 提供了两种互补的方法。

`on_conflict_replace` - SQLite 和 MySQL¶

SQLite 和 MySQL 支持 `REPLACE` 查询，该查询在发生冲突时会替换记录

class User(BaseModel):
    username = TextField(unique=True)
    last_login = DateTimeField(null=True)

# Insert, or replace the entire existing row.
User.replace(username='huey', last_login=datetime.datetime.now()).execute()

# Equivalent using insert():
(User
 .insert(username='huey', last_login=datetime.datetime.now())
 .on_conflict_replace()
 .execute())

replace 会删除并重新插入，这会更改主键。当主键必须保留，或仅应更新某些列时，请使用 `on_conflict` (如下)。

`on_conflict` - 所有后端¶

a on_conflict() 方法功能强大得多。

class User(BaseModel):
    username = TextField(unique=True)
    last_login = DateTimeField(null=True)
    login_count = IntegerField(default=0)

now = datetime.datetime.now()

(User
 .insert(username='huey', last_login=now, login_count=1)
 .on_conflict(
     # Postgresql and SQLite require identifying the conflicting constraint.
     # MySQL does not need this.
     conflict_target=[User.username],

     # Columns whose values should come from the incoming row:
     preserve=[User.last_login],

     # Columns to update using an expression:
     update={User.login_count: User.login_count + 1})
 .execute())

反复调用此查询将原子地递增 login_count 并在每次调用时更新 last_login，而不会创建重复的记录。

a EXCLUDED 命名空间引用如果约束未触发将会插入的值。这允许条件更新

class KV(BaseModel):
    key = TextField(unique=True)
    value = IntegerField()

KV.create(key='k1', value=1)

# Demonstrate usage of EXCLUDED.
# Here we will attempt to insert a new value for a given key. If that
# key already exists, then we will update its value with the *sum* of its
# original value and the value we attempted to insert -- provided that
# the new value is larger than the original value.
query = (KV.insert(key='k1', value=10)
         .on_conflict(conflict_target=[KV.key],
                      update={KV.value: KV.value + EXCLUDED.value},
                      where=(EXCLUDED.value > KV.value)))

# Executing the above query will result in the following data being
# present in the "kv" table:
# (key='k1', value=11)
query.execute()

# If we attempted to execute the query *again*, then nothing would be
# updated, as the new value (10) is now less than the value in the
# original row (11).

使用 ON CONFLICT 时有几个重要概念需要理解

conflict_target=: 哪个(些)列具有 UNIQUE 约束。对于用户表，这可能是用户的电子邮件 (仅限 SQLite 和 Postgresql)。
preserve=: 如果发生冲突，此参数用于指示我们希望更新的 **新** 数据中的哪些值。
update=: 如果发生冲突，这是一个应用于预先存在的记录的数据映射。
EXCLUDED: 这个“魔术”命名空间允许您引用如果约束未失败将会插入的新数据。

完整示例

class User(Model):
    email = CharField(unique=True)  # Unique identifier for user.
    last_login = DateTimeField()
    login_count = IntegerField(default=0)
    ip_log = TextField(default='')


# Demonstrates the above 4 concepts.
def login(email, ip):
    rowid = (User
             .insert({User.email: email,
                      User.last_login: datetime.now(),
                      User.login_count: 1,
                      User.ip_log: ip})
             .on_conflict(
                 # If the INSERT fails due to a constraint violation on the
                 # user email, then perform an UPDATE instead.
                 conflict_target=[User.email],

                 # Set the "last_login" to the value we would have inserted
                 # (our call to datetime.now()).
                 preserve=[User.last_login],

                 # Increment the user's login count and prepend the new IP
                 # to the user's ip history.
                 update={User.login_count: User.login_count + 1,
                         User.ip_log: fn.CONCAT(EXCLUDED.ip_log, ',', User.ip_log)})
             .execute())

    return rowid

# This will insert the initial row, returning the new row id (1).
print(login('test@example.com', '127.1'))

# Because test@example.com exists, this will trigger the UPSERT. The row id
# from above is returned again (1).
print(login('test@example.com', '127.2'))

u = User.get()
print(u.login_count, u.ip_log)

# Prints "2 127.2,127.1"

另请参阅

Insert.on_conflict() 和 OnConflict。

`on_conflict_ignore`¶

插入记录，如果违反约束则静默无操作

# Insert if username does not exist; ignore if it does.
User.insert(username='huey').on_conflict_ignore().execute()

支持 SQLite、MySQL 和 Postgresql。

删除记录¶

使用 delete_instance() 删除单个已获取的实例

tweet = Tweet.get_by_id(42)
tweet.delete_instance()   # Returns number of rows deleted.

要删除一个记录及其所有依赖行 (在其他表中通过外键引用它的行)，请传递 recursive=True

# Deletes the user and all their tweets, favorites, etc.
with db.atomic():
    user.delete_instance(recursive=True)

recursive=True 通过查询依赖行并先删除它们来工作——它不依赖于 ON DELETE CASCADE。对于大型相关数据图，这可能会很慢。请务必将调用包装在事务中，并考虑在数据库级别使用外键的级联约束。

要删除任意一组记录而不获取它们

# Delete all unpublished tweets older than 30 days.
cutoff = datetime.datetime.now() - datetime.timedelta(days=30)
nrows = (Tweet
         .delete()
         .where(
             (Tweet.is_published == False) &
             (Tweet.timestamp < cutoff))
         .execute())

DELETE 查询支持与 Postgresql 和 SQLite 一起使用 returning() 来获取已删除的记录

query = (User
         .delete()
         .where(User.username.contains('billing'))
         .returning(User))
for user in query:
    print(f'Deleted: {user.username}')

因为 DELETE 查询不支持 JOIN，所以我们可以使用子查询来根据相关表中的值删除记录。例如，删除用户名中包含 'billing' 的用户的推文

spammers = User.select().where(User.username.contains('billing'))

(Tweet
 .delete()
 .where(Tweet.user.in_(spammers))
 .execute())

另请参阅

Returning Clause¶

PostgresqlDatabase 和 SqliteDatabase (3.35.0+) 支持 `UPDATE`、`INSERT` 和 `DELETE` 查询上的 RETURNING 子句。指定 `RETURNING` 子句允许您迭代查询访问的行。

默认情况下，不同查询执行时的返回值是

INSERT - 新插入记录的自动增量主键值。当不使用自动增量主键时，Postgres 会返回新记录的主键，但 SQLite 和 MySQL 不会。
UPDATE - 修改的行数
DELETE - 删除的行数

当使用 returning 子句时，执行查询的返回值将是一个可迭代的游标对象，提供对查询插入、更新或删除的数据的访问。

例如，假设您有一个 Update 查询，该查询将停用所有注册已过期的用户帐户。在停用它们之后，您想向每个用户发送一封电子邮件，告知他们帐户已被停用。您可以将此操作与一个 `SELECT` 查询和一个 `UPDATE` 查询合并，使用一个带 `RETURNING` 子句的 `UPDATE` 查询来完成。

query = (User
         .update(is_active=False)
         .where(User.registration_expired == True)
         .returning(User))

# Send an email to every user that was deactivated.
for deactivate_user in query.execute():
    send_deactivation_email(deactivated_user.email)

query = (User
         .delete()
         .where(User.is_spam == True)
         .returning(User.id))
for user in query.execute():
    print(f'Deleted spam user id: {user.id}')

`RETURNING` 子句可用于

作为另一个例子，让我们添加一个用户并将他们的创建日期设置为服务器生成的当前时间戳。我们将使用一个查询来创建并检索新用户的 ID、电子邮件和创建时间戳。

query = (User
         .insert(email='foo@bar.com', created=fn.now())
         .returning(User))  # Shorthand for all columns on User.

# When using RETURNING, execute() returns a cursor.
cursor = query.execute()

# Get the user object we just inserted and log the data:
user = cursor[0]
logger.info('Created user %s (id=%s) at %s', user.email, user.id, user.created)

默认情况下，游标将返回 Model 实例，但您可以指定不同的行类型

data = [{'name': 'charlie'}, {'name': 'huey'}, {'name': 'mickey'}]
query = (User
         .insert_many(data)
         .returning(User.id, User.username)
         .dicts())

for new_user in query.execute():
    print('Added user "%s", id=%s' % (new_user['username'], new_user['id']))

与 Select 查询一样，您可以指定各种结果行类型。

写入数据¶

插入记录¶

创建单行¶

批量插入¶

批量处理大数据集¶

批量创建模型实例¶

从另一个表加载¶

更新记录¶

更新模型实例¶

更新多行¶

原子更新¶

批量更新模型实例¶

Upsert¶

`on_conflict_replace` - SQLite 和 MySQL¶

`on_conflict` - 所有后端¶

`on_conflict_ignore`¶

删除记录¶

Returning Clause¶

peewee

导航

相关主题

写入数据¶

插入记录¶

创建单行¶

批量插入¶

批量处理大数据集¶

批量创建模型实例¶

从另一个表加载¶

更新记录¶

更新模型实例¶

更新多行¶

原子更新¶

批量更新模型实例¶

Upsert¶

on_conflict_replace - SQLite 和 MySQL¶

on_conflict - 所有后端¶

on_conflict_ignore¶

删除记录¶

Returning Clause¶

`on_conflict_replace` - SQLite 和 MySQL¶

`on_conflict` - 所有后端¶

`on_conflict_ignore`¶