swap_db_sample.py

# -*- coding:utf-8 -*-
import sqlalchemy as sa
import sqlalchemy.orm as orm
from sqlalchemy.ext.declarative import declarative_base

Session = orm.sessionmaker()
engine = sa.create_engine('sqlite://')
Session.configure(bind=engine, autoflush=False)

Base = declarative_base(bind=engine)

class Group(Base):
    __tablename__ = "groups"
    id = sa.Column(sa.Integer(), primary_key=True, nullable=False)
    name = sa.Column(sa.String(255), unique=True, nullable=False)

## first
Base.metadata.create_all()

session = Session()

session.add(Group(name="foo"))
session.commit()

## lookup
session = Session()
assert session.query(Group).count() == 1

## second (swapped)
engine2 = sa.create_engine("sqlite://")
Base.metadata.bind = engine2

Session.configure(bind=engine2, autoflush=False)
session = Session()
Base.metadata.create_all()

assert session.query(Group).count() == 0

## first (swapped)
Session.configure(bind=engine, autoflush=False)
session = Session()


assert session.query(Group).count() == 1

## second (swapped)
Session.configure(bind=engine2, autoflush=False)
session = Session()
assert session.query(Group).count() == 0

swap_db_sample2.py

# -*- coding:utf-8 -*-
import sqlalchemy as sa
import sqlalchemy.orm as orm
from sqlalchemy.ext.declarative import declarative_base

Session = orm.scoped_session(orm.sessionmaker())
engine = sa.create_engine('sqlite://')
Session.configure(bind=engine, autoflush=False)

Base = declarative_base(bind=engine)

class Group(Base):
    __tablename__ = "groups"
    query = Session.query_property()
    id = sa.Column(sa.Integer(), primary_key=True, nullable=False)
    name = sa.Column(sa.String(255), unique=True, nullable=False)

## first
Base.metadata.create_all()

Session.add(Group(name="foo"))
Session.commit()

### lookup
assert Session.query(Group).count() == 1
assert Group.query.count() == 1

## second (swapped)
engine2 = sa.create_engine("sqlite://")
Base.metadata.bind = engine2

Session.remove()
Session.configure(bind=engine2, autoflush=False)
Session.bind=engine2
Base.metadata.create_all()

assert Session.query(Group).count() == 0
assert Group.query.count() == 0

## first (swapped)
Session.remove()
Session.configure(bind=engine, autoflush=False)
Session.bind=engine

assert Session.query(Group).count() == 1
assert Group.query.count() == 1

## second (swapped)
Session.remove()
Session.configure(bind=engine2, autoflush=False)
Session.bind=engine2
Session = Session()
assert Session.query(Group).count() == 0
assert Group.query.count() == 0