Working with the same metadata in Multiple Databases

In some applications, it may be desirable for there to be multiple instances of a database all sharing the same metadata (for example, if each user gets a new database for each user session). This recipe describes the basic technique using Elixir to manage multiple databases with only one metadata.

The Recipe

By default, Elixir creates a session and metadata for the context of the application. In this recipe, we continue to use the application-wide metadata, but disable the application-wide session and instead create two session instances, each connected to a distinct engine. This recipe has been tested in Elixir 0.5.2 and 0.6.0 using SQLAlchemy 0.4.6.

from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy import create_engine
import elixir

import logging
log = logging.getLogger('demo')

# I want to create any number of databases that all use the same schema (metadata).

# first, I have to make sure that Elixir doesn't automatically bind the various
#  classes to the default session.
__session__ = None

# Here's an example metadata... keeping it simple
elixir.options_defaults['shortnames'] = True

class Person(elixir.Entity):
        name = elixir.Field(elixir.Unicode(256), required=True)
        
        def __repr__(self):
                return 'Person(name="%s")' % self.name

elixir.setup_all()

# Since __session__ is None, Person should not have a query attribute
assert not hasattr(Person, 'query')

# For this example, I'll just use two engines/sessions, a & b.
a_engine = create_engine('sqlite:///', echo=True)
a_session = scoped_session(sessionmaker(autoflush=True, transactional=True, bind=a_engine))

b_engine = create_engine('sqlite:///', echo=True)
b_session = scoped_session(sessionmaker(autoflush=True, transactional=True, bind=b_engine))

# now that I've got two sessions, let's create the schema in each.
elixir.create_all(a_engine)
elixir.create_all(b_engine)

# get all of the people in a_session
assert a_session.query(Person).all() == []

# get all of the people in b_session
assert b_session.query(Person).all() == []

# create a new person in a_session
a_person = Person(name=u"George Washington")
a_session.save(a_person)
a_session.commit()
a_session.clear()

# create a new person in b_session
b_person = Person(name=u"Thomas Jefferson")
b_session.save(b_person)
b_session.commit()
b_session.clear()

# get all of the people in a_session
a_people = a_session.query(Person).all()
print 'session a contains', a_people

# get all of the people in b_session
b_people = b_session.query(Person).all()
print 'session b contains', b_people