Peewee ORM class 每个实例数据库的定义

Peewee ORM class definition for per-instance databases

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

我的问题似乎归结为从内部 class 访问 class 实例属性。使用 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')

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

我认为简短的回答是 "peewee isn't really designed for your use case"。但我试了一下,虽然必须有更好的解决方案,但这里有一些可行的方法。但这不是一个好主意,你不应该这样做。

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

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'))

虽然这种方法可以满足您的需求,但您始终需要使您的 class 合格(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)

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