如何编写一个 peewee class 在其自身初始化期间成功初始化其他 peewee 对象?

How to write a peewee class that successfully initialises other peewee objects during its own initialisation?

我的类是:

class B(Base_Model):
    b_attribute_1 = peewee.TextField(null=True)
    a = peewee.ForeignKey(A, null=True)

class A(Base_Model):
    a_attribute_1 = peewee.IntegerField(null=True)
    a_attribute_2 = peewee.DoubleField(null=True)
    class Meta:
        friend_server_address = "103.11.399.002"
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.save()
        from B import B
        # Option (1) -- semi-works: creates incomplete object b,
        #               a is initialised correctly.
        B.create(a=self)
        # Option (2) -- does not work: b is created successfully,
        #               a not created, giving KeyError = "ba1".
        # B.create(b_attribute_1 = kwargs["ba1"], a=self)
        return

我查看代码注释中描述的问题的方式(参见选项 1 和 2)是通过测试代码:

class test_a_and_b(unittest.TestCase):
    def setUp(self):
        config.database.init("test.db")
        config.database.create_tables([A, B])
        A(a_attribute_1 = 9705,
          a_attribute_2 = 0.77,
          ba1 = "this is b")
        # This is the breakpoint where I study objects in memory.
        pdb.set_trace()
        return
    def tearDownModule()
        A.get().delete_instance(recursive=True, delete_nullable=True)  
        B.get().delete_instance(recursive=True, delete_nullable=True)
        config.database.drop_tables([A, B])
        os.remove("test.db")

因此,根据我在 A__init__ 中选择的选项,我(在断点处调用 A.get()B.get() 之后)可以看到要么完成 a 和不完整的 b 已创建,或仅完成 b 和错误消息 KeyError:"ba1"我想要的是正在创建两个完整的对象(数据库行)ab

我试着查看 peewee 库,他们的 Model 似乎使用了 __new__,我没有看到任何 __init__,所以也许这就是我的问题所在。在我的 A 中尝试了不同的组合,但这似乎没有帮助。还尝试使用初始化,以编程方式定义属性和 .save() 而不是 .create() for b,但这也没有帮助。

很确定您遇到的问题是,当您需要 A 的实例时,您并不总是会包含 ba1。只有在创建新实例时才会传递在 ba1.当您 select 一个或当 peewee 在从 B 引用它时在内部实例化一个时,您的 'ba1' 关键字参数将不存在并且您的代码将失败并出现您收到的错误。

在使用前检查 'ba1' 是否存在(因为只有在您自己明确创建新 A 时才会使用它)。换句话说,试试这个:

import peewee
import unittest

database_proxy = peewee.Proxy()
class BaseModel(peewee.Model):
    class Meta:
        database = database_proxy

class A(BaseModel):
    a_attribute_1 = peewee.IntegerField(null=True)
    a_attribute_2 = peewee.IntegerField(null=True)
    def __init__(self, *args, **kwargs):
        super(A, self).__init__(*args, **kwargs)
        if "ba1" in kwargs:
            self.save()
            B.create(a=self, b_attribute_1 = kwargs["ba1"])

class B(BaseModel):
    b_attribute_1 = peewee.TextField(null=True)
    a = peewee.ForeignKeyField(A, null=True)

class test_a_and_b(unittest.TestCase):
    def setUp(self):
        db = peewee.SqliteDatabase(':memory:')
        database_proxy.initialize(db)
        db.create_tables([A, B])
        A(a_attribute_1 = 9705,
          a_attribute_2 = 0.77,
          ba1 = "this is b")

    def test_a(self):
        b = B.select().where(B.b_attribute_1 == 'this is b').get()
        self.assertEqual(b.b_attribute_1, 'this is b')
        self.assertEqual(b.a.a_attribute_1, 9705)


if __name__ == '__main__':
    unittest.main()