Flask应用能否动态连接SQL Server？多服务器数据库查询方案问询

问题背景 #

• 开发工作用数据库管理微API，涉及多台SQL Server，每台服务器挂载多个数据库
• 需求：用户在UI选择服务器后，API查询该服务器下所有数据库名称并返回
• 当前痛点：SQLALCHEMY_DATABASE_URI需在Flask启动前配置，无法运行时动态更新，无法实现服务器切换查询

当前代码实现 #

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from omegaconf.dictconfig import DictConfig

db = SQLAlchemy()


def init_extensions(app: Flask) -> Flask:
    db.init_app(app)
    return app


def register_blueprints(app: Flask) -> Flask:
    pass


def init_app(config: DictConfig) -> Flask:
    app = Flask(__name__.split(".")[0])
    app.config.update(
        DEBUG=config.debug,
        SECRET_KEY=config.secret_key,
        SQLALCHEMY_DATABASE_URI=config.database_uri,
    )

    app = init_extensions(app)
    app = register_blueprints(app)

    return app


if __name__ == "__main__":
    app = init_app(config)
    app.run(host=config.server.host, port=config.server.port, debug=config.debug)

解决方案 #

核心思路 #

放弃依赖Flask-SQLAlchemy全局配置，改用SQLAlchemy核心引擎动态创建连接。针对用户选择的服务器实时生成连接字符串，执行查询后立即释放资源，完全规避启动时绑定单一URI的限制。

重构后代码示例 #

from flask import Flask, request, jsonify
from sqlalchemy import create_engine, text
from omegaconf.dictconfig import DictConfig

def register_blueprints(app: Flask) -> Flask:
    @app.route('/databases', methods=['POST'])
    def get_databases():
        # 从请求体获取用户选择的服务器信息
        server_info = request.json
        # 构建SQL Server连接字符串（根据实际认证方式调整）
        conn_str = f"mssql+pyodbc://{server_info['username']}:{server_info['password']}@{server_info['host']}/master?driver=ODBC+Driver+17+for+SQL+Server"
        
        try:
            # 动态创建引擎并建立连接
            engine = create_engine(conn_str)
            with engine.connect() as conn:
                # 查询非系统数据库列表
                query = text("SELECT name FROM sys.databases WHERE name NOT IN ('master', 'tempdb', 'model', 'msdb')")
                result = conn.execute(query)
                databases = [row[0] for row in result]
            return jsonify({"databases": databases})
        except Exception as e:
            return jsonify({"error": str(e)}), 500
        finally:
            # 销毁引擎释放资源
            if 'engine' in locals():
                engine.dispose()
    return app


def init_app(config: DictConfig) -> Flask:
    app = Flask(__name__.split(".")[0])
    app.config.update(
        DEBUG=config.debug,
        SECRET_KEY=config.secret_key,
        # 无需全局配置SQLALCHEMY_DATABASE_URI
    )

    app = register_blueprints(app)
    return app


if __name__ == "__main__":
    app = init_app(config)
    app.run(host=config.server.host, port=config.server.port, debug=config.debug)

关键说明 #

• 动态连接：每次请求根据用户提交的服务器信息生成专属连接字符串，目标库指定为master（查询系统数据库列表必须访问该库）
• 资源隔离：每次请求创建临时引擎，查询完成后立即销毁，避免连接池占用和线程安全问题
• 系统库过滤：默认排除SQL Server自带的4个系统数据库，可根据业务需求调整过滤条件
• 错误捕获：捕获连接、查询过程中的异常，返回明确错误信息和500状态码

替代方案（保留Flask-SQLAlchemy实例） #

如果必须使用Flask-SQLAlchemy的全局实例，可动态修改配置并重新绑定引擎，但多线程环境下存在线程安全风险，仅推荐在单线程测试场景使用：

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from omegaconf.dictconfig import DictConfig

db = SQLAlchemy()

def init_extensions(app: Flask) -> Flask:
    # 初始化时不绑定特定URI
    db.init_app(app)
    return app

def register_blueprints(app: Flask) -> Flask:
    @app.route('/databases', methods=['POST'])
    def get_databases():
        server_info = request.json
        conn_str = f"mssql+pyodbc://{server_info['username']}:{server_info['password']}@{server_info['host']}/master?driver=ODBC+Driver+17+for+SQL+Server"
        
        # 动态更新配置并重新绑定引擎
        app.config['SQLALCHEMY_DATABASE_URI'] = conn_str
        db.engine.dispose()
        db.init_app(app)
        
        try:
            with db.engine.connect() as conn:
                result = conn.execute(db.text("SELECT name FROM sys.databases WHERE name NOT IN ('master', 'tempdb', 'model', 'msdb')"))
                databases = [row[0] for row in result]
            return jsonify({"databases": databases})
        except Exception as e:
            return jsonify({"error": str(e)}), 500
    return app

# init_app函数保持原有结构，移除SQLALCHEMY_DATABASE_URI配置项

内容的提问来源于stack exchange，提问作者noahward