【问题标题】:Peewee ORM class definition for per-instance databases每个实例数据库的 Peewee ORM 类定义
【发布时间】:2017-12-12 19:29:36
【问题描述】:

我希望每个 Peewee ORM 实例都有不同的数据库文件。 Peewee 使用嵌套的“元”类将数据库引擎分配给实例。

我的问题似乎归结为从内部类访问类实例属性。使用 Peewee 快速入门示例,这就是我试图在(损坏的)Python 中实现的目标:

from peewee import *

class Person(Model):

    def __init__(self, database):
        self.database = database

    name = CharField()
    birthday = DateField()
    is_relative = BooleanField()

    class Meta:

        # The following is incorrect; I'm trying to access the instance
        # variable for the database filename string
        database = SqliteDatabase(Person.database)

# Create two instances with different databases:
john = Person('john-database.db')
jane = Person('jane-database.db')

我找到了一些关于嵌套类的一般答案,但很难将他们的课程转化为这个特定的应用程序。感谢您的帮助!

【问题讨论】:

    标签: python peewee


    【解决方案1】:

    我认为简短的回答是“peewee 并不是真正为您的用例设计的”。但是我玩了一下,虽然必须有更好的解决方案,但这里有一些工作。但这不是一个好主意,你不应该这样做。

    首先,我们使用标准 peewee 示例模型,除了我们使用Proxy 类进行数据库连接:

    from peewee import *
    from playhouse import *
    
    db = Proxy()
    
    
    class Person(Model):
    
        name = CharField()
        birthday = DateField()
        is_relative = BooleanField()
    
        class Meta:
            database = db
    

    假设我们在model.py 中有这个。

    现在,为了完成这项工作,我们需要model 模块的两个实例,我们可以通过(ab)使用importlib 模块获得:

    import importlib.util
    import peewee
    import sys
    
    def load_module_as(modname, alias):
        mod_spec = importlib.util.find_spec(modname)
        mod = importlib.util.module_from_spec(mod_spec)
        mod_spec.loader.exec_module(mod)
        sys.modules[alias] = mod
        return mod
    

    这允许我们加载model 的两个独立实例:

    model1 = load_module_as('model', 'model1')
    model2 = load_module_as('model', 'model2')
    

    然后我们可以初始化两个不同的数据库:

    model1.db.intitialize(pwee.SqliteDatabase('db1.db'))
    model2.db.intitialize(pwee.SqliteDatabase('db2.db'))
    

    虽然这种方式可以满足您的需求,但您始终需要对您的课程进行限定(model1.Personmodel2.Person)。

    这是一个完整的示例,带有单元测试:

    import datetime
    import importlib.util
    import os
    import peewee
    import shutil
    import sys
    import tempfile
    import unittest
    
    
    def load_module_as(modname, alias):
        mod_spec = importlib.util.find_spec(modname)
        mod = importlib.util.module_from_spec(mod_spec)
        mod_spec.loader.exec_module(mod)
        sys.modules[alias] = mod
        return mod
    
    model1 = load_module_as('model', 'model1')
    model2 = load_module_as('model', 'model2')
    
    
    class TestDatabase(unittest.TestCase):
        def setUp(self):
            self.workdir = tempfile.mkdtemp('testXXXXXX')
            self.db1_path = os.path.join(self.workdir, 'db1.db')
            self.db1 = peewee.SqliteDatabase(self.db1_path)
            self.db1.connect()
    
            self.db2_path = os.path.join(self.workdir, 'db2.db')
            self.db2 = peewee.SqliteDatabase(self.db2_path)
            self.db2.connect()
    
            model1.db.initialize(self.db1)
            model2.db.initialize(self.db2)
    
            self.db1.create_tables([model1.Person])
            self.db2.create_tables([model2.Person])
    
        def test_different_instances(self):
            assert model1.db != model2.db
    
        def test_create_model1_person(self):
            p = model1.Person(name='testperson',
                              birthday=datetime.datetime.now().date(),
                              is_relative=True)
            p.save()
    
        def test_create_model2_person(self):
            p = model2.Person(name='testperson',
                              birthday=datetime.datetime.now().date(),
                              is_relative=True)
            p.save()
    
        def test_create_both(self):
            p1 = model1.Person(name='testperson',
                               birthday=datetime.datetime.now().date(),
                               is_relative=True)
            p2 = model2.Person(name='testperson',
                               birthday=datetime.datetime.now().date(),
                               is_relative=False)
            p1.save()
            p2.save()
    
            p1 = model1.Person.select().where(model1.Person.name == 'testperson').get()
            p2 = model2.Person.select().where(model2.Person.name == 'testperson').get()
    
            assert p1.is_relative
            assert not p2.is_relative
    
        def tearDown(self):
            self.db1.close()
            self.db2.close()
            shutil.rmtree(self.workdir)
    
    if __name__ == '__main__':
        unittest.main(verbosity=2)
    

    【讨论】:

    • 谢谢,感谢您的详尽回答!我试试看。
    • 因为我不需要同时访问两个数据库,所以就像您在第一个示例中指出的那样使用代理对象一样简单。谢谢!
    【解决方案2】:

    我还找到了this thread 并提供了一些可能的答案。

    【讨论】:

      猜你喜欢
      • 2018-07-19
      • 2017-06-04
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2020-09-30
      • 1970-01-01
      • 1970-01-01
      • 2013-08-14
      相关资源
      最近更新 更多