python - 在 Flask API 中处理多个连接

Question

我正在使用 Flask 为我们的解决方案编写一个简单的内部 REST API，通过 get 调用(包括身份验证)提供 JSON 对象。我们有多个后端可以从中获取数据。据我了解，这些应该连接到一个用 @app.before_request 修饰的函数中，并分配给 g 全局以用于被请求的特定路由。这不是我习惯的模式。

这是我正在做的一个玩具示例:

@app.before_request
def before_request():
    g.some_conn_a = create_connection('a')
    g.some_conn_b = create_connection('b')
    g.some_client = create_client()


@app.route('/get_some_data')
@requires_auth
def get_some_data():
    # Fetch something from all connections in g
    payload = ... # Construct payload using above connections
    return jsonify(payload)


@app.route('/get_some_other_data')
@requires_auth
def get_some_other_data():
    # Fetch something from maybe just g.some_conn_b
    payload = ... # Construct payload using g.some_conn_b
    return jsonify(payload)

如果用户请求仅驻留在这些连接/客户端中的一个或两个中的数据，这对我来说似乎很浪费，就像在 get_some_other_data 路由示例中一样。

我正在考虑只在路由函数中创建连接/客户端，或者延迟加载它。什么是“正确”的方式？我希望这不是制作一个新模块，这对我正在做的事情来说似乎很极端。

Answer 1

重复 the Flask docs Database Connections example您可以修改 get_db() 以接受多个连接中的每一个的参数。

def get_db(conn):
    """Open specificied connection if none yet for the current app context. """
    if conn == 'some_conn_a':
        if not hasattr(g, 'some_conn_a'):
            g.some_conn_a = create_connection('a')
        db = g.some_conn_a
    elif conn == 'some_conn_b':
        if not hasattr(g, 'some_conn_b'):
            g.some_conn_b = create_connection('b')
        db = g.some_conn_b 
    elif conn == 'some_client':
        if not hasattr(g, 'some_client'):
            g.some_client = create_client()
        db = g.some_client
    else:
        raise Exception("Unknown connection: %s" % conn)

    return db

@app.teardown_appcontext
def close_db(error):
    """Closes the db connections. """
    if hasattr(g, 'some_conn_a'):
        g.some_conn_a.close()
    if hasattr(g, 'some_conn_b'):
        g.some_conn_b.close()
    if hasattr(g, 'some_client'):
        g.some_client.close()

然后您可以根据需要查询每个连接:

@app.route('/get_some_data')
def get_some_data():
    data_a = get_db('some_conn_a').query().something()
    data_b = get_db('some_conn_b').query().something()
    data_c = get_db('some_client').query().something()
    payload = {'a': data_a, 'b': data_b, 'c': data_c}
    return jsonify(payload)

对于延迟加载数据库连接，get_db() 模式优于 before_request 模式。 Flask 0.11 及更高版本的文档示例在更大程度上利用了 get_db() 模式。