<p>虽然没有官方支持,但是您可以定制Flask SQLalchemy会话来使用主从连接</p>
<pre><code>from functools import partial
from sqlalchemy import orm
from flask import current_app
from flask_sqlalchemy import SQLAlchemy, get_state
class RoutingSession(orm.Session):
def __init__(self, db, autocommit=False, autoflush=True, **options):
self.app = db.get_app()
self.db = db
self._bind_name = None
orm.Session.__init__(
self, autocommit=autocommit, autoflush=autoflush,
bind=db.engine,
binds=db.get_binds(self.app),
**options,
)
def get_bind(self, mapper=None, clause=None):
try:
state = get_state(self.app)
except (AssertionError, AttributeError, TypeError) as err:
current_app.logger.info(
'cant get configuration. default bind. Error:' + err)
return orm.Session.get_bind(self, mapper, clause)
# If there are no binds configured, use default SQLALCHEMY_DATABASE_URI
if not state or not self.app.config['SQLALCHEMY_BINDS']:
return orm.Session.get_bind(self, mapper, clause)
# if want to user exact bind
if self._bind_name:
return state.db.get_engine(self.app, bind=self._bind_name)
else:
# if no bind is used connect to default
return orm.Session.get_bind(self, mapper, clause)
def using_bind(self, name):
bind_session = RoutingSession(self.db)
vars(bind_session).update(vars(self))
bind_session._bind_name = name
return bind_session
class RouteSQLAlchemy(SQLAlchemy):
def __init__(self, *args, **kwargs):
SQLAlchemy.__init__(self, *args, **kwargs)
self.session.using_bind = lambda s: self.session().using_bind(s)
def create_scoped_session(self, options=None):
if options is None:
options = {}
scopefunc = options.pop('scopefunc', None)
return orm.scoped_session(
partial(RoutingSession, self, **options),
scopefunc=scopefunc,
)
</code></pre>
<p>而默认会话将是<code>master</code>,当您想从slave中选择时,可以直接调用它,下面是示例:</p>
<p>在你的应用程序中:</p>
^{pr2}$
<p>从主控形状中选择</p>
<pre><code>session.query(User).filter_by(id=1).first()
</code></pre>
<p>从从属服务器选择</p>
<pre><code>session.using_bind('slave').query(User).filter_by(id=1).first()
</code></pre>