Search code examples
pythonpeewee

Peewee ORM class definition for per-instance databases


I'd like to have different database files for each Peewee ORM instance. Peewee assigns the database engine to an instance using a nested "Meta" class.

My issue seems to come down to accessing a class instance attribute from an inner class. Using the Peewee quickstart example, this is what I'm trying to achieve in (broken) 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')

I've found a few general answers regarding nested classes, but struggle to translate their lessons to this specific application.


Solution

  • I think the short answer is "peewee isn't really designed for your use case". But I played around with it a bit, and while there has to be a better solution out there, here's something worked. But it's not a good idea, and you shouldn't do it.

    First, we use the standard peewee example model, except we use the Proxy class for the database connection:

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

    Assume we have this in model.py.

    Now, to make this work, we're going to need two instances of the model module, which we can get by (ab)using the importlib module:

    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
    

    This allows us to load in two separate instances of the model:

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

    And we can then initialize two different databases:

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

    While this sort of gets you what you want, you will always need to qualify your classes (model1.Person, model2.Person).

    Here's a complete example, with unit tests:

    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)