flask-sqlalchemy 同时连接多个数据库，动态获得不同的session

目前正在用 Flask 做一个小项目，持久层用的是 flask-sqlalchemy，这个和直接使用 sqlalchemy 还略有些不同。

背景

项目中有个需求，要从另外一个 mssqlserver 数据库中读取数据。查到的很多资料都是需要再次在代码中定义数据库的表结构，通过定义不同的 bind，然后使用关键字 __bind_key__ 把表映射到不同的数据库中。这样略有些麻烦。

要对接的 mssqlserver 是另外一个 C/S 系统的数据库，主要是读取报表功能。已经找到了一个方法，可以通过在界面上操作，直接获取到所有对数据库的操作：

SELECT st.text as sql_statement,
       qs.creation_time as plan_last_compiled,
       qs.last_execution_time as plan_last_executed,
       qs.execution_count as plan_executed_count
FROM sys.dm_exec_query_stats qs
CROSS APPLY sys.dm_exec_sql_text(qs.plan_handle) st
order by plan_last_compiled desc

执行这条 SQL 就可以知道这个客户端对 sqlserver 最终执行了什么。所以简单的操作就是直接把 SQL 语句拿到我们的系统里直接执行就可以了，因此就需要一个方法传入一个变量，就直接拿到特殊的 session，而这个 session 又是 ORM 框架管理。

查了很多资料，包括 flask-sqlalchemy 的官网，都没有找到直接答案。自己翻了翻源码，发现这样实现是可以的。

配置部分

from contextlib import contextmanager

from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy

from greenwater_app import app


app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///db/database1.db'
app.config['SQLALCHEMY_BINDS'] = {
    'database2': 'sqlite:///db/database2.db'
}
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
app.config['SQLALCHEMY_ECHO'] = True

db = SQLAlchemy(app)

migrate = Migrate(app, db, compare_type=True)


@contextmanager
def session_scope(bind='default'):
    """Provide a transactional scope around a series of operations."""

    if bind != 'default':
        # 切换到指定的数据库，获取 scope session
        s = db.create_scoped_session(options={'bind': db.get_engine(db.get_app(), bind)})
    else:
        s = db.session

    s.expire_on_commit = False
    try:
        yield s
        s.commit()
    except:
        s.rollback()
        raise
    finally:
        s.close()

按照官网的正常逻辑是可以使用的。具体参考这里：https://flask-sqlalchemy.palletsprojects.com/en/2.x/binds/

使用方式

如果想直接拿到一个 session 去执行 SQL 语句，可以这样：

sql = 'select * from user'
with session_scope(bind='database2') as s:
    res = s.execute(sql).fetchall()
    print([dict(r) for r in res])

with session_scope() as s:
    res = s.execute(sql).fetchall()
    print([dict(r) for r in res])

通过给 session_scope 传入不同的变量，就可以获得一个针对不同的数据库的 session。