瓶SQLAlchemy的设置动态URI(Flask SQLAlchemy setup dynamic

我有一个瓶的应用程序下WSGI在数据库URI随时间变化担任了。 每隔两个小时的URI切换到另一个数据库。 我使用的是时间，而另一种是应用服务数据备份到填满一个数据库。

我有一个很难弄清楚如何最好地配置会话，这样，当发生切换时，客户将获得他们的下一个请求的正确（不同的）数据库。 从我的测试中，如果我在顶层初始化数据库，当发生切换时，客户仍留有指向旧数据库。

我认为有关设置页面中的会话（指数等）本身，而是什么是痛苦，然后我担心打开和关闭过多的数据库连接，并让他们躺在身边。 我想我大概可以使它在启动时初始化两个会话，然后只需选择其中在每一页中切换使用工作。 这似乎效率不高，我敢肯定有一个更好的办法。

救命？！

~~~~~~~~~

这个目前我在做什么的一般概念，但有没有办法改变的请求之间的URI。 顶层代码只运行一次，或每隔一段时间。

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)

文章来源: Flask SQLAlchemy setup dynamic URI