Flask SQLAlchemy 动态 URI 设置

Question

我有一个用Flask做的应用程序，它通过WSGI来提供服务，数据库的地址会随着时间变化。每隔两个小时，数据库的地址就会切换到另一个数据库。在这段时间里，我会把一个数据库填充数据，而另一个数据库则在为应用提供数据。

我现在很难找到最佳的方式来配置会话，这样在切换发生时，用户在下一个请求时能获取到正确（不同）的数据库。根据我的测试，如果我在最上层初始化数据库，当切换发生时，用户仍然会指向旧的数据库。

我考虑过在页面（比如首页等）内部设置会话，但这太麻烦了，我还担心打开和关闭太多数据库连接，导致连接闲置。我想我可能可以在启动时初始化两个会话，然后在每个页面的切换时选择使用哪个。这似乎效率不高，我相信还有更好的方法。

求助？！

~~~~~~~~~

这就是我目前正在做的一个大致想法，但在请求之间没有办法更改数据库地址。最上层的代码只运行一次，或者偶尔运行。

if now.hour % 2:
    db_name = 'db1'
else:
    db_name = 'db2'

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql://foo:poo@localhost:3306/%s" % db_name

def init_db(uri, **kwargs):
    engine = create_engine(uri, **kwargs)

    Base.metadata.create_all(bind=engine)

    global db_session
    db_session = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))
    Base.query = db_session.query_property()


init_db(app.config['SQLALCHEMY_DATABASE_URI'], pool_recycle=3600)

@app.teardown_request
def shutdown_session(exception=None):
    db_session.remove()

@app.route('/')
def index():
    ...etc...

工作示例 - 很棒。

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql://foo:poo@localhost:3306/%s"

class SessionManager(object):
    def __init__(self, base_uri=None, **kwargs):
        self.session = None
        self.base_uri = base_uri
        self.kwargs = kwargs

    def get_session(self):
        if now.hour % 2:
            db_name = 'db1'
        else:
            db_name = 'db2'

        if self.session:
            if self.session.name == db_name:
                return self.session
            else:
                self.session.remove()
                self.session = None

        if not self.session:
            engine = create_engine(self.base_uri % db_name, **self.kwargs)
            db_session = scoped_session(sessionmaker(bind=engine))
            db_session.name = db_name
            self.session = db_session
        return db_session

session_manager = SessionManager(base_uri=app.config['SQLALCHEMY_DATABASE_URI'], pool_recycle=3600)

db_session = LocalProxy(session_manager.get_session)

Answer 1

你可以创建一个自定义的构建器，用来在你设定的规则下重新创建引擎和作用域会话。就像下面这样：

class SessionManager(object):

    def __init__(self):
        self.session = None

    def get_session(self):
        # return existing session or make a new engine and scoped_session

为了让这个类更透明，可以使用Werkzeug的LocalProxy。这样，使用会话的代码就完全不需要改动了。

session_manager = SessionManager()
db_session = LocalProxy(session_manager.get_session)