peewee 检查自动创建的 id 不在子查询的结果中
peewee check automatically created id not in result of subquery
我有下一个数据结构:
from enum import IntEnum, unique
from pathlib import Path
from datetime import datetime
from peewee import *
@unique
class Status(IntEnum):
CREATED = 0
FAIL = -1
SUCCESS = 1
db_path = Path(__file__).parent / "test.sqlite"
database = SqliteDatabase(db_path)
class BaseModel(Model):
class Meta:
database = database
class Unit(BaseModel):
name = TextField(unique=True)
some_field = TextField(null=True)
created_at = DateTimeField(default=datetime.now)
class Campaign(BaseModel):
id_ = AutoField()
created_at = DateTimeField(default=datetime.now)
class Task(BaseModel):
id_ = AutoField()
status = IntegerField(default=Status.CREATED)
unit = ForeignKeyField(Unit, backref="tasks")
campaign = ForeignKeyField(Campaign, backref="tasks")
下一个代码创建单位、活动和任务:
def fill_units(count):
units = []
with database.atomic():
for i in range(count):
units.append(Unit.create(name=f"unit{i}"))
return units
def init_campaign(count):
units = Unit.select().limit(count)
with database.atomic():
campaign = Campaign.create()
for unit in units:
Task.create(unit=unit, campaign=campaign)
return campaign
当我尝试向现有 广告系列 添加更多 单位 时出现问题。我需要 select 单位,在这个 活动 中还没有使用过。在 SQL 中,我可以使用下一个查询来做到这一点:
SELECT * FROM unit WHERE id NOT IN (SELECT unit_id FROM task WHERE campaign_id = 1) LIMIT 10
但是如何使用 peewee 做到这一点?
我找到的唯一方法是:
def get_new_units_for_campaign(campaign, count):
unit_names = [task.unit.name for task in campaign.tasks]
units = Unit.select().where(Unit.name.not_in(unit_names)).limit(count)
return units
它以某种方式起作用,但我 100% 确定这是实现它的最愚蠢的方法。你能告诉我实现这个的正确方法吗?
我终于找到了这个:
Unit.select().where(Unit.id.not_in(campaign.tasks.select(Task.unit))).limit(10)
产生
SELECT "t1"."id", "t1"."name", "t1"."some_field", "t1"."created_at" FROM "unit" AS "t1" WHERE ("t1"."id" NOT IN (SELECT "t2"."unit_id" FROM "task" AS "t2" WHERE ("t2"."campaign_id" = 1))) LIMIT 10
与我在问题中提供的 SQL 查询匹配。
P.S。我做了一些研究,它似乎是一个正确的实现,但如果有人纠正我并展示更好的方法,我将不胜感激 (if exist).
我有下一个数据结构:
from enum import IntEnum, unique
from pathlib import Path
from datetime import datetime
from peewee import *
@unique
class Status(IntEnum):
CREATED = 0
FAIL = -1
SUCCESS = 1
db_path = Path(__file__).parent / "test.sqlite"
database = SqliteDatabase(db_path)
class BaseModel(Model):
class Meta:
database = database
class Unit(BaseModel):
name = TextField(unique=True)
some_field = TextField(null=True)
created_at = DateTimeField(default=datetime.now)
class Campaign(BaseModel):
id_ = AutoField()
created_at = DateTimeField(default=datetime.now)
class Task(BaseModel):
id_ = AutoField()
status = IntegerField(default=Status.CREATED)
unit = ForeignKeyField(Unit, backref="tasks")
campaign = ForeignKeyField(Campaign, backref="tasks")
下一个代码创建单位、活动和任务:
def fill_units(count):
units = []
with database.atomic():
for i in range(count):
units.append(Unit.create(name=f"unit{i}"))
return units
def init_campaign(count):
units = Unit.select().limit(count)
with database.atomic():
campaign = Campaign.create()
for unit in units:
Task.create(unit=unit, campaign=campaign)
return campaign
当我尝试向现有 广告系列 添加更多 单位 时出现问题。我需要 select 单位,在这个 活动 中还没有使用过。在 SQL 中,我可以使用下一个查询来做到这一点:
SELECT * FROM unit WHERE id NOT IN (SELECT unit_id FROM task WHERE campaign_id = 1) LIMIT 10
但是如何使用 peewee 做到这一点?
我找到的唯一方法是:
def get_new_units_for_campaign(campaign, count):
unit_names = [task.unit.name for task in campaign.tasks]
units = Unit.select().where(Unit.name.not_in(unit_names)).limit(count)
return units
它以某种方式起作用,但我 100% 确定这是实现它的最愚蠢的方法。你能告诉我实现这个的正确方法吗?
我终于找到了这个:
Unit.select().where(Unit.id.not_in(campaign.tasks.select(Task.unit))).limit(10)
产生
SELECT "t1"."id", "t1"."name", "t1"."some_field", "t1"."created_at" FROM "unit" AS "t1" WHERE ("t1"."id" NOT IN (SELECT "t2"."unit_id" FROM "task" AS "t2" WHERE ("t2"."campaign_id" = 1))) LIMIT 10
与我在问题中提供的 SQL 查询匹配。
P.S。我做了一些研究,它似乎是一个正确的实现,但如果有人纠正我并展示更好的方法,我将不胜感激 (if exist).