peewee orm：使用子查询进行批量插入，但基于 python-side-data

Question

peewee 允许通过insert_many() 和insert_from() 进行批量插入，但是insert_many() 允许插入数据列表，但不允许从数据库的其他部分计算数据。 insert_from() 允许从数据库的其他部分计算数据，但不允许从 python 发送任何数据。

示例：

假设模型结构如下：

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

class Person(BaseModel):
    name = CharField(max_length=100, unique=True)

class StatusUpdate(BaseModel):
    person = ForeignKeyField(Person, related_name='statuses')
    status = TextField()
    timestamp = DateTimeField(constraints=[SQL('DEFAULT CURRENT_TIMESTAMP')], index=True)

还有一些初始数据：

Person.insert_many(rows=[{'name': 'Frank'}, {'name': 'Joe'}, {'name': 'Arnold'}]).execute()
print ('Person.select().count():',Person.select().count())

输出：

Person.select().count(): 3

假设我们想添加一堆新的状态更新，就像这个列表中的那些：

new_status_updates = [ ('Frank', 'wat')
                     , ('Frank', 'nooo')
                     , ('Joe', 'noooo')
                     , ('Arnold', 'nooooo')]

我们可能会尝试像这样使用insert_many()：

StatusUpdate.insert_many( rows=[{'person': 'Frank', 'status': 'wat'}
                              , {'person': 'Frank', 'status': 'nooo'}
                              , {'person': 'Joe', 'status': 'noooo'}
                              , {'person': 'Arnold', 'status': 'nooooo'}]).execute()

但这会失败：person 字段需要 Person 模型或 Person.id，我们必须进行额外查询才能从名称中检索它们。

我们可以通过insert_from() 来避免这种情况，允许我们进行子查询，但insert_from() 无法处理我们的列表或字典。怎么办？

Answer 1

一个想法是使用 SQL VALUES 子句作为 SELECT 语句的一部分。

如果您熟悉 SQL，您可能以前见过 VALUES 子句，它通常用作 INSERT 语句的一部分，如下所示：

INSERT INTO statusupdate (person_id,status)
VALUES (1, 'my status'), (1, 'another status'), (2, 'his status');

这告诉数据库将三行 - AKA 元组 - 插入到表 statusupdate 中。

另一种插入内容的方法是：

INSERT INTO statusupdate (person_id,status)
SELECT ..., ... FROM <elsewhere or subquery>;

这相当于 peewee 提供的insert_from() 功能。

但是您可以做另一件不太常见的事情：您可以在 any select 中使用VALUES 子句来提供文字值。示例：

SELECT *
FROM (VALUES (1,2,3), (4,5,6)) as my_literal_values;

这将返回一个包含两个行/元组的结果集，每个具有 3 个值。

因此，如果您可以将“批量”插入转换为 SELECT/FROM/VALUES 语句，那么您就可以进行任何您需要做的转换（即将 Person.name 值转换为相应的 Person.id 值），然后将其组合起来使用 peewee 'insert_from()` 功能。

那么让我们看看这会是什么样子。

首先让我们开始构建VALUES 子句本身。我们想要正确转义的值，所以我们现在将使用问号而不是值，稍后将实际值放入。

#this is gonna look like '(?,?), (?,?), (?,?)'
# or '(%s,%s), (%s,%s), (%s,%s)' depending on the database type
values_question_marks = ','.join(['(%s, %s)' % (db.interpolation,db.interpolation)]*len(new_status_updates))

下一步是构造 values 子句。这是我们的第一次尝试：

--the %s here will be replaced by the question marks of the clause
--in postgres, you must have a name for every item in `FROM`
SELECT * FROM (VALUES %s) someanonymousname

好的，现在我们的结果集如下所示：

name | status
-----|-------
...  | ...

除了！没有列名。这很快就会让我们有点心痛，所以我们必须想办法给结果集正确的列名。

postgres 的方式是改变AS 子句：

SELECT * FROM (VALUES %s) someanonymousname(name,status)

sqlite3 不支持 (grr)。

所以我们变成了一个杂物。幸运的是 stackoverflow 提供了：Is it possible to select sql server data using column ordinal position，我们可以这样构造：

SELECT NULL as name, NULL as status WHERE 1=0
UNION ALL
SELECT * FROM (VALUES %s) someanonymousname

首先使用正确的列名创建一个空结果集，然后将VALUES 子句中的结果集连接到它。这将产生一个具有正确列名的结果集，可以在 sqlite3 和 postgres 中工作。

现在把这个带回peewee：

values_query = """
(
    --a trick to make an empty query result with two named columns, to more portably name the resulting
    --VALUES clause columns (grr sqlite)
    SELECT NULL as name, NULL as status WHERE 1=0
    UNION ALL
    SELECT * FROM (VALUES %s) someanonymousname
)
"""

values_query %= (values_question_marks,)

#unroll the parameters into one large list
#this is gonna look like ['Frank', 'wat', 'Frank', 'nooo', 'Joe', 'noooo' ...]
values_query_params = [value for values in new_status_updates for value in values]

#turn it into peewee SQL
values_query = SQL(values_query,*values_query_params)
data_query = (Person
                .select(Person.id, SQL('values_list.status').alias('status'))
                .from_(Person,values_query.alias('values_list'))
                .where(SQL('values_list.name') == Person.name))


insert_query = StatusUpdate.insert_from([StatusUpdate.person, StatusUpdate.status], data_query)

print (insert_query)
insert_query.execute()
print ('StatusUpdate.select().count():',StatusUpdate.select().count())

输出：

StatusUpdate.select().count(): 4